#!/usr/bin/env python # -*- coding: utf-8 -*- """ 数据提取标签页 从 ERP 系统提取备料计划数据。 """ import os import sys import threading 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 from gui.utils import RealtimeOutput 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 self.extracting = False self.extractor = None self.extraction_thread = None self.progress_calculator = ProgressCalculator() self.progress_queue = queue.Queue() # 进度更新队列 # 启动进度更新轮询 self._poll_progress_queue() self.create_widgets() # 应用字体设置 self._apply_ui_config() # 稍后显示就绪消息 try: self.log_text.info("数据提取标签页已就绪") except: 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", 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) # 进度显示 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.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.pack(side=tk.LEFT, padx=5) 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 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 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.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 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 # 创建提取器实例 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") else: self._update_log("提取失败", "ERROR") finally: # 删除临时文件 try: os.unlink(temp_file) except: pass except Exception as e: self._update_log(f"提取过程中发生错误：{str(e)}", "ERROR") finally: # 更新 UI 状态 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 self.progress_bar["value"] = value self.status_label.config(text=message) except queue.Empty: break finally: # 继续轮询（每 50ms 检查一次） 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 # 队列满时忽略 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) 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) def reload_config(self): """配置更新后重新应用 UI 设置""" self._apply_ui_config() # 通知主窗口当前的 Production ID self._on_production_ids_changed()