""" 离散备料计划维护数据清理工具 功能：自动登录 ERP 系统，根据负责人姓名批量清理指定的备料计划物料。 优化点： 1. 数据库预取：从 $O(n)$ 次数据库查询优化为 $O(1)$ 内存匹配（HashSet）。 2. 日志规范：使用 logging 模块替代 print。 3. 代码整洁：移除方法内导入，增加通用定位辅助函数。 """ import os import re import time import logging from playwright.sync_api import sync_playwright, TimeoutError # 统一顶部导入 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 # --- 日志配置 --- logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger(__name__) class DiscreteMaterialPlanCleaner: """离散备料计划维护数据清理器""" def __init__(self, username, password, manager_name, headless=False, verbose=True): self.username = username self.password = password self.manager_name = manager_name self.headless = headless self.verbose = verbose # 核心优化：使用 set 存储待删除编码，查询复杂度为 $O(1)$ self.to_delete_set = set() def _log(self, message, level="info"): """统一日志输出控制""" if self.verbose: if level == "info": logger.info(message) elif level == "warn": logger.warning(message) elif level == "error": logger.error(message) def _is_button_enabled(self, button_locator): """判定按钮是否可用""" try: return button_locator.is_enabled() except Exception as e: self._log(f"检查按钮状态时出错: {e}", "error") return False def _get_input_value(self, container, label_regex): """通用辅助函数：根据 Label 正则获取 Input 的值""" return container.locator("div").filter( has_text=re.compile(label_regex, re.MULTILINE) ).locator("input").first.input_value() def preload_data(self): """批量预取数据库数据""" self._log(f"正在从数据库提取负责人 [{self.manager_name}] 的待删除物料清单...") # 假设返回的是 material_code 列表 raw_list = get_materials_to_delete(self.manager_name) self.to_delete_set = set(raw_list) self._log(f"预加载完成，共计 {len(self.to_delete_set)} 条不合规物料编码。") def get_production_order_numbers(self, production_id_file): """读取文件并查询生产订单号""" production_ids = read_production_ids(production_id_file) order_ids = query_production_order_numbers(production_ids) self._log(f"读取到 {len(production_ids)} 个总排号 -> 匹配到 {len(order_ids)} 个生产订单号") return order_ids def process_order(self, inner_frame, order_id, order_index, page1): """清理单个订单的数据""" # 1. 查询订单 textbox = inner_frame.get_by_role("textbox", name="生产订单号") textbox.fill(order_id) inner_frame.locator(".search-component-searchBtn").click() # 2. 等待加载（改进：增加 60s 安全超时，防止死锁） 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=60000) except TimeoutError: pass # 3. 进入备料计划详情 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 # 4. 穿透嵌套 Iframe detail_main_frame = page2.locator("#forwardFrame").content_frame detail_inner_frame = detail_main_frame.locator("#mainiframe").content_frame # 5. 提取订单状态信息 plan_code_locator = detail_inner_frame.get_by_text(re.compile(r"^离散备料计划维护：")) plan_code_locator.wait_for(state="visible", timeout=15000) detail_count_text = detail_inner_frame.get_by_text(re.compile(r"^详细信息 \(\d+\)$")).inner_text() detail_count = int(re.search(r"\((\d+)\)", detail_count_text).group(1)) status_text = detail_inner_frame.get_by_text(re.compile(r"^备料状态:.+$")).inner_text() detail_status = re.search(r"备料状态:(.+)$", status_text.replace("\n", "")).group(1) # 6. 执行清理逻辑 if detail_count > 0 and detail_status == "审批通过": # --- 点击修改并等待状态切换 (保留原逻辑) --- detail_inner_frame.get_by_role("button", name="修改").click() # 关键判断：等待保存按钮出现，确认进入编辑模式 save_button_locator = detail_inner_frame.get_by_role("button", name="保存") save_button_locator.wait_for(state="visible", timeout=10000) self._log("已进入编辑模式（保存按钮已就绪）") # --------------------------------------- detail_inner_frame.get_by_text("展开").first.click() child_form = detail_inner_frame.locator(".card-table-side-box") button_wrapper = child_form.locator(".button-wrapper") delete_row_btn = button_wrapper.get_by_role("button", name="删行") next_btn = button_wrapper.locator(".icon-jiantouyou") collapse_btn = button_wrapper.locator(".icon-celashouqi") last_row_number = None while True: # 稳定性检查：等待行号更新 current_row = self._get_input_value(child_form, r"^行号$") if current_row == last_row_number: time.sleep(0.5) material_code = self._get_input_value(child_form, r"^材料编码") material_name = self._get_input_value(child_form, r"^材料名称") pending_qty = self._get_input_value(child_form, r"^累计待发数量$") if material_code in self.to_delete_set: self._log(f"发现匹配物料: {material_name} ({material_code})") if not pending_qty or float(pending_qty) == 0: delete_row_btn.click() self._log(f"✅ 已点击删行") continue else: self._log(f"⚠️ 待发数量为 {pending_qty}，跳过删除", "warn") if self._is_button_enabled(next_btn): last_row_number = current_row next_btn.click() else: break collapse_btn.click() # 执行最终保存逻辑（如业务需要） # save_button_locator.click() page2.close() def setup_query_interface(self, inner_frame): """初始化查询界面配置""" inner_frame.locator(".search-name-wrapper > .iconfont").click() inner_frame.get_by_text("订单号查询").click() inner_frame.get_by_role("tab", name="全部").click() # 填充每页显示条数（5000条测试值） input_el = inner_frame.locator("#rc_select_0") input_el.fill("5000") input_el.press("Enter") def clean(self, production_id_file): """执行完整清理流程""" # 0. 预加载数据库数据 self.preload_data() 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._log("="*30 + " 开始清理任务 " + "="*30) # 进入功能页面 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 # 定位主 Iframe work_main_frame = page1.locator("#forwardFrame").content_frame inner_frame = work_main_frame.locator("#mainiframe").content_frame inner_frame.locator("#hot-key-head_list").wait_for(state="visible", timeout=15000) self.setup_query_interface(inner_frame) order_ids = self.get_production_order_numbers(production_id_file) # 遍历处理 for index, order_id in enumerate(order_ids): self._log(f"进度: [{index+1}/{len(order_ids)}] 处理单号: {order_id}") try: self.process_order(inner_frame, order_id, index, page1) except Exception as e: self._log(f"处理单号 {order_id} 时发生异常: {e}", "error") continue # 单个失败不影响整体执行 # 登出清理 logout(work_main_frame, verbose=self.verbose) context.close() browser.close() self._log("="*30 + " 任务全部完成 " + "="*30) def main(): # 路径配置 base_dir = os.path.dirname(__file__) id_file = os.path.join(base_dir, "productionID.txt") cleaner = DiscreteMaterialPlanCleaner( username="BLDpengqiangqiang", password="your_password_here", manager_name="彭羽", headless=False ) cleaner.clean(id_file) input("执行完毕，按回车键退出程序...") if __name__ == "__main__": main()