admin/playwrite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: rename Chinese-named Python files to English

Browse Source 
Rename utility files to use English names for better cross-platform compatibility:
- utils/离散备料计划维护数据提取.py → utils/discrete_material_plan_extractor.py
- utils/离散备料计划维护数据清理.py → utils/discrete_material_plan_cleaner.py

Update all import statements across the codebase and documentation references.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
2026-02-24 17:35:23 +08:00
parent 7a8a38a3b6
commit 19dfe5b09e
7 changed files with 10 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

246
utils/discrete_material_plan_cleaner.py Normal file
View File

@@ -0,0 +1,246 @@
"""
离散备料计划维护数据清理工具
功能：自动登录 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_status == "审批通过":
if detail_count >0:
# --- 点击修改并等待状态切换 (保留原逻辑) ---
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")
else:
self._log(f"物料 {material_name} ({material_code}) 不在删除列表中，不做处理")
if self._is_button_enabled(next_btn):
last_row_number = current_row
next_btn.click()
else:
break
collapse_btn.click()
# 执行最终保存逻辑（如业务需要）
# save_button_locator.click()
else:
self._log("订单无备料计划数据，无需处理")
elif detail_status == "完成":
self._log("订单已完成，无需处理")
else:
self._log(f"订单状态为 [{detail_status}]，不符合处理条件", "warn")
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()