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

"""入库：调 KnowHub ingest API。开发期默认关闭，只组装+存库（见 技术文档/技术架构.md）。"""
from __future__ import annotations

from typing import Any, Callable, Optional

import httpx

from creation_knowledge.config import Settings
from creation_knowledge.models import IngestPayload

INGEST_PATH = "/api/v1/knowledge/ingest"


class IngestError(RuntimeError):
    pass


def ingest(
    payload: IngestPayload,
    *,
    settings: Optional[Settings] = None,
    env_file: str = ".env",
    http_post: Callable[..., Any] = httpx.post,
    timeout: float = 30.0,
) -> dict:
    """真实发送一条知识到 KnowHub。仅在 INGEST_ENABLED=true 时允许调用。"""
    settings = settings or Settings.from_env(env_file)
    if not settings.ingest_enabled:
        raise IngestError("INGEST_ENABLED=false：开发期拒绝真实入库")
    url = settings.knowhub_api.rstrip("/") + INGEST_PATH
    try:
        resp = http_post(
            url,
            json=payload.model_dump(),
            headers={"Content-Type": "application/json"},
            timeout=timeout,
        )
        resp.raise_for_status()
        return resp.json()
    except httpx.HTTPError as exc:
        raise IngestError(f"ingest_http_error: {exc}") from exc