"""配置加载：从 .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）：小红书/微信等走 aiddit；抖音单独走 piaoquantv。 aiddit_crawler_base_url: str crawler_timeout: int # OpenRouter / Gemini（遗留视频链路和兼容烟测） openrouter_model: str openrouter_timeout_seconds: float 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 = "综合" # 抖音走独立后端（piaoquantv，绕开 aiddit 限流）：host / account_id / cookie_batch 单独配 piaoquantv_douyin_base_url: str = "http://crawapi.piaoquantv.com" piaoquantv_douyin_account_id: str = "7450041106378522636" piaoquantv_douyin_cookie_batch: str = "default" # Aliyun Bailian / Qwen（旧解构图文读懂 + 文本闸门/拆解）。 # 带默认值，保护旧测试里直接 Settings(...) 的关键字构造。 bailian_api_key: str = "" bailian_base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1" bailian_vl_model: str = "qwen-vl-plus" bailian_text_model: str = "qwen-plus" bailian_timeout_seconds: float = 120.0 demo_data_dir: 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), aiddit_crawler_base_url=env_value( "AIDDIT_CRAWLER_BASE_URL", file_env, "http://crawler.aiddit.com", ), crawler_timeout=int(env_value( "AIDDIT_CRAWLER_TIMEOUT_SECONDS", file_env, "30", )), openrouter_model=env_value( "OPENROUTER_MODEL", file_env, "google/gemini-3-flash-preview", ), openrouter_timeout_seconds=float(env_value( "OPENROUTER_TIMEOUT_SECONDS", file_env, "90", )), 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), bailian_api_key=env_value("ALIYUN_BAILIAN_API_KEY", file_env), bailian_base_url=env_value( "ALIYUN_BAILIAN_BASE_URL", file_env, "https://dashscope.aliyuncs.com/compatible-mode/v1", ), bailian_vl_model=env_value( "ALIYUN_BAILIAN_VL_MODEL", file_env, env_value("ALIYUN_BAILIAN_MODEL", file_env, "qwen-vl-plus"), ), bailian_text_model=env_value( "ALIYUN_BAILIAN_TEXT_MODEL", file_env, env_value("ALIYUN_BAILIAN_MODEL", file_env, "qwen-plus"), ), bailian_timeout_seconds=float(env_value( "ALIYUN_BAILIAN_TIMEOUT_SECONDS", file_env, "120", )), # 文本步（筛选/拆分/解构）默认走百炼 Qwen；可用 CK_STAGE_MODEL 覆盖 llm_model=env_value( "CK_STAGE_MODEL", file_env, env_value("ALIYUN_BAILIAN_TEXT_MODEL", file_env, env_value("ALIYUN_BAILIAN_MODEL", file_env, "qwen-plus")), ), 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"), demo_data_dir=env_value( "CK_DEMO_DATA_DIR", file_env, str(Path(env_value("CK_DATA_DIR", file_env, "data")) / "demo"), ), oss_upload_url=env_value( "CRAWLER_OSS_UPLOAD_URL", file_env, "http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream", ), oss_upload_timeout_seconds=int( env_value( "CRAWLER_OSS_UPLOAD_TIMEOUT_SECONDS", file_env, "60", ) ), search_default_limit=int(env_value( "CREATION_SEARCH_DEFAULT_LIMIT", file_env, "5", )), search_content_type=env_value( "CREATION_SEARCH_CONTENT_TYPE", file_env, "图文", ), search_sort_type=env_value( "CREATION_SEARCH_SORT_TYPE", file_env, "综合", ), piaoquantv_douyin_base_url=env_value( "PIAOQUANTV_DOUYIN_BASE_URL", file_env, "http://crawapi.piaoquantv.com", ), piaoquantv_douyin_account_id=env_value( "PIAOQUANTV_DOUYIN_ACCOUNT_ID", file_env, "7450041106378522636", ), piaoquantv_douyin_cookie_batch=env_value( "PIAOQUANTV_DOUYIN_COOKIE_BATCH", file_env, "default", ), ) @property def crawler_base_url(self) -> str: return self.aiddit_crawler_base_url @property def douyin_base_url(self) -> str: return self.piaoquantv_douyin_base_url @property def douyin_account_id(self) -> str: return self.piaoquantv_douyin_account_id @property def douyin_cookie_batch(self) -> str: return self.piaoquantv_douyin_cookie_batch @property def video_model(self) -> str: return self.openrouter_model