3b7c00377f
Apply Black formatter to the entire codebase for consistent code style. Co-Authored-By: Claude (glm-5) <noreply@anthropic.com>
332 lines
12 KiB
Python
332 lines
12 KiB
Python
"""
|
||
离散备料计划维护数据提取工具 - 日志同步优化版
|
||
功能：负责登录 ERP、批量下载数据、转换并合并数据，支持与 UI 实时同步标准格式日志。
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
import time
|
||
import logging
|
||
import pandas as pd
|
||
from typing import Callable, Optional, List
|
||
from playwright.sync_api import sync_playwright, TimeoutError
|
||
|
||
# 统一顶部导入
|
||
from utils.excel_converter import ExcelConverter
|
||
from utils.auth import login, logout
|
||
from db.production_order_query import (
|
||
read_production_ids,
|
||
query_production_order_numbers,
|
||
)
|
||
|
||
# --- 进度条对象导入 (保持容错) ---
|
||
try:
|
||
from gui.progress import ProgressInfo
|
||
except ImportError:
|
||
ProgressInfo = None
|
||
|
||
# --- 全局日志配置 ---
|
||
# 调整格式：增加 [] 使其与 UI 控件的默认风格保持一致
|
||
LOG_FORMAT = "[%(asctime)s] [%(levelname)s] %(message)s"
|
||
DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
|
||
|
||
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT, datefmt=DATE_FORMAT)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class DiscreteMaterialPlanExtractor:
|
||
"""离散备料计划维护数据提取器"""
|
||
|
||
def __init__(
|
||
self,
|
||
username,
|
||
password,
|
||
headless=False,
|
||
verbose=True,
|
||
batch_size=100,
|
||
enable_db_persistence=False,
|
||
):
|
||
self.username = username
|
||
self.password = password
|
||
self.headless = headless
|
||
self.verbose = verbose
|
||
self.batch_size = batch_size
|
||
self.progress_callback = None
|
||
self.converter = ExcelConverter(verbose=verbose)
|
||
self.enable_db_persistence = enable_db_persistence
|
||
self.dao = None
|
||
|
||
if self.enable_db_persistence:
|
||
try:
|
||
from db.discrete_material_plan_dao import DiscreteMaterialPlanDAO
|
||
|
||
self.dao = DiscreteMaterialPlanDAO()
|
||
except ImportError:
|
||
self._log("无法加载数据库 DAO 模块，持久化功能将不可用", "error")
|
||
|
||
def _log(self, message, level="info"):
|
||
"""
|
||
统一日志出口：同步分发到控制台和 UI 回调
|
||
"""
|
||
level = level.lower()
|
||
# 1. 记录到标准控制台
|
||
log_map = {"info": logger.info, "warn": logger.warning, "error": logger.error}
|
||
log_func = log_map.get(level, logger.info)
|
||
log_func(message)
|
||
|
||
# 2. 同步到 UI
|
||
# 优化：发送原始 message，让 UI 控件自行添加时间戳，确保格式统一且不报错
|
||
if self.progress_callback:
|
||
self._report_progress("log", 0, 0, message, log_level=level.upper())
|
||
|
||
def _report_progress(
|
||
self, stage: str, current: int, total: int, message: str, **detail
|
||
):
|
||
"""标准化进度汇报"""
|
||
if self.progress_callback and ProgressInfo:
|
||
try:
|
||
progress_info = ProgressInfo(
|
||
stage=stage,
|
||
current=current,
|
||
total=total,
|
||
message=message,
|
||
detail=detail,
|
||
)
|
||
self.progress_callback(progress_info)
|
||
except Exception:
|
||
pass
|
||
|
||
def get_production_order_numbers(self, production_id_file, report_progress=False):
|
||
"""读取总排号并查询数据库获取生产订单号"""
|
||
if report_progress:
|
||
self._report_progress(
|
||
"query", 1, 3, "正在读取总排号文件...", action="read_file"
|
||
)
|
||
|
||
production_ids = read_production_ids(production_id_file)
|
||
self._log(f"文件读取完成: 找到 {len(production_ids)} 个 Production ID")
|
||
|
||
if report_progress:
|
||
self._report_progress(
|
||
"query",
|
||
2,
|
||
3,
|
||
"正在查询数据库获取生产订单号...",
|
||
action="query_database",
|
||
)
|
||
|
||
order_ids = query_production_order_numbers(production_ids)
|
||
self._log(f"数据库查询完成: 共匹配到 {len(order_ids)} 条生产订单号")
|
||
|
||
if report_progress:
|
||
self._report_progress(
|
||
"query", 3, 3, "订单号查询阶段结束", action="query_complete"
|
||
)
|
||
|
||
return order_ids
|
||
|
||
def group_order_ids(self, order_ids, group_size=100):
|
||
"""生成器：按批次切割订单号"""
|
||
for i in range(0, len(order_ids), group_size):
|
||
yield order_ids[i : i + group_size]
|
||
|
||
def download_batch(self, inner_frame, order_ids, batch_index, total_batches, page1):
|
||
"""执行单批次数据的下载流程"""
|
||
self._report_progress(
|
||
"download",
|
||
batch_index * 7 + 1,
|
||
total_batches * 7,
|
||
f"第 {batch_index + 1} 批: 正在填充订单号",
|
||
action="fill_orders",
|
||
)
|
||
|
||
textbox = inner_frame.get_by_role("textbox", name="来源生产订单号")
|
||
textbox.fill("")
|
||
textbox.fill(",".join(order_ids))
|
||
|
||
inner_frame.locator(".search-component-searchBtn").click()
|
||
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
|
||
|
||
inner_frame.get_by_role("row", name="序号").get_by_label("").click()
|
||
inner_frame.get_by_role("button", name="更多").hover()
|
||
inner_frame.get_by_text("输出", exact=True).click()
|
||
|
||
threshold_box = (
|
||
inner_frame.locator("div")
|
||
.filter(has_text=re.compile(r"^行数阈值$"))
|
||
.locator("input[type='text']")
|
||
)
|
||
threshold_box.fill("300000")
|
||
|
||
download_path = f"D:/python/playwrite/data/temp_batch_{batch_index + 1}.xlsx"
|
||
with page1.expect_download() as download_info:
|
||
inner_frame.get_by_role("button", name="确定(Y)").click()
|
||
|
||
download = download_info.value
|
||
download.save_as(download_path)
|
||
self._log(f"批次 {batch_index + 1} 下载成功 -> {download_path}")
|
||
|
||
time.sleep(1)
|
||
return download_path
|
||
|
||
def convert_and_merge_files(self, file_paths, output_path):
|
||
"""合并 Excel 文件并清理临时文件"""
|
||
output_path = os.path.normpath(output_path)
|
||
output_dir = os.path.dirname(output_path)
|
||
if output_dir and not os.path.exists(output_dir):
|
||
os.makedirs(output_dir)
|
||
|
||
all_dfs = []
|
||
total_steps = len(file_paths) * 2 + 3
|
||
|
||
for i, path in enumerate(file_paths, 1):
|
||
self._report_progress(
|
||
"convert",
|
||
1 + (i - 1) * 2 + 1,
|
||
total_steps,
|
||
f"正在转换 Excel {i}/{len(file_paths)}",
|
||
)
|
||
df = self.converter.convert(path, output_file=None)
|
||
all_dfs.append(df)
|
||
self._log(f"文件 {i} 转换完成: 提取到 {len(df)} 条记录")
|
||
|
||
if all_dfs:
|
||
self._report_progress(
|
||
"convert", total_steps - 1, total_steps, "正在进行最终数据合并..."
|
||
)
|
||
merged_df = pd.concat(all_dfs, ignore_index=True)
|
||
merged_df.to_excel(output_path, index=False)
|
||
|
||
for p in file_paths:
|
||
try:
|
||
os.remove(p)
|
||
except:
|
||
pass
|
||
|
||
return output_path, merged_df
|
||
return None, None
|
||
|
||
def _save_to_database(self, df: pd.DataFrame):
|
||
"""将结果存入数据库并打印详细统计信息"""
|
||
if not self.dao:
|
||
return
|
||
try:
|
||
self._report_progress("database", 1, 3, "正在将数据同步至数据库...")
|
||
# 使用 with 关键字确保资源安全释放
|
||
with self.dao as db:
|
||
stats = db.save_dataframe_with_replace(df)
|
||
|
||
# 保留并输出完整的处理细节：删除条数和新增条数
|
||
msg = f"数据库保存完成: 删除 {stats.get('deleted', 0)} 条, 新增 {stats.get('inserted', 0)} 条"
|
||
self._log(msg, "info")
|
||
|
||
except Exception as e:
|
||
self._log(f"数据库保存失败: {str(e)}", "error")
|
||
|
||
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()
|
||
|
||
input_box = inner_frame.locator("#rc_select_0")
|
||
input_box.fill("5000")
|
||
input_box.press("Enter")
|
||
|
||
def extract(
|
||
self,
|
||
production_id_file,
|
||
output_file="D:/python/playwrite/data/离散备料计划维护_合并.xlsx",
|
||
progress_callback=None,
|
||
):
|
||
"""主入口：执行全流程数据提取任务"""
|
||
self.progress_callback = progress_callback
|
||
downloaded_files = []
|
||
|
||
try:
|
||
with sync_playwright() as playwright:
|
||
self._report_progress("login", 1, 3, "启动浏览器并尝试登录 ERP...")
|
||
browser, context, page, main_frame = login(
|
||
playwright=playwright,
|
||
username=self.username,
|
||
password=self.password,
|
||
headless=self.headless,
|
||
ignore_https_errors=True,
|
||
)
|
||
|
||
self._log(
|
||
"======================================== 开始执行数据提取任务 ========================================"
|
||
)
|
||
|
||
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
|
||
|
||
f_frame = page1.locator("#forwardFrame").content_frame
|
||
inner_frame_locator = f_frame.locator("#mainiframe")
|
||
inner_frame_locator.wait_for(state="visible", timeout=15000)
|
||
work_frame = inner_frame_locator.content_frame
|
||
|
||
self.setup_query_interface(work_frame)
|
||
order_ids = self.get_production_order_numbers(
|
||
production_id_file, report_progress=True
|
||
)
|
||
|
||
batch_list = list(self.group_order_ids(order_ids, self.batch_size))
|
||
for i, batch_ids in enumerate(batch_list):
|
||
self._log(f"正在处理第 {i+1} 批次 (共 {len(batch_list)} 批)")
|
||
try:
|
||
f_path = self.download_batch(
|
||
work_frame, batch_ids, i, len(batch_list), page1
|
||
)
|
||
downloaded_files.append(f_path)
|
||
except Exception as e:
|
||
self._log(
|
||
f"批次 {i+1} 处理异常，已跳过。详细错误: {e}", "error"
|
||
)
|
||
continue
|
||
|
||
self._log("正在注销并关闭浏览器环境...")
|
||
logout(f_frame, verbose=self.verbose)
|
||
context.close()
|
||
browser.close()
|
||
|
||
if downloaded_files:
|
||
final_path, final_df = self.convert_and_merge_files(
|
||
downloaded_files, output_file
|
||
)
|
||
if self.enable_db_persistence and final_df is not None:
|
||
self._save_to_database(final_df)
|
||
|
||
self._log(f"所有流程已顺利结束，结果文件: {final_path}")
|
||
self._report_progress("complete", 1, 1, "任务完成")
|
||
return final_path
|
||
|
||
self._log("未获得任何有效数据，任务终止", "warn")
|
||
return None
|
||
|
||
finally:
|
||
self.progress_callback = None
|
||
|
||
|
||
def main():
|
||
extractor = DiscreteMaterialPlanExtractor(
|
||
username="BLDpengqiangqiang",
|
||
password="your_password",
|
||
enable_db_persistence=True,
|
||
)
|
||
id_file = os.path.join(os.path.dirname(__file__), "productionID.txt")
|
||
extractor.extract(id_file)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|