Create-knowledge-find-decode/core/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 ""


def env_float_tuple(
    key: str,
    file_env: dict[str, str],
    default: str,
) -> tuple[float, ...]:
    raw = env_value(key, file_env, default)
    values: list[float] = []
    for part in raw.split(","):
        text = part.strip()
        if not text:
            continue
        values.append(float(text))
    return tuple(values)


@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 CreationDbConfig:
    """Formal creation-knowledge state database, backed by cloud PostgreSQL."""

    host: str
    port: int
    user: str
    password: str
    database: str
    schema: str = DEFAULT_PG_SCHEMA
    timeout: int = 10
    application_name: str = "creation-knowledge"
    pool_min: int = 1
    pool_max: int = 10

    @classmethod
    def from_env(cls, env_file: str | Path = ".env") -> "CreationDbConfig":
        file_env = load_env_file(env_file)
        return cls(
            host=env_value("CK_DB_HOST", file_env, required=True),
            port=int(env_value("CK_DB_PORT", file_env, "5432")),
            user=env_value("CK_DB_USER", file_env, required=True),
            password=env_value("CK_DB_PASSWORD", file_env, required=True),
            database=env_value(
                "CK_DB_NAME",
                file_env,
                "creation_knowledge_prod",
            ),
            schema=env_value("CK_DB_SCHEMA", file_env, DEFAULT_PG_SCHEMA),
            pool_min=int(env_value("CK_DB_POOL_MIN", file_env, "1")),
            pool_max=int(env_value("CK_DB_POOL_MAX", file_env, "10")),
        )


@dataclass
class IngestApiConfig:
    """External knowledge ingest API configuration.

    Empty url means real ingest is disabled; dry-run paths do not need it.
    """

    url: str = ""
    token: str = ""
    timeout_seconds: float = 30.0
    retry_delays_seconds: tuple[float, ...] = (1.0, 3.0)

    @classmethod
    def from_env(cls, env_file: str | Path = ".env") -> "IngestApiConfig":
        file_env = load_env_file(env_file)
        return cls(
            url=env_value("CK_INGEST_API_URL", file_env),
            token=env_value("CK_INGEST_TOKEN", file_env),
            timeout_seconds=float(env_value("CK_INGEST_TIMEOUT_SECONDS", file_env, "30")),
            retry_delays_seconds=env_float_tuple(
                "CK_INGEST_RETRY_DELAYS_SECONDS",
                file_env,
                "1,3",
            ),
        )


@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
    video_oss_retry_delays_seconds: tuple[float, ...] = (5.0, 10.0)
    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",
                )
            ),
            video_oss_retry_delays_seconds=env_float_tuple(
                "CK_VIDEO_OSS_RETRY_DELAYS_SECONDS",
                file_env,
                "5,10",
            ),
            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