Create-knowledge-find-decode/creation_knowledge/config.py

"""配置加载：从 .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, "综合"),
        )