"""실제 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) ]