admin/playwrite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: centralize configuration management with type-safe schema

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
2026-02-05 14:27:28 +08:00
parent 885d87dff8
commit 13bc4520bf
11 changed files with 431 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
config/database_config.py
View File

@@ -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'
}

51
config/defaults.py Normal file
View File

@@ -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()

138
config/loader.py Normal file
View File

@@ -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

143
config/schema.py Normal file
View File

@@ -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,
}
}

55
config/user_settings.py
View File

@@ -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,
)

12
db/connection.py
View File

@@ -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:

121
gui/config_manager.py
View File

@@ -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
try:
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
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

5
gui/data_extraction_tab.py
View File

@@ -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

2
gui/material_validation_tab.py
View File

@@ -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
)

16
gui/settings_tab.py
View File

@@ -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())

11
utils/离散备料计划维护数据提取.py
View File

@@ -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)} 个订单号 ===")
# 报告开始下载批次