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

"""External knowledge ingest API client and draft-level orchestration."""
from __future__ import annotations

import time
from dataclasses import dataclass
from typing import Any, Callable

import httpx

from core.config import IngestApiConfig
from core.text_limits import ERROR_MESSAGE_MAX_CHARS, clip_text
from decode_content.models import IngestRecord, PayloadDraft
from decode_content.payloads import IngestPayloadValidationError, validate_ingest_payload
from decode_content.repository import DecodeRepository

PostFn = Callable[..., Any]


class IngestApiError(RuntimeError):
    def __init__(
        self,
        message: str,
        *,
        response_payload: dict[str, Any] | None = None,
        http_status: int | None = None,
        attempt_count: int = 1,
    ) -> None:
        super().__init__(message)
        self.response_payload = response_payload or {}
        self.http_status = http_status
        self.attempt_count = attempt_count


@dataclass(frozen=True)
class IngestApiResult:
    knowledge_id: str | None
    source_id: str | None
    http_status: int
    attempt_count: int
    response_payload: dict[str, Any]


class KnowledgeIngestClient:
    """Small synchronous client for POST /api/v1/knowledge/ingest."""

    def __init__(
        self,
        config: IngestApiConfig,
        *,
        post_fn: PostFn | None = None,
        sleep_fn: Callable[[float], None] = time.sleep,
    ) -> None:
        self.config = config
        self.post_fn = post_fn or httpx.post
        self.sleep_fn = sleep_fn

    def ingest(self, payload: dict[str, Any]) -> IngestApiResult:
        validate_ingest_payload(payload)
        if not self.config.url:
            raise IngestApiError(
                "CK_INGEST_API_URL 未配置，真实 ingest 已关闭",
                response_payload={"dry_run": False, "missing_config": "CK_INGEST_API_URL"},
            )
        headers = {"Content-Type": "application/json"}
        if self.config.token:
            headers["Authorization"] = f"Bearer {self.config.token}"
        delays = tuple(self.config.retry_delays_seconds)
        last_error: IngestApiError | None = None
        for attempt_index in range(len(delays) + 1):
            attempt_count = attempt_index + 1
            try:
                response = self.post_fn(
                    self.config.url,
                    json=payload,
                    headers=headers,
                    timeout=self.config.timeout_seconds,
                )
                raw = _response_payload(response)
                status_code = int(getattr(response, "status_code", 0) or 0)
                if 200 <= status_code < 300:
                    knowledge_id = raw.get("knowledge_id") if isinstance(raw, dict) else None
                    source_id = raw.get("source_id") if isinstance(raw, dict) else None
                    return IngestApiResult(
                        knowledge_id=str(knowledge_id) if knowledge_id else None,
                        source_id=str(source_id) if source_id else None,
                        http_status=status_code,
                        attempt_count=attempt_count,
                        response_payload={
                            "dry_run": False,
                            "http_status": status_code,
                            "knowledge_id": knowledge_id,
                            "source_id": source_id,
                            "attempt_count": attempt_count,
                            "response": raw,
                        },
                    )
                last_error = IngestApiError(
                    f"ingest API 返回 {status_code}",
                    response_payload={
                        "dry_run": False,
                        "http_status": status_code,
                        "attempt_count": attempt_count,
                        "response": raw,
                    },
                    http_status=status_code,
                    attempt_count=attempt_count,
                )
            except IngestApiError as exc:
                last_error = exc
            except Exception as exc:  # network/client errors
                last_error = IngestApiError(
                    str(exc) or exc.__class__.__name__,
                    response_payload={
                        "dry_run": False,
                        "exception_type": exc.__class__.__name__,
                        "attempt_count": attempt_count,
                    },
                    attempt_count=attempt_count,
                )
            if attempt_index < len(delays):
                self.sleep_fn(delays[attempt_index])
        assert last_error is not None
        raise last_error


def _response_payload(response: Any) -> dict[str, Any] | str:
    try:
        value = response.json()
    except Exception:
        value = getattr(response, "text", "")
    return value if isinstance(value, (dict, str)) else {"response": value}


def ingest_payload_draft(
    repo: DecodeRepository,
    draft: PayloadDraft,
    *,
    dry_run: bool = True,
    client: KnowledgeIngestClient | None = None,
) -> IngestRecord:
    payload = draft.payload or {}
    source_id = ((payload.get("source") or {}).get("id") or "") if isinstance(payload, dict) else ""
    try:
        validate_ingest_payload(payload)
        if dry_run:
            updated = _mark_ingested(repo, draft)
            return repo.save_ingest_record(
                payload_draft_id=draft.id,
                target_system="dry-run",
                target_id=str(draft.id) if draft.id else None,
                status="ingested",
                attempt_count=1,
                response_payload={
                    "dry_run": True,
                    "source_id": source_id,
                    "note": "payload validated by formal ingest path; external ingest API not called",
                    "payload": updated.payload if hasattr(updated, "payload") else payload,
                },
            )
        if client is None:
            raise IngestApiError("真实 ingest 需要 KnowledgeIngestClient")
        result = client.ingest(payload)
        _mark_ingested(repo, draft)
        return repo.save_ingest_record(
            payload_draft_id=draft.id,
            target_system="knowledge-ingest-api",
            target_id=result.knowledge_id,
            status="ingested",
            attempt_count=result.attempt_count,
            response_payload=result.response_payload,
        )
    except (IngestPayloadValidationError, IngestApiError) as exc:
        response_payload = getattr(exc, "response_payload", {}) or {}
        response_payload.setdefault("dry_run", dry_run)
        response_payload.setdefault("source_id", source_id)
        return repo.save_ingest_record(
            payload_draft_id=draft.id,
            target_system="dry-run" if dry_run else "knowledge-ingest-api",
            target_id=None,
            status="failed",
            attempt_count=getattr(exc, "attempt_count", 1),
            response_payload=response_payload,
            error_message=clip_text(str(exc), ERROR_MESSAGE_MAX_CHARS),
        )


def _mark_ingested(repo: DecodeRepository, draft: PayloadDraft) -> PayloadDraft:
    if draft.id is None:
        return draft
    marker = getattr(repo, "mark_payload_draft_ingested", None)
    if marker is None:
        return draft
    return marker(draft.id)