"""配置加载模块 -- 从 config.yaml / 环境变量读取配置""" from __future__ import annotations import os from pathlib import Path from typing import Any import yaml from pydantic import BaseModel, Field _CONFIG_PATH = os.getenv("CONFIG_PATH", "config.yaml") class ServerConfig(BaseModel): host: str = "0.0.0.0" port: int = 8080 proxy_api_key: str = "sk-plan-manage-change-me" class DatabaseConfig(BaseModel): path: str = "./data/plan_manage.db" class StorageConfig(BaseModel): path: str = "./data/files" class SchedulerConfig(BaseModel): """调度器配置""" task_processing_limit: int = 5 # 每次处理的最大任务数 loop_interval_seconds: int = 30 # 主循环间隔(秒) class QuotaRuleSeed(BaseModel): """config.yaml 中单条 QuotaRule 种子""" rule_name: str quota_total: int quota_unit: str = "requests" refresh_type: str = "calendar_cycle" interval_hours: float | None = None calendar_unit: str | None = None calendar_anchor: dict[str, Any] | None = None class PlanSeed(BaseModel): """config.yaml 中单个 Plan 种子配置""" name: str provider: str api_key: str = "" api_base: str = "" plan_type: str = "coding" supported_models: list[str] = Field(default_factory=list) extra_headers: dict[str, str] = Field(default_factory=dict) extra_config: dict[str, Any] = Field(default_factory=dict) quota_rules: list[QuotaRuleSeed] = Field(default_factory=list) class AppConfig(BaseModel): server: ServerConfig = Field(default_factory=ServerConfig) database: DatabaseConfig = Field(default_factory=DatabaseConfig) storage: StorageConfig = Field(default_factory=StorageConfig) scheduler: SchedulerConfig = Field(default_factory=SchedulerConfig) plans: list[PlanSeed] = Field(default_factory=list) def load_config(path: str | None = None) -> AppConfig: """加载配置文件,不存在则返回默认值""" cfg_path = Path(path or _CONFIG_PATH) if cfg_path.exists(): with open(cfg_path, "r", encoding="utf-8") as f: raw = yaml.safe_load(f) or {} return AppConfig(**raw) return AppConfig() # 全局单例 settings: AppConfig = load_config()