60 lines
2.2 KiB
Python
60 lines
2.2 KiB
Python
"""系统配置加载 + Profile 覆盖。
|
|
|
|
读取 configs/system.yaml，如果设置了 EN_NEWS_PROFILE 环境变量，
|
|
则加载 configs/profiles/{name}.yaml 并深度合并覆盖。
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SYSTEM_CONFIG_PATH = Path("configs/system.yaml")
|
|
PROFILES_DIR = Path("configs/profiles")
|
|
|
|
|
|
def _deep_merge(base: dict, override: dict) -> dict:
|
|
"""深度合并两个字典，override 的值覆盖 base。"""
|
|
result = base.copy()
|
|
for key, value in override.items():
|
|
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
|
|
result[key] = _deep_merge(result[key], value)
|
|
else:
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def load_system_config() -> dict[str, Any]:
|
|
"""加载系统配置，自动应用 EN_NEWS_PROFILE 覆盖。
|
|
|
|
Returns:
|
|
合并后的完整配置字典（已应用 profile 覆盖）
|
|
"""
|
|
if not SYSTEM_CONFIG_PATH.exists():
|
|
raise FileNotFoundError(f"系统配置文件不存在: {SYSTEM_CONFIG_PATH}")
|
|
|
|
with open(SYSTEM_CONFIG_PATH, encoding="utf-8") as f:
|
|
config: dict[str, Any] = yaml.safe_load(f) or {}
|
|
|
|
# ── Profile 覆盖 ──────────────────────────────────
|
|
profile_name = os.environ.get("EN_NEWS_PROFILE", "")
|
|
if profile_name:
|
|
profile_path = PROFILES_DIR / f"{profile_name}.yaml"
|
|
if profile_path.exists():
|
|
logger.info("📋 加载 Profile: %s (%s)", profile_name, profile_path)
|
|
with open(profile_path, encoding="utf-8") as f:
|
|
profile_cfg = yaml.safe_load(f) or {}
|
|
config = _deep_merge(config, profile_cfg)
|
|
logger.info("📋 Profile 覆盖已应用: proxy=%s, headful=%s, max_memory=%s",
|
|
config.get("proxy", {}).get("enabled"),
|
|
config.get("crawler", {}).get("headful"),
|
|
config.get("crawler", {}).get("max_memory_mb"))
|
|
else:
|
|
logger.warning("⚠️ Profile 文件不存在: %s", profile_path)
|
|
|
|
return config
|