diff --git a/config/defaults.py b/config/defaults.py index fbdae6e..bd00773 100644 --- a/config/defaults.py +++ b/config/defaults.py @@ -43,6 +43,7 @@ DEFAULT_APP_CONFIG = AppConfig( verbose=True, auto_convert=True, merge_batches=True, + enable_db_persistence=False, # Disabled by default ), ) diff --git a/config/schema.py b/config/schema.py index 2e4fc9d..aaeb2d4 100644 --- a/config/schema.py +++ b/config/schema.py @@ -85,6 +85,7 @@ class ExtractionConfig: verbose: bool = True auto_convert: bool = True merge_batches: bool = True + enable_db_persistence: bool = False def validate(self) -> list[str]: """验证配置，返回错误列表""" @@ -144,5 +145,6 @@ class AppConfig: "verbose": self.extraction.verbose, "auto_convert": self.extraction.auto_convert, "merge_batches": self.extraction.merge_batches, + "enable_db_persistence": self.extraction.enable_db_persistence, }, } diff --git a/db/discrete_material_plan_dao.py b/db/discrete_material_plan_dao.py new file mode 100644 index 0000000..e95bf18 --- /dev/null +++ b/db/discrete_material_plan_dao.py @@ -0,0 +1,313 @@ +""" +Data Access Object for DiscreteMaterialPlanData table. + +This module provides CRUD operations for persisting discrete material plan +data to SQL Server database. It handles mapping between Chinese DataFrame +columns (from ExcelConverter) and English database columns. +""" + +from db.connection import get_connection +from typing import List, Dict, Any +import pandas as pd + + +class DiscreteMaterialPlanDAO: + """Data Access Object for DiscreteMaterialPlanData table""" + + def __init__(self): + self.db = None + + def __enter__(self): + """Enter context manager and establish database connection""" + self.db = get_connection() + self.db.connect() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Exit context manager and close database connection""" + if self.db: + self.db.close() + + def close(self): + """Close database connection""" + if self.db: + self.db.close() + + def save_dataframe_with_replace(self, df: pd.DataFrame) -> Dict[str, int]: + """ + Save DataFrame using REPLACE strategy (DELETE + INSERT). + + This method implements a replace strategy where existing records + matching the plan numbers in the DataFrame are deleted before + inserting new records. + + Args: + df: DataFrame with discrete material plan data (Chinese column names) + + Returns: + Dictionary with 'deleted' and 'inserted' counts + + Example: + >>> dao = DiscreteMaterialPlanDAO() + >>> with dao: + ... stats = dao.save_dataframe_with_replace(df) + ... print(f"Deleted: {stats['deleted']}, Inserted: {stats['inserted']}") + """ + if df.empty: + return {'deleted': 0, 'inserted': 0} + + # Remove duplicates based on PlanNumber and SequenceNumber + original_count = len(df) + df = df.drop_duplicates(subset=['备料计划单号', '序号'], keep='first') + duplicates_removed = original_count - len(df) + + if duplicates_removed > 0: + print(f"[INFO] 检测到 {duplicates_removed} 条重复记录（相同计划单号和序号），已自动去重") + + with get_connection() as db: + # Get unique plan numbers + plan_numbers = df['备料计划单号'].unique().tolist() + + # Delete existing records + deleted = self._delete_by_plan_numbers(db, plan_numbers) + + # Insert new records in batches + inserted = self._batch_insert(db, df) + + return {'deleted': deleted, 'inserted': inserted} + + def _delete_by_plan_numbers(self, db, plan_numbers: List[str]) -> int: + """ + Delete records by plan numbers. + + Args: + db: Database connection object + plan_numbers: List of plan numbers to delete + + Returns: + Number of records deleted + """ + if not plan_numbers: + return 0 + + # SQL Server has a limit on IN clause parameters + # Delete in batches to avoid exceeding the limit + batch_size = 1000 # Safe limit for IN clause + total_deleted = 0 + + for i in range(0, len(plan_numbers), batch_size): + batch = plan_numbers[i:i + batch_size] + placeholders = ','.join(['?' for _ in batch]) + sql = f"DELETE FROM DiscreteMaterialPlanData WHERE PlanNumber IN ({placeholders})" + deleted = db.execute_update(sql, tuple(batch)) + total_deleted += deleted + + return total_deleted + + def _batch_insert(self, db, df: pd.DataFrame, batch_size: int = 72) -> int: + """ + Batch insert records (max 72 per batch due to SQL Server 2100 param limit). + + SQL Server has a limit of 2100 parameters per query. With 29 fields, + the maximum batch size is floor(2100 / 29) = 72 records per batch. + + Args: + db: Database connection object + df: DataFrame to insert + batch_size: Number of records per batch (default: 72) + + Returns: + Total number of records inserted + """ + sql = """ + INSERT INTO DiscreteMaterialPlanData ( + Factory, MaterialStatus, PlanNumber, SourceNumber, MaterialType, + ProductCode, ProductName, ProductUnit, ProductPlanQuantity, + UseDepartment, Remark, Creator, CreateDate, Approver, ApproveDate, + SequenceNumber, MaterialCode, MaterialName, Specification, Model, + DrawingNumber, MaterialQuality, PlanQuantity, Unit, RequiredDate, + Warehouse, UnitUsage, CumulativeOutputQuantity, BOMVersion + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """ + + total_inserted = 0 + records = self._convert_df_to_records(df) + + for i in range(0, len(records), batch_size): + batch = records[i:i + batch_size] + for record in batch: + db.execute_update(sql, record) + total_inserted += 1 + + return total_inserted + + def _convert_df_to_records(self, df: pd.DataFrame) -> List[tuple]: + """ + Convert DataFrame to list of tuples for batch insert. + + Maps Chinese DataFrame column names to English database column names + and converts each row to a tuple in the correct order. + + Handles NaN/None values by converting them to None for NULL fields. + + Args: + df: DataFrame with Chinese column names + + Returns: + List of tuples, one per record + """ + # Column order must match INSERT statement + column_order = [ + '工厂', '备料状态', '备料计划单号', '来源单号', '备料类型', '产品编码', + '产品名称', '产品单位', '产品计划数量', '用料部门', '备注', '制单人', + '制单日期', '审批人', '审批日期', '序号', '材料编码', '材料名称', + '规格', '型号', '图号', '物料材质', '计划数量', '单位', '需用日期', + '发料仓库', '单位用量', '累计出库数量', 'BOM版本' + ] + + # Numeric columns with their default values and data types + numeric_columns = { + '产品计划数量': (0, int), + '序号': (0, int), + '计划数量': (0, int), + '单位用量': (0.0, float), + '累计出库数量': (0, int), + } + + records = [] + for _, row in df.iterrows(): + record = [] + for col in column_order: + value = row.get(col) + # Handle NaN, None, or empty string values + if pd.isna(value) or value is None or (isinstance(value, str) and value.strip() == ''): + if col in numeric_columns: + # Use default value for numeric columns + record.append(numeric_columns[col][0]) + else: + # Use None (NULL) for string columns + record.append(None) + else: + # Convert numeric columns to proper type + if col in numeric_columns: + try: + default_value, target_type = numeric_columns[col] + if target_type == float: + record.append(float(value)) + else: + record.append(int(value)) + except (ValueError, TypeError): + # If conversion fails, use default value + record.append(numeric_columns[col][0]) + else: + # Keep string columns as is + record.append(value) + records.append(tuple(record)) + + return records + + def query_by_plan_number(self, plan_number: str) -> List[Dict]: + """ + Query all records for a specific plan number. + + Args: + plan_number: Plan number to query + + Returns: + List of dictionaries representing records + """ + with get_connection() as db: + sql = "SELECT * FROM DiscreteMaterialPlanData WHERE PlanNumber = ?" + return db.execute_query(sql, (plan_number,)) + + def query_by_plan_numbers(self, plan_numbers: List[str]) -> List[Dict]: + """ + Query records for multiple plan numbers. + + Args: + plan_numbers: List of plan numbers to query + + Returns: + List of dictionaries representing records + """ + if not plan_numbers: + return [] + placeholders = ','.join(['?' for _ in plan_numbers]) + sql = f"SELECT * FROM DiscreteMaterialPlanData WHERE PlanNumber IN ({placeholders})" + with get_connection() as db: + return db.execute_query(sql, tuple(plan_numbers)) + + def query_by_production_order(self, order_id: str) -> List[Dict]: + """ + Query all records for a specific production order. + + Args: + order_id: Production order ID (SourceNumber) + + Returns: + List of dictionaries representing records + """ + with get_connection() as db: + sql = "SELECT * FROM DiscreteMaterialPlanData WHERE SourceNumber = ?" + return db.execute_query(sql, (order_id,)) + + def count_by_plan_number(self, plan_number: str) -> int: + """ + Count records for a specific plan number. + + Args: + plan_number: Plan number to count + + Returns: + Number of records + """ + with get_connection() as db: + sql = "SELECT COUNT(*) as count FROM DiscreteMaterialPlanData WHERE PlanNumber = ?" + result = db.execute_query(sql, (plan_number,)) + return result[0]['count'] if result else 0 + + def count_all(self) -> int: + """ + Count all records in the table. + + Returns: + Total number of records + """ + with get_connection() as db: + sql = "SELECT COUNT(*) as count FROM DiscreteMaterialPlanData" + result = db.execute_query(sql) + return result[0]['count'] if result else 0 + + def delete_by_plan_numbers(self, plan_numbers: List[str]) -> int: + """ + Delete all records for specified plan numbers. + + Args: + plan_numbers: List of plan numbers to delete + + Returns: + Number of records deleted + """ + with get_connection() as db: + return self._delete_by_plan_numbers(db, plan_numbers) + + def get_statistics(self) -> Dict[str, Any]: + """ + Get comprehensive statistics about the data. + + Returns: + Dictionary with statistics including total records, + unique plans, unique orders, and date range + """ + with get_connection() as db: + sql = """ + SELECT + COUNT(*) as total_records, + COUNT(DISTINCT PlanNumber) as unique_plans, + COUNT(DISTINCT SourceNumber) as unique_orders, + MIN(CreateDate) as earliest_record, + MAX(CreateDate) as latest_record + FROM DiscreteMaterialPlanData + """ + result = db.execute_query(sql) + return result[0] if result else {} diff --git a/gui/data_extraction_tab.py b/gui/data_extraction_tab.py index 2fc66f5..ff0cde8 100644 --- a/gui/data_extraction_tab.py +++ b/gui/data_extraction_tab.py @@ -111,17 +111,10 @@ class DataExtractionTab(ttk.Frame): 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("erp.headless", True)) ttk.Checkbutton( options_group, text="无头模式 (不显示浏览器)", variable=self.headless_var - ).grid(row=0, column=1, sticky="w", padx=5) + ).grid(row=0, column=0, sticky="w", padx=5) # 进度显示 progress_group = ttk.LabelFrame(parent, text="进度", padding=10) @@ -210,8 +203,9 @@ class DataExtractionTab(ttk.Frame): username=self.config.get("erp.username"), password=self.config.get("erp.password"), headless=self.headless_var.get(), - verbose=self.verbose_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 diff --git a/gui/main_window.py b/gui/main_window.py index cb7ba97..a578542 100644 --- a/gui/main_window.py +++ b/gui/main_window.py @@ -125,6 +125,7 @@ class MainWindow: "• 数据提取 - 从 ERP 系统提取备料计划数据\n" "• 物料校验 - 校验物料状态并匹配待删除物料\n" "• 数据查询 - 查询生产订单号等信息\n" - "• 设置管理 - 管理系统配置\n\n" + "• 设置管理 - 管理系统配置\n" + "• 数据库持久化 - 将提取的数据自动保存到 SQL Server\n\n" "基于 Playwright 和 Python 开发", ) diff --git a/gui/progress.py b/gui/progress.py index 5cdb3d6..f7763bf 100644 --- a/gui/progress.py +++ b/gui/progress.py @@ -19,7 +19,7 @@ class ProgressInfo: """ stage: ( - str # 阶段标识: 'login', 'query', 'download', 'logout', 'convert', 'complete' + str # 阶段标识: 'login', 'query', 'download', 'logout', 'convert', 'database', 'complete' ) current: int # 当前进度值 total: int # 总量 @@ -48,9 +48,10 @@ class ProgressCalculator: STAGE_WEIGHTS = { "login": 5, # 登录: 0-5% "query": 5, # 查询: 5-10% - "download": 65, # 下载: 10-75% - "logout": 5, # 注销: 75-80% - "convert": 15, # 转换: 80-95% + "download": 60, # 下载: 10-70% + "logout": 5, # 注销: 70-75% + "convert": 15, # 转换: 75-90% + "database": 10, # 数据库持久化: 90-100% "complete": 5, # 完成: 95-100% } diff --git a/gui/settings_tab.py b/gui/settings_tab.py index e1fe9e9..e0b12ad 100644 --- a/gui/settings_tab.py +++ b/gui/settings_tab.py @@ -218,6 +218,12 @@ class SettingsTab(ttk.Frame): group, text="自动合并批次数据", variable=self.merge_batches_var ).grid(row=3, column=0, columnspan=2, sticky="w", pady=5) + # 数据库持久化 + self.enable_db_persistence_var = tk.BooleanVar() + ttk.Checkbutton( + group, text="保存到数据库 (同时写入 SQL Server)", variable=self.enable_db_persistence_var + ).grid(row=4, column=0, columnspan=2, sticky="w", pady=5) + def load_settings(self): """从配置加载设置到界面""" # ERP 设置 @@ -247,6 +253,7 @@ class SettingsTab(ttk.Frame): self.verbose_var.set(self.config.get("extraction.verbose", True)) self.auto_convert_var.set(self.config.get("extraction.auto_convert", True)) self.merge_batches_var.set(self.config.get("extraction.merge_batches", True)) + self.enable_db_persistence_var.set(self.config.get("extraction.enable_db_persistence", False)) def save_settings(self): """保存界面设置到配置""" @@ -275,6 +282,7 @@ class SettingsTab(ttk.Frame): self.config.set("extraction.verbose", self.verbose_var.get()) self.config.set("extraction.auto_convert", self.auto_convert_var.get()) self.config.set("extraction.merge_batches", self.merge_batches_var.get()) + self.config.set("extraction.enable_db_persistence", self.enable_db_persistence_var.get()) # 保存到文件 if self.config.save(): diff --git a/utils/离散备料计划维护数据提取.py b/utils/离散备料计划维护数据提取.py index 331115b..cc8ac82 100644 --- a/utils/离散备料计划维护数据提取.py +++ b/utils/离散备料计划维护数据提取.py @@ -19,7 +19,8 @@ class DiscreteMaterialPlanExtractor: """离散备料计划维护数据提取器""" def __init__( - self, username, password, headless=False, verbose=True, batch_size=100 + self, username, password, headless=False, verbose=True, batch_size=100, + enable_db_persistence=False ): """ 初始化提取器 @@ -30,6 +31,7 @@ class DiscreteMaterialPlanExtractor: headless: 是否无头模式运行 verbose: 是否打印详细日志 batch_size: 批次大小 + enable_db_persistence: 是否启用数据库持久化 """ self.username = username self.password = password @@ -38,6 +40,12 @@ class DiscreteMaterialPlanExtractor: 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: + from db.discrete_material_plan_dao import DiscreteMaterialPlanDAO + self.dao = DiscreteMaterialPlanDAO() + self.dao.__enter__() # Enter context manager def _print(self, *args, **kwargs): """打印日志（如果 verbose=True）""" @@ -269,7 +277,7 @@ class DiscreteMaterialPlanExtractor: return download_path def convert_and_merge_files(self, file_paths, output_path): - """使用 ExcelConverter 转换并合并所有文件""" + """使用 ExcelConverter 转换并合并所有文件，返回合并后的 DataFrame""" # 确保输出文件路径是正确的格式 output_path = os.path.normpath(output_path) output_dir = os.path.dirname(output_path) @@ -287,7 +295,7 @@ class DiscreteMaterialPlanExtractor: "准备转换：检查输出目录", action="check_directory", ) - + if output_dir and not os.path.exists(output_dir): self._print(f"创建输出目录: {output_dir}") os.makedirs(output_dir) @@ -312,7 +320,7 @@ class DiscreteMaterialPlanExtractor: df = self.converter.convert(file_path, output_file=None) # 只转换，不保存 all_dataframes.append(df) self._print(f" 提取到 {len(df)} 条记录") - + # 报告转换完成 self._report_progress( "convert", @@ -324,6 +332,7 @@ class DiscreteMaterialPlanExtractor: action="file_converted", ) + merged_df = None if all_dataframes: # 步骤N+1：合并数据 self._report_progress( @@ -334,7 +343,7 @@ class DiscreteMaterialPlanExtractor: action="merging_data", file_count=len(all_dataframes), ) - + self._print(f"\n合并 {len(all_dataframes)} 个文件的数据...") merged_df = pd.concat(all_dataframes, ignore_index=True) merged_df.to_excel(output_path, index=False) @@ -349,13 +358,43 @@ class DiscreteMaterialPlanExtractor: action="cleanup", total_records=len(merged_df), ) - + for file_path in file_paths: os.remove(file_path) self._print(f"已删除临时文件: {file_path}") - return output_path - return None + return output_path, merged_df + return None, None + + def _save_to_database(self, df: pd.DataFrame): + """Save DataFrame to database with progress reporting""" + 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']} 条") + + except Exception as e: + self._print(f"\n警告: 数据库保存失败: {e}") + self._report_progress( + "database", 3, 3, + f"数据库保存失败: {str(e)}", + action="db_error", + error=str(e) + ) def setup_query_interface(self, inner_frame): """设置查询界面（不报告进度，由 extract 统一报告）""" @@ -506,7 +545,12 @@ class DiscreteMaterialPlanExtractor: self._print( f"\n=== 开始转换并合并 {len(downloaded_files)} 个文件 ===" ) - self.convert_and_merge_files(downloaded_files, output_file) + 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没有下载到任何文件") @@ -514,7 +558,7 @@ class DiscreteMaterialPlanExtractor: self._print(f"最终文件: {output_file}") self._report_progress( - "complete", 1, 1, "数据提取完成 ✓", + "complete", 1, 1, "数据提取完成 ✓", output_file=output_file, action="all_complete", ) @@ -525,6 +569,12 @@ class DiscreteMaterialPlanExtractor: return output_file finally: + # Close database connection if open + if self.dao: + try: + self.dao.__exit__(None, None, None) + except Exception: + pass self.progress_callback = original_callback