#!/usr/bin/env python3
"""Shared helpers for the knowledge supply chain MVP."""

from __future__ import annotations

import fnmatch
import hashlib
import json
import re
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Final, cast

try:
    import jsonschema
except ImportError as exc:
    raise SystemExit(
        f"缺少依赖 jsonschema，请使用 uvx --with jsonschema --with pyyaml python <script>\n{exc}"
    ) from exc

try:
    import yaml
except ImportError as exc:
    raise SystemExit(
        f"缺少依赖 PyYAML，请使用 uvx --with pyyaml --with jsonschema python <script>\n{exc}"
    ) from exc


JSONMapping = dict[str, object]

SOURCE_CONFIG_SCHEMA_RELATIVE: Final[Path] = Path(
    "platform/contracts/knowledge-source-config.schema.yaml"
)
REVIEW_LEDGER_SCHEMA_RELATIVE: Final[Path] = Path(
    "platform/contracts/knowledge-review-ledger.schema.yaml"
)
ARTIFACT_SCHEMA_RELATIVE: Final[Path] = Path(
    "platform/contracts/knowledge-artifact.schema.yaml"
)
PUBLISH_RECORD_SCHEMA_RELATIVE: Final[Path] = Path(
    "platform/contracts/knowledge-publish-record.schema.yaml"
)
DOCUMENT_SCHEMA_RELATIVE: Final[Path] = Path(
    "platform/contracts/knowledge-document.schema.yaml"
)
CHUNK_SCHEMA_RELATIVE: Final[Path] = Path(
    "platform/contracts/knowledge-chunk.schema.yaml"
)
SOURCE_REGISTRY_RELATIVE: Final[Path] = Path("platform/knowledge/source-registry.yaml")
ACL_POLICIES_RELATIVE: Final[Path] = Path("platform/knowledge/acl-policies.yaml")
PUBLISH_TARGETS_RELATIVE: Final[Path] = Path("platform/knowledge/publish-targets.yaml")
DEFAULT_MAX_CHUNK_CHARACTERS: Final[int] = 1200
DEFAULT_REVIEW_STANDARDS_VERSION: Final[str] = "knowledge-mvp-1"
MARKDOWN_MIN_BODY_CHARACTERS: Final[int] = 80
MARKDOWN_MAX_TITLE_CHARACTERS: Final[int] = 120
PLACEHOLDER_PATTERN: Final[re.Pattern[str]] = re.compile(
    r"\b(todo|tbd|placeholder|coming soon)\b|待补充|待完善|占位",
    re.IGNORECASE,
)


class KnowledgeContractError(Exception):
    """Raised when a knowledge contract is invalid."""


@dataclass(frozen=True, slots=True)
class ChangeEntry:
    """Represents one git diff path change."""

    change_type: str
    path: str


@dataclass(frozen=True, slots=True)
class SourceRule:
    """Represents one source rule declared by a business repository."""

    source_id: str
    include: tuple[str, ...]
    exclude: tuple[str, ...]
    visibility: str
    source_type: str
    doc_type: str
    product_area: str
    owner: str
    review_status: str
    tags: tuple[str, ...]
    title_strategy: str
    title_prefix: str
    review_mode: str
    approved_by: str | None
    effective_from: str | None
    expires_at: str | None


@dataclass(frozen=True, slots=True)
class ReviewEntry:
    """Represents one document-level review ledger entry."""

    path: str
    source_id: str
    checksum: str
    review_status: str
    approved_by: str | None
    effective_from: str | None
    expires_at: str | None
    standards_version: str
    review_notes: str | None


@dataclass(frozen=True, slots=True)
class ResolvedReview:
    """Represents effective review metadata for one document."""

    review_status: str
    approved_by: str | None
    effective_from: str | None
    expires_at: str | None
    checksum: str
    standards_version: str
    review_notes: str | None
    standard_issues: tuple[str, ...]
    review_issues: tuple[str, ...]
    reason: str | None


@dataclass(frozen=True, slots=True)
class KnowledgePaths:
    """Filesystem layout inside ljwx-deploy for knowledge state."""

    artifacts_root: Path
    events_root: Path
    publish_records_root: Path


