feat: add discrete material plan data cleaning tool

Implement a new automation script and utility to handle the cleaning of
discrete material plan maintenance data.

- Add main_clean.py as the primary entry point, configuring the cleaner
  instance and reading order IDs.
- Add DiscreteMaterialPlanCleaner in utils/ utilizing Playwright to automate
  browser interactions.
- Implement logic for user authentication (login/logout), UI setup for
  queries, and processing individual orders.
- Include verbose logging and debug mode support for troubleshooting.
This commit is contained in:
Misaka_Company
2026-01-22 15:55:38 +08:00
parent 24acc39411
commit 8708709cac
2 changed files with 217 additions and 0 deletions

View File

@@ -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()