diff --git a/main_clean.py b/main_clean.py new file mode 100644 index 0000000..8514850 --- /dev/null +++ b/main_clean.py @@ -0,0 +1,27 @@ +""" +主程序 - 使用离散备料计划维护数据清理工具 +""" +import os +from utils.离散备料计划维护数据清理 import DiscreteMaterialPlanCleaner + + +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() diff --git a/utils/离散备料计划维护数据清理.py b/utils/离散备料计划维护数据清理.py new file mode 100644 index 0000000..5d32d2f --- /dev/null +++ b/utils/离散备料计划维护数据清理.py @@ -0,0 +1,190 @@ +""" +离散备料计划维护数据清理工具 +负责登录、逐个清理订单数据 +""" +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() + page1.pause() + 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()