def utc_now() -> str:
    """Return an ISO-8601 timestamp in UTC."""

    return datetime.now(tz=UTC).isoformat()


def ensure_directory(path: Path) -> None:
    """Create a directory tree when it does not exist."""

    path.mkdir(parents=True, exist_ok=True)


def ensure_parent(path: Path) -> None:
    """Create the parent directory for a file path."""

    ensure_directory(path.parent)


def write_json(path: Path, payload: JSONMapping | list[JSONMapping]) -> None:
    """Write a JSON file with deterministic formatting."""

    ensure_parent(path)
    path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )


def load_json_mapping(path: Path) -> JSONMapping:
    """Load a JSON object from disk."""

    raw = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(raw, dict):
        raise KnowledgeContractError(f"{path} 根节点必须是 object")
    return {str(key): value for key, value in raw.items()}


def load_yaml_mapping(path: Path) -> JSONMapping:
    """Load a YAML mapping from disk."""

    raw = yaml.safe_load(path.read_text(encoding="utf-8"))
    if raw is None:
        return {}
    if not isinstance(raw, dict):
        raise KnowledgeContractError(f"{path} 根节点必须是 mapping")
    return {str(key): value for key, value in raw.items()}


def write_jsonl(path: Path, rows: list[JSONMapping]) -> None:
    """Write JSONL rows to disk."""

    ensure_parent(path)
    with path.open("w", encoding="utf-8") as handle:
        for row in rows:
            handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True))
            handle.write("\n")


def append_jsonl(path: Path, row: JSONMapping) -> None:
    """Append one JSONL row to disk."""

    ensure_parent(path)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True))
        handle.write("\n")


def load_jsonl(path: Path) -> list[JSONMapping]:
    """Load JSONL records from disk when present."""

    if not path.exists():
        return []
    rows: list[JSONMapping] = []
    for line in path.read_text(encoding="utf-8").splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        raw = json.loads(stripped)
        if not isinstance(raw, dict):
            raise KnowledgeContractError(f"{path} 存在非 mapping JSONL 记录")
        rows.append({str(key): value for key, value in raw.items()})
    return rows


def load_validator(path: Path) -> jsonschema.Draft202012Validator:
    """Load a JSON schema validator from YAML."""

    schema = load_yaml_mapping(path)
    jsonschema.Draft202012Validator.check_schema(schema)
    return jsonschema.Draft202012Validator(schema)


def validate_payload(
    validator: jsonschema.Draft202012Validator,
    payload: JSONMapping,
    payload_name: str,
) -> None:
    """Validate one payload against a schema."""

    try:
        validator.validate(payload)
    except jsonschema.ValidationError as exc:
        raise KnowledgeContractError(
            f"{payload_name} 不符合 schema: {exc.message}"
        ) from exc


def require_mapping(value: object, field_path: str) -> JSONMapping:
    """Require a JSON mapping."""

    if not isinstance(value, dict):
        raise KnowledgeContractError(f"{field_path} 必须是 mapping")
    return {str(key): item for key, item in value.items()}


def require_list(value: object, field_path: str) -> list[object]:
    """Require a list."""

    if not isinstance(value, list):
        raise KnowledgeContractError(f"{field_path} 必须是列表")
    return value


def require_non_empty_str(value: object, field_path: str) -> str:
    """Require a non-empty string."""

    if not isinstance(value, str) or not value.strip():
        raise KnowledgeContractError(f"{field_path} 必须是非空字符串")
    return value.strip()


def optional_non_empty_str(value: object, field_path: str) -> str | None:
    """Read an optional non-empty string."""

    if value in ("", None):
        return None
    return require_non_empty_str(value, field_path)


def require_str_list(value: object, field_path: str) -> list[str]:
    """Require a list of strings."""

    items = require_list(value, field_path)
    results: list[str] = []
    for index, item in enumerate(items):
        results.append(require_non_empty_str(item, f"{field_path}[{index}]"))
    return results


def sha256_text(text: str) -> str:
    """Return the SHA-256 digest of a text payload."""

    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def control_files_changed(
    diff_entries: list[ChangeEntry], controlled_paths: set[str]
) -> bool:
    """Return True when a diff touched control files that require full reevaluation."""

    return any(entry.path in controlled_paths for entry in diff_entries)


