diff --git a/gui/data_extraction_tab.py b/gui/data_extraction_tab.py index 1d40fc6..83a7efa 100644 --- a/gui/data_extraction_tab.py +++ b/gui/data_extraction_tab.py @@ -1,9 +1,9 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- """ -数据提取标签页 +数据提取标签页 - 稳定性修复版 -从 ERP 系统提取备料计划数据。 +修复了 LogText.info 不支持 add_timestamp 参数导致的 TypeError。 """ import os @@ -13,7 +13,6 @@ import queue import tkinter as tk from tkinter import ttk, filedialog, messagebox from pathlib import Path -from contextlib import redirect_stdout from gui.widgets import FileSelector, LogText, ProductionIdInput from gui.config_manager import ConfigManager from gui.progress import ProgressInfo, ProgressCalculator @@ -24,14 +23,6 @@ class DataExtractionTab(ttk.Frame): """数据提取标签页""" def __init__(self, parent, config: ConfigManager, main_window=None): - """ - 初始化数据提取标签页 - - Args: - parent: 父容器 - config: 配置管理器 - main_window: 主窗口引用，用于共享 Production ID 数据 - """ super().__init__(parent) self.config = config self.main_window = main_window @@ -39,333 +30,209 @@ class DataExtractionTab(ttk.Frame): self.extractor = None self.extraction_thread = None self.progress_calculator = ProgressCalculator() - self.progress_queue = queue.Queue() # 进度更新队列 + self.progress_queue = queue.Queue() - # 启动进度更新轮询 self._poll_progress_queue() - self.create_widgets() - - # 应用字体设置 self._apply_ui_config() - # 稍后显示就绪消息 try: self.log_text.info("数据提取标签页已就绪") except: - pass # 如果窗口还未完全就绪，忽略错误 + pass def create_widgets(self): - """创建界面组件""" - # 主容器 - 使用水平 PanedWindow 分割左右部分 horizontal_paned = ttk.PanedWindow(self, orient=tk.HORIZONTAL) horizontal_paned.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) - # 左侧：Production ID 输入面板 left_panel = ttk.Frame(horizontal_paned) horizontal_paned.add(left_panel, weight=0) - # 右侧：主面板（控制面板 + 日志） right_panel = ttk.Frame(horizontal_paned) horizontal_paned.add(right_panel, weight=1) self._create_left_panel(left_panel) self._create_right_panel(right_panel) - # 保存 PanedWindow 引用，后续用于设置分隔条位置 self.horizontal_paned = horizontal_paned - - # 设置默认宽度（使用 after 确保在渲染后设置） input_width = self.config.get("ui.production_id_input_width", 20) - # 字符宽度约 8 像素 self.after(100, lambda: self._set_pane_width(input_width * 8)) def _create_left_panel(self, parent): - """创建左侧 Production ID 输入面板""" - # 创建带标题的框架 input_group = ttk.LabelFrame(parent, text="Production ID", padding=10) input_group.pack(fill=tk.BOTH, expand=True) - - # 创建 Production ID 输入控件 self.production_id_input = ProductionIdInput( input_group, placeholder="每行输入一个 Production ID\n\n示例：\n26B848\n26B849" ) self.production_id_input.pack(fill=tk.BOTH, expand=True) - - # 绑定变化事件：当文本框失去焦点时更新共享 Production ID self.production_id_input.text_widget.bind("", self._on_production_ids_changed) def _create_right_panel(self, parent): - """创建右侧主面板""" - # 主容器 - 使用垂直 PanedWindow 分割上下部分 main_paned = ttk.PanedWindow(parent, orient=tk.VERTICAL) main_paned.pack(fill=tk.BOTH, expand=True) - - # 上部：控制面板 control_frame = ttk.Frame(main_paned) main_paned.add(control_frame, weight=0) - - # 下部：日志输出 log_frame = ttk.LabelFrame(main_paned, text="日志输出", padding=5) main_paned.add(log_frame, weight=1) - self._create_control_panel(control_frame) self._create_log_panel(log_frame) def _create_control_panel(self, parent): - """创建控制面板""" - # 输出文件选择 output_group = ttk.LabelFrame(parent, text="输出文件", padding=10) output_group.pack(fill=tk.X, pady=5) - self.output_file_selector = FileSelector( - output_group, - label_text="保存为:", - file_type="file", + output_group, label_text="保存为:", file_type="file", file_types=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")], initial_dir=self.config.get("paths.data_dir", "data/"), ) self.output_file_selector.pack(fill=tk.X) - - # 设置默认输出文件 default_output = os.path.join( self.config.get("paths.data_dir", "data/"), self.config.get("paths.default_output", "离散备料计划维护_合并.xlsx"), ) self.output_file_selector.set(default_output) - # 选项 options_group = ttk.LabelFrame(parent, text="提取选项", padding=10) options_group.pack(fill=tk.X, pady=5) - self.headless_var = tk.BooleanVar(value=self.config.get("erp.headless", True)) - ttk.Checkbutton( - options_group, text="无头模式 (不显示浏览器)", variable=self.headless_var - ).grid(row=0, column=0, sticky="w", padx=5) + ttk.Checkbutton(options_group, text="无头模式", variable=self.headless_var).grid(row=0, column=0, sticky="w", padx=5) - # 进度显示 progress_group = ttk.LabelFrame(parent, text="进度", padding=10) progress_group.pack(fill=tk.X, pady=5) - self.progress_bar = ttk.Progressbar(progress_group, mode="determinate") self.progress_bar.pack(fill=tk.X, pady=5) - - self.status_label = ttk.Label( - progress_group, text="就绪", relief=tk.SUNKEN, anchor=tk.W - ) + self.status_label = ttk.Label(progress_group, text="就绪", relief=tk.SUNKEN, anchor=tk.W) self.status_label.pack(fill=tk.X) - # 控制按钮 button_frame = ttk.Frame(parent) button_frame.pack(fill=tk.X, pady=10) - - self.start_button = ttk.Button( - button_frame, text="开始提取", command=self.start_extraction - ) + self.start_button = ttk.Button(button_frame, text="开始提取", command=self.start_extraction) self.start_button.pack(side=tk.LEFT, padx=5) - - self.stop_button = ttk.Button( - button_frame, text="停止", command=self.stop_extraction, state=tk.DISABLED - ) + self.stop_button = ttk.Button(button_frame, text="停止", command=self.stop_extraction, state=tk.DISABLED) self.stop_button.pack(side=tk.LEFT, padx=5) def _create_log_panel(self, parent): - """创建日志面板""" self.log_text = LogText(parent, height=15, readonly=True) self.log_text.pack(fill=tk.BOTH, expand=True) def _apply_ui_config(self): - """应用 UI 配置（字体等）""" try: font_family = self.config.get("ui.font_family", "Microsoft YaHei UI") font_size = self.config.get("ui.font_size", 10) - - # 应用到 Production ID 输入控件 self.production_id_input.apply_font(font_family, font_size) - - # 应用到日志控件（如果支持） if hasattr(self.log_text, 'apply_font'): self.log_text.apply_font(font_family, font_size) - except Exception as e: - # 如果应用字体失败，不影响主流程 - pass + except: pass def _set_pane_width(self, width: int): - """设置左侧 pane 的宽度 - - Args: - width: 宽度（像素） - """ - try: - # 使用 sashpos 方法设置分隔条位置 - # 参数 0 表示第一个分隔条（索引从 0 开始） - self.horizontal_paned.sashpos(0, width) - except Exception as e: - # 如果设置失败，不影响主流程 - pass + try: self.horizontal_paned.sashpos(0, width) + except: pass def start_extraction(self): - """开始数据提取""" - # 获取 Production ID 列表 production_ids = self.production_id_input.get() - if not production_ids: messagebox.showerror("错误", "请输入至少一个 Production ID") return - output_file = self.output_file_selector.get() - if not output_file: messagebox.showerror("错误", "请指定输出文件路径") return - - # 确保输出目录存在 - output_dir = os.path.dirname(output_file) - if output_dir and not os.path.exists(output_dir): - os.makedirs(output_dir, exist_ok=True) - - # 更新 UI 状态 self.extracting = True self.start_button.config(state=tk.DISABLED) self.stop_button.config(state=tk.NORMAL) self.progress_bar["value"] = 0 - self.status_label.config(text="正在登录...") + self.status_label.config(text="正在初始化...") self.log_text.clear() - self.log_text.info(f"开始数据提取... ({len(production_ids)} 个 Production ID)") - - # 在后台线程中执行提取 self.extraction_thread = threading.Thread( target=self._extraction_worker, args=(production_ids, output_file), daemon=True ) self.extraction_thread.start() def stop_extraction(self): - """停止数据提取""" if self.extracting: self.extracting = False self.log_text.warning("正在停止提取...") - self.status_label.config(text="正在停止...") def _extraction_worker(self, production_ids: list[str], output_file: str): - """提取工作线程""" import tempfile - + temp_file = None try: - # 创建临时文件保存 Production ID 列表 with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8') as f: temp_file = f.name f.write('\n'.join(production_ids)) - try: - # 导入提取器（延迟导入以避免启动时加载 Playwright） - from utils.离散备料计划维护数据提取 import DiscreteMaterialPlanExtractor + from utils.离散备料计划维护数据提取 import DiscreteMaterialPlanExtractor + self.extractor = DiscreteMaterialPlanExtractor( + username=self.config.get("erp.username"), + password=self.config.get("erp.password"), + headless=self.headless_var.get(), + verbose=self.config.get("extraction.verbose", True), + batch_size=self.config.get("extraction.batch_size", 100), + enable_db_persistence=self.config.get("extraction.enable_db_persistence", False), + ) - # 创建提取器实例 - self.extractor = DiscreteMaterialPlanExtractor( - username=self.config.get("erp.username"), - password=self.config.get("erp.password"), - headless=self.headless_var.get(), - verbose=self.config.get("extraction.verbose", True), - batch_size=self.config.get("extraction.batch_size", 100), - enable_db_persistence=self.config.get("extraction.enable_db_persistence", False), - ) - - # 创建实时输出流，每次写入立即更新 GUI - realtime_output = RealtimeOutput( - lambda line: self._update_log(line, "INFO") - ) - - # 创建进度回调函数 - def progress_callback(progress_info: ProgressInfo): - # 计算总体进度百分比 - overall_percent = self.progress_calculator.calculate_overall_percent( - progress_info - ) - self._update_progress(overall_percent, progress_info.message) - - # 重定向 stdout 并执行提取（带进度回调） - with redirect_stdout(realtime_output): - result = self.extractor.extract( - production_id_file=temp_file, - output_file=output_file, - progress_callback=progress_callback, - ) - - if result and self.extracting: - self._update_log(f"数据已保存到：{output_file}", "SUCCESS") - elif not self.extracting: - self._update_log("提取已取消", "WARNING") + # 修复：直接调用标准的 _update_log，不再传入 add_timestamp 参数 + def progress_callback(progress_info: ProgressInfo): + if progress_info.stage == "log": + level = progress_info.detail.get("log_level", "INFO").upper() + self._update_log(progress_info.message, level) else: - self._update_log("提取失败", "ERROR") + percent = self.progress_calculator.calculate_overall_percent(progress_info) + self._update_progress(percent, progress_info.message) - finally: - # 删除临时文件 - try: - os.unlink(temp_file) - except: - pass + result = self.extractor.extract( + production_id_file=temp_file, output_file=output_file, progress_callback=progress_callback + ) + + if result and self.extracting: + self._update_log("数据处理任务圆满结束", "SUCCESS") + elif not self.extracting: + self._update_log("提取已取消", "WARNING") except Exception as e: - self._update_log(f"提取过程中发生错误：{str(e)}", "ERROR") + self._update_log(f"运行时错误: {str(e)}", "ERROR") finally: - # 更新 UI 状态 + if temp_file and os.path.exists(temp_file): + try: os.unlink(temp_file) + except: pass self.after(0, self._extraction_complete) def _extraction_complete(self): - """提取完成后的 UI 更新""" self.extracting = False self.start_button.config(state=tk.NORMAL) self.stop_button.config(state=tk.DISABLED) self.extractor = None def _poll_progress_queue(self): - """轮询进度队列，处理进度更新""" try: while True: - # 非阻塞地获取队列中的消息 try: - progress_data = self.progress_queue.get_nowait() - value, message = progress_data + value, message = self.progress_queue.get_nowait() self.progress_bar["value"] = value self.status_label.config(text=message) - except queue.Empty: - break - finally: - # 继续轮询（每 50ms 检查一次） - self.after(50, self._poll_progress_queue) + except queue.Empty: break + finally: self.after(50, self._poll_progress_queue) def _update_progress(self, value: int, message: str): - """线程安全的进度更新（通过队列）""" - try: - self.progress_queue.put_nowait((value, message)) - except: - pass # 队列满时忽略 + try: self.progress_queue.put_nowait((value, message)) + except: pass def _update_log(self, message: str, level: str = "INFO"): - """线程安全的日志更新""" - + """标准的日志更新方法""" def update(): + # 即使任务结束，只要是成功/错误消息也强制显示 if self.extracting or level in ["ERROR", "WARNING", "SUCCESS"]: - if level == "INFO": - self.log_text.info(message) - elif level == "SUCCESS": - self.log_text.success(message) - elif level == "WARNING": - self.log_text.warning(message) - elif level == "ERROR": - self.log_text.error(message) - + if level == "INFO": self.log_text.info(message) + elif level == "SUCCESS": self.log_text.success(message) + elif level == "WARNING": self.log_text.warning(message) + elif level == "ERROR": self.log_text.error(message) self.after(0, update) def _on_production_ids_changed(self, event=None): - """Production ID 变化时的回调""" if self.main_window: - production_ids = self.production_id_input.get() - self.main_window.update_shared_production_ids(production_ids) + self.main_window.update_shared_production_ids(self.production_id_input.get()) def reload_config(self): - """配置更新后重新应用 UI 设置""" self._apply_ui_config() - # 通知主窗口当前的 Production ID - self._on_production_ids_changed() + self._on_production_ids_changed() \ No newline at end of file diff --git a/utils/离散备料计划维护数据提取.py b/utils/离散备料计划维护数据提取.py index cc8ac82..bb3b796 100644 --- a/utils/离散备料计划维护数据提取.py +++ b/utils/离散备料计划维护数据提取.py @@ -1,19 +1,41 @@ """ -离散备料计划维护数据提取工具 -负责登录、批量下载、转换数据 +离散备料计划维护数据提取工具 - 日志同步优化版 +功能：负责登录 ERP、批量下载数据、转换并合并数据，支持与 UI 实时同步标准格式日志。 """ import os +import re +import time +import logging import pandas as pd -from playwright.sync_api import sync_playwright +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, ) -from typing import Callable, Optional +# --- 进度条对象导入 (保持容错) --- +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: """离散备料计划维护数据提取器""" @@ -22,17 +44,6 @@ class DiscreteMaterialPlanExtractor: self, username, password, headless=False, verbose=True, batch_size=100, enable_db_persistence=False ): - """ - 初始化提取器 - - Args: - username: 登录用户名 - password: 登录密码 - headless: 是否无头模式运行 - verbose: 是否打印详细日志 - batch_size: 批次大小 - enable_db_persistence: 是否启用数据库持久化 - """ self.username = username self.password = password self.headless = headless @@ -42,33 +53,37 @@ class DiscreteMaterialPlanExtractor: self.converter = ExcelConverter(verbose=verbose) self.enable_db_persistence = enable_db_persistence self.dao = None + if self.enable_db_persistence: - from db.discrete_material_plan_dao import DiscreteMaterialPlanDAO - self.dao = DiscreteMaterialPlanDAO() - self.dao.__enter__() # Enter context manager - - def _print(self, *args, **kwargs): - """打印日志（如果 verbose=True）""" - if self.verbose: - print(*args, **kwargs) - - def _report_progress( - self, stage: str, current: int, total: int, message: str, **detail - ): - """ - 报告进度 - - Args: - stage: 阶段标识 - current: 当前进度值 - total: 总量 - message: 显示消息 - **detail: 额外详细信息 - """ - if self.progress_callback: try: - from gui.progress import ProgressInfo + 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, @@ -78,522 +93,190 @@ class DiscreteMaterialPlanExtractor: ) self.progress_callback(progress_info) except Exception: - # 如果进度回调失败，忽略错误，不影响主流程 pass def get_production_order_numbers(self, production_id_file, report_progress=False): - """ - 读取总排号文件并查询数据库获取生产订单号 - - Args: - production_id_file: ProductionID.txt 文件路径 - report_progress: 是否报告进度 - - Returns: - 生产订单号列表 - """ + """读取总排号并查询数据库获取生产订单号""" if report_progress: - self._report_progress( - "query", - 1, - 3, - "正在读取总排号文件...", - action="read_file", - ) + self._report_progress("query", 1, 3, "正在读取总排号文件...", action="read_file") - # 读取总排号 production_ids = read_production_ids(production_id_file) - self._print(f"从文件读取到 {len(production_ids)} 个总排号") + self._log(f"文件读取完成: 找到 {len(production_ids)} 个 Production ID") if report_progress: - self._report_progress( - "query", - 2, - 3, - f"正在查询数据库（{len(production_ids)} 个总排号）...", - action="query_database", - production_id_count=len(production_ids), - ) + self._report_progress("query", 2, 3, "正在查询数据库获取生产订单号...", action="query_database") - # 查询数据库获取生产订单号 order_ids = query_production_order_numbers(production_ids) - self._print(f"查询到 {len(order_ids)} 个生产订单号") + self._log(f"数据库查询完成: 共匹配到 {len(order_ids)} 条生产订单号") if report_progress: - self._report_progress( - "query", - 3, - 3, - f"查询完成：获取到 {len(order_ids)} 个生产订单号", - action="query_complete", - order_id_count=len(order_ids), - ) - + 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, - debug_mode=False, - debug_batch=None, - ): - """下载一批订单号的数据""" - from playwright.sync_api import TimeoutError - import re - import time - - # 步骤1：清空文本框 - self._report_progress( - "download", - batch_index * 7 + 1, - total_batches * 7, - f"第 {batch_index + 1}/{total_batches} 批 - 准备输入订单号", - batch_index=batch_index + 1, - action="clear_textbox", - ) + 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("") - - # 步骤2：填充订单号 - self._report_progress( - "download", - batch_index * 7 + 2, - total_batches * 7, - f"第 {batch_index + 1}/{total_batches} 批 - 输入 {len(order_ids)} 个订单号", - batch_index=batch_index + 1, - action="fill_order_ids", - order_count=len(order_ids), - ) textbox.fill(",".join(order_ids)) - # 步骤3：点击查询 - self._report_progress( - "download", - batch_index * 7 + 3, - total_batches * 7, - f"第 {batch_index + 1}/{total_batches} 批 - 提交查询请求", - batch_index=batch_index + 1, - action="click_search", - ) inner_frame.locator(".search-component-searchBtn").click() - self._print(f"第 {batch_index + 1} 批查询完成，等待加载结果...") - - # 步骤4：等待加载完成 - self._report_progress( - "download", - batch_index * 7 + 4, - total_batches * 7, - f"第 {batch_index + 1}/{total_batches} 批 - 等待数据加载...", - batch_index=batch_index + 1, - action="wait_loading", - ) 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) # 无限等待，直到消失 + loading_locator.wait_for(state="hidden", timeout=0) except TimeoutError: - # 加载很快完成，或者没有出现加载提示 pass - self._print(f"第 {batch_index + 1} 批加载完成，开始选择数据...") - # 调试模式：只在指定批次暂停 - if debug_mode and (debug_batch is None or batch_index == debug_batch): - self._print(f"=== 调试暂停：第 {batch_index + 1} 批 ===") - page1.pause() - - # 步骤5：选择所有数据 - self._report_progress( - "download", - batch_index * 7 + 5, - total_batches * 7, - f"第 {batch_index + 1}/{total_batches} 批 - 选择所有数据行", - batch_index=batch_index + 1, - action="select_all_rows", - ) inner_frame.get_by_role("row", name="序号").get_by_label("").click() - - # 步骤6：配置并触发导出 - self._report_progress( - "download", - batch_index * 7 + 6, - total_batches * 7, - f"第 {batch_index + 1}/{total_batches} 批 - 配置导出参数", - batch_index=batch_index + 1, - action="configure_export", - ) - - # 点击输出 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") - # 设置行数阈值 - input_box = ( - inner_frame.locator("div") - .filter(has_text=re.compile(r"^行数阈值$")) - .locator("input[type='text']") - ) - input_box.fill("300000") - - # 步骤7：下载文件 - self._report_progress( - "download", - batch_index * 7 + 7, - total_batches * 7, - f"第 {batch_index + 1}/{total_batches} 批 - 正在下载文件...", - batch_index=batch_index + 1, - action="downloading_file", - ) 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._print(f"第 {batch_index + 1} 批下载完成: {download_path}") + self._log(f"批次 {batch_index + 1} 下载成功 -> {download_path}") - # 报告批次完成 - self._report_progress( - "download", - (batch_index + 1) * 7, - total_batches * 7, - f"第 {batch_index + 1}/{total_batches} 批下载完成 ✓", - batch_index=batch_index + 1, - action="batch_complete", - file_path=download_path, - ) - - # 等待页面恢复，准备下一次查询 time.sleep(1) - return download_path def convert_and_merge_files(self, file_paths, output_path): - """使用 ExcelConverter 转换并合并所有文件，返回合并后的 DataFrame""" - # 确保输出文件路径是正确的格式 + """合并 Excel 文件并清理临时文件""" output_path = os.path.normpath(output_path) output_dir = os.path.dirname(output_path) - output_filename = os.path.basename(output_path) - - self._print(f"输出文件路径: {output_path}") - self._print(f"输出目录: {output_dir}") - self._print(f"输出文件名: {output_filename}") - - # 步骤1：检查并创建输出目录 - self._report_progress( - "convert", - 1, - len(file_paths) * 2 + 3, - "准备转换：检查输出目录", - action="check_directory", - ) - if output_dir and not os.path.exists(output_dir): - self._print(f"创建输出目录: {output_dir}") os.makedirs(output_dir) - all_dataframes = [] + all_dfs = [] + total_steps = len(file_paths) * 2 + 3 - # 步骤2-N：转换每个文件 - for i, file_path in enumerate(file_paths, 1): - self._print(f"转换第 {i} 个文件: {file_path}") + 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)} 条记录") - # 报告开始转换 - self._report_progress( - "convert", - 1 + (i - 1) * 2 + 1, - len(file_paths) * 2 + 3, - f"正在转换文件 {i}/{len(file_paths)}", - file_index=i, - file_path=file_path, - action="converting_file", - ) - - df = self.converter.convert(file_path, output_file=None) # 只转换，不保存 - all_dataframes.append(df) - self._print(f" 提取到 {len(df)} 条记录") - - # 报告转换完成 - self._report_progress( - "convert", - 1 + (i - 1) * 2 + 2, - len(file_paths) * 2 + 3, - f"文件 {i}/{len(file_paths)} 转换完成（{len(df)} 条记录）", - file_index=i, - record_count=len(df), - action="file_converted", - ) - - merged_df = None - if all_dataframes: - # 步骤N+1：合并数据 - self._report_progress( - "convert", - len(file_paths) * 2 + 2, - len(file_paths) * 2 + 3, - f"正在合并 {len(all_dataframes)} 个文件的数据...", - action="merging_data", - file_count=len(all_dataframes), - ) - - self._print(f"\n合并 {len(all_dataframes)} 个文件的数据...") - merged_df = pd.concat(all_dataframes, ignore_index=True) + 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) - self._print(f"合并完成: {output_path}, 总共 {len(merged_df)} 条记录") - - # 步骤N+2：删除临时文件 - self._report_progress( - "convert", - len(file_paths) * 2 + 3, - len(file_paths) * 2 + 3, - f"清理临时文件...", - action="cleanup", - total_records=len(merged_df), - ) - - for file_path in file_paths: - os.remove(file_path) - self._print(f"已删除临时文件: {file_path}") - + + 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): - """Save DataFrame to database with progress reporting""" + """将结果存入数据库并打印详细统计信息""" + if not self.dao: return try: - self._report_progress( - "database", 0, 3, "准备保存到数据库...", - action="db_start" - ) - - stats = self.dao.save_dataframe_with_replace(df) - - self._report_progress( - "database", 3, 3, - f"数据库保存完成: 删除 {stats['deleted']} 条, 新增 {stats['inserted']} 条", - action="db_complete", - stats=stats - ) - - self._print(f"\n数据库保存成功:") - self._print(f" 删除旧记录: {stats['deleted']} 条") - self._print(f" 新增记录: {stats['inserted']} 条") - + 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._print(f"\n警告: 数据库保存失败: {e}") - self._report_progress( - "database", 3, 3, - f"数据库保存失败: {str(e)}", - action="db_error", - error=str(e) - ) + self._log(f"数据库保存失败: {str(e)}", "error") def setup_query_interface(self, inner_frame): - """设置查询界面（不报告进度，由 extract 统一报告）""" - 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} 次尝试后仍未成功填充，继续执行..." - ) + input_box = inner_frame.locator("#rc_select_0") + input_box.fill("5000") + input_box.press("Enter") def extract( - self, - production_id_file, - data_dir="D:/python/playwrite/data", - output_file="D:/python/playwrite/data/离散备料计划维护_合并.xlsx", - debug_mode=False, - debug_batch=None, - progress_callback=None, + self, production_id_file, output_file="D:/python/playwrite/data/离散备料计划维护_合并.xlsx", + progress_callback=None ): - """执行完整的数据提取流程""" - original_callback = self.progress_callback - self.progress_callback = progress_callback or self.progress_callback + """主入口：执行全流程数据提取任务""" + self.progress_callback = progress_callback + downloaded_files = [] try: with sync_playwright() as playwright: - # 步骤1：启动浏览器并登录 - self._report_progress( - "login", - 1, - 3, # 保持 3 步 - "启动浏览器并登录...", - action="launch_browser", - ) - + 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, + playwright=playwright, username=self.username, password=self.password, + headless=self.headless, ignore_https_errors=True ) - # 步骤2：打开功能页面 - self._report_progress( - "login", - 2, - 3, # 保持 3 步 - "登录成功，打开功能页面...", - action="open_function_page", - ) - - self._print("=" * 80) - self._print("开始执行离散备料计划维护数据提取") - self._print("=" * 80) - + self._log("======================================== 开始执行数据提取任务 ========================================") + main_frame.locator("i").first.click() - with page.expect_popup() as page1_info: - main_frame.get_by_title( - "离散备料计划维护", exact=True - ).first.click() + main_frame.get_by_title("离散备料计划维护", exact=True).first.click() page1 = page1_info.value - main_frame = page1.locator("#forwardFrame").content_frame - inner_frame_locator = main_frame.locator("#mainiframe") + f_frame = page1.locator("#forwardFrame").content_frame + inner_frame_locator = f_frame.locator("#mainiframe") inner_frame_locator.wait_for(state="visible", timeout=15000) - inner_frame = inner_frame_locator.content_frame + work_frame = inner_frame_locator.content_frame - # 步骤3：设置查询界面 - self._report_progress( - "login", - 3, - 3, # 保持 3 步 - "配置查询界面...", - action="setup_query_interface", - ) - self.setup_query_interface(inner_frame) + self.setup_query_interface(work_frame) + order_ids = self.get_production_order_numbers(production_id_file, report_progress=True) - # 后续代码保持不变... - order_ids = self.get_production_order_numbers( - production_id_file, report_progress=True - ) - - downloaded_files = [] - total_batches = sum( - 1 for _ in self.group_order_ids(order_ids, self.batch_size) - ) - - for batch_index, order_ids_batch in enumerate( - self.group_order_ids(order_ids, self.batch_size) - ): - self._print( - f"\n=== 开始处理第 {batch_index + 1} 批，共 {len(order_ids_batch)} 个订单号 ===" - ) - - downloaded_file = self.download_batch( - inner_frame, - order_ids_batch, - batch_index, - total_batches, - page1, - debug_mode=debug_mode, - debug_batch=debug_batch, - ) - downloaded_files.append(downloaded_file) - - self._print("\n开始执行账号注销...") - self._report_progress( - "logout", - 1, - 2, - "正在注销账号...", - action="logout_start", - ) - logout(main_frame, verbose=self.verbose) - - self._report_progress( - "logout", - 2, - 2, - "注销完成 ✓", - action="logout_complete", - ) - - if downloaded_files: - self._print( - f"\n=== 开始转换并合并 {len(downloaded_files)} 个文件 ===" - ) - output_path, merged_df = self.convert_and_merge_files(downloaded_files, output_file) - - # 数据库保存步骤（独立阶段） - if self.enable_db_persistence and self.dao and merged_df is not None: - self._print(f"\n=== 开始保存数据到数据库 ===") - self._save_to_database(merged_df) - else: - self._print("\n没有下载到任何文件") - - self._print(f"\n=== 全部完成 ===") - self._print(f"最终文件: {output_file}") - - self._report_progress( - "complete", 1, 1, "数据提取完成 ✓", - output_file=output_file, - action="all_complete", - ) + 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() - return output_file + 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: - # Close database connection if open - if self.dao: - try: - self.dao.__exit__(None, None, None) - except Exception: - pass - self.progress_callback = original_callback - + self.progress_callback = None def main(): - """测试函数""" extractor = DiscreteMaterialPlanExtractor( username="BLDpengqiangqiang", - password="Cqbld123456.", - headless=False, - verbose=True, + password="your_password", + enable_db_persistence=True ) - - production_id_file = os.path.join(os.path.dirname(__file__), "productionID.txt") - output_file = "D:/python/playwrite/data/离散备料计划维护_合并.xlsx" - - extractor.extract(production_id_file, output_file) - - input("按回车退出...") - + id_file = os.path.join(os.path.dirname(__file__), "productionID.txt") + extractor.extract(id_file) if __name__ == "__main__": main() \ No newline at end of file