src/p_formula_intake06.py

"""사람이 확정한 AIFlow Ink v1 수식만 P Formula v1 학습 후보로 변환한다."""

from __future__ import annotations

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


TRAINABLE_LABEL_STATUS06 = frozenset({"human_verified"})
SPLIT_MAP06 = {
    "train": "training",
    "training": "training",
    "validation": "validation",
    "test": "test",
}


@dataclass(frozen=True, slots=True)
class PFormulaAnnotation06:
    """필요 변수: sample·기기·cell token. 작동 원리: raw ink와 분리된 사람 확정 label 계약을 보존한다."""

    sample_id: str
    device_id: str
    tokens: dict[str, str]
    label_status: str = "human_verified"


def parse_p_formula_annotation06(value: dict[str, Any]) -> PFormulaAnnotation06:
    """필요 변수: annotation JSON object. 작동 원리: 비어 있는 ID·token과 자동 label을 fail-closed로 거부한다."""

    sample_id = str(value.get("sample_id") or "").strip()
    device_id = str(value.get("device_id") or "").strip()
    label_status = str(value.get("label_status") or "").strip()
    raw_tokens = value.get("tokens")
    if not sample_id or not device_id:
        raise ValueError("annotation에는 sample_id와 실제 device_id가 필요합니다.")
    if label_status not in TRAINABLE_LABEL_STATUS06:
        raise ValueError("P Formula token은 human_verified 상태만 허용합니다.")
    if not isinstance(raw_tokens, dict) or not raw_tokens:
        raise ValueError("annotation tokens는 비어 있지 않은 cell_id→token 객체여야 합니다.")
    tokens = {
        str(cell_id).strip(): str(token).strip()
        for cell_id, token in raw_tokens.items()
    }
    if any(not cell_id or not token for cell_id, token in tokens.items()):
        raise ValueError("annotation cell ID와 token은 비어 있을 수 없습니다.")
    if len(tokens) != len(raw_tokens):
        raise ValueError("정규화 후 중복되는 annotation cell ID가 있습니다.")
    return PFormulaAnnotation06(
        sample_id=sample_id,
        device_id=device_id,
        tokens=tokens,
        label_status=label_status,
    )


def read_p_formula_annotations06(path: Path) -> dict[str, PFormulaAnnotation06]:
    """필요 변수: UTF-8 annotation JSONL. 작동 원리: 행 위치 오류와 sample 중복을 명시하며 전체를 읽는다."""

    annotations: dict[str, PFormulaAnnotation06] = {}
    for line_number, raw in enumerate(
        path.read_text(encoding="utf-8").splitlines(),
        start=1,
    ):
        if not raw.strip():
            continue
        try:
            value = json.loads(raw)
        except json.JSONDecodeError as error:
            raise ValueError(
                f"{path}:{line_number} UTF-8 JSON 파싱 실패: {error.msg}",
            ) from error
        if not isinstance(value, dict):
            raise ValueError(f"{path}:{line_number} annotation이 JSON object가 아닙니다.")
        try:
            annotation = parse_p_formula_annotation06(value)
        except ValueError as error:
            raise ValueError(f"{path}:{line_number} {error}") from error
        if annotation.sample_id in annotations:
            raise ValueError(f"중복 annotation sample_id입니다: {annotation.sample_id}")
        annotations[annotation.sample_id] = annotation
    if not annotations:
        raise ValueError("P Formula annotation이 없습니다.")
    return annotations


def _point06(point: dict[str, Any]) -> dict[str, float | None]:
    """필요 변수: Ink v1 point. 작동 원리: 좌표·관측 timestamp·pressure를 P Formula 이름으로 손실 없이 옮긴다."""

    try:
        x, y = float(point["x"]), float(point["y"])
    except (KeyError, TypeError, ValueError) as error:
        raise ValueError("모든 point에는 유효한 x/y가 필요합니다.") from error
    timestamp = point.get("t_ms")
    pressure = point.get("pressure")
    return {
        "x": x,
        "y": y,
        "t": None if timestamp is None else float(timestamp),
        "pressure": None if pressure is None else float(pressure),
    }


