#!/usr/bin/env python3 """ NEXT11 keypose-image handover utility. Purpose ------- Read a NEXT11 root and build an exercise/IK index that later tools can use for MediaPipe image candidate sampling. This script is intentionally read-only: it does not patch Nex, ota_youtube_proxy.php, runtime files, or canonical IK data. Output ------ JSON manifest with likely exercise records, IK/exercise files, keypose/phase hints, and confidence/provenance. """ from __future__ import annotations import argparse import csv import hashlib import json import os import re from dataclasses import dataclass, asdict from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple TEXT_EXTS = {".json", ".js", ".mjs", ".ts", ".csv", ".md", ".txt"} IGNORE_DIR_PARTS = { "node_modules", ".git", "vendor", "dist", "build", "cache", ".next", "three", "three_vendor", "__pycache__", "screenshots", "source_images" } EXERCISE_KEY_FIELDS = [ "exerciseKey", "exercise_key", "key", "slug", "id", "movementId", "movement_id", "name", "displayName", "title", "exercise", "exerciseName" ] IK_MARKER_FIELDS = [ "ikKeypoints", "keypoints", "landmarks", "targetsMeters", "sacredPoints", "lockedPoints", "contactContracts", "objectContracts", "phases", "phaseId", "mediapipe", "MediaPipe", "poseLandmarks" ] def sha1_file(path: Path, max_bytes: int = 2_000_000) -> str: h = hashlib.sha1() with path.open("rb") as f: remaining = max_bytes while remaining > 0: chunk = f.read(min(65536, remaining)) if not chunk: break h.update(chunk) remaining -= len(chunk) return h.hexdigest() def should_scan(path: Path) -> bool: if path.suffix.lower() not in TEXT_EXTS: return False parts = set(path.parts) if parts & IGNORE_DIR_PARTS: return False try: return path.stat().st_size <= 8_000_000 except OSError: return False def safe_read_text(path: Path) -> str: for enc in ("utf-8", "utf-8-sig", "latin-1"): try: return path.read_text(encoding=enc, errors="replace") except Exception: pass return "" def normalize_key(value: Any) -> Optional[str]: if value is None: return None s = str(value).strip() if not s: return None s = re.sub(r"[^A-Za-z0-9]+", "_", s).strip("_").lower() return s or None def walk_dict(obj: Any, path: str = "") -> Iterable[Tuple[str, Any]]: yield path, obj if isinstance(obj, dict): for k, v in obj.items(): next_path = f"{path}.{k}" if path else str(k) yield from walk_dict(v, next_path) elif isinstance(obj, list): for i, v in enumerate(obj): yield from walk_dict(v, f"{path}[{i}]") def extract_candidate_records_from_json(data: Any, source: Path) -> List[Dict[str, Any]]: records: List[Dict[str, Any]] = [] def is_exercise_like(d: Dict[str, Any]) -> bool: keys = set(d.keys()) has_name = any(k in keys for k in EXERCISE_KEY_FIELDS) has_ik = any(k in keys for k in IK_MARKER_FIELDS) has_exercise_shape = bool({"movementFamily", "equipment", "forbiddenBehaviors", "tolerances"} & keys) return has_name and (has_ik or has_exercise_shape or len(keys) >= 4) for obj_path, obj in walk_dict(data): if not isinstance(obj, dict): continue if not is_exercise_like(obj): continue key = None display = None for field in EXERCISE_KEY_FIELDS: if field in obj: cand = normalize_key(obj.get(field)) if cand and not key: key = cand if obj.get(field) and not display: display = str(obj.get(field)).strip() if not key: continue phases = [] if isinstance(obj.get("phases"), list): for p in obj.get("phases", []): if isinstance(p, str): phases.append(p) elif isinstance(p, dict): phases.append(str(p.get("phaseId") or p.get("id") or p.get("name") or "phase")) sacred = obj.get("sacredPoints") if isinstance(obj.get("sacredPoints"), list) else [] locked = obj.get("lockedPoints") if isinstance(obj.get("lockedPoints"), list) else [] equipment = obj.get("equipment") if isinstance(obj.get("equipment"), list) else [] movement_family = obj.get("movementFamily") if isinstance(obj.get("movementFamily"), list) else [] marker_hits = [m for m in IK_MARKER_FIELDS if m in obj] records.append({ "exerciseKey": key, "displayName": display or key, "sourceFile": str(source), "jsonPath": obj_path, "movementFamily": movement_family, "equipment": equipment, "phases": phases, "sacredPointCount": len(sacred), "lockedPointCount": len(locked), "ikMarkerFields": marker_hits, "candidateConfidence": 0.9 if marker_hits else 0.65, }) return records def extract_json_blobs_from_js(text: str) -> List[Any]: """Best-effort extractor for common JS assignment/export JSON-ish blobs.""" blobs: List[Any] = [] # Prefer explicit arrays/objects around exercise-like terms to avoid parsing whole bundles. patterns = [ r"(?:export\s+default|module\.exports\s*=|const\s+\w+\s*=|window\.\w+\s*=)\s*(\[.*?\]);", r"(?:export\s+default|module\.exports\s*=|const\s+\w+\s*=|window\.\w+\s*=)\s*(\{.*?\});", ] for pat in patterns: for m in re.finditer(pat, text, flags=re.S): raw = m.group(1) if not any(tok in raw for tok in ("exercise", "Exercise", "ik", "IK", "phases", "targetsMeters")): continue cleaned = re.sub(r",\s*([}\]])", r"\1", raw) try: blobs.append(json.loads(cleaned)) except Exception: continue return blobs def regex_extract_names(text: str, source: Path) -> List[Dict[str, Any]]: records: List[Dict[str, Any]] = [] # Conservative fallback: capture exerciseKey/displayName pairs if found close together. for m in re.finditer(r'"exerciseKey"\s*:\s*"([^"]+)"', text): key = normalize_key(m.group(1)) if key: window = text[max(0, m.start()-800):m.end()+800] dm = re.search(r'"displayName"\s*:\s*"([^"]+)"', window) records.append({ "exerciseKey": key, "displayName": dm.group(1) if dm else key, "sourceFile": str(source), "jsonPath": "regex:exerciseKey", "movementFamily": [], "equipment": [], "phases": [], "sacredPointCount": len(re.findall(r'"sacredPoints"', window)), "lockedPointCount": len(re.findall(r'"lockedPoints"', window)), "ikMarkerFields": [h for h in IK_MARKER_FIELDS if h in window], "candidateConfidence": 0.55, }) return records def parse_csv(path: Path) -> List[Dict[str, Any]]: records: List[Dict[str, Any]] = [] text = safe_read_text(path) try: sample = text[:2048] dialect = csv.Sniffer().sniff(sample) except Exception: dialect = csv.excel try: rows = csv.DictReader(text.splitlines(), dialect=dialect) for i, row in enumerate(rows): key = None display = None for field in EXERCISE_KEY_FIELDS: for actual in row.keys(): if actual and actual.lower() == field.lower(): val = row.get(actual) key = key or normalize_key(val) display = display or (str(val).strip() if val else None) if key: records.append({ "exerciseKey": key, "displayName": display or key, "sourceFile": str(path), "jsonPath": f"csv:row[{i}]", "movementFamily": [], "equipment": [], "phases": [], "sacredPointCount": 0, "lockedPointCount": 0, "ikMarkerFields": [], "candidateConfidence": 0.5, }) except Exception: pass return records def scan_root(root: Path) -> Dict[str, Any]: records: List[Dict[str, Any]] = [] files_scanned = [] ik_files = [] for path in root.rglob("*"): if not path.is_file() or not should_scan(path): continue rel = path.relative_to(root) text = safe_read_text(path) if not text: continue lowered = text.lower() marker_score = sum(1 for m in ["exercisekey", "targetsmeters", "ikkeypoints", "sacredpoints", "mediapipe", "phases"] if m.lower() in lowered) if marker_score: ik_files.append({"path": str(rel), "markerScore": marker_score, "sha1": sha1_file(path)}) files_scanned.append(str(rel)) if path.suffix.lower() == ".json": try: data = json.loads(text) records.extend(extract_candidate_records_from_json(data, path)) except Exception: records.extend(regex_extract_names(text, path)) elif path.suffix.lower() == ".csv": records.extend(parse_csv(path)) else: for blob in extract_json_blobs_from_js(text): records.extend(extract_candidate_records_from_json(blob, path)) records.extend(regex_extract_names(text, path)) merged: Dict[str, Dict[str, Any]] = {} for rec in records: key = rec["exerciseKey"] if key not in merged: merged[key] = rec merged[key]["sources"] = [{"file": rec["sourceFile"], "path": rec["jsonPath"], "confidence": rec["candidateConfidence"]}] else: cur = merged[key] cur["sources"].append({"file": rec["sourceFile"], "path": rec["jsonPath"], "confidence": rec["candidateConfidence"]}) cur["candidateConfidence"] = max(cur.get("candidateConfidence", 0), rec.get("candidateConfidence", 0)) for list_field in ("phases", "movementFamily", "equipment", "ikMarkerFields"): seen = list(cur.get(list_field, [])) for v in rec.get(list_field, []): if v not in seen: seen.append(v) cur[list_field] = seen cur["sacredPointCount"] = max(cur.get("sacredPointCount", 0), rec.get("sacredPointCount", 0)) cur["lockedPointCount"] = max(cur.get("lockedPointCount", 0), rec.get("lockedPointCount", 0)) exercises = sorted(merged.values(), key=lambda r: r["exerciseKey"]) return { "schema": "next11.keypose_image.exercise_ik_index.v1", "root": str(root.resolve()), "generatedBy": "discover_next11_ik_data.py", "mode": "read_only_no_runtime_patch", "summary": { "filesScanned": len(files_scanned), "ikLikeFiles": len(ik_files), "exerciseCandidates": len(exercises), }, "ikLikeFiles": ik_files, "exercises": exercises, "notes": [ "This is a discovery index, not a truth source.", "MediaPipe image sampling must write candidates only; IK approval remains separate.", "Do not patch ota_youtube_proxy.php or Nex runtime from this tool." ] } def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("root", help="NEXT11 server/root directory") ap.add_argument("--out", default="tools/keypose_image_sampler/out/exercise_ik_index.json") args = ap.parse_args() root = Path(args.root).resolve() manifest = scan_root(root) out = Path(args.out) if not out.is_absolute(): out = root / out out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8") print(json.dumps(manifest["summary"], indent=2, ensure_ascii=False)) print(f"Wrote {out}") if __name__ == "__main__": main()