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（M4）
    llm_model: str
    # 入库（M5）
    knowhub_api: str
    ingest_enabled: bool
    # 卡片 / 抽帧
    max_cards: int
    frames_dir: str
    douyin_ratio: str  # 视频下载偏好码率（控成本/体积），如 540p

    @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),
            llm_model=env_value("MODEL", file_env, "anthropic/claude-sonnet-4.5"),
            knowhub_api=env_value("KNOWHUB_API", file_env, "http://localhost:8000"),
            # 开发期默认关闭真实入库；显式置 true 才发送
            ingest_enabled=env_value("INGEST_ENABLED", file_env, "false").lower()
            in ("1", "true", "yes"),
            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"),
        )