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

"""Formal single-item decode orchestration."""
from __future__ import annotations

import re
from dataclasses import dataclass, field
from typing import Any, Callable
from uuid import UUID

from core.config import Settings
from core.llm import chat_json as default_chat_json
from core.models import Post
from decode_content.contracts import SkillContract, load_contract
from decode_content.framing import frame_and_clean
from decode_content.gates import ChatJsonFn, creation_gate
from decode_content.models import DecodeResult, GateResult, PayloadDraft, ReadResult
from decode_content.payloads import build_payloads
from decode_content.readers.service import read_post
from decode_content.repository import DecodeRepository
from decode_content.scoping import ScopeLinker, apply_scopes, nounify_scopes, scope_candidates


@dataclass
class DecodeWorkflowOutput:
    item_id: UUID
    read_result: ReadResult
    gate_result: GateResult
    knowledges: list[dict[str, Any]] = field(default_factory=list)
    payloads: list[dict[str, Any]] = field(default_factory=list)
    decode_result: DecodeResult | None = None
    payload_drafts: list[PayloadDraft] = field(default_factory=list)
    status: str = "draft"


def slugify(text: str) -> str:
    s = re.sub(r"[^\w一-鿿]+", "-", text or "").strip("-").lower()
    return s or "run"


class DecodeService:
    """Decode one creation candidate without depending on web files or SQLite."""

    def __init__(
        self,
        *,
        settings: Settings | None = None,
        repository: DecodeRepository | None = None,
        contract: SkillContract | None = None,
        scope_linker: ScopeLinker | None = None,
        chat_json_fn: ChatJsonFn | None = None,
        reader: Callable[[Post], ReadResult] | None = None,
    ) -> None:
        self.settings = settings
        self.repository = repository
        self.contract = contract or load_contract()
        self.scope_linker = scope_linker
        self.chat_json_fn = chat_json_fn or default_chat_json
        self.reader = reader

    def _read(self, post: Post) -> ReadResult:
        if self.reader is not None:
            return self.reader(post)
        return read_post(post, settings=self.settings)

    def decode_post(
        self,
        *,
        item_id: UUID,
        post: Post,
        read_result: ReadResult | None = None,
    ) -> DecodeWorkflowOutput:
        repo = self.repository
        job = repo.create_decode_job(item_id=item_id, status="running") if repo else None
        self._save_contract_snapshots()
        read = read_result or self._read(post)
        if read.is_empty:
            gate = GateResult(passed=False, reason="读懂结果为空", details={"is_empty": True})
            decode_result = self._save_decode_result(
                item_id=item_id,
                job_id=getattr(job, "id", None),
                read=read,
                gate=gate,
                knowledges=[],
                status="skipped",
            )
            return DecodeWorkflowOutput(
                item_id=item_id,
                read_result=read,
                gate_result=gate,
                decode_result=decode_result,
                status="skipped",
            )

        gate = creation_gate(read.text, chat_json_fn=self.chat_json_fn)
        if not gate.passed:
            decode_result = self._save_decode_result(
                item_id=item_id,
                job_id=getattr(job, "id", None),
                read=read,
                gate=gate,
                knowledges=[],
                status="rejected",
            )
            return DecodeWorkflowOutput(
                item_id=item_id,
                read_result=read,
                gate_result=gate,
                decode_result=decode_result,
                status="rejected",
            )

        knowledges = frame_and_clean(
            post,
            read.text,
            contract=self.contract,
            chat_json_fn=self.chat_json_fn,
        )
        scopes = scope_candidates(knowledges, contract=self.contract, chat_json_fn=self.chat_json_fn)
        scopes = nounify_scopes(scopes, chat_json_fn=self.chat_json_fn)
        apply_scopes(knowledges, scopes, self.scope_linker)
        payloads = build_payloads(post, knowledges)
        decode_result = self._save_decode_result(
            item_id=item_id,
            job_id=getattr(job, "id", None),
            read=read,
            gate=gate,
            knowledges=knowledges,
            status="decoded",
        )
        payload_drafts = self._save_particles_and_payloads(
            item_id=item_id,
            decode_result=decode_result,
            knowledges=knowledges,
            payloads=payloads,
        )
        return DecodeWorkflowOutput(
            item_id=item_id,
            read_result=read,
            gate_result=gate,
            knowledges=knowledges,
            payloads=payloads,
            decode_result=decode_result,
            payload_drafts=payload_drafts,
            status="decoded",
        )

    def _save_contract_snapshots(self) -> None:
        if self.repository is None:
            return
        for snapshot in self.contract.snapshots():
            self.repository.save_contract_snapshot(
                contract_name=snapshot.contract_name,
                contract_type=snapshot.contract_type,
                version_label=snapshot.version_label,
                content_hash=snapshot.content_hash,
                source_path=snapshot.source_path,
                snapshot=snapshot.snapshot,
            )

    def _save_decode_result(
        self,
        *,
        item_id: UUID,
        job_id: UUID | None,
        read: ReadResult,
        gate: GateResult,
        knowledges: list[dict[str, Any]],
        status: str,
    ) -> DecodeResult:
        framing_result = {"knowledges": knowledges, "contract_hash": self.contract.content_hash}
        if self.repository is None:
            return DecodeResult(
                item_id=item_id,
                decode_job_id=job_id,
                read_result=read.model_dump(mode="json"),
                gate_result=gate.model_dump(mode="json"),
                framing_result=framing_result,
                status=status,
            )
        return self.repository.save_decode_result(
            item_id=item_id,
            decode_job_id=job_id,
            read_result=read.model_dump(mode="json"),
            gate_result=gate.model_dump(mode="json"),
            framing_result=framing_result,
            status=status,
        )

    def _save_particles_and_payloads(
        self,
        *,
        item_id: UUID,
        decode_result: DecodeResult,
        knowledges: list[dict[str, Any]],
        payloads: list[dict[str, Any]],
    ) -> list[PayloadDraft]:
        if self.repository is None:
            return [PayloadDraft(item_id=item_id, payload=payload) for payload in payloads]
        drafts: list[PayloadDraft] = []
        for knowledge, payload in zip(knowledges, payloads):
            particle = self.repository.save_knowledge_particle(
                item_id=item_id,
                decode_result_id=decode_result.id,
                particle_type=knowledge.get("type") or "how",
                title=knowledge.get("title") or "",
                content=knowledge,
                status="draft",
            )
            for scope in payload.get("scopes") or []:
                source_scope = self._scope_source(knowledge, scope)
                self.repository.save_scope_result(
                    item_id=item_id,
                    particle_id=particle.id,
                    scope_type=scope["scope_type"],
                    scope_value=scope["value"],
                    evidence={k: v for k, v in source_scope.items() if k not in {"scope_type", "value"}},
                    status="draft",
                )
            drafts.append(
                self.repository.save_payload_draft(
                    item_id=item_id,
                    particle_id=particle.id,
                    payload=payload,
                    review_status="pending",
                    ingest_ready=False,
                    status="draft",
                )
            )
        return drafts

    @staticmethod
    def _scope_source(knowledge: dict[str, Any], scope: dict[str, Any]) -> dict[str, Any]:
        source_scopes = list(knowledge.get("作用域", []))
        for step in knowledge.get("steps", []):
            source_scopes.extend(step.get("作用域", []))
        return next(
            (
                item
                for item in source_scopes
                if item.get("scope_type") == scope["scope_type"] and item.get("value") == scope["value"]
            ),
            scope,
        )


def decode_post(
    *,
    item_id: UUID,
    post: Post,
    settings: Settings | None = None,
    repository: DecodeRepository | None = None,
    scope_linker: ScopeLinker | None = None,
    chat_json_fn: ChatJsonFn | None = None,
    reader: Callable[[Post], ReadResult] | None = None,
    read_result: ReadResult | None = None,
) -> DecodeWorkflowOutput:
    service = DecodeService(
        settings=settings,
        repository=repository,
        scope_linker=scope_linker,
        chat_json_fn=chat_json_fn,
        reader=reader,
    )
    return service.decode_post(item_id=item_id, post=post, read_result=read_result)


def decode_item(
    item_id: UUID,
    *,
    load_post: Callable[[UUID], Post],
    service: DecodeService | None = None,
) -> DecodeWorkflowOutput:
    svc = service or DecodeService()
    return svc.decode_post(item_id=item_id, post=load_post(item_id))