From 13bc4520bf128938bffcf259f8220a909b0a07e5 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Thu, 5 Feb 2026 14:27:28 +0800 Subject: [PATCH] refactor: centralize configuration management with type-safe schema MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Major changes: - Add dataclass-based configuration schema with validation (config/schema.py) - Create centralized config loader and default values (config/defaults.py, config/loader.py) - Remove duplicate database_config.py, merge into unified structure - Consolidate browser settings into ERP config - Add batch_size parameter support to extractor Bug fixes: - Fix settings save error by updating config paths (browser.* → erp.*) - Fix batch_size not being applied in data extraction Co-Authored-By: Claude Sonnet 4.5 --- config/database_config.py | 9 -- config/defaults.py | 51 +++++++++++ config/loader.py | 138 ++++++++++++++++++++++++++++ config/schema.py | 143 ++++++++++++++++++++++++++++++ config/user_settings.py | 55 ++++-------- db/connection.py | 12 ++- gui/config_manager.py | 125 +++++++++++--------------- gui/data_extraction_tab.py | 5 +- gui/material_validation_tab.py | 2 +- gui/settings_tab.py | 16 ++-- utils/离散备料计划维护数据提取.py | 11 +-- 11 files changed, 431 insertions(+), 136 deletions(-) delete mode 100644 config/database_config.py create mode 100644 config/defaults.py create mode 100644 config/loader.py create mode 100644 config/schema.py diff --git a/config/database_config.py b/config/database_config.py deleted file mode 100644 index 39e370f..0000000 --- a/config/database_config.py +++ /dev/null @@ -1,9 +0,0 @@ -# ================= SQL Server 配置 ================= -SQL_SERVER_CONFIG = { - 'driver': 'ODBC Driver 18 for SQL Server', - 'server': '192.168.110.114', - 'database': 'CompanyDB', - 'username': 'peng', - 'password': 'Cqbld123456.', - 'TrustServerCertificate': 'yes' -} diff --git a/config/defaults.py b/config/defaults.py new file mode 100644 index 0000000..332d314 --- /dev/null +++ b/config/defaults.py @@ -0,0 +1,51 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +默认配置值 + +定义所有配置项的默认值。 +""" +from config.schema import ( + ERPConfig, + DatabaseConfig, + PathConfig, + ExtractionConfig, + AppConfig +) + + +# 默认配置 +DEFAULT_APP_CONFIG = AppConfig( + erp=ERPConfig( + url="https://68.11.34.30:8082/", + username="BLDpengqiangqiang", + password="Cqbld123456.", + headless=True, + ignore_https_errors=True, + auto_close_browser=True, + ), + database=DatabaseConfig( + server="192.168.110.114", + database="CompanyDB", + username="peng", + password="Cqbld123456.", + driver="ODBC Driver 18 for SQL Server", + trust_server_certificate="yes", + ), + paths=PathConfig( + data_dir="D:/python/playwrite/data/", + production_id_file="ProductionID.txt", + default_output="离散备料计划维护_合并.xlsx", + validation_output="物料状态校验结果.xlsx", + ), + extraction=ExtractionConfig( + batch_size=100, + verbose=True, + auto_convert=True, + merge_batches=True, + ) +) + + +# 兼容旧版本的字典格式 +DEFAULT_SETTINGS_DICT = DEFAULT_APP_CONFIG.to_dict() diff --git a/config/loader.py b/config/loader.py new file mode 100644 index 0000000..ad466e9 --- /dev/null +++ b/config/loader.py @@ -0,0 +1,138 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +配置加载器 + +负责加载、合并和验证配置。 +""" +import json +import os +from typing import Any, Dict +from config.schema import AppConfig +from config.defaults import DEFAULT_APP_CONFIG, DEFAULT_SETTINGS_DICT + + +class ConfigLoader: + """配置加载器""" + + @staticmethod + def load(config_file: str = "config/user_settings.json") -> AppConfig: + """ + 加载配置文件 + + Args: + config_file: 配置文件路径 + + Returns: + 应用配置对象 + """ + if os.path.exists(config_file): + try: + with open(config_file, 'r', encoding='utf-8') as f: + loaded_settings = json.load(f) + # 合并默认配置和加载的配置 + merged_settings = ConfigLoader._merge_settings(DEFAULT_SETTINGS_DICT, loaded_settings) + return ConfigLoader._dict_to_config(merged_settings) + except (json.JSONDecodeError, IOError) as e: + print(f"加载配置文件失败: {e}，使用默认配置") + return DEFAULT_APP_CONFIG + else: + # 首次运行，创建默认配置文件 + ConfigLoader.save(DEFAULT_APP_CONFIG, config_file) + return DEFAULT_APP_CONFIG + + @staticmethod + def save(config: AppConfig, config_file: str = "config/user_settings.json") -> bool: + """ + 保存配置到文件 + + Args: + config: 应用配置对象 + config_file: 配置文件路径 + + Returns: + 保存是否成功 + """ + try: + # 确保配置目录存在 + os.makedirs(os.path.dirname(config_file), exist_ok=True) + + with open(config_file, 'w', encoding='utf-8') as f: + json.dump(config.to_dict(), f, ensure_ascii=False, indent=2) + return True + except IOError as e: + print(f"保存配置文件失败: {e}") + return False + + @staticmethod + def _merge_settings(defaults: Dict, loaded: Dict) -> Dict: + """ + 合并默认配置和加载的配置 + + Args: + defaults: 默认配置 + loaded: 加载的配置 + + Returns: + 合并后的配置 + """ + result = defaults.copy() + + for key, value in loaded.items(): + if key in result and isinstance(result[key], dict) and isinstance(value, dict): + result[key] = ConfigLoader._merge_settings(result[key], value) + else: + result[key] = value + + return result + + @staticmethod + def _dict_to_config(settings: Dict) -> AppConfig: + """ + 将字典转换为配置对象 + + Args: + settings: 配置字典 + + Returns: + 应用配置对象 + """ + erp_dict = settings.get("erp", {}) + database_dict = settings.get("database", {}) + paths_dict = settings.get("paths", {}) + extraction_dict = settings.get("extraction", {}) + + return AppConfig( + erp=ERPConfig( + url=erp_dict.get("url", ""), + username=erp_dict.get("username", ""), + password=erp_dict.get("password", ""), + headless=erp_dict.get("headless", True), + ignore_https_errors=erp_dict.get("ignore_https_errors", True), + auto_close_browser=erp_dict.get("auto_close_browser", True), + ), + database=DatabaseConfig( + server=database_dict.get("server", ""), + database=database_dict.get("database", ""), + username=database_dict.get("username", ""), + password=database_dict.get("password", ""), + driver=database_dict.get("driver", "ODBC Driver 18 for SQL Server"), + trust_server_certificate=database_dict.get("trust_server_certificate", "yes"), + ), + paths=PathConfig( + data_dir=paths_dict.get("data_dir", ""), + production_id_file=paths_dict.get("production_id_file", ""), + default_output=paths_dict.get("default_output", "离散备料计划维护_合并.xlsx"), + validation_output=paths_dict.get("validation_output", "物料状态校验结果.xlsx"), + ), + extraction=ExtractionConfig( + batch_size=extraction_dict.get("batch_size", 100), + verbose=extraction_dict.get("verbose", True), + auto_convert=extraction_dict.get("auto_convert", True), + merge_batches=extraction_dict.get("merge_batches", True), + ) + ) + + +# 为了兼容旧代码，导入必要的类型 +from config.schema import ERPConfig, DatabaseConfig, PathConfig, ExtractionConfig diff --git a/config/schema.py b/config/schema.py new file mode 100644 index 0000000..9cee8a4 --- /dev/null +++ b/config/schema.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +配置结构定义 + +使用 dataclass 定义所有配置项的结构和类型。 +""" +from dataclasses import dataclass, field +from typing import Optional +from pathlib import Path + + +@dataclass +class ERPConfig: + """ERP 系统配置""" + url: str + username: str + password: str + headless: bool = True + ignore_https_errors: bool = True + auto_close_browser: bool = True + + def validate(self) -> list[str]: + """验证配置，返回错误列表""" + errors = [] + if not self.url: + errors.append("ERP URL 不能为空") + if not self.username: + errors.append("ERP 用户名不能为空") + if not self.password: + errors.append("ERP 密码不能为空") + return errors + + +@dataclass +class DatabaseConfig: + """数据库配置""" + server: str + database: str + username: str + password: str + driver: str = "ODBC Driver 18 for SQL Server" + trust_server_certificate: str = "yes" + + def validate(self) -> list[str]: + """验证配置，返回错误列表""" + errors = [] + if not self.server: + errors.append("数据库服务器地址不能为空") + if not self.database: + errors.append("数据库名称不能为空") + if not self.username: + errors.append("数据库用户名不能为空") + if not self.password: + errors.append("数据库密码不能为空") + return errors + + +@dataclass +class PathConfig: + """文件路径配置""" + data_dir: str + production_id_file: str + default_output: str = "离散备料计划维护_合并.xlsx" + validation_output: str = "物料状态校验结果.xlsx" + + def validate(self) -> list[str]: + """验证配置，返回错误列表""" + errors = [] + if not self.data_dir: + errors.append("数据目录路径不能为空") + if not self.production_id_file: + errors.append("ProductionID 文件路径不能为空") + return errors + + +@dataclass +class ExtractionConfig: + """数据提取配置""" + batch_size: int = 100 + verbose: bool = True + auto_convert: bool = True + merge_batches: bool = True + + def validate(self) -> list[str]: + """验证配置，返回错误列表""" + errors = [] + if self.batch_size <= 0: + errors.append("批次大小必须大于 0") + if self.batch_size > 1000: + errors.append("批次大小不应超过 1000") + return errors + + +@dataclass +class AppConfig: + """应用总配置""" + erp: ERPConfig + database: DatabaseConfig + paths: PathConfig + extraction: ExtractionConfig + + def validate(self) -> list[str]: + """验证所有配置，返回错误列表""" + errors = [] + errors.extend(self.erp.validate()) + errors.extend(self.database.validate()) + errors.extend(self.paths.validate()) + errors.extend(self.extraction.validate()) + return errors + + def to_dict(self) -> dict: + """转换为字典格式（用于保存到 JSON）""" + return { + "erp": { + "url": self.erp.url, + "username": self.erp.username, + "password": self.erp.password, + "headless": self.erp.headless, + "ignore_https_errors": self.erp.ignore_https_errors, + "auto_close_browser": self.erp.auto_close_browser, + }, + "database": { + "server": self.database.server, + "database": self.database.database, + "username": self.database.username, + "password": self.database.password, + "driver": self.database.driver, + "trust_server_certificate": self.database.trust_server_certificate, + }, + "paths": { + "data_dir": self.paths.data_dir, + "production_id_file": self.paths.production_id_file, + "default_output": self.paths.default_output, + "validation_output": self.paths.validation_output, + }, + "extraction": { + "batch_size": self.extraction.batch_size, + "verbose": self.extraction.verbose, + "auto_convert": self.extraction.auto_convert, + "merge_batches": self.extraction.merge_batches, + } + } diff --git a/config/user_settings.py b/config/user_settings.py index 1f973ed..5909cb5 100644 --- a/config/user_settings.py +++ b/config/user_settings.py @@ -1,40 +1,23 @@ """ -用户配置模板 +用户配置模板（已废弃，保留用于向后兼容） -此文件包含用户配置的默认值和结构说明。 -实际配置保存在 config/user_settings.json +此文件已迁移到： +- config/schema.py - 配置结构定义 +- config/defaults.py - 默认配置值 + +请使用以下方式导入： +```python +from config.defaults import DEFAULT_APP_CONFIG +from config.schema import AppConfig, ERPConfig, DatabaseConfig +``` """ -DEFAULT_SETTINGS = { - "erp": { - "url": "https://68.11.34.30:8082/", - "username": "BLDpengqiangqiang", - "password": "Cqbld123456.", - "headless": True, - "ignore_https_errors": True, - "auto_close_browser": True - }, - "database": { - "server": "192.168.110.114", - "database": "CompanyDB", - "username": "peng", - "password": "Cqbld123456." - }, - "browser": { - "headless": True, - "ignore_https_errors": True, - "auto_close": True - }, - "paths": { - "data_dir": "D:/python/playwrite/data/", - "production_id_file": "ProductionID.txt", - "default_output": "离散备料计划维护_合并.xlsx", - "validation_output": "物料状态校验结果.xlsx" - }, - "extraction": { - "batch_size": 100, - "verbose": True, - "auto_convert": True, - "merge_batches": True - } -} +# 为了向后兼容，保留旧版本导入 +from config.defaults import DEFAULT_SETTINGS_DICT as DEFAULT_SETTINGS +from config.schema import ( + AppConfig, + ERPConfig, + DatabaseConfig, + PathConfig, + ExtractionConfig, +) diff --git a/db/connection.py b/db/connection.py index 0cd5ea0..e1988d7 100644 --- a/db/connection.py +++ b/db/connection.py @@ -13,7 +13,17 @@ project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if project_root not in sys.path: sys.path.insert(0, project_root) -from config.database_config import SQL_SERVER_CONFIG +from config.defaults import DEFAULT_APP_CONFIG + +# 从默认配置获取数据库配置 +SQL_SERVER_CONFIG = { + 'driver': DEFAULT_APP_CONFIG.database.driver, + 'server': DEFAULT_APP_CONFIG.database.server, + 'database': DEFAULT_APP_CONFIG.database.database, + 'username': DEFAULT_APP_CONFIG.database.username, + 'password': DEFAULT_APP_CONFIG.database.password, + 'TrustServerCertificate': DEFAULT_APP_CONFIG.database.trust_server_certificate, +} class DatabaseConnection: diff --git a/gui/config_manager.py b/gui/config_manager.py index 0a4e862..e8ce551 100644 --- a/gui/config_manager.py +++ b/gui/config_manager.py @@ -5,11 +5,15 @@ 负责加载、保存和管理用户配置。 """ - -import json import os -from typing import Any, Dict -from config.user_settings import DEFAULT_SETTINGS +from typing import TYPE_CHECKING +from config.loader import ConfigLoader +from config.schema import AppConfig +from config.defaults import DEFAULT_SETTINGS_DICT + +# 类型提示时导入，避免循环引用 +if TYPE_CHECKING: + from config.schema import ERPConfig, DatabaseConfig, PathConfig, ExtractionConfig class ConfigManager: @@ -23,54 +27,25 @@ class ConfigManager: config_file: 配置文件路径 """ self.config_file = config_file - self.settings = self.load() + self.config: AppConfig = ConfigLoader.load(config_file) - def load(self) -> Dict[str, Any]: - """ - 加载配置文件 + # 验证配置 + errors = self.config.validate() + if errors: + print("配置验证失败：") + for error in errors: + print(f" - {error}") - Returns: - 配置字典，如果文件不存在则返回默认配置 - """ - if os.path.exists(self.config_file): - try: - with open(self.config_file, 'r', encoding='utf-8') as f: - loaded_settings = json.load(f) - # 合并默认配置，确保所有必需的键都存在 - return self._merge_settings(DEFAULT_SETTINGS, loaded_settings) - except (json.JSONDecodeError, IOError) as e: - print(f"加载配置文件失败: {e}") - return DEFAULT_SETTINGS.copy() - else: - # 首次运行，创建默认配置文件 - self.save(DEFAULT_SETTINGS.copy()) - return DEFAULT_SETTINGS.copy() - - def save(self, settings: Dict[str, Any] = None) -> bool: + def save(self) -> bool: """ 保存配置到文件 - Args: - settings: 要保存的配置字典，如果为 None 则保存当前配置 - Returns: 保存是否成功 """ - if settings is not None: - self.settings = settings + return ConfigLoader.save(self.config, self.config_file) - try: - # 确保配置目录存在 - os.makedirs(os.path.dirname(self.config_file), exist_ok=True) - - with open(self.config_file, 'w', encoding='utf-8') as f: - json.dump(self.settings, f, ensure_ascii=False, indent=2) - return True - except IOError as e: - print(f"保存配置文件失败: {e}") - return False - - def get(self, key: str, default=None) -> Any: + def get(self, key: str, default=None): """ 获取配置项 @@ -84,17 +59,16 @@ class ConfigManager: 配置值 """ keys = key.split('.') - value = self.settings + value = self.config - for k in keys: - if isinstance(value, dict) and k in value: - value = value[k] - else: - return default + try: + for k in keys: + value = getattr(value, k) + return value + except (AttributeError, TypeError): + return default - return value - - def set(self, key: str, value: Any) -> None: + def set(self, key: str, value) -> None: """ 设置配置项 @@ -105,37 +79,40 @@ class ConfigManager: value: 配置值 """ keys = key.split('.') - settings = self.settings + obj = self.config + # 导航到父对象 for k in keys[:-1]: - if k not in settings: - settings[k] = {} - settings = settings[k] + obj = getattr(obj, k) - settings[keys[-1]] = value + # 设置最终值 + setattr(obj, keys[-1], value) def reset_to_defaults(self) -> None: """重置为默认配置""" - self.settings = DEFAULT_SETTINGS.copy() + self.config = ConfigLoader.load("default") # 重新加载默认配置 self.save() - def _merge_settings(self, defaults: Dict, loaded: Dict) -> Dict: - """ - 合并默认配置和加载的配置 + @property + def erp(self): + """获取 ERP 配置""" + return self.config.erp - Args: - defaults: 默认配置 - loaded: 加载的配置 + @property + def database(self): + """获取数据库配置""" + return self.config.database - Returns: - 合并后的配置 - """ - result = defaults.copy() + @property + def paths(self): + """获取路径配置""" + return self.config.paths - for key, value in loaded.items(): - if key in result and isinstance(result[key], dict) and isinstance(value, dict): - result[key] = self._merge_settings(result[key], value) - else: - result[key] = value + @property + def extraction(self): + """获取提取配置""" + return self.config.extraction - return result + +# 为了向后兼容，保留旧版本的导入 +DEFAULT_SETTINGS = DEFAULT_SETTINGS_DICT diff --git a/gui/data_extraction_tab.py b/gui/data_extraction_tab.py index 37ea3b5..76a3f54 100644 --- a/gui/data_extraction_tab.py +++ b/gui/data_extraction_tab.py @@ -114,7 +114,7 @@ class DataExtractionTab(ttk.Frame): 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)) + 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) # 进度显示 @@ -200,7 +200,8 @@ 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.verbose_var.get(), + batch_size=self.config.get('extraction.batch_size', 100) ) # 创建实时输出流，每次写入立即更新 GUI diff --git a/gui/material_validation_tab.py b/gui/material_validation_tab.py index 0bab25f..3c6eba4 100644 --- a/gui/material_validation_tab.py +++ b/gui/material_validation_tab.py @@ -253,7 +253,7 @@ class MaterialValidationTab(ttk.Frame): validator = MaterialStatusValidator( username=self.config.get('erp.username'), password=self.config.get('erp.password'), - headless=self.config.get('browser.headless', True), + headless=self.config.get('erp.headless', True), verbose=True ) diff --git a/gui/settings_tab.py b/gui/settings_tab.py index 3eb3b58..e61daf3 100644 --- a/gui/settings_tab.py +++ b/gui/settings_tab.py @@ -198,10 +198,10 @@ class SettingsTab(ttk.Frame): self.db_username_var.set(self.config.get('database.username', '')) self.db_password_var.set(self.config.get('database.password', '')) - # 浏览器设置 - self.browser_headless_var.set(self.config.get('browser.headless', True)) - self.browser_ignore_https_var.set(self.config.get('browser.ignore_https_errors', True)) - self.browser_auto_close_var.set(self.config.get('browser.auto_close', True)) + # 浏览器设置（已合并到 ERP 配置中） + self.browser_headless_var.set(self.config.get('erp.headless', True)) + self.browser_ignore_https_var.set(self.config.get('erp.ignore_https_errors', True)) + self.browser_auto_close_var.set(self.config.get('erp.auto_close_browser', True)) # 路径设置 self.data_dir_selector.set(self.config.get('paths.data_dir', '')) @@ -226,10 +226,10 @@ class SettingsTab(ttk.Frame): self.config.set('database.username', self.db_username_var.get()) self.config.set('database.password', self.db_password_var.get()) - # 浏览器设置 - self.config.set('browser.headless', self.browser_headless_var.get()) - self.config.set('browser.ignore_https_errors', self.browser_ignore_https_var.get()) - self.config.set('browser.auto_close', self.browser_auto_close_var.get()) + # 浏览器设置（已合并到 ERP 配置中） + self.config.set('erp.headless', self.browser_headless_var.get()) + self.config.set('erp.ignore_https_errors', self.browser_ignore_https_var.get()) + self.config.set('erp.auto_close_browser', self.browser_auto_close_var.get()) # 路径设置 self.config.set('paths.data_dir', self.data_dir_selector.get()) diff --git a/utils/离散备料计划维护数据提取.py b/utils/离散备料计划维护数据提取.py index a40e2c6..4b9417d 100644 --- a/utils/离散备料计划维护数据提取.py +++ b/utils/离散备料计划维护数据提取.py @@ -14,7 +14,7 @@ from typing import Callable, Optional class DiscreteMaterialPlanExtractor: """离散备料计划维护数据提取器""" - def __init__(self, username, password, headless=False, verbose=True, progress_callback=None): + def __init__(self, username, password, headless=False, verbose=True, batch_size=100): """ 初始化提取器 @@ -23,13 +23,14 @@ class DiscreteMaterialPlanExtractor: password: 登录密码 headless: 是否无头模式运行 verbose: 是否打印详细日志 - progress_callback: 进度回调函数，接收 ProgressInfo 对象 + batch_size: 批次大小 """ self.username = username self.password = password self.headless = headless self.verbose = verbose - self.progress_callback = progress_callback + self.batch_size = batch_size + self.progress_callback = None self.converter = ExcelConverter(verbose=verbose) def _print(self, *args, **kwargs): @@ -302,9 +303,9 @@ class DiscreteMaterialPlanExtractor: # 按批次下载 downloaded_files = [] # 计算总批次数 - total_batches = sum(1 for _ in self.group_order_ids(order_ids, 100)) + 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, 100)): + 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)} 个订单号 ===") # 报告开始下载批次