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

"""Skill contract loading for the formal decode-content workflow."""
from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable

from core.prompts import PROMPTS_DIR
from decode_content.models import ContractSnapshot


ROOT = Path(__file__).resolve().parent.parent
DEFAULT_SKILL_DIR = ROOT / "创作知识提取-skill"
SCOPE_TREE_DIR = ROOT / "scope_trees"

KNOWLEDGE_TYPES = ("how", "what", "why")
BUSINESS_STAGES = ("灵感", "选题", "脚本")
CREATION_STAGES = ("定向", "构思", "结构", "成文", "打磨")
WHAT_KINDS = ("子集", "多维关系", "序列")
SCOPE_TYPES = ("substance", "form", "feeling", "effect", "intent")
REUSE_THRESHOLD = 0.90
SCOPE_TYPE_CN = {
    "substance": "实质",
    "form": "形式",
    "feeling": "感受",
    "effect": "作用",
    "intent": "意图",
}
DIM_CREATIONS = ("创作",)
DIM_ATTRIBUTE_BY_TYPE = {"how": "how", "what": "what", "why": "why"}
CUSTOM_EXT_KEYS_BY_TYPE = {
    "how": ("业务阶段", "创作阶段", "动作", "出自"),
    "what": ("业务阶段", "概要", "出自"),
    "why": ("业务阶段", "出自"),
}
PROMPT_NAMES = (
    "gate_admit",
    "gate_refute",
    "gate_tiebreak",
    "normalize_scope",
    "gate_how_admit",
    "gate_how_refute",
    "gate_how_tiebreak",
    "gate_why_refute",
)
SCOPE_TREE_FILES = ("trees.json", "trees_index.json", "trees_embeddings.npy")
DRIFT_NOTES = (
    {
        "key": "content_shape_drift",
        "note": "formal ingest boundary uses string content; what/why objects are serialized before payload draft.",
    },
    {
        "key": "how_step_wording_drift",
        "note": "formal code uses input/directive/output even when older phase text mentions 目的/指引/产出.",
    },
)


@dataclass(frozen=True)
class ContractArtifact:
    name: str
    contract_type: str
    source_path: Path
    content: str
    content_hash: str

    def snapshot(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "contract_type": self.contract_type,
            "relative_path": str(self.source_path.relative_to(ROOT)),
        }


@dataclass(frozen=True)
class SkillContract:
    """Immutable view of the extraction skill docs used by one decode run."""

    skill_dir: Path
    artifacts: tuple[ContractArtifact, ...]
    scope_tree_cache: tuple[dict[str, Any], ...] = ()

    @property
    def content_hash(self) -> str:
        h = hashlib.sha256()
        for artifact in self.artifacts:
            h.update(artifact.name.encode("utf-8"))
            h.update(artifact.content_hash.encode("ascii"))
        for item in self.scope_tree_cache:
            h.update(str(item.get("path", "")).encode("utf-8"))
            h.update(str(item.get("content_hash", "")).encode("ascii"))
        return h.hexdigest()

    def get(self, name: str) -> ContractArtifact:
        for artifact in self.artifacts:
            if artifact.name == name:
                return artifact
        raise KeyError(name)

    @property
    def phase1(self) -> str:
        return self.get("extraction/phase1-frame.md").content

    @property
    def phase2(self) -> str:
        return self.get("extraction/phase2-scope.md").content

    @property
    def phase3(self) -> str:
        return self.get("extraction/phase3-assemble.md").content

    def schema_for_particle_type(self, particle_type: str) -> dict[str, Any]:
        artifact = self.get(f"format/ingest-payload-{particle_type}.schema.json")
        return json.loads(artifact.content)

    def prompt_sources(self) -> list[dict[str, Any]]:
        return [a.snapshot() | {"content_hash": a.content_hash} for a in self.artifacts if a.contract_type == "prompt"]

    def scope_tree_sources(self) -> list[dict[str, Any]]:
        return list(self.scope_tree_cache)

    def constants(self) -> dict[str, Any]:
        return {
            "knowledge_types": list(KNOWLEDGE_TYPES),
            "business_stages": list(BUSINESS_STAGES),
            "creation_stages": list(CREATION_STAGES),
            "what_kinds": list(WHAT_KINDS),
            "scope_types": list(SCOPE_TYPES),
            "scope_type_cn": SCOPE_TYPE_CN,
            "dim_creations": list(DIM_CREATIONS),
            "dim_attribute_by_type": DIM_ATTRIBUTE_BY_TYPE,
            "custom_ext_keys_by_type": {k: list(v) for k, v in CUSTOM_EXT_KEYS_BY_TYPE.items()},
        }

    def snapshots(self, *, version_label: str | None = None) -> list[ContractSnapshot]:
        snapshots = [
            ContractSnapshot(
                contract_name="创作知识提取-skill",
                contract_type="skill",
                version_label=version_label,
                content_hash=self.content_hash,
                source_path=str(self.skill_dir.relative_to(ROOT)),
                snapshot={
                    "artifact_count": len(self.artifacts),
                    "artifacts": [a.snapshot() for a in self.artifacts],
                    "scope_tree_cache": self.scope_tree_sources(),
                    "constants": self.constants(),
                    "drift_notes": list(DRIFT_NOTES),
                },
            )
        ]
        snapshots.extend(
            ContractSnapshot(
                contract_name=artifact.name,
                contract_type=artifact.contract_type,
                version_label=version_label,
                content_hash=artifact.content_hash,
                source_path=str(artifact.source_path.relative_to(ROOT)),
                snapshot=artifact.snapshot(),
            )
            for artifact in self.artifacts
        )
        return snapshots


