"""配置加载:从 .env 文件或环境变量读取,风格对齐 ContentFindAgentNew。 只读环境变量,不硬编码密钥。PG 是 M1 唯一必填项;其余环节配置随里程碑推进逐步使用。 """ from __future__ import annotations import os from dataclasses import dataclass from pathlib import Path # 我们自己的过程库 schema(与 OPEN_AIGC_PG_SCHEMA=public 区分开) DEFAULT_PG_SCHEMA = "creation_knowledge" def load_env_file(env_path: str | Path = ".env") -> dict[str, str]: """解析 .env 为 dict;文件不存在则返回空。与 crawapi_http._load_env_file 同款。""" path = Path(env_path) if not path.exists(): return {} env: dict[str, str] = {} for line in path.read_text(encoding="utf-8").splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#") or "=" not in stripped: continue key, value = stripped.split("=", 1) env[key.strip()] = value.strip().strip('"').strip("'") return env def env_value( key: str, file_env: dict[str, str], default: str | None = None, required: bool = False, ) -> str: """取值优先级:os.environ > .env 文件 > default。required 缺失时报错。""" value = os.getenv(key) or file_env.get(key) or default if required and not value: raise RuntimeError(f"missing required env: {key}") return value or "" @dataclass class PgConfig: """过程库(Greenplum open_aigc,schema creation_knowledge)连接配置。""" host: str port: int user: str password: str database: str schema: str = DEFAULT_PG_SCHEMA timeout: int = 10 @classmethod def from_env(cls, env_file: str | Path = ".env") -> "PgConfig": file_env = load_env_file(env_file) return cls( host=env_value("OPEN_AIGC_PG_HOST", file_env, required=True), port=int(env_value("OPEN_AIGC_PG_PORT", file_env, "5432")), user=env_value("OPEN_AIGC_PG_USER", file_env, required=True), password=env_value("OPEN_AIGC_PG_PASSWORD", file_env, required=True), database=env_value( "OPEN_AIGC_PG_DB_NAME", file_env, "open_aigc" ), schema=env_value("CK_PG_SCHEMA", file_env, DEFAULT_PG_SCHEMA), ) @dataclass class Settings: """全流程配置聚合。M1 只用到 pg;其余字段供 M2+ 使用。""" pg: PgConfig # 爬虫(M2) crawler_base_url: str crawler_key: str crawler_timeout: int # 多模态提取(M3) video_model: str gemini_api_key: str openrouter_base_url: str openrouter_api_key: str # 文本步 LLM(多模态读取的辅助文本调用) llm_model: str # 卡片 / 抽帧 max_cards: int frames_dir: str douyin_ratio: str # 视频下载偏好码率(控成本/体积),如 540p data_dir: str # 媒体本地落盘根目录;空字符串=关闭媒体保存(解耦开关) # query 搜索 + OSS 转存(默认值必填:tests/test_video_extract.py 用关键字构造 Settings 且不传这些) oss_upload_url: str = "http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream" oss_upload_timeout_seconds: int = 60 search_default_limit: int = 5 search_content_type: str = "图文" search_sort_type: str = "综合" @classmethod def from_env(cls, env_file: str | Path = ".env") -> "Settings": file_env = load_env_file(env_file) return cls( pg=PgConfig.from_env(env_file), crawler_base_url=env_value( "CONTENTFIND_API_CRAWAPI_BASE_URL", file_env, "http://crawler.aiddit.com", ), crawler_key=env_value("CONTENTFIND_API_CRAWAPI_KEY", file_env), crawler_timeout=int( env_value("CONTENTFIND_API_CRAWAPI_TIMEOUT_SECONDS", file_env, "30") ), video_model=env_value( "CONTENT_AGENT_VIDEO_LLM_MODEL", file_env, "google/gemini-3-flash-preview", ), gemini_api_key=env_value("GEMINI_API_KEY", file_env), openrouter_base_url=env_value( "OPENROUTER_BASE_URL", file_env, "https://openrouter.ai/api/v1" ), openrouter_api_key=env_value("OPENROUTER_API_KEY", file_env) or env_value("OPEN_ROUTER_API_KEY", file_env), # 文本步(筛选/拆分/解构)默认也用 Gemini(彻底 Gemini);可用 CK_STAGE_MODEL 覆盖 llm_model=env_value( "CK_STAGE_MODEL", file_env, env_value("CONTENT_AGENT_VIDEO_LLM_MODEL", file_env, "google/gemini-3-flash-preview"), ), max_cards=int(env_value("CK_MAX_CARDS", file_env, "12")), frames_dir=env_value("CK_FRAMES_DIR", file_env, "runtime/frames"), douyin_ratio=env_value("CK_DOUYIN_RATIO", file_env, "540p"), data_dir=env_value("CK_DATA_DIR", file_env, "data"), oss_upload_url=env_value( "CK_OSS_UPLOAD_URL", file_env, "http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream", ), oss_upload_timeout_seconds=int( env_value("CK_OSS_UPLOAD_TIMEOUT_SECONDS", file_env, "60") ), search_default_limit=int(env_value("CK_SEARCH_DEFAULT_LIMIT", file_env, "5")), search_content_type=env_value("CK_SEARCH_CONTENT_TYPE", file_env, "图文"), search_sort_type=env_value("CK_SEARCH_SORT_TYPE", file_env, "综合"), )