#!/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