def _hash_text(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def _hash_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _artifact_type(path: Path) -> str:
    parts = path.parts
    if "extraction" in parts:
        return "prompt_phase"
    if "format" in parts:
        return "schema"
    if "taxonomy" in parts:
        return "taxonomy"
    if path.name == "README.md":
        return "readme"
    if "tools" in parts:
        return "tool_contract"
    return "contract"


def _scope_tree_cache(scope_tree_dir: Path = SCOPE_TREE_DIR) -> tuple[dict[str, Any], ...]:
    out: list[dict[str, Any]] = []
    for name in SCOPE_TREE_FILES:
        path = scope_tree_dir / name
        if not path.exists():
            continue
        stat = path.stat()
        out.append(
            {
                "path": str(path.relative_to(ROOT)),
                "size": stat.st_size,
                "mtime": int(stat.st_mtime),
                "content_hash": _hash_bytes(path.read_bytes()),
            }
        )
    return tuple(out)


def _iter_contract_files(skill_dir: Path) -> Iterable[Path]:
    preferred = [
        skill_dir / "README.md",
        skill_dir / "extraction" / "phase1-frame.md",
        skill_dir / "extraction" / "phase2-scope.md",
        skill_dir / "extraction" / "phase3-assemble.md",
        skill_dir / "format" / "ingest-payload-how.schema.json",
        skill_dir / "format" / "ingest-payload-what.schema.json",
        skill_dir / "format" / "ingest-payload-why.schema.json",
        skill_dir / "taxonomy" / "业务阶段.json",
        skill_dir / "taxonomy" / "作用域.md",
        skill_dir / "taxonomy" / "创作阶段.json",
        skill_dir / "taxonomy" / "知识类型.json",
    ]
    seen: set[Path] = set()
    for path in preferred:
        if path.exists():
            seen.add(path)
            yield path
    for path in sorted(skill_dir.rglob("*")):
        if path.is_file() and path.suffix in {".md", ".json"} and path not in seen:
            yield path


def _iter_prompt_files() -> Iterable[Path]:
    for name in PROMPT_NAMES:
        path = PROMPTS_DIR / f"{name}.txt"
        if path.exists():
            yield path


def load_contract(skill_dir: Path | str = DEFAULT_SKILL_DIR) -> SkillContract:
    root = Path(skill_dir)
    if not root.exists():
        raise FileNotFoundError(f"skill contract directory not found: {root}")
    artifacts: list[ContractArtifact] = []
    for path in _iter_contract_files(root):
        text = path.read_text(encoding="utf-8")
        artifacts.append(
            ContractArtifact(
                name=str(path.relative_to(root)),
                contract_type=_artifact_type(path.relative_to(root)),
                source_path=path,
                content=text,
                content_hash=_hash_text(text),
            )
        )
    for path in _iter_prompt_files():
        text = path.read_text(encoding="utf-8")
        artifacts.append(
            ContractArtifact(
                name=f"prompts/{path.name}",
                contract_type="prompt",
                source_path=path,
                content=text,
                content_hash=_hash_text(text),
            )
        )
    return SkillContract(skill_dir=root, artifacts=tuple(artifacts), scope_tree_cache=_scope_tree_cache())


def load_decode_contracts(skill_dir: Path | str = DEFAULT_SKILL_DIR) -> SkillContract:
    return load_contract(skill_dir)


def compute_contract_hash(contract: SkillContract | None = None) -> str:
    return (contract or load_contract()).content_hash


def iter_contract_artifacts(
    contract: SkillContract | None = None,
    *,
    version_label: str | None = None,
) -> list[ContractSnapshot]:
    return (contract or load_contract()).snapshots(version_label=version_label)


def iter_contract_snapshots(
    contract: SkillContract | None = None,
    *,
    version_label: str | None = None,
) -> list[ContractSnapshot]:
    """Backward-compatible alias for tests and callers that need artifact payloads."""
    return iter_contract_artifacts(contract, version_label=version_label)