aaa46ef282
This commit implements a complete migration from JSON-based configuration to .env environment variables, providing better security and flexibility. Key Changes: - Add python-dotenv dependency for environment variable support - Create config/env_loader.py with type conversion utilities - Add from_env() class methods to all config dataclasses - Update ConfigLoader to prioritize environment variables - Add save_to_env() method for .env file management - Implement database connection factory pattern - Add base DAO and connection classes for better abstraction - Support both SQL Server and MySQL with unified interface - Create migration script (scripts/migrate_to_env.py) - Update GUI to read/write .env files - Add comprehensive migration documentation New Files: - config/env_loader.py - Environment variable loader - db/base_connection.py - Base database connection interface - db/base_dao.py - Base DAO with common utilities - db/connection_factory.py - Factory for creating connections - db/mysql_connection.py - MySQL-specific connection - db/sqlserver_connection.py - SQL Server-specific connection - db/table_name_converter.py - SQL dialect converter - scripts/migrate_to_env.py - Configuration migration tool - docs/ENV_MIGRATION.md - Complete migration guide - .env.example - Environment variable template Testing: - Verified MySQL connection (8.0.44) - Tested all DAO operations - Confirmed 150 tables accessible - Validated configuration loading Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
352 lines
12 KiB
Python
352 lines
12 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
配置结构定义
|
||
|
||
使用 dataclass 定义所有配置项的结构和类型。
|
||
"""
|
||
from dataclasses import dataclass, field
|
||
from typing import Optional
|
||
from pathlib import Path
|
||
from enum import Enum
|
||
|
||
|
||
class DatabaseType(str, Enum):
|
||
"""数据库类型枚举"""
|
||
SQLSERVER = "sqlserver"
|
||
MYSQL = "mysql"
|
||
|
||
|
||
@dataclass
|
||
class ERPConfig:
|
||
"""ERP 系统配置"""
|
||
|
||
url: str
|
||
username: str
|
||
password: str
|
||
headless: bool = True
|
||
ignore_https_errors: bool = True
|
||
auto_close_browser: bool = True
|
||
|
||
@classmethod
|
||
def from_env(cls) -> "ERPConfig":
|
||
"""从环境变量创建配置"""
|
||
from config.env_loader import get_env, get_env_bool
|
||
|
||
return cls(
|
||
url=get_env("ERP_URL", "https://68.11.34.30:8082/"),
|
||
username=get_env("ERP_USERNAME", "BLDpengqiangqiang"),
|
||
password=get_env("ERP_PASSWORD", ""),
|
||
headless=get_env_bool("ERP_HEADLESS", True),
|
||
ignore_https_errors=get_env_bool("ERP_IGNORE_HTTPS_ERRORS", True),
|
||
auto_close_browser=get_env_bool("ERP_AUTO_CLOSE_BROWSER", 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 SQLServerConfig:
|
||
"""SQL Server 特定配置"""
|
||
driver: str = "ODBC Driver 18 for SQL Server"
|
||
trust_server_certificate: str = "yes"
|
||
|
||
@classmethod
|
||
def from_env(cls) -> "SQLServerConfig":
|
||
"""从环境变量创建配置"""
|
||
from config.env_loader import get_env
|
||
|
||
return cls(
|
||
driver=get_env("DB_SQLSERVER_DRIVER", "ODBC Driver 18 for SQL Server"),
|
||
trust_server_certificate=get_env("DB_TRUST_SERVER_CERTIFICATE", "yes"),
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class MySQLConfig:
|
||
"""MySQL 特定配置"""
|
||
host: str = ""
|
||
port: int = 3306
|
||
charset: str = "utf8mb4"
|
||
|
||
@classmethod
|
||
def from_env(cls) -> "MySQLConfig":
|
||
"""从环境变量创建配置"""
|
||
from config.env_loader import get_env, get_env_int
|
||
|
||
return cls(
|
||
host=get_env("DB_MYSQL_HOST", "192.168.31.83"),
|
||
port=get_env_int("DB_MYSQL_PORT", 3306),
|
||
charset=get_env("DB_MYSQL_CHARSET", "utf8mb4"),
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class DatabaseConfig:
|
||
"""数据库配置"""
|
||
|
||
db_type: DatabaseType = DatabaseType.SQLSERVER
|
||
server: str = "" # SQL Server 服务器地址
|
||
database: str = ""
|
||
username: str = ""
|
||
password: str = ""
|
||
sqlserver: Optional[SQLServerConfig] = None
|
||
mysql: Optional[MySQLConfig] = None
|
||
|
||
@classmethod
|
||
def from_env(cls) -> "DatabaseConfig":
|
||
"""从环境变量创建配置"""
|
||
from config.env_loader import get_env, get_env_int
|
||
|
||
db_type_str = get_env("DB_TYPE", "sqlserver")
|
||
try:
|
||
db_type = DatabaseType(db_type_str)
|
||
except ValueError:
|
||
db_type = DatabaseType.SQLSERVER
|
||
|
||
return cls(
|
||
db_type=db_type,
|
||
server=get_env("DB_SERVER", "192.168.110.114"),
|
||
database=get_env("DB_NAME", "CompanyDB"),
|
||
username=get_env("DB_USERNAME", "peng"),
|
||
password=get_env("DB_PASSWORD", ""),
|
||
sqlserver=SQLServerConfig.from_env(),
|
||
mysql=MySQLConfig.from_env(),
|
||
)
|
||
|
||
def validate(self) -> list[str]:
|
||
"""验证配置，返回错误列表"""
|
||
errors = []
|
||
|
||
if self.db_type == DatabaseType.SQLSERVER:
|
||
if not self.server:
|
||
errors.append("SQL Server 服务器地址不能为空")
|
||
if not self.database:
|
||
errors.append("数据库名称不能为空")
|
||
if not self.username:
|
||
errors.append("数据库用户名不能为空")
|
||
if not self.password:
|
||
errors.append("数据库密码不能为空")
|
||
elif self.db_type == DatabaseType.MYSQL:
|
||
if self.mysql and not self.mysql.host:
|
||
errors.append("MySQL 主机地址不能为空")
|
||
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"
|
||
|
||
@classmethod
|
||
def from_env(cls) -> "PathConfig":
|
||
"""从环境变量创建配置"""
|
||
from config.env_loader import get_env
|
||
|
||
return cls(
|
||
data_dir=get_env("PATH_DATA_DIR", "D:/python/playwrite/data/"),
|
||
production_id_file=get_env("PATH_PRODUCTION_ID_FILE", "ProductionID.txt"),
|
||
default_output=get_env("PATH_DEFAULT_OUTPUT", "离散备料计划维护_合并.xlsx"),
|
||
validation_output=get_env("PATH_VALIDATION_OUTPUT", "物料状态校验结果.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
|
||
enable_db_persistence: bool = False
|
||
|
||
@classmethod
|
||
def from_env(cls) -> "ExtractionConfig":
|
||
"""从环境变量创建配置"""
|
||
from config.env_loader import get_env_int, get_env_bool
|
||
|
||
return cls(
|
||
batch_size=get_env_int("EXTRACTION_BATCH_SIZE", 100),
|
||
verbose=get_env_bool("EXTRACTION_VERBOSE", True),
|
||
auto_convert=get_env_bool("EXTRACTION_AUTO_CONVERT", True),
|
||
merge_batches=get_env_bool("EXTRACTION_MERGE_BATCHES", True),
|
||
enable_db_persistence=get_env_bool("EXTRACTION_ENABLE_DB_PERSISTENCE", False),
|
||
)
|
||
|
||
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 ValidationConfig:
|
||
"""物料校验配置"""
|
||
|
||
data_source: str = "database_full"
|
||
use_database: bool = True
|
||
batch_size: int = 2000
|
||
enable_crud_operations: bool = False
|
||
default_manager: str = ""
|
||
match_mode: str = "substring"
|
||
|
||
@classmethod
|
||
def from_env(cls) -> "ValidationConfig":
|
||
"""从环境变量创建配置"""
|
||
from config.env_loader import get_env, get_env_int, get_env_bool
|
||
|
||
return cls(
|
||
data_source=get_env("VALIDATION_DATA_SOURCE", "database_full"),
|
||
use_database=get_env_bool("VALIDATION_USE_DATABASE", True),
|
||
batch_size=get_env_int("VALIDATION_BATCH_SIZE", 2000),
|
||
enable_crud_operations=get_env_bool("VALIDATION_ENABLE_CRUD", False),
|
||
default_manager=get_env("VALIDATION_DEFAULT_MANAGER", ""),
|
||
match_mode=get_env("VALIDATION_MATCH_MODE", "substring"),
|
||
)
|
||
|
||
def validate(self) -> list[str]:
|
||
"""验证配置，返回错误列表"""
|
||
errors = []
|
||
|
||
valid_sources = [
|
||
"database_full",
|
||
"database_filtered",
|
||
"excel_existing",
|
||
"excel_full"
|
||
]
|
||
if self.data_source not in valid_sources:
|
||
errors.append(
|
||
f"无效的数据源: {self.data_source}。"
|
||
f"有效选项: {', '.join(valid_sources)}"
|
||
)
|
||
|
||
if self.batch_size <= 0:
|
||
errors.append("批次大小必须大于 0")
|
||
if self.batch_size > 2000:
|
||
errors.append("批次大小不应超过 2000（SQL Server 参数限制）")
|
||
|
||
valid_match_modes = ["substring", "exact"]
|
||
if self.match_mode not in valid_match_modes:
|
||
errors.append(
|
||
f"无效的匹配模式: {self.match_mode}。"
|
||
f"有效选项: {', '.join(valid_match_modes)}"
|
||
)
|
||
|
||
return errors
|
||
|
||
|
||
@dataclass
|
||
class AppConfig:
|
||
"""应用总配置"""
|
||
|
||
erp: ERPConfig
|
||
database: DatabaseConfig
|
||
paths: PathConfig
|
||
extraction: ExtractionConfig
|
||
validation: ValidationConfig
|
||
|
||
@classmethod
|
||
def from_env(cls) -> "AppConfig":
|
||
"""从环境变量创建配置"""
|
||
return cls(
|
||
erp=ERPConfig.from_env(),
|
||
database=DatabaseConfig.from_env(),
|
||
paths=PathConfig.from_env(),
|
||
extraction=ExtractionConfig.from_env(),
|
||
validation=ValidationConfig.from_env(),
|
||
)
|
||
|
||
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())
|
||
errors.extend(self.validation.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": {
|
||
"db_type": self.database.db_type.value,
|
||
"server": self.database.server,
|
||
"database": self.database.database,
|
||
"username": self.database.username,
|
||
"password": self.database.password,
|
||
"sqlserver": {
|
||
"driver": self.database.sqlserver.driver if self.database.sqlserver else "ODBC Driver 18 for SQL Server",
|
||
"trust_server_certificate": self.database.sqlserver.trust_server_certificate if self.database.sqlserver else "yes",
|
||
},
|
||
"mysql": {
|
||
"host": self.database.mysql.host if self.database.mysql else "",
|
||
"port": self.database.mysql.port if self.database.mysql else 3306,
|
||
"charset": self.database.mysql.charset if self.database.mysql else "utf8mb4",
|
||
},
|
||
},
|
||
"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,
|
||
"enable_db_persistence": self.extraction.enable_db_persistence,
|
||
},
|
||
"validation": {
|
||
"data_source": self.validation.data_source,
|
||
"use_database": self.validation.use_database,
|
||
"batch_size": self.validation.batch_size,
|
||
"enable_crud_operations": self.validation.enable_crud_operations,
|
||
"default_manager": self.validation.default_manager,
|
||
"match_mode": self.validation.match_mode,
|
||
},
|
||
}
|