""" 离散备料计划维护数据清理工具 负责登录、逐个清理订单数据 """ import os from playwright.sync_api import sync_playwright from utils.auth import login, logout class DiscreteMaterialPlanCleaner: """离散备料计划维护数据清理器""" def __init__(self, username, password, headless=False, verbose=True): """ 初始化清理器 Args: username: 登录用户名 password: 登录密码 headless: 是否无头模式运行 verbose: 是否打印详细日志 """ self.username = username self.password = password self.headless = headless self.verbose = verbose def _print(self, *args, **kwargs): """打印日志（如果 verbose=True）""" if self.verbose: print(*args, **kwargs) def read_order_ids(self, file_path): """读取订单号文件""" with open(file_path, 'r', encoding='utf-8') as f: # 去除空白行和空格 order_ids = [line.strip() for line in f if line.strip()] return order_ids def process_order(self, inner_frame, order_id, order_index, page1, debug_mode=False, debug_order=None): """清理单个订单的数据 Args: inner_frame: 内部 iframe order_id: 订单号 order_index: 订单索引 page1: 页面对象 debug_mode: 是否启用调试模式 debug_order: 调试订单号 """ 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+\)$")) detail_text = detail_element.inner_text() # 使用正则表达式提取括号中的数字 match = re.search(r"详细信息 \((\d+)\)", detail_text) if match: detail_count = int(match.group(1)) self._print(f"详细信息数量: {detail_count}") # 提取"备料状态"信息 detail_element = inner_frame.get_by_text(re.compile(r"^备料状态:.+$")) detail_text = detail_element.inner_text().replace("\n", "") # 使用正则表达式提取括号中的数字 match = re.search(r"^备料状态:(.+)$", detail_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 已找到") page2.pause() for id in range(detail_count): id_lable_locator = child_form.get_by_text("序号 " + str(id + 1)) id_lable_locator.wait_for(state="visible", timeout=10000) self._print(f"处理 {id_lable_locator.inner_text()} ") # 获取材料编码（通过文本定位，取第一个input） input_box = child_form.locator("div").filter(has_text=re.compile(r"^材料编码\d{11}$", re.MULTILINE)).locator("input").first self._print(f"材料编码:{input_box.input_value()}") # 获取材料名称 input_box = child_form.locator("div").filter(has_text=re.compile(r"^材料名称$")).locator("input[type='text']") self._print(f"材料名称:{input_box.input_value()}") # 获取累计待发数量 input_box = child_form.locator("div").filter(has_text=re.compile(r"^累计待发数量$")).locator("input[type='text']") self._print(f"累计待发数量:{input_box.input_value()}") # 获取累计出库数量 input_box = child_form.locator("div").filter(has_text=re.compile(r"^累计出库数量$")).locator("input[type='text']") self._print(f"累计出库数量:{input_box.input_value()}") if id != detail_count - 1: child_form.get_by_role("button").filter(has_text=re.compile(r"^$")).nth(2).click() else: child_form.get_by_role("button").filter(has_text=re.compile(r"^$")).nth(4).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, order_id_file, debug_mode=False, debug_order=None): """ 执行完整的数据清理流程 Args: order_id_file: 订单号文件路径 debug_mode: 是否启用调试模式 debug_order: 调试订单索引 """ 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.read_order_ids(order_id_file) self._print(f"共读取到 {len(order_ids)} 个订单号") # 按订单清理 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, 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.", headless=False, verbose=True ) order_id_file = os.path.join(os.path.dirname(__file__), "orderID.txt") cleaner.clean(order_id_file) input("按回车退出...") if __name__ == "__main__": main()