def build_p_formula_record06(
    intake_record: dict[str, Any],
    annotation: PFormulaAnnotation06,
    *,
    allowed_labels: Sequence[str] | None = None,
) -> dict[str, Any]:
    """필요 변수: curated Ink v1·사람 annotation·선택 vocabulary. 작동 원리: cell/stroke/token 전단사만 P Formula v1로 변환한다."""

    sample_id = str(intake_record.get("sample_id") or "").strip()
    if sample_id != annotation.sample_id:
        raise ValueError("intake sample_id와 annotation sample_id가 다릅니다.")
    if str(intake_record.get("format")) != "aiflow-ink/v1":
        raise ValueError("AIFlow Ink v1 원본만 P Formula v1로 변환할 수 있습니다.")
    if annotation.label_status not in TRAINABLE_LABEL_STATUS06:
        raise ValueError("사람이 확정하지 않은 token은 학습 후보로 변환할 수 없습니다.")
    consent = str(intake_record.get("consent_scope") or "")
    if not consent.startswith("model_training"):
        raise ValueError("model_training 동의가 없는 intake record입니다.")
    source_id = str(intake_record.get("source") or "").strip()
    writer_id = str(intake_record.get("writer_hash") or "").strip()
    split = SPLIT_MAP06.get(str(intake_record.get("split") or ""))
    canvas = intake_record.get("canvas")
    if not source_id or not writer_id or split is None:
        raise ValueError("source·writer·training/validation/test split이 필요합니다.")
    if not isinstance(canvas, dict):
        raise ValueError("canvas 객체가 필요합니다.")
    try:
        canvas_width = float(canvas["width"])
        canvas_height = float(canvas["height"])
    except (KeyError, TypeError, ValueError) as error:
        raise ValueError("유효한 canvas width/height가 필요합니다.") from error
    if canvas_width <= 0 or canvas_height <= 0:
        raise ValueError("canvas width/height는 양수여야 합니다.")

    raw_strokes = intake_record.get("strokes")
    cells = intake_record.get("formula_cells")
    if not isinstance(raw_strokes, list) or not raw_strokes:
        raise ValueError("비어 있지 않은 raw strokes가 필요합니다.")
    if not isinstance(cells, list) or not cells:
        raise ValueError("비어 있지 않은 formula_cells가 필요합니다.")
    stroke_map: dict[int, dict[str, Any]] = {}
    for stroke in raw_strokes:
        if not isinstance(stroke, dict):
            raise ValueError("stroke는 JSON object여야 합니다.")
        stroke_id = int(stroke["stroke_id"])
        if stroke_id in stroke_map:
            raise ValueError(f"중복 stroke_id입니다: {stroke_id}")
        stroke_map[stroke_id] = stroke

    cell_ids = [str(cell.get("formula_id") or "").strip() for cell in cells]
    if any(not cell_id for cell_id in cell_ids) or len(set(cell_ids)) != len(cell_ids):
        raise ValueError("formula cell ID는 비어 있지 않고 고유해야 합니다.")
    if set(cell_ids) != set(annotation.tokens):
        missing = sorted(set(cell_ids) - set(annotation.tokens))
        extra = sorted(set(annotation.tokens) - set(cell_ids))
        raise ValueError(f"cell token 전단사가 아닙니다: missing={missing}, extra={extra}")
    allowed = None if allowed_labels is None else set(str(label) for label in allowed_labels)
    unknown = sorted({
        annotation.tokens[cell_id]
        for cell_id in cell_ids
        if allowed is not None and annotation.tokens[cell_id] not in allowed
    })
    if unknown:
        raise ValueError(f"0.6 vocabulary 밖 token입니다: {unknown}")

    assigned: list[int] = []
    symbols = []
    for cell, cell_id in zip(cells, cell_ids, strict=True):
        raw_ids = cell.get("stroke_ids")
        if not isinstance(raw_ids, list) or not raw_ids:
            raise ValueError(f"cell {cell_id}에 stroke가 없습니다.")
        stroke_ids = [int(value) for value in raw_ids]
        missing_strokes = sorted(set(stroke_ids) - set(stroke_map))
        if missing_strokes:
            raise ValueError(f"cell {cell_id}가 없는 stroke를 참조합니다: {missing_strokes}")
        assigned.extend(stroke_ids)
        ordered = sorted(
            (stroke_map[stroke_id] for stroke_id in stroke_ids),
            key=lambda stroke: (int(stroke.get("order", 0)), int(stroke["stroke_id"])),
        )
        converted_strokes = []
        for stroke in ordered:
            points = stroke.get("points")
            if not isinstance(points, list) or not points:
                raise ValueError(f"stroke {stroke['stroke_id']}에 point가 없습니다.")
            converted_strokes.append([_point06(point) for point in points])
        symbols.append({
            "token": annotation.tokens[cell_id],
            "strokes": converted_strokes,
            "source_cell_id": cell_id,
        })
    if len(assigned) != len(set(assigned)) or set(assigned) != set(stroke_map):
        raise ValueError("모든 raw stroke는 정확히 한 symbol cell에 속해야 합니다.")

    return {
        "formula_id": sample_id,
        "origin_id": f"{source_id}:{sample_id}",
        "writer_id": writer_id,
        "device_id": annotation.device_id,
        "source_id": source_id,
        "split": split,
        "canvas_width": canvas_width,
        "canvas_height": canvas_height,
        "symbols": symbols,
        "rights_track": "P",
        "commercial_training_allowed": True,
        "license_id": str(intake_record.get("license_id") or ""),
        "label_status": annotation.label_status,
    }


