#!/usr/bin/env python # -*- coding: utf-8 -*- """ 数据提取标签页 从 ERP 系统提取备料计划数据。 """ import os import sys import threading import tkinter as tk from tkinter import ttk, filedialog, messagebox from pathlib import Path from io import StringIO from contextlib import redirect_stdout from gui.widgets import FileSelector, LogText from gui.config_manager import ConfigManager class DataExtractionTab(ttk.Frame): """数据提取标签页""" def __init__(self, parent, config: ConfigManager): """ 初始化数据提取标签页 Args: parent: 父容器 config: 配置管理器 """ super().__init__(parent) self.config = config self.extracting = False self.extractor = None self.extraction_thread = None self.create_widgets() # 稍后显示就绪消息 try: self.log_text.info("数据提取标签页已就绪") except: pass # 如果窗口还未完全就绪，忽略错误 def create_widgets(self): """创建界面组件""" # 主容器 - 使用 PanedWindow 分割上下部分 main_paned = ttk.PanedWindow(self, orient=tk.VERTICAL) main_paned.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) # 上部：控制面板 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): """创建控制面板""" # 输入文件选择 input_group = ttk.LabelFrame(parent, text="输入文件", padding=10) input_group.pack(fill=tk.X, pady=5) self.input_file_selector = FileSelector( input_group, label_text="ProductionID 文件:", file_type="file", file_types=[("文本文件", "*.txt"), ("所有文件", "*.*")], initial_dir="D:/python/playwrite/" ) self.input_file_selector.pack(fill=tk.X) # 设置默认文件 default_input = self.config.get('paths.production_id_file', 'ProductionID.txt') if os.path.exists(default_input): self.input_file_selector.set(default_input) # 输出文件选择 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.verbose_var = tk.BooleanVar(value=self.config.get('extraction.verbose', True)) ttk.Checkbutton(options_group, text="详细日志", variable=self.verbose_var).grid(row=0, column=0, sticky="w", padx=5) self.headless_var = tk.BooleanVar(value=self.config.get('browser.headless', True)) ttk.Checkbutton(options_group, text="无头模式 (不显示浏览器)", variable=self.headless_var).grid(row=0, column=1, 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 start_extraction(self): """开始数据提取""" # 验证输入 input_file = self.input_file_selector.get() output_file = self.output_file_selector.get() if not input_file: messagebox.showerror("错误", "请选择 ProductionID 输入文件") return if not os.path.exists(input_file): messagebox.showerror("错误", f"输入文件不存在：{input_file}") return 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.log_text.clear() self.log_text.info("开始数据提取...") # 在后台线程中执行提取 self.extraction_thread = threading.Thread( target=self._extraction_worker, args=(input_file, 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, input_file: str, output_file: str): """提取工作线程""" 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.verbose_var.get() ) # 捕获 stdout 输出 captured_output = StringIO() # 重定向 stdout 并执行提取 with redirect_stdout(captured_output): result = self.extractor.extract( production_id_file=input_file, output_file=output_file ) # 获取捕获的输出并显示到日志 output_text = captured_output.getvalue() if output_text: for line in output_text.split('\n'): if line.strip(): self._update_log(line, "INFO") if result and self.extracting: self._update_progress(100, "提取完成") self._update_log(f"数据已保存到：{output_file}", "SUCCESS") elif not self.extracting: self._update_log("提取已取消", "WARNING") else: self._update_log("提取失败", "ERROR") 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 _update_progress(self, value: int, message: str): """线程安全的进度更新""" def update(): if self.extracting: self.progress_bar['value'] = value self.status_label.config(text=message) self.after(0, update) 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)