src/ink06_source_registry.py

"""Math Ink 0.6 데이터 출처의 발견·권리·중복·학습 준비 단계를 검증한다."""

from __future__ import annotations

from dataclasses import dataclass
import json
from pathlib import Path
from typing import Sequence


SOURCE_STAGES = ("discovered", "rights_review", "deduplicated", "approved")
TRAINING_ROLES = (
    "supervised_symbol", "geometry_pretrain", "raster_verifier", "raster_pseudo_stroke", "evaluation_only",
)
VOCABULARY_POLICIES = (
    "exact_378", "intersection_378", "digits_latin_intersection", "geometry_only", "evaluation_only",
)


@dataclass(frozen=True, slots=True)
class SourceRegistryEntry06:
    """필요 변수: 출처 권리·중복·로컬 상태. 작동 원리: 배포 checkpoint 유입 여부를 명시적인 한 행으로 표현한다."""

    source_id: str
    stage: str
    official_url: str
    license_id: str | None
    commercial_allowed: bool | None
    allowed_tracks: tuple[str, ...]
    independent_source_group: str | None
    deployment_role: str | None
    vocabulary_policy: str | None
    local_materialized: bool
    notes: str


def load_source_registry06(path: Path) -> tuple[SourceRegistryEntry06, ...]:
    """필요 변수: UTF-8 source registry 경로. 작동 원리: fail-closed schema 검증 후 불변 entry 묶음을 반환한다."""

    payload = json.loads(path.read_text(encoding="utf-8"))
    if payload.get("schema_version") != "0.6":
        raise ValueError("Math Ink source registry schema_version은 0.6이어야 합니다.")
    raw_sources = payload.get("sources")
    if not isinstance(raw_sources, list):
        raise ValueError("Math Ink source registry sources가 배열이 아닙니다.")
    entries: list[SourceRegistryEntry06] = []
    seen: set[str] = set()
    for raw in raw_sources:
        source_id = str(raw.get("source_id") or "").strip()
        if not source_id or source_id in seen:
            raise ValueError(f"source_id가 비었거나 중복입니다: {source_id!r}")
        seen.add(source_id)
        stage = str(raw.get("stage") or "")
        if stage not in SOURCE_STAGES:
            raise ValueError(f"{source_id}의 stage가 올바르지 않습니다: {stage}")
        official_url = str(raw.get("official_url") or "")
        if not official_url.startswith("https://"):
            raise ValueError(f"{source_id}의 공식 HTTPS URL이 없습니다.")
        license_id = raw.get("license_id")
        commercial_allowed = raw.get("commercial_allowed")
        group = raw.get("independent_source_group")
        role = raw.get("deployment_role")
        vocabulary_policy = raw.get("vocabulary_policy")
        if SOURCE_STAGES.index(stage) >= SOURCE_STAGES.index("rights_review"):
            if not license_id or not isinstance(commercial_allowed, bool):
                raise ValueError(f"{source_id}의 권리 검토 결과가 불완전합니다.")
        if SOURCE_STAGES.index(stage) >= SOURCE_STAGES.index("deduplicated") and not group:
            raise ValueError(f"{source_id}의 independent_source_group이 없습니다.")
        if stage == "approved":
            if commercial_allowed is not True or "P" not in raw.get("allowed_tracks", []):
                raise ValueError(f"{source_id}는 상용 P-track 승인 조건을 충족하지 못했습니다.")
            if role not in TRAINING_ROLES or vocabulary_policy not in VOCABULARY_POLICIES:
                raise ValueError(f"{source_id}의 배포 역할 또는 vocabulary 정책이 올바르지 않습니다.")
            if not bool(raw.get("local_materialized", False)):
                raise ValueError(f"{source_id}는 content dedup 전이라 approved가 될 수 없습니다.")
        entries.append(SourceRegistryEntry06(
            source_id=source_id, stage=stage, official_url=official_url,
            license_id=str(license_id) if license_id else None,
            commercial_allowed=commercial_allowed if isinstance(commercial_allowed, bool) else None,
            allowed_tracks=tuple(str(value) for value in raw.get("allowed_tracks", [])),
            independent_source_group=str(group) if group else None,
            deployment_role=str(role) if role else None,
            vocabulary_policy=str(vocabulary_policy) if vocabulary_policy else None,
            local_materialized=bool(raw.get("local_materialized", False)), notes=str(raw.get("notes") or ""),
        ))
    return tuple(entries)


def approved_training_source_ids06(
    entries: Sequence[SourceRegistryEntry06], *, role: str = "supervised_symbol",
) -> tuple[str, ...]:
    """필요 변수: 검증된 entry·학습 역할. 작동 원리: 승인·상용·로컬 준비 조건을 모두 만족한 출처만 반환한다."""

    if role not in TRAINING_ROLES:
        raise ValueError(f"지원하지 않는 deployment role입니다: {role}")
    return tuple(sorted(
        entry.source_id for entry in entries
        if entry.stage == "approved" and entry.commercial_allowed is True and "P" in entry.allowed_tracks
        and entry.local_materialized and entry.deployment_role == role
    ))


def source_registry_audit06(
    entries: Sequence[SourceRegistryEntry06],
    *,
    required_discovered_sources: int = 200,
    required_approved_groups: int = 30,
) -> dict:
    """필요 변수: 검증된 출처·목표 수. 작동 원리: 조사 수와 미러 제거 독립 배포 그룹을 별도 hard gate로 계산한다."""

    approved = [entry for entry in entries if entry.stage == "approved"]
    approved_deployment = [
        entry for entry in approved
        if entry.deployment_role != "evaluation_only"
    ]
    rights_cleared = [
        entry for entry in entries
        if entry.commercial_allowed is True and entry.license_id is not None
    ]
    groups = {entry.independent_source_group for entry in approved if entry.independent_source_group}
    deployment_groups = {
        entry.independent_source_group
        for entry in approved_deployment
        if entry.independent_source_group
    }
    group_members: dict[str, list[str]] = {}
    for entry in entries:
        if entry.independent_source_group:
            group_members.setdefault(entry.independent_source_group, []).append(entry.source_id)
    ready = approved_training_source_ids06(entries)
    materialized_by_role = {
        role: sorted(entry.source_id for entry in approved if entry.local_materialized and entry.deployment_role == role)
        for role in TRAINING_ROLES
    }
    return {
        "discovered_sources": len(entries),
        "required_discovered_sources": required_discovered_sources,
        "discovery_gate_passed": len(entries) >= required_discovered_sources,
        "stage_counts": {stage: sum(entry.stage == stage for entry in entries) for stage in SOURCE_STAGES},
        "approved_sources": len(approved),
        "approved_independent_groups": len(groups),
        "approved_deployment_sources": len(approved_deployment),
        "approved_deployment_independent_groups": len(deployment_groups),
        "required_approved_independent_groups": required_approved_groups,
        "approved_group_gate_passed": len(deployment_groups) >= required_approved_groups,
        "source_release_gate_passed": (
            len(entries) >= required_discovered_sources
            and len(deployment_groups) >= required_approved_groups
        ),
        "shared_independent_groups": {
            group: sorted(members)
            for group, members in sorted(group_members.items())
            if len(members) > 1
        },
        "rights_cleared_sources": sorted(entry.source_id for entry in rights_cleared),
        "rights_cleared_source_count": len(rights_cleared),
        "rights_cleared_pending_materialization": sorted(
            entry.source_id for entry in rights_cleared if entry.stage != "approved"
        ),
        "materialized_supervised_sources": list(ready),
        "materialized_supervised_source_count": len(ready),
        "materialized_by_role": materialized_by_role,
    }