de25841364
Apply consistent code formatting including: - Single quotes to double quotes for string literals - Multi-line formatting for long expressions - Trailing whitespace cleanup - Blank line adjustments for PEP 8 compliance Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
272 lines
11 KiB
Python
272 lines
11 KiB
Python
"""
|
||
离散备料计划维护数据清理工具
|
||
功能：自动登录 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()
|