def slugify_fragment(value: str) -> str:
    """Convert a string into a filesystem-safe slug."""

    normalized = re.sub(r"[^a-zA-Z0-9]+", "-", value).strip("-").lower()
    return normalized or "unknown"


def repo_slug(repo: str) -> str:
    """Return a stable directory slug for an owner/repo name."""

    return repo.replace("/", "__")


def build_document_id(source_repo: str, source_path: str) -> str:
    """Build a deterministic document identifier from repo and path."""

    repo_name = source_repo.split("/")[-1]
    base_slug = slugify_fragment(f"{repo_name}-{source_path}")
    if len(base_slug) <= 96:
        return base_slug
    digest = hashlib.sha1(f"{source_repo}:{source_path}".encode("utf-8")).hexdigest()[
        :12
    ]
    return f"{base_slug[:80]}-{digest}"


def match_globs(
    path: str, includes: tuple[str, ...], excludes: tuple[str, ...]
) -> bool:
    """Return True when a path matches include/exclude rules."""

    included = any(fnmatch.fnmatch(path, pattern) for pattern in includes)
    if not included:
        return False
    return not any(fnmatch.fnmatch(path, pattern) for pattern in excludes)


def parse_diff_entries(path: Path) -> list[ChangeEntry]:
    """Parse git diff --name-status output into flattened path changes."""

    if not path.exists():
        return []

    results: list[ChangeEntry] = []
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line:
            continue
        parts = line.split("\t")
        status = require_non_empty_str(parts[0], "diff.status")
        base_status = status[0]
        if base_status in {"A", "C", "M", "T"}:
            if len(parts) < 2:
                raise KnowledgeContractError(f"无效 diff 行: {line}")
            results.append(ChangeEntry(change_type="upsert", path=parts[-1]))
            continue
        if base_status == "D":
            if len(parts) != 2:
                raise KnowledgeContractError(f"无效删除 diff 行: {line}")
            results.append(ChangeEntry(change_type="delete", path=parts[1]))
            continue
        if base_status == "R":
            if len(parts) != 3:
                raise KnowledgeContractError(f"无效重命名 diff 行: {line}")
            results.append(ChangeEntry(change_type="delete", path=parts[1]))
            results.append(ChangeEntry(change_type="upsert", path=parts[2]))
            continue
        raise KnowledgeContractError(f"暂不支持的 diff 状态: {status}")
    return results


def extract_markdown_title(content: str, source_path: str, strategy: str) -> str:
    """Resolve a document title from markdown content."""

    if strategy == "filename":
        return Path(source_path).stem.replace("-", " ").replace("_", " ").strip()

    for line in content.splitlines():
        stripped = line.strip()
        if stripped.startswith("# "):
            return stripped.removeprefix("# ").strip()

    return Path(source_path).stem.replace("-", " ").replace("_", " ").strip()


def chunk_markdown(
    content: str, max_characters: int = DEFAULT_MAX_CHUNK_CHARACTERS
) -> list[str]:
    """Chunk markdown into paragraph groups."""

    paragraphs = [item.strip() for item in content.split("\n\n") if item.strip()]
    if not paragraphs:
        return []

    chunks: list[str] = []
    current_parts: list[str] = []
    current_length = 0
    for paragraph in paragraphs:
        paragraph_length = len(paragraph)
        if current_parts and current_length + paragraph_length > max_characters:
            chunks.append("\n\n".join(current_parts))
            current_parts = [paragraph]
            current_length = paragraph_length
            continue
        current_parts.append(paragraph)
        current_length += paragraph_length

    if current_parts:
        chunks.append("\n\n".join(current_parts))

    return chunks


def knowledge_paths(deploy_root: Path, environment: str) -> KnowledgePaths:
    """Resolve the filesystem layout used by central knowledge state."""

    base_root = deploy_root / "knowledge"
    return KnowledgePaths(
        artifacts_root=base_root / "artifacts" / environment,
        events_root=base_root / "events" / environment,
        publish_records_root=base_root / "publish-records" / environment,
    )