def materialize_p_formula_records06(
    intake_records: Iterable[dict[str, Any]],
    annotations: dict[str, PFormulaAnnotation06],
    *,
    allowed_labels: Sequence[str] | None = None,
) -> list[dict[str, Any]]:
    """필요 변수: intake 순회열·sample별 annotation. 작동 원리: 양쪽 sample 집합이 정확히 같을 때만 순서를 보존해 변환한다."""

    records = list(intake_records)
    sample_ids = [str(record.get("sample_id") or "").strip() for record in records]
    if any(not sample_id for sample_id in sample_ids):
        raise ValueError("모든 intake record에 sample_id가 필요합니다.")
    if len(set(sample_ids)) != len(sample_ids):
        raise ValueError("중복 intake sample_id가 있습니다.")
    if set(sample_ids) != set(annotations):
        missing = sorted(set(sample_ids) - set(annotations))
        extra = sorted(set(annotations) - set(sample_ids))
        raise ValueError(f"intake/annotation sample 집합이 다릅니다: missing={missing}, extra={extra}")
    return [
        build_p_formula_record06(
            record,
            annotations[sample_id],
            allowed_labels=allowed_labels,
        )
        for record, sample_id in zip(records, sample_ids, strict=True)
    ]


def write_p_formula_jsonl06(
    path: Path,
    records: Sequence[dict[str, Any]],
    *,
    overwrite: bool = False,
) -> None:
    """필요 변수: 출력 경로·검증된 record. 작동 원리: UTF-8 임시 파일을 원자 교체해 부분 JSONL을 남기지 않는다."""

    if not records:
        raise ValueError("저장할 P Formula record가 없습니다.")
    if path.exists() and not overwrite:
        raise FileExistsError(f"기존 P Formula 파일을 덮어쓰지 않습니다: {path}")
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".part")
    payload = "".join(
        json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n"
        for record in records
    )
    try:
        temporary.write_text(payload, encoding="utf-8", newline="\n")
        temporary.replace(path)
    finally:
        if temporary.exists():
            temporary.unlink()