""" 离散备料计划维护数据清理工具 负责登录、逐个清理订单数据 """ import os from playwright.sync_api import sync_playwright from utils.auth import login, logout from db.production_order_query import ( read_production_ids, query_production_order_numbers, ) from db.materials_to_delete import get_materials_to_delete class DiscreteMaterialPlanCleaner: """离散备料计划维护数据清理器""" def __init__(self, username, password, manager_name, headless=False, verbose=True): """ 初始化清理器 Args: username: 登录用户名 password: 登录密码 manager_name: 负责人姓名 headless: 是否无头模式运行 verbose: 是否打印详细日志 """ self.username = username self.password = password self.manager_name = manager_name self.headless = headless self.verbose = verbose def _print(self, *args, **kwargs): """打印日志（如果 verbose=True）""" if self.verbose: print(*args, **kwargs) def _is_button_enabled(self, button_locator): """ 判定按钮是否可用 Args: button_locator: Playwright Locator 对象 Returns: bool: True 表示按钮可用，False 表示按钮不可用（有 disabled 属性） """ try: # 使用 Playwright 内置的 is_enabled() 方法 # 这个方法会正确检查元素及其父元素的可启用状态 return button_locator.is_enabled() except Exception as e: self._print(f"检查按钮状态时出错: {e}") return False def get_production_order_numbers(self, production_id_file): """ 读取总排号文件并查询数据库获取生产订单号 Args: production_id_file: ProductionID.txt 文件路径 Returns: 生产订单号列表 """ # 读取总排号 production_ids = read_production_ids(production_id_file) self._print(f"从文件读取到 {len(production_ids)} 个总排号") # 查询数据库获取生产订单号 order_ids = query_production_order_numbers(production_ids) self._print(f"查询到 {len(order_ids)} 个生产订单号") return order_ids def process_order( self, inner_frame, order_id, order_index, page1, manager_name=None, debug_mode=False, debug_order=None, ): """清理单个订单的数据 Args: inner_frame: 内部 iframe order_id: 订单号 order_index: 订单索引 page1: 页面对象 manager_name: 负责人姓名（用于查询待删除物料） debug_mode: 是否启用调试模式 debug_order: 调试订单号 """ if manager_name is None: manager_name = self.manager_name from playwright.sync_api import TimeoutError import re import time # 清空文本框 textbox = inner_frame.get_by_role("textbox", name="生产订单号") textbox.fill("") # 填充订单号 textbox.fill(order_id) # 点击查询 inner_frame.locator(".search-component-searchBtn").click() self._print(f"第 {order_index + 1} 个订单查询完成，等待加载结果...") # 等待加载完成 loading_locator = inner_frame.locator("div").filter(has_text="加载中").nth(1) try: loading_locator.wait_for(state="visible", timeout=3000) loading_locator.wait_for(state="hidden", timeout=0) # 无限等待，直到消失 except TimeoutError: # 加载很快完成，或者没有出现加载提示 pass self._print(f"第 {order_index + 1} 个订单加载完成，开始清理数据...") # 调试模式：只在指定订单暂停 if debug_mode and (debug_order is None or order_index == debug_order): self._print(f"=== 调试暂停：第 {order_index + 1} 个订单 ===") page1.pause() inner_frame.locator("#hot-key-head_list").get_by_text("更多").click() with page1.expect_popup() as page2_info: inner_frame.get_by_text("备料计划").click() page2 = page2_info.value # 获取 nested iframe main_frame = page2.locator("#forwardFrame").content_frame inner_frame_locator = main_frame.locator("#mainiframe") inner_frame_locator.wait_for(state="visible", timeout=15000) inner_frame = inner_frame_locator.content_frame # 等待备料计划页面加载完成（等待序列号加载出来） self._print("等待备料计划页面加载完成...") plan_code_locator = inner_frame.get_by_text(re.compile(r"^离散备料计划维护：")) plan_code_locator.wait_for(state="visible", timeout=30000) # 循环检查编码是否已加载 max_wait = 30 # 最多等待30秒 wait_interval = 0.5 # 每0.5秒检查一次 waited = 0 plan_code = None while waited < max_wait: plan_text = plan_code_locator.inner_text() match = re.search(r"离散备料计划维护：(.+)", plan_text) if match and match.group(1).strip(): plan_code = match.group(1).strip() break time.sleep(wait_interval) waited += wait_interval if plan_code: self._print(f"备料计划页面加载完成，编码: {plan_code}") else: self._print(f"警告: 备料计划页面加载超时") # 提取"详细信息"中的数字 detail_element = inner_frame.get_by_text(re.compile(r"^详细信息 \(\d+\)$")) # 使用正则表达式提取括号中的数字 match = re.search(r"详细信息 \((\d+)\)", detail_element.inner_text()) if match: detail_count = int(match.group(1)) self._print(f"详细信息数量: {detail_count}") # 提取"备料状态"信息 status_element = inner_frame.get_by_text(re.compile(r"^备料状态:.+$")) status_text = status_element.inner_text().replace("\n", "") # 使用正则表达式提取括号中的数字 match = re.search(r"^备料状态:(.+)$", status_text) if match: detail_status = match.group(1) self._print(f"备料状态: {detail_status}") # page2.pause() if detail_count > 0 and detail_status == "审批通过": inner_frame.get_by_role("button", name="修改").click() save_button_locator = inner_frame.get_by_role("button", name="保存") save_button_locator.wait_for(state="visible", timeout=10000) inner_frame.get_by_text("展开").first.click() # 获取展开后的父容器，基于它定位子元素更加精确 # 父元素 class="card-table-side-box undefined" child_form = inner_frame.locator(".card-table-side-box") # 等待父容器变为可见 child_form.wait_for(state="visible", timeout=5000) self._print(f"父容器 .card-table-side-box 已找到") # 获取按钮容器对象 button_wrapper = child_form.locator(".button-wrapper") button_wrapper.wait_for(state="visible", timeout=5000) self._print(f"按钮容器 .button-wrapper 已找到") # 创建按钮对象 # 1. 删行按钮 delete_row_button = button_wrapper.get_by_role("button", name="删行") # 2. 下一个物料按钮 (class="icon-jiantouyou") next_material_button = button_wrapper.locator(".icon-jiantouyou") # 3. 侧拉收起按钮 (class="icon-celashouqi") collapse_button = button_wrapper.locator(".icon-celashouqi") last_row_number = None # 用于记录上一个物料的行号，初始值为 None # page2.pause() while self._is_button_enabled(next_material_button): # 等待页面数据加载完成（带超时机制） max_wait_time = 10 # 最大等待10秒 start_time = time.time() while True: row_number_locator = ( child_form.locator("div") .filter(has_text=re.compile(r"^行号$", re.MULTILINE)) .locator("input") .first ) row_number = row_number_locator.input_value() # 检查行号是否发生变化 if row_number != last_row_number: break # 检查是否超时 if time.time() - start_time > max_wait_time: self._print(f"⚠️ 等待数据加载超时，强制继续 (行号: {row_number})") break # 等待后重新检查 child_form.wait_for_timeout(500) self._print(f"当前行号: {row_number} (上一次: {last_row_number})") # ... 后续代码逻辑 ... # 处理完当前行后，更新 last_row_number last_row_number = row_number # 获取材料编码（用于精确匹配） input_box = ( child_form.locator("div") .filter(has_text=re.compile(r"^材料编码\d{11}$", re.MULTILINE)) .locator("input") .first ) material_code = input_box.input_value() self._print(f"材料编码:{material_code}") # 获取材料名称（仅用于日志） input_box = ( child_form.locator("div") .filter(has_text=re.compile(r"^材料名称$")) .locator("input[type='text']") ) material_name = input_box.input_value() self._print(f"材料名称:{material_name}") # 获取累计待发数量 # 创建一个变量来保存累计待发数量的值，以便后续使用 cumulative_pending_quantity = 0 input_box = ( child_form.locator("div") .filter(has_text=re.compile(r"^累计待发数量$")) .locator("input[type='text']") ) cumulative_pending_quantity = input_box.input_value() self._print(f"累计待发数量:{cumulative_pending_quantity}") # 获取累计出库数量 input_box = ( child_form.locator("div") .filter(has_text=re.compile(r"^累计出库数量$")) .locator("input[type='text']") ) self._print(f"累计出库数量:{input_box.input_value()}") # 检查是否需要清理该物料（直接数据库查询） from db.materials_to_delete import should_delete_material should_delete = should_delete_material(manager_name, material_code) if should_delete: if not cumulative_pending_quantity : self._print(f"✅ 可以删除") delete_row_button.click() # 使用正则表达式提取括号中的数字 match = re.search(r"详细信息 \((\d+)\)", detail_element.inner_text()) if match: detail_count = int(match.group(1)) self._print(f"详细信息数量: {detail_count}") break else: self._print(f"❌ 累计待发数量不为空，无法删除") self._print( f">>> 需要清理：材料名称【{material_name}】材料编码【{material_code}】" ) # TODO: 执行删除操作 else: self._print(f"保留：材料名称【{material_name}】材料编码【{material_code}】无需清理") # 检查"下一个物料"按钮是否可用 if self._is_button_enabled(next_material_button): self._print("按钮可用，可以点击") next_material_button.click() else: self._print("按钮不可用，有 disabled 属性") collapse_button.click() # page2.pause() elif detail_count == 0: self._print(f"第 {order_index + 1} 个订单无数据需要清理，跳过...") page2.close() return elif detail_status != "审批通过": self._print(f"第 {order_index + 1} 个订单备料状态: {detail_status}") page2.close() return page2.close() time.sleep(1) pass def setup_query_interface(self, inner_frame): """设置查询界面""" import re # 点击图标按钮打开查询界面 inner_frame.locator(".search-name-wrapper > .iconfont").click() inner_frame.get_by_text("订单号查询").click() inner_frame.get_by_role("tab", name="全部").click() # 填充并验证，如果失败则重试 max_retries = 3 expected_value = "5000" for attempt in range(max_retries): inner_frame.locator("#rc_select_0").fill(expected_value) inner_frame.locator("#rc_select_0").press("Enter") # 检查填充是否成功 actual_value = inner_frame.locator("#rc_select_0").input_value() if actual_value == expected_value: self._print(f"文本框填充成功: {expected_value}") break else: self._print( f"第 {attempt + 1} 次填充失败，实际值: {actual_value}，重试..." ) if attempt == max_retries - 1: self._print( f"警告: {max_retries} 次尝试后仍未成功填充，继续执行..." ) def clean(self, production_id_file, debug_mode=False, debug_order=None): """ 执行完整的数据清理流程 Args: production_id_file: ProductionID.txt 文件路径 debug_mode: 是否启用调试模式 debug_order: 调试订单索引 """ self._print(f"使用负责人 [{self.manager_name}] 进行数据清理") with sync_playwright() as playwright: # 调用登录模块 browser, context, page, main_frame = login( playwright=playwright, username=self.username, password=self.password, headless=self.headless, ignore_https_errors=True, ) self._print("=" * 80) self._print("开始执行离散备料计划维护数据清理") self._print("=" * 80) # 登录成功后可以进行后续操作 # 点击打开"功能菜单" main_frame.locator("i").first.click() # 点击打开"离散生产订单维护" with page.expect_popup() as page1_info: main_frame.get_by_title("离散生产订单维护", exact=True).first.click() page1 = page1_info.value # 获取 nested iframe main_frame = page1.locator("#forwardFrame").content_frame inner_frame_locator = main_frame.locator("#mainiframe") inner_frame_locator.wait_for(state="visible", timeout=15000) inner_frame = inner_frame_locator.content_frame # 设置查询界面 self.setup_query_interface(inner_frame) # 读取总排号并查询生产订单号 order_ids = self.get_production_order_numbers(production_id_file) # 按订单清理 for order_index, order_id in enumerate(order_ids): self._print( f"\n=== 开始处理第 {order_index + 1} 个订单，订单号: {order_id} ===" ) self.process_order( inner_frame, order_id, order_index, page1, self.manager_name, debug_mode=debug_mode, debug_order=debug_order, ) # 执行账号注销 self._print("\n开始执行账号注销...") logout(main_frame, verbose=self.verbose) self._print(f"\n=== 全部完成 ===") # 关闭浏览器 context.close() browser.close() def main(): """测试函数""" cleaner = DiscreteMaterialPlanCleaner( username="BLDpengqiangqiang", password="Cqbld123456.", manager_name="彭羽", headless=False, verbose=True, ) production_id_file = os.path.join(os.path.dirname(__file__), "productionID.txt") cleaner.clean(production_id_file) input("按回车退出...") if __name__ == "__main__": main()