def load_source_rules(
    config_path: Path, deploy_root: Path
) -> tuple[str, list[SourceRule]]:
    """Load and validate source config rules."""

    validator = load_validator(deploy_root / SOURCE_CONFIG_SCHEMA_RELATIVE)
    payload = load_yaml_mapping(config_path)
    validate_payload(validator, payload, config_path.as_posix())

    service = require_non_empty_str(
        payload.get("service"), "knowledge.sources.yaml.service"
    )
    raw_sources = require_list(payload.get("sources"), "knowledge.sources.yaml.sources")

    rules: list[SourceRule] = []
    for index, raw_source in enumerate(raw_sources):
        source = require_mapping(raw_source, f"knowledge.sources.yaml.sources[{index}]")
        rules.append(
            SourceRule(
                source_id=require_non_empty_str(
                    source.get("id"), f"knowledge.sources.yaml.sources[{index}].id"
                ),
                include=tuple(
                    require_str_list(
                        source.get("include"),
                        f"knowledge.sources.yaml.sources[{index}].include",
                    )
                ),
                exclude=tuple(
                    require_str_list(
                        source.get("exclude", []),
                        f"knowledge.sources.yaml.sources[{index}].exclude",
                    )
                ),
                visibility=require_non_empty_str(
                    source.get("visibility"),
                    f"knowledge.sources.yaml.sources[{index}].visibility",
                ),
                source_type=require_non_empty_str(
                    source.get("source_type"),
                    f"knowledge.sources.yaml.sources[{index}].source_type",
                ),
                doc_type=require_non_empty_str(
                    source.get("doc_type"),
                    f"knowledge.sources.yaml.sources[{index}].doc_type",
                ),
                product_area=require_non_empty_str(
                    source.get("product_area"),
                    f"knowledge.sources.yaml.sources[{index}].product_area",
                ),
                owner=require_non_empty_str(
                    source.get("owner"),
                    f"knowledge.sources.yaml.sources[{index}].owner",
                ),
                review_status=require_non_empty_str(
                    source.get("review_status"),
                    f"knowledge.sources.yaml.sources[{index}].review_status",
                ),
                tags=tuple(
                    require_str_list(
                        source.get("tags", []),
                        f"knowledge.sources.yaml.sources[{index}].tags",
                    )
                ),
                title_strategy=optional_non_empty_str(
                    source.get("title_strategy"),
                    f"knowledge.sources.yaml.sources[{index}].title_strategy",
                )
                or "first_heading",
                title_prefix=optional_non_empty_str(
                    source.get("title_prefix"),
                    f"knowledge.sources.yaml.sources[{index}].title_prefix",
                )
                or "",
                review_mode=optional_non_empty_str(
                    source.get("review_mode"),
                    f"knowledge.sources.yaml.sources[{index}].review_mode",
                )
                or "legacy",
                approved_by=optional_non_empty_str(
                    source.get("approved_by"),
                    f"knowledge.sources.yaml.sources[{index}].approved_by",
                ),
                effective_from=optional_non_empty_str(
                    source.get("effective_from"),
                    f"knowledge.sources.yaml.sources[{index}].effective_from",
                ),
                expires_at=optional_non_empty_str(
                    source.get("expires_at"),
                    f"knowledge.sources.yaml.sources[{index}].expires_at",
                ),
            )
        )
    return service, rules


