2a784e66c0
Track materials that are not in the delete list and display them in a new "未处理物料" (Unmatched Materials) section in the report. Co-Authored-By: Claude (glm-5) <noreply@anthropic.com>
573 lines
25 KiB
Python
573 lines
25 KiB
Python
"""
|
||
离散备料计划维护数据清理工具
|
||
功能：自动登录 ERP 系统，根据负责人姓名批量清理指定的备料计划物料。
|
||
优化点：
|
||
1. 数据库预取：从 $O(n)$ 次数据库查询优化为 $O(1)$ 内存匹配（HashSet）。
|
||
2. 日志规范：使用 logging 模块替代 print。
|
||
3. 代码整洁：移除方法内导入，增加通用定位辅助函数。
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
import time
|
||
import logging
|
||
from typing import Union, List, Optional, Callable, Dict, Any
|
||
from datetime import datetime
|
||
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_by_managers
|
||
|
||
# --- 日志配置 ---
|
||
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_names: Union[str, List[str], None] = None,
|
||
headless=False,
|
||
verbose=True,
|
||
dryrun=False,
|
||
progress_callback: Optional[Callable[[int, int, str], None]] = None,
|
||
):
|
||
self.username = username
|
||
self.password = password
|
||
self.headless = headless
|
||
self.verbose = verbose
|
||
self.dryrun = dryrun
|
||
self.progress_callback = progress_callback
|
||
|
||
# 参数规范化：支持 str、List[str]、None
|
||
if manager_names is None:
|
||
self.manager_names = None # 表示全部
|
||
elif isinstance(manager_names, str):
|
||
self.manager_names = [manager_names] if manager_names.strip() else None
|
||
else:
|
||
self.manager_names = manager_names if manager_names else None
|
||
|
||
# 核心优化：使用 set 存储待删除编码，查询复杂度为 $O(1)$
|
||
self.to_delete_set = set()
|
||
|
||
# 统计信息
|
||
self.stats = {
|
||
'total_orders': 0,
|
||
'processed_orders': 0,
|
||
'total_materials': 0, # 总物料数
|
||
'processed_materials': 0, # 已处理物料数
|
||
'deleted_materials': [], # [{order_id, material_code, material_name}]
|
||
'skipped_materials': [], # [{order_id, material_code, material_name, reason}]
|
||
'unmatched_materials': [], # [{order_id, material_code, material_name}] 不在删除列表的物料
|
||
'errors': [], # [{order_id, error_message}]
|
||
'start_time': None,
|
||
'end_time': None,
|
||
}
|
||
|
||
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 _report_progress(self, current: int, total: int, message: str):
|
||
"""报告进度"""
|
||
if self.progress_callback:
|
||
self.progress_callback(current, total, message)
|
||
|
||
def _report_material_progress(
|
||
self,
|
||
order_idx: int,
|
||
total_orders: int,
|
||
material_idx: int,
|
||
total_materials: int,
|
||
order_id: str,
|
||
material_name: str,
|
||
action: str
|
||
):
|
||
"""报告物料处理进度
|
||
|
||
进度计算逻辑：
|
||
- 每个订单占 1/total_orders 的固定进度配额
|
||
- 订单内物料进度按比例分配（material_idx / total_materials）
|
||
- 总体进度 = (order_idx + material_idx / total_materials) / total_orders
|
||
"""
|
||
if self.progress_callback:
|
||
order_progress = f"订单 [{order_idx + 1}/{total_orders}]"
|
||
if total_materials > 0:
|
||
material_progress = f"物料 [{material_idx}/{total_materials}]"
|
||
message = f"{order_progress} {material_progress} - {order_id} - {action}: {material_name}"
|
||
|
||
# 计算总体进度比例（0.0 到 1.0）
|
||
order_internal_ratio = material_idx / total_materials
|
||
overall_ratio = (order_idx + order_internal_ratio) / total_orders
|
||
|
||
# 使用固定精度整数表示进度（范围 0-10000，显示时除以 100 即为百分比）
|
||
PROGRESS_SCALE = 10000
|
||
overall_current = int(overall_ratio * PROGRESS_SCALE)
|
||
overall_total = PROGRESS_SCALE
|
||
self.progress_callback(overall_current, overall_total, message)
|
||
else:
|
||
message = f"{order_progress} - {order_id} - {action}: {material_name}"
|
||
self.progress_callback(order_idx + 1, total_orders, 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):
|
||
"""批量预取数据库数据"""
|
||
if self.manager_names is None:
|
||
self._log("正在从数据库提取所有负责人的待删除物料清单...")
|
||
else:
|
||
names_str = "、".join(self.manager_names)
|
||
self._log(f"正在从数据库提取负责人 [{names_str}] 的待删除物料清单...")
|
||
|
||
raw_list = get_materials_to_delete_by_managers(self.manager_names)
|
||
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, total_orders: int = 1):
|
||
"""清理单个订单的数据
|
||
|
||
Args:
|
||
inner_frame: 内层 iframe
|
||
order_id: 订单 ID
|
||
order_index: 订单索引（从 0 开始）
|
||
page1: 页面对象
|
||
total_orders: 总订单数（用于进度报告）
|
||
"""
|
||
# 报告订单进度
|
||
# self._report_progress(order_index + 1, total_orders, f"正在打开订单: {order_id}")
|
||
|
||
# 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=30000)
|
||
|
||
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. 执行清理逻辑
|
||
try:
|
||
if detail_status == "审批通过":
|
||
if detail_count > 0:
|
||
# 更新总物料数统计
|
||
self.stats['total_materials'] += detail_count
|
||
|
||
# --- 点击修改并等待状态切换 (保留原逻辑) ---
|
||
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=30000)
|
||
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
|
||
material_idx = 0 # 物料计数器
|
||
# page2.pause() # 调试用，正式运行时可删除
|
||
while True:
|
||
material_idx += 1
|
||
self.stats['processed_materials'] += 1
|
||
|
||
# 稳定性检查：等待行号更新
|
||
current_row = self._get_input_value(child_form, r"^行号$")
|
||
row_num_int = int(current_row)
|
||
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"^累计待发数量$")
|
||
|
||
# 报告物料进度
|
||
self._report_material_progress(
|
||
order_index, total_orders,
|
||
material_idx, detail_count,
|
||
order_id, material_name, "检查"
|
||
)
|
||
|
||
if material_code in self.to_delete_set:
|
||
self._log(f"发现匹配物料: {material_name} ({material_code})")
|
||
if (not pending_qty) and not (
|
||
row_num_int >= 7000 and row_num_int < 8000
|
||
):
|
||
# 报告删除进度
|
||
self._report_material_progress(
|
||
order_index, total_orders,
|
||
material_idx, detail_count,
|
||
order_id, material_name, "删除"
|
||
)
|
||
# 记录删除前的行号
|
||
old_row_number = current_row
|
||
delete_row_btn.click()
|
||
self._log(f"✅ 已点击删行，等待删除完成...")
|
||
|
||
# 等待行号变化（表示删除完成且新数据已加载）
|
||
max_wait_time = 10 # 最大等待10秒
|
||
start_time = time.time()
|
||
delete_success = False
|
||
while time.time() - start_time < max_wait_time:
|
||
try:
|
||
new_row_number = self._get_input_value(
|
||
child_form, r"^行号$"
|
||
)
|
||
if new_row_number != old_row_number:
|
||
self._log(
|
||
f"✓ 删除完成，行号已从 {old_row_number} 变更为 {new_row_number}"
|
||
)
|
||
delete_success = True
|
||
break
|
||
time.sleep(0.2) # 每200ms检查一次
|
||
except Exception as e:
|
||
self._log(f"获取新行号时出错: {e}", "warn")
|
||
time.sleep(0.2)
|
||
else:
|
||
self._log(
|
||
f"⚠️ 等待删除完成超时（{max_wait_time}秒）", "warn"
|
||
)
|
||
|
||
# 记录删除统计
|
||
if delete_success:
|
||
self.stats['deleted_materials'].append({
|
||
'order_id': order_id,
|
||
'material_code': material_code,
|
||
'material_name': material_name
|
||
})
|
||
|
||
continue
|
||
elif row_num_int >= 7000 and row_num_int < 8000:
|
||
reason = f"行号 {row_num_int} 在 7000-8000 范围内"
|
||
self._report_material_progress(
|
||
order_index, total_orders,
|
||
material_idx, detail_count,
|
||
order_id, material_name, "跳过"
|
||
)
|
||
self._log(f"⚠️ {reason}，跳过删除", "warn")
|
||
self.stats['skipped_materials'].append({
|
||
'order_id': order_id,
|
||
'material_code': material_code,
|
||
'material_name': material_name,
|
||
'reason': reason
|
||
})
|
||
elif pending_qty:
|
||
reason = f"待发数量为 {pending_qty}"
|
||
self._report_material_progress(
|
||
order_index, total_orders,
|
||
material_idx, detail_count,
|
||
order_id, material_name, "跳过"
|
||
)
|
||
self._log(f"⚠️ {reason}，跳过删除", "warn")
|
||
self.stats['skipped_materials'].append({
|
||
'order_id': order_id,
|
||
'material_code': material_code,
|
||
'material_name': material_name,
|
||
'reason': reason
|
||
})
|
||
|
||
else:
|
||
reason = "不满足删除条件"
|
||
self._report_material_progress(
|
||
order_index, total_orders,
|
||
material_idx, detail_count,
|
||
order_id, material_name, "跳过"
|
||
)
|
||
self._log(
|
||
f"⚠️ {reason}，跳过物料 {material_name} ({material_code})",
|
||
"warn",
|
||
)
|
||
self.stats['skipped_materials'].append({
|
||
'order_id': order_id,
|
||
'material_code': material_code,
|
||
'material_name': material_name,
|
||
'reason': reason
|
||
})
|
||
|
||
else:
|
||
self._log(
|
||
f"ℹ️ 物料 {material_name} ({material_code}) 不在删除列表中，不做处理"
|
||
)
|
||
# 记录到 unmatched_materials
|
||
self.stats['unmatched_materials'].append({
|
||
'order_id': order_id,
|
||
'material_code': material_code,
|
||
'material_name': material_name
|
||
})
|
||
if self._is_button_enabled(next_btn):
|
||
last_row_number = current_row
|
||
next_btn.click()
|
||
else:
|
||
break
|
||
collapse_btn.click()
|
||
|
||
# 执行最终保存逻辑（如业务需要）
|
||
if self.dryrun:
|
||
self._log("[DRYRUN] 跳过保存操作")
|
||
else:
|
||
save_button_locator.click()
|
||
self._log("已点击保存，等待保存完成...")
|
||
save_button_locator.wait_for(
|
||
state="hidden", timeout=60000
|
||
) # 等待按钮消失，超时60秒
|
||
self._log("✅ 保存成功（保存按钮已消失）")
|
||
else:
|
||
self._log("订单无备料计划数据，无需处理")
|
||
elif detail_status == "完成":
|
||
self._log("订单已完成，无需处理")
|
||
else:
|
||
self._log(f"订单状态为 [{detail_status}]，不符合处理条件", "warn")
|
||
finally:
|
||
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):
|
||
"""执行完整清理流程"""
|
||
# 初始化统计
|
||
self.stats['start_time'] = datetime.now()
|
||
|
||
# 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=30000
|
||
)
|
||
|
||
self.setup_query_interface(inner_frame)
|
||
order_ids = self.get_production_order_numbers(production_id_file)
|
||
|
||
# 设置总订单数
|
||
self.stats['total_orders'] = len(order_ids)
|
||
|
||
# 遍历处理
|
||
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, len(order_ids))
|
||
self.stats['processed_orders'] += 1
|
||
except Exception as e:
|
||
self._log(f"处理单号 {order_id} 时发生异常: {e}", "error")
|
||
self.stats['errors'].append({
|
||
'order_id': order_id,
|
||
'error_message': str(e)
|
||
})
|
||
continue # 单个失败不影响整体执行
|
||
|
||
# 登出清理
|
||
logout(work_main_frame, verbose=self.verbose)
|
||
context.close()
|
||
browser.close()
|
||
self._log("=" * 30 + " 任务全部完成 " + "=" * 30)
|
||
|
||
# 记录结束时间
|
||
self.stats['end_time'] = datetime.now()
|
||
|
||
def generate_report(self) -> str:
|
||
"""生成 Markdown 格式的执行报告
|
||
|
||
Returns:
|
||
Markdown 格式的报告字符串
|
||
"""
|
||
report_lines = []
|
||
|
||
# 标题
|
||
report_lines.append("# 执行报告")
|
||
report_lines.append("")
|
||
|
||
# 概述
|
||
report_lines.append("## 概述")
|
||
report_lines.append("")
|
||
start_time = self.stats.get('start_time')
|
||
end_time = self.stats.get('end_time')
|
||
duration = None
|
||
if start_time and end_time:
|
||
duration = end_time - start_time
|
||
report_lines.append(f"- 开始时间: {start_time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
report_lines.append(f"- 结束时间: {end_time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
report_lines.append(f"- 执行时长: {duration}")
|
||
report_lines.append(f"- 处理订单: {self.stats['processed_orders']}/{self.stats['total_orders']} 个")
|
||
report_lines.append(f"- 处理物料: {self.stats['processed_materials']}/{self.stats['total_materials']} 条")
|
||
report_lines.append(f"- 删除物料: {len(self.stats['deleted_materials'])} 条")
|
||
report_lines.append(f"- 跳过物料: {len(self.stats['skipped_materials'])} 条")
|
||
report_lines.append(f"- 未处理物料: {len(self.stats['unmatched_materials'])} 条")
|
||
report_lines.append(f"- 错误数量: {len(self.stats['errors'])} 个")
|
||
report_lines.append(f"- 执行模式: {'预览模式 (dryrun)' if self.dryrun else '正常执行'}")
|
||
if self.manager_names:
|
||
report_lines.append(f"- 负责人: {', '.join(self.manager_names)}")
|
||
report_lines.append("")
|
||
|
||
# 删除明细
|
||
if self.stats['deleted_materials']:
|
||
report_lines.append("## 删除明细")
|
||
report_lines.append("")
|
||
report_lines.append("| 订单号 | 物料编码 | 物料名称 |")
|
||
report_lines.append("|--------|----------|----------|")
|
||
for item in self.stats['deleted_materials']:
|
||
report_lines.append(f"| {item['order_id']} | {item['material_code']} | {item['material_name']} |")
|
||
report_lines.append("")
|
||
|
||
# 跳过明细
|
||
if self.stats['skipped_materials']:
|
||
report_lines.append("## 跳过明细")
|
||
report_lines.append("")
|
||
report_lines.append("| 订单号 | 物料编码 | 物料名称 | 跳过原因 |")
|
||
report_lines.append("|--------|----------|----------|----------|")
|
||
for item in self.stats['skipped_materials']:
|
||
report_lines.append(f"| {item['order_id']} | {item['material_code']} | {item['material_name']} | {item['reason']} |")
|
||
report_lines.append("")
|
||
|
||
# 未处理物料明细
|
||
if self.stats['unmatched_materials']:
|
||
report_lines.append("## 未处理物料")
|
||
report_lines.append("")
|
||
report_lines.append("| 订单号 | 物料编码 | 物料名称 |")
|
||
report_lines.append("|--------|----------|----------|")
|
||
for item in self.stats['unmatched_materials']:
|
||
report_lines.append(f"| {item['order_id']} | {item['material_code']} | {item['material_name']} |")
|
||
report_lines.append("")
|
||
|
||
# 错误明细
|
||
if self.stats['errors']:
|
||
report_lines.append("## 错误明细")
|
||
report_lines.append("")
|
||
report_lines.append("| 订单号 | 错误信息 |")
|
||
report_lines.append("|--------|----------|")
|
||
for item in self.stats['errors']:
|
||
report_lines.append(f"| {item['order_id']} | {item['error_message']} |")
|
||
report_lines.append("")
|
||
|
||
return "\n".join(report_lines)
|
||
|
||
|
||
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_names="彭羽", # 支持字符串、列表或 None
|
||
headless=False,
|
||
)
|
||
|
||
cleaner.clean(id_file)
|
||
input("执行完毕，按回车键退出程序...")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|