src/p_formula_gate06.py

"""실제 P 연속 수식의 권리·분할·symbol-group 계약을 fail-closed로 검사한다."""

from __future__ import annotations

from collections import defaultdict
from collections.abc import Iterable, Sequence
from typing import Any


REQUIRED_SPLITS_06 = frozenset({"training", "validation", "test"})
REQUIRED_RECORD_FIELDS_06 = (
    "formula_id", "origin_id", "writer_id", "device_id", "source_id", "split",
    "canvas_width", "canvas_height", "symbols", "rights_track",
    "commercial_training_allowed",
)


def _identity_overlap06(records: Sequence[dict[str, Any]], field: str) -> dict[str, list[str]]:
    """필요 변수: formula record·identity field. 작동 원리: 하나의 identity가 둘 이상의 split에 나타난 경우만 반환한다."""

    split_by_identity: dict[str, set[str]] = defaultdict(set)
    for record in records:
        value = str(record.get(field) or "").strip()
        if value:
            split_by_identity[value].add(str(record.get("split") or ""))
    return {
        identity: sorted(splits)
        for identity, splits in split_by_identity.items()
        if len(splits) > 1
    }


def _valid_strokes06(strokes: Any) -> bool:
    """필요 변수: symbol stroke payload. 작동 원리: 비어 있지 않은 획과 유한 좌표 필드를 구조적으로 검사한다."""

    if not isinstance(strokes, list) or not strokes:
        return False
    for stroke in strokes:
        if not isinstance(stroke, list) or not stroke:
            return False
        for point in stroke:
            if not isinstance(point, dict) or "x" not in point or "y" not in point:
                return False
            try:
                x, y = float(point["x"]), float(point["y"])
            except (TypeError, ValueError):
                return False
            if not (-1e9 < x < 1e9 and -1e9 < y < 1e9):
                return False
    return True


def audit_p_formula_records06(
    records: Sequence[dict[str, Any]], *, minimum_independent_sources: int = 2,
) -> dict[str, Any]:
    """필요 변수: 실제 연속식 record. 작동 원리: P 권리·필수 metadata·identity 누수·group 정답을 한 번에 감사한다."""

    issues: list[dict[str, Any]] = []
    formula_ids: set[str] = set()
    origin_splits: dict[str, set[str]] = defaultdict(set)
    split_counts: dict[str, int] = defaultdict(int)
    symbol_counts: dict[str, int] = defaultdict(int)
    missing_timestamp_symbols = missing_pressure_symbols = total_symbols = 0

    for index, record in enumerate(records):
        missing = [
            field for field in REQUIRED_RECORD_FIELDS_06
            if field not in record or record[field] is None or record[field] == ""
        ]
        if missing:
            issues.append({"type": "missing_fields", "record_index": index, "fields": missing})
            continue
        formula_id = str(record["formula_id"])
        if formula_id in formula_ids:
            issues.append({"type": "duplicate_formula_id", "formula_id": formula_id})
        formula_ids.add(formula_id)
        split = str(record["split"])
        split_counts[split] += 1
        if split not in REQUIRED_SPLITS_06:
            issues.append({"type": "invalid_split", "formula_id": formula_id, "split": split})
        origin_splits[str(record["origin_id"])].add(split)
        if str(record["rights_track"]) != "P" or record["commercial_training_allowed"] is not True:
            issues.append({"type": "rights_not_product_approved", "formula_id": formula_id})
        try:
            if float(record["canvas_width"]) <= 0 or float(record["canvas_height"]) <= 0:
                raise ValueError
        except (TypeError, ValueError):
            issues.append({"type": "invalid_canvas", "formula_id": formula_id})

        symbols = record["symbols"]
        if not isinstance(symbols, list) or not symbols:
            issues.append({"type": "empty_symbol_groups", "formula_id": formula_id})
            continue
        for symbol_index, symbol in enumerate(symbols):
            total_symbols += 1
            token = str(symbol.get("token") or "").strip() if isinstance(symbol, dict) else ""
            if not token or not isinstance(symbol, dict) or not _valid_strokes06(symbol.get("strokes")):
                issues.append({
                    "type": "invalid_symbol_group", "formula_id": formula_id,
                    "symbol_index": symbol_index,
                })
                continue
            symbol_counts[token] += 1
            points = [point for stroke in symbol["strokes"] for point in stroke]
            if not all(point.get("t") is not None for point in points):
                missing_timestamp_symbols += 1
            if not all(point.get("pressure") is not None for point in points):
                missing_pressure_symbols += 1

    for origin, splits in origin_splits.items():
        if len(splits) > 1:
            issues.append({"type": "origin_split_leakage", "origin_id": origin, "splits": sorted(splits)})
    identity_overlap = {
        field: _identity_overlap06(records, field)
        for field in ("writer_id", "device_id", "source_id")
    }
    for field, overlaps in identity_overlap.items():
        if overlaps:
            issues.append({
                "type": f"{field}_split_leakage",
                "count": len(overlaps),
                "examples": dict(list(overlaps.items())[:10]),
            })
    missing_splits = sorted(REQUIRED_SPLITS_06 - set(split_counts))
    if missing_splits:
        issues.append({"type": "missing_required_splits", "splits": missing_splits})
    independent_sources = len({str(record.get("source_id") or "") for record in records if record.get("source_id")})
    if independent_sources < minimum_independent_sources:
        issues.append({
            "type": "insufficient_independent_sources",
            "observed": independent_sources, "required": minimum_independent_sources,
        })

    return {
        "schema": "aiflow-p-formula-preflight-v1",
        "records": len(records),
        "symbols": total_symbols,
        "split_counts": dict(sorted(split_counts.items())),
        "label_count": len(symbol_counts),
        "independent_sources": independent_sources,
        "identity_overlap_counts": {
            field: len(overlaps) for field, overlaps in identity_overlap.items()
        },
        "missing_metadata_slices": {
            "timestamp_symbols": missing_timestamp_symbols,
            "pressure_symbols": missing_pressure_symbols,
        },
        "issues": issues,
        "eligible_for_product_evaluation": not issues,
        "product_validation": False,
    }


def formula_boundary_candidates06(record: dict[str, Any]) -> list[dict[str, Any]]:
    """필요 변수: 검증된 formula record. 작동 원리: 실제 단일 group과 인접 두 group 결합을 boundary 정답 후보로 만든다."""

    symbols = record["symbols"]
    candidates = [
        {
            "formula_id": str(record["formula_id"]),
            "symbol_indices": [index],
            "strokes": symbol["strokes"],
            "boundary_target": 0,
        }
        for index, symbol in enumerate(symbols)
    ]
    for index in range(len(symbols) - 1):
        candidates.append({
            "formula_id": str(record["formula_id"]),
            "symbol_indices": [index, index + 1],
            "strokes": [*symbols[index]["strokes"], *symbols[index + 1]["strokes"]],
            "boundary_target": 1,
        })
    return candidates


def collect_formula_boundary_candidates06(
    records: Iterable[dict[str, Any]],
) -> list[dict[str, Any]]:
    """필요 변수: 검증된 formula iterable. 작동 원리: formula별 실제 group 후보를 평가용 평탄 목록으로 만든다."""

    return [
        candidate
        for record in records
        for candidate in formula_boundary_candidates06(record)
    ]