def load_review_entries(
    review_path: Path, deploy_root: Path
) -> dict[tuple[str, str], ReviewEntry]:
    """Load an optional document-level review ledger."""

    if not review_path.exists():
        return {}

    validator = load_validator(deploy_root / REVIEW_LEDGER_SCHEMA_RELATIVE)
    payload = load_yaml_mapping(review_path)
    validate_payload(validator, payload, review_path.as_posix())

    raw_documents = require_list(
        payload.get("documents", []),
        f"{review_path.name}.documents",
    )
    entries: dict[tuple[str, str], ReviewEntry] = {}
    for index, raw_document in enumerate(raw_documents):
        document = require_mapping(
            raw_document, f"{review_path.name}.documents[{index}]"
        )
        entry = ReviewEntry(
            path=require_non_empty_str(
                document.get("path"), f"{review_path.name}.documents[{index}].path"
            ),
            source_id=require_non_empty_str(
                document.get("source_id"),
                f"{review_path.name}.documents[{index}].source_id",
            ),
            checksum=require_non_empty_str(
                document.get("checksum"),
                f"{review_path.name}.documents[{index}].checksum",
            ),
            review_status=require_non_empty_str(
                document.get("review_status"),
                f"{review_path.name}.documents[{index}].review_status",
            ),
            approved_by=optional_non_empty_str(
                document.get("approved_by"),
                f"{review_path.name}.documents[{index}].approved_by",
            ),
            effective_from=optional_non_empty_str(
                document.get("effective_from"),
                f"{review_path.name}.documents[{index}].effective_from",
            ),
            expires_at=optional_non_empty_str(
                document.get("expires_at"),
                f"{review_path.name}.documents[{index}].expires_at",
            ),
            standards_version=optional_non_empty_str(
                document.get("standards_version"),
                f"{review_path.name}.documents[{index}].standards_version",
            )
            or DEFAULT_REVIEW_STANDARDS_VERSION,
            review_notes=optional_non_empty_str(
                document.get("review_notes"),
                f"{review_path.name}.documents[{index}].review_notes",
            ),
        )
        key = (entry.source_id, entry.path)
        if key in entries:
            raise KnowledgeContractError(
                f"{review_path.name} 存在重复审核记录: {entry.source_id}/{entry.path}"
            )
        entries[key] = entry
    return entries


def markdown_standard_issues(source_path: str, content: str) -> list[str]:
    """Validate the minimum markdown standard required by the knowledge workflow."""

    issues: list[str] = []
    stripped_content = content.strip()
    if not stripped_content:
        return ["文档内容为空"]

    title = extract_markdown_title(content, source_path, "first_heading").strip()
    if "# " not in content:
        issues.append("缺少一级标题（# Heading）")
    if len(title) < 3:
        issues.append("标题过短")
    if len(title) > MARKDOWN_MAX_TITLE_CHARACTERS:
        issues.append("标题过长")

    body_parts: list[str] = []
    in_code_fence = False
    for raw_line in content.splitlines():
        stripped_line = raw_line.strip()
        if stripped_line.startswith("```"):
            in_code_fence = not in_code_fence
            continue
        if in_code_fence or not stripped_line:
            continue
        if stripped_line.startswith("#"):
            continue
        if re.fullmatch(r"[-_*`~ ]+", stripped_line):
            continue
        body_parts.append(stripped_line)

    body_text = " ".join(body_parts).strip()
    if len(body_text) < MARKDOWN_MIN_BODY_CHARACTERS:
        issues.append("正文内容过短，未达到最小信息量要求")
    if PLACEHOLDER_PATTERN.search(body_text) and len(body_text) < (
        MARKDOWN_MIN_BODY_CHARACTERS * 2
    ):
        issues.append("正文仍包含明显占位内容")
    return issues


