"""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))