"""全局配置:仅从 JSON 文件加载,不使用环境变量。""" from __future__ import annotations import json import logging from typing import Any, Dict, List from pydantic import BaseModel, ConfigDict from app.core.config_paths import project_root, resolved_config_path logger = logging.getLogger("ProxyPool") _DEFAULTS: Dict[str, Any] = { "db_path": "db/proxies.sqlite", "host": "127.0.0.1", "port": 18080, "validator_timeout": 5, "validator_max_concurrency": 200, "validator_connect_timeout": 3, "crawler_num_validators": 50, "crawler_max_queue_size": 48, "log_level": "INFO", "log_dir": "logs", "ws_stats_interval_seconds": 1, "export_max_records": 10000, "score_valid": 10, "score_invalid": -5, "score_min": 0, "score_max": 100, "score_latency_ref_ms": 500.0, "score_use_penalty_per_pick": 2.5, "score_max_use_penalty": 70.0, "score_default_latency_ms": 1500.0, "validator_test_urls": [ "http://httpbin.org/ip", "https://httpbin.org/ip", "http://api.ipify.org", "https://api.ipify.org", "http://www.baidu.com", "http://www.qq.com", ], "plugins_dir": "plugins", "cors_origins": [ "http://localhost:8080", "http://localhost:5173", "http://127.0.0.1:18081", "http://localhost:18081", ], "run_network_tests": False, } def _load_merged_dict() -> Dict[str, Any]: data = dict(_DEFAULTS) path = resolved_config_path() if not path.is_file(): logger.warning("配置文件不存在,使用内置默认项: %s", path) return data try: with path.open(encoding="utf-8") as f: file_data = json.load(f) if not isinstance(file_data, dict): logger.error("配置文件须为 JSON 对象,已忽略: %s", path) return data data.update(file_data) except (json.JSONDecodeError, OSError) as e: logger.error("读取配置文件失败,使用内置默认项: %s (%s)", path, e) return data class AppSettings(BaseModel): """应用配置(与 config/app.json 字段一致)""" model_config = ConfigDict(extra="ignore") db_path: str host: str port: int validator_timeout: int validator_max_concurrency: int validator_connect_timeout: int crawler_num_validators: int crawler_max_queue_size: int log_level: str log_dir: str ws_stats_interval_seconds: int export_max_records: int score_valid: int score_invalid: int score_min: int score_max: int score_latency_ref_ms: float score_use_penalty_per_pick: float score_max_use_penalty: float score_default_latency_ms: float validator_test_urls: List[str] plugins_dir: str cors_origins: List[str] run_network_tests: bool = False @property def base_dir(self) -> str: return str(project_root()) # 全局单例(进程内首次导入时按当前 resolved_config_path() 加载) settings = AppSettings.model_validate(_load_merged_dict()) # 历史代码别名 Settings = AppSettings