def resolve_document_review(
    *,
    rule: SourceRule,
    relative_path: str,
    content: str,
    review_entries: dict[tuple[str, str], ReviewEntry],
) -> ResolvedReview:
    """Resolve effective review metadata for one document."""

    checksum = sha256_text(content)
    standard_issues = tuple(markdown_standard_issues(relative_path, content))
    if rule.review_mode != "required":
        return ResolvedReview(
            review_status=rule.review_status,
            approved_by=rule.approved_by,
            effective_from=rule.effective_from,
            expires_at=rule.expires_at,
            checksum=checksum,
            standards_version=DEFAULT_REVIEW_STANDARDS_VERSION,
            review_notes=None,
            standard_issues=standard_issues,
            review_issues=(),
            reason=None if not standard_issues else "legacy_standard_warning",
        )

    review_issues: list[str] = []
    review_status = "reviewing"
    approved_by: str | None = None
    effective_from: str | None = None
    expires_at: str | None = None
    standards_version = DEFAULT_REVIEW_STANDARDS_VERSION
    review_notes: str | None = None

    entry = review_entries.get((rule.source_id, relative_path))
    if entry is None:
        review_issues.append("缺少 knowledge.review.yaml 审核记录")
    else:
        review_status = entry.review_status
        approved_by = entry.approved_by
        effective_from = entry.effective_from
        expires_at = entry.expires_at
        standards_version = entry.standards_version
        review_notes = entry.review_notes
        if entry.checksum != checksum:
            review_issues.append("审核记录 checksum 与当前文档不一致")
        if entry.review_status != "approved":
            review_issues.append(f"审核状态不是 approved: {entry.review_status}")

    if standard_issues:
        review_status = "reviewing"
    if review_issues and review_status == "approved":
        review_status = "reviewing"

    reason_parts = list(standard_issues) + review_issues
    reason = "; ".join(reason_parts) if reason_parts else None
    return ResolvedReview(
        review_status=review_status,
        approved_by=approved_by,
        effective_from=effective_from,
        expires_at=expires_at,
        checksum=checksum,
        standards_version=standards_version,
        review_notes=review_notes,
        standard_issues=standard_issues,
        review_issues=tuple(review_issues),
        reason=reason,
    )


def load_source_registry(deploy_root: Path) -> dict[str, dict[str, object]]:
    """Load artifact source registry entries keyed by repository."""

    payload = load_yaml_mapping(deploy_root / SOURCE_REGISTRY_RELATIVE)
    raw_registry = require_list(
        payload.get("artifact_sources", []),
        "platform/knowledge/source-registry.yaml.artifact_sources",
    )
    registry: dict[str, dict[str, object]] = {}
    for index, raw_entry in enumerate(raw_registry):
        entry = require_mapping(
            raw_entry,
            f"platform/knowledge/source-registry.yaml.artifact_sources[{index}]",
        )
        repo_name = require_non_empty_str(
            entry.get("repo"),
            f"platform/knowledge/source-registry.yaml.artifact_sources[{index}].repo",
        )
        registry[repo_name] = {
            "service": require_non_empty_str(
                entry.get("service"),
                "platform/knowledge/source-registry.yaml.artifact_sources."
                f"[{index}].service",
            ),
            "source_ids": require_str_list(
                entry.get("source_ids", []),
                "platform/knowledge/source-registry.yaml.artifact_sources."
                f"[{index}].source_ids",
            ),
            "allowed_visibilities": require_str_list(
                entry.get("allowed_visibilities", []),
                "platform/knowledge/source-registry.yaml.artifact_sources."
                f"[{index}].allowed_visibilities",
            ),
        }
    return registry


def load_acl_policies(deploy_root: Path) -> dict[str, dict[str, object]]:
    """Load ACL policies keyed by visibility."""

    payload = load_yaml_mapping(deploy_root / ACL_POLICIES_RELATIVE)
    raw_policies = require_mapping(
        payload.get("policies"),
        "platform/knowledge/acl-policies.yaml.policies",
    )
    policies: dict[str, dict[str, object]] = {}
    for visibility, raw_policy in raw_policies.items():
        policy = require_mapping(
            raw_policy,
            f"platform/knowledge/acl-policies.yaml.policies.{visibility}",
        )
        policies[visibility] = {
            "acl_expression_template": require_non_empty_str(
                policy.get("acl_expression_template"),
                "platform/knowledge/acl-policies.yaml.policies."
                f"{visibility}.acl_expression_template",
            ),
            "allowed_targets": require_str_list(
                policy.get("allowed_targets", []),
                "platform/knowledge/acl-policies.yaml.policies."
                f"{visibility}.allowed_targets",
            ),
            "review_required": bool(policy.get("review_required", False)),
        }
    return policies


def load_publish_targets(deploy_root: Path) -> dict[str, dict[str, object]]:
    """Load publish targets keyed by target id."""

    payload = load_yaml_mapping(deploy_root / PUBLISH_TARGETS_RELATIVE)
    raw_targets = require_mapping(
        payload.get("targets"),
        "platform/knowledge/publish-targets.yaml.targets",
    )
    targets: dict[str, dict[str, object]] = {}
    for target_id, raw_target in raw_targets.items():
        target = require_mapping(
            raw_target, f"platform/knowledge/publish-targets.yaml.targets.{target_id}"
        )
        targets[target_id] = {
            "type": require_non_empty_str(
                target.get("type"),
                f"platform/knowledge/publish-targets.yaml.targets.{target_id}.type",
            ),
            "visibility": require_non_empty_str(
                target.get("visibility"),
                "platform/knowledge/publish-targets.yaml.targets."
                f"{target_id}.visibility",
            ),
            "dataset_name": optional_non_empty_str(
                target.get("dataset_name"),
                "platform/knowledge/publish-targets.yaml.targets."
                f"{target_id}.dataset_name",
            ),
            "indexing_technique": optional_non_empty_str(
                target.get("indexing_technique"),
                "platform/knowledge/publish-targets.yaml.targets."
                f"{target_id}.indexing_technique",
            ),
        }
    return targets


def load_publish_record(path: Path) -> JSONMapping:
    """Load one publish record file."""

    return load_json_mapping(path)


def load_publish_records(root: Path) -> dict[str, JSONMapping]:
    """Load all publish records under a directory."""

    if not root.exists():
        return {}
    records: dict[str, JSONMapping] = {}
    for path in sorted(root.rglob("*.json")):
        record = load_publish_record(path)
        document_id = require_non_empty_str(
            record.get("document_id"), f"{path}.document_id"
        )
        records[document_id] = record
    return records


def build_acl_expression(
    visibility: str, policies: dict[str, dict[str, object]]
) -> str:
    """Resolve the ACL expression for a visibility."""

    policy = policies.get(visibility)
    if policy is None:
        raise KnowledgeContractError(f"未注册 visibility ACL policy: {visibility}")
    acl_template = policy.get("acl_expression_template")
    return require_non_empty_str(acl_template, f"acl policy for {visibility}")


def build_event(
    *,
    action: str,
    artifact_id: str,
    document_id: str,
    source_repo: str,
    status: str,
    reason: str,
    target_ids: list[str],
) -> JSONMapping:
    """Build one append-only event record."""

    compact_time = utc_now().replace(":", "").replace("-", "")
    return {
        "event_id": f"knowledge-{action}-{document_id}-{compact_time}",
        "timestamp": utc_now(),
        "artifact_id": artifact_id,
        "document_id": document_id,
        "source_repo": source_repo,
        "action": action,
        "status": status,
        "reason": reason,
        "target_ids": target_ids,
    }


def artifact_document_name(document_id: str) -> str:
    """Return the stable remote document name used in Dify."""

    return document_id


def artifact_snapshot_root(
    deploy_root: Path,
    environment: str,
    source_repo: str,
    artifact_id: str,
) -> Path:
    """Return the snapshot root for one artifact."""

    return (
        knowledge_paths(deploy_root, environment).artifacts_root
        / repo_slug(source_repo)
        / artifact_id
    )


def record_path_for_document(
    deploy_root: Path,
    environment: str,
    source_repo: str,
    document_id: str,
) -> Path:
    """Return the publish-record path for a document."""

    return (
        knowledge_paths(deploy_root, environment).publish_records_root
        / repo_slug(source_repo)
        / f"{document_id}.json"
    )


def event_path_for_record(
    deploy_root: Path,
    environment: str,
    source_repo: str,
    event_id: str,
) -> Path:
    """Return the audit event path for one immutable knowledge event."""

    return (
        knowledge_paths(deploy_root, environment).events_root
        / repo_slug(source_repo)
        / f"{event_id}.json"
    )


def cast_mapping_list(value: object, field_path: str) -> list[JSONMapping]:
    """Cast a list payload into a list of mappings."""

    items = require_list(value, field_path)
    mappings: list[JSONMapping] = []
    for index, item in enumerate(items):
        mappings.append(require_mapping(item, f"{field_path}[{index}]"))
    return mappings


def optional_mapping(value: object, field_path: str) -> JSONMapping | None:
    """Return an optional mapping."""

    if value is None:
        return None
    return require_mapping(value, field_path)


def cast_str(value: object) -> str:
    """Cast a value known to be a string."""

    return cast(str, value)