"""Data loaders for the sync_pilot dashboard. All loaders are Streamlit-cached so multiple pages and reruns are cheap. The ``SYNC_PILOT_DATA_SOURCE`` env var picks between ``"local"`` (default — read from the on-disk ``data/outputs/median`` directory under the package project root) and ``"hf"`` (snapshot-download from the private HuggingFace dataset and read from the local cache). Page code never inspects the source. Lightweight markdown parsing for the taxonomy lives here too — we extract structured fields per dimension so the taxonomy browser can render rich controls rather than just dumping the raw text. The parser falls back to raw markdown for any section it can't decompose, so a malformed section never blocks rendering of the rest. """ from __future__ import annotations import json import os import re from dataclasses import dataclass, field from datetime import datetime, timezone from functools import lru_cache from pathlib import Path from typing import Any, Literal import streamlit as st from sync_pilot import config from sync_pilot.schema import TrackRecord SummaryName = Literal["tagging", "clap", "description", "transcription"] _SUMMARY_FILENAMES: dict[str, str] = { "tagging": "_batch_summary.json", "clap": "_clap_summary.json", "description": "_description_summary.json", "transcription": "_transcription_summary.json", } # --------------------------------------------------------------------------- # Local-vs-HF data source resolution # --------------------------------------------------------------------------- def _data_source() -> str: """Return ``"local"`` (default) or ``"hf"``. Setting ``SYNC_PILOT_DATA_SOURCE=hf`` triggers a snapshot-download of the private dataset on first call (cached for the session via ``lru_cache``) and routes every subsequent loader to the local snapshot directory. """ return os.getenv("SYNC_PILOT_DATA_SOURCE", "local").strip().lower() @lru_cache(maxsize=1) def _resolved_data_root() -> Path: """Where data lives on disk for this session. Local mode: ``/data``. HF mode: a directory inside ``HF_CACHE_DIR / "sync_pilot_dashboard"`` that we populate via ``huggingface_hub.snapshot_download``. The HF snapshot mirrors the project's ``data/`` subtree under a ``sync_pilot/`` prefix inside the repo (see the ``publish`` CLI), so we point callers at ``/sync_pilot`` to keep the on-disk layout identical to local. """ mode = _data_source() if mode == "local": return config.DATA_DIR if mode != "hf": raise RuntimeError( f"SYNC_PILOT_DATA_SOURCE must be 'local' or 'hf', got {mode!r}" ) # Lazy import so local-only users don't pay the import cost. from huggingface_hub import snapshot_download repo_id = os.getenv("PRIVATE_DATASET_REPO", "").strip() if not repo_id: raise RuntimeError( "SYNC_PILOT_DATA_SOURCE=hf but PRIVATE_DATASET_REPO is unset" ) revision = os.getenv("PRIVATE_DATASET_REVISION", "main").strip() or "main" token = os.getenv("HF_TOKEN") or None local_dir = config.HF_CACHE_DIR / "sync_pilot_dashboard" / repo_id.replace("/", "__") local_dir.mkdir(parents=True, exist_ok=True) snapshot_dir = snapshot_download( repo_id=repo_id, repo_type="dataset", revision=revision, token=token, local_dir=str(local_dir), # ``_audit_hf.jsonl`` is a Space-only sidecar (curator edits made # from the HF dashboard). It must come down on boot so we append # rather than overwriting previous Space-side history. # ``_review.json`` is the INFER-curation sidecar; same story. allow_patterns=[ "sync_pilot/outputs/median/*.json", "sync_pilot/groundtruth/taxonomy.md", "sync_pilot/groundtruth/median/*.json", "sync_pilot/groundtruth/median/_audit_hf.jsonl", "sync_pilot/audio/median/manifest.jsonl", "sync_pilot/audio/median/*.m4a", # Extension (GT-expansion) set — display + edit in the GT-review # "Extension" tab. ``_audit_hf.jsonl`` is the Space-only edit # history; it must come down on boot so writebacks append rather # than overwrite prior Space-side edits (same story as median). "sync_pilot/outputs/gt_expansion/median_adjacent_combined_500/*.json", "sync_pilot/gt_expansion/median_adjacent_combined_500/groundtruth/*.json", "sync_pilot/gt_expansion/median_adjacent_combined_500/groundtruth/_audit_hf.jsonl", "sync_pilot/gt_expansion/median_adjacent_combined_500/manifest.jsonl", "sync_pilot/gt_expansion/median_adjacent_combined_500/_review.json", # Cohort map (catalog / ext / ext-median) — without it load_cohorts() # is empty and the GT-review Ext-Median tab can't appear. "sync_pilot/gt_ingest/cohorts.json", ], ) # The downloaded snapshot mirrors the repo tree, so the project-equivalent # data root is the ``sync_pilot`` subdirectory inside it. return Path(snapshot_dir) / "sync_pilot" def _outputs_dir() -> Path: return _resolved_data_root() / "outputs" / "median" def _groundtruth_dir() -> Path: return _resolved_data_root() / "groundtruth" / "median" def _hf_writeback( files: list[tuple[Path, str]], *, commit_message: str, ) -> bool: """Push one or more locally-edited files to the private dataset. ``files`` is ``[(local_path, path_in_repo), ...]``. We use ``HfApi.create_commit`` with ``CommitOperationAdd`` operations so the set lands as a single atomic commit (e.g. updated track JSON + audit log entry in one shot). Returns True on success, False on failure; callers surface a Streamlit warning so the curator knows the local edit succeeded but the push didn't (state will revert on Space restart in that case). Lazy imports ``huggingface_hub`` so local-mode users never pay the import cost and our ``BlockTorch``-style smoke tests stay clean. """ repo_id = os.getenv("PRIVATE_DATASET_REPO", "").strip() token = os.getenv("HF_TOKEN") or None if not repo_id or not token: st.warning( "HF writeback skipped — PRIVATE_DATASET_REPO or HF_TOKEN not set " "in the Space environment. Edit persisted to the snapshot cache " "only and will be lost on Space restart." ) return False revision = os.getenv("PRIVATE_DATASET_REVISION", "main").strip() or "main" try: from huggingface_hub import CommitOperationAdd, HfApi except Exception as exc: # noqa: BLE001 st.error(f"HF writeback failed: huggingface_hub import error ({exc})") return False operations = [ CommitOperationAdd(path_in_repo=repo_path, path_or_fileobj=str(local)) for local, repo_path in files if local.exists() ] if not operations: return False try: HfApi(token=token).create_commit( repo_id=repo_id, repo_type="dataset", revision=revision, operations=operations, commit_message=commit_message, ) return True except Exception as exc: # noqa: BLE001 st.error(f"HF writeback failed: {exc}") return False def _review_local_path() -> Path: """Where the GT-review sidecar is written. Always under the LOCAL data dir — never inside the HF snapshot cache (which is read-only). The reader still consults _resolved_data_root() first so an HF snapshot can bundle a published review state.""" return config.DATA_DIR / "groundtruth" / "median" / "_review.json" def _review_read_path() -> Path: return _resolved_data_root() / "groundtruth" / "median" / "_review.json" def _manifest_path() -> Path: return _resolved_data_root() / "audio" / "median" / "manifest.jsonl" # Expansion paths resolve via ``_resolved_data_root()`` so the GT-review # "Extension" tab reads the same way the catalog does: from the on-disk # ``data/`` tree locally, and from the downloaded HF snapshot in ``hf`` mode. # In local mode ``_resolved_data_root()`` == ``config.DATA_DIR``, so existing # local read/write behaviour is unchanged. def _expansion_outputs_dir() -> Path: return _resolved_data_root() / "outputs" / "gt_expansion" / "median_adjacent_combined_500" def _expansion_groundtruth_dir() -> Path: return ( _resolved_data_root() / "gt_expansion" / "median_adjacent_combined_500" / "groundtruth" ) def _expansion_review_path() -> Path: return ( _resolved_data_root() / "gt_expansion" / "median_adjacent_combined_500" / "_review.json" ) def _expansion_manifest_path() -> Path: return ( _resolved_data_root() / "gt_expansion" / "median_adjacent_combined_500" / "manifest.jsonl" ) def _taxonomy_path() -> Path: # The taxonomy lives under the package source tree, not under data/, so # it ships with the install and is the same in local & HF modes (the HF # publish copies it into the dataset alongside outputs/ so the dashboard # works even when run with no source checkout). local = config.PACKAGE_ROOT / "groundtruth" / "taxonomy.md" if local.exists(): return local # HF mode fallback — taxonomy.md was published to the dataset. return _resolved_data_root() / "groundtruth" / "taxonomy.md" # --------------------------------------------------------------------------- # Track + summary loaders # --------------------------------------------------------------------------- @st.cache_data(ttl=300, show_spinner=False) def load_tracks() -> list[dict[str, Any]]: """Load every per-track JSON, validated through the schema. Returns dicts (not the Pydantic model itself) because Streamlit's data cache works best on JSON-serialisable types and pages downstream don't need the Pydantic features. We still round-trip through ``TrackRecord`` so a structurally bad file fails loudly here rather than crashing a page deep in rendering. """ out: list[dict[str, Any]] = [] outputs_dir = _outputs_dir() if not outputs_dir.exists(): return out for p in sorted(outputs_dir.glob("*.json")): if p.name.startswith("_"): continue try: raw = json.loads(p.read_text()) # Validate but don't drop fields we didn't model — the schema is # ``extra='forbid'`` for the record itself but ``metadata`` is # ``dict[str, Any]`` so all the provenance keys survive. TrackRecord.model_validate(raw) out.append(raw) except Exception as e: # noqa: BLE001 — render a warning, keep going st.warning(f"Skipping unparseable track {p.name}: {e}") continue out.sort(key=lambda r: r.get("track_id", "")) return out @st.cache_data(ttl=300, show_spinner=False) def load_summary(name: SummaryName) -> dict[str, Any]: """Load one of the four ``_*_summary.json`` files. Returns an empty dict if the summary is missing (e.g. that stage hasn't been run yet in the snapshot) so pages can degrade gracefully. """ fname = _SUMMARY_FILENAMES[name] p = _outputs_dir() / fname if not p.exists(): return {} try: return json.loads(p.read_text()) except Exception as e: # noqa: BLE001 st.warning(f"Could not parse {fname}: {e}") return {} @st.cache_data(ttl=300, show_spinner=False) def load_groundtruth() -> dict[str, dict[str, Any]]: """Load every ``GroundTruthRecord`` JSON under ``groundtruth/median/``, keyed by ``track_id`` for O(1) lookup from the Tracks page. Returns plain dicts (not the Pydantic model) for the same reason ``load_tracks()`` does — Streamlit's data cache works best on JSON- serialisable types. We don't round-trip through the Pydantic schema here because the GT records are written by sync_pilot's own extractor (already validated on write) and the dashboard never mutates them. A bad file still degrades gracefully — we just skip it with a warning. Returns ``{}`` when the directory doesn't exist (e.g. a fresh checkout that hasn't run ``research-batch`` yet), so caller code can check ``gt_by_id.get(track_id)`` without further guards. """ out: dict[str, dict[str, Any]] = {} gt_dir = _groundtruth_dir() if not gt_dir.exists(): return out for p in sorted(gt_dir.glob("*.json")): if p.name.startswith("_"): continue try: raw = json.loads(p.read_text()) except Exception as e: # noqa: BLE001 st.warning(f"Skipping unparseable ground-truth {p.name}: {e}") continue tid = raw.get("track_id") if tid: out[tid] = raw return out @st.cache_data(ttl=300, show_spinner=False) def load_expansion_tracks() -> list[dict[str, Any]]: """Load expansion TrackRecords for read-only GT review display.""" out: list[dict[str, Any]] = [] outputs_dir = _expansion_outputs_dir() if not outputs_dir.exists(): return out for p in sorted(outputs_dir.glob("*.json")): if p.name.startswith("_"): continue try: raw = json.loads(p.read_text()) TrackRecord.model_validate(raw) out.append(raw) except Exception as e: # noqa: BLE001 st.warning(f"Skipping unparseable expansion track {p.name}: {e}") continue out.sort(key=lambda r: r.get("track_id", "")) return out @st.cache_data(ttl=300, show_spinner=False) def load_expansion_groundtruth() -> dict[str, dict[str, Any]]: """Load metadata GT records for the expansion set.""" out: dict[str, dict[str, Any]] = {} gt_dir = _expansion_groundtruth_dir() if not gt_dir.exists(): return out for p in sorted(gt_dir.glob("*.json")): if p.name.startswith("_"): continue try: raw = json.loads(p.read_text()) except Exception as e: # noqa: BLE001 st.warning(f"Skipping unparseable expansion ground-truth {p.name}: {e}") continue tid = raw.get("track_id") if tid: out[tid] = raw return out @st.cache_data(ttl=300, show_spinner=False) def load_expansion_triage() -> dict[str, dict[str, Any]]: """Load the ``_triage.json`` review-priority ranking for the expansion set. Produced by ``scripts/triage_disagreement.py`` — a per-track priority score (model disagreement + uncertainty) used to order the GT-review queue so the highest-information tracks surface first. Returns ``{track_id: {priority, disagreement, uncertainty, reasons, ...}}``; ``{}`` when the sidecar is absent (triage not yet run / not in the snapshot), so the page degrades to plain alphabetical order. The ``_triage.json`` sidecar rides the same ``outputs/.../*.json`` publish + snapshot globs as the TrackRecords. """ path = _expansion_outputs_dir() / "_triage.json" if not path.exists(): return {} try: data = json.loads(path.read_text()) except Exception as e: # noqa: BLE001 st.warning(f"Could not parse _triage.json: {e}") return {} out: dict[str, dict[str, Any]] = {} for row in data.get("ranked", []): tid = row.get("track_id") if tid: out[tid] = row return out @st.cache_data(ttl=300, show_spinner=False) def load_cohorts() -> dict[str, str]: """Load the ``gt_ingest/cohorts.json`` track→cohort map. Cohorts: ``catalog`` (Median eval set), ``ext`` (prior expansion), and ``ext-median`` (the Spotify-sourced Median playlist tracks, pooled into the combined_500 expansion but tagged distinctly for focused annotation). Built by ``scripts/build_cohorts.py``. Returns ``{track_id: cohort}``; ``{}`` when absent so the GT-review page degrades to its undivided Extension view. """ path = _resolved_data_root() / "gt_ingest" / "cohorts.json" if not path.exists(): return {} try: data = json.loads(path.read_text()) except Exception as e: # noqa: BLE001 st.warning(f"Could not parse cohorts.json: {e}") return {} return data.get("cohorts", {}) @st.cache_data(ttl=600, show_spinner=False) def load_subtypes_by_family() -> dict[str, list[str]]: """Parse dim 2 (genre subtype) from ``taxonomy.md`` grouped by parent family. Returns ``{family: [bare_subtype, ...]}`` — e.g. ``{'arabesk': ['acılı-arabesk', 'fantezi-arabesk', ...], 'halk': [...]}``. The existing ``load_taxonomy`` parser flattens dim 2 into a single ``controlled_vocab`` list and loses the ``#### Under \\`\\``` sub-heading grouping. The GT-review sheet needs the grouping to build cascading dropdowns (subtype options filtered by current family), so this loader re-parses the same section with a sub-heading-aware regex. Nested forms like ``oyun-havası.halay`` are kept verbatim — the dotted form is preserved through to the editable dropdown. """ p = _taxonomy_path() if not p.exists(): return {} raw = p.read_text() dim2 = re.search( r"###\s+Dimension\s+2[^\n]*\n(.+?)(?=###\s+Dimension\s+3)", raw, re.DOTALL, ) if not dim2: return {} body = dim2.group(1) out: dict[str, list[str]] = {} # Stop each section at the next `#### Under` block, the next bold # callout (e.g. ``**Source authority:**``, ``**Example tracks…**``), # or end-of-string. Without the bold-callout boundary the LAST family # (here ``fantezi``) eats the catalog-examples list that follows it. section_re = re.compile( r"####\s+Under\s+`([^`]+)`[^\n]*\n(.+?)(?=####|^\*\*|\Z)", re.DOTALL | re.MULTILINE, ) for m in section_re.finditer(body): family = m.group(1).strip() terms: list[str] = [] for line in m.group(2).splitlines(): tm = re.match(r"^\s*-\s*`([^`]+)`", line) if tm: terms.append(tm.group(1).strip()) if terms: out[family] = terms return out def save_gt_edit( track_id: str, field: str, new_value: Any, *, old_value: Any = None, confidence: str = "high", cascade_family: str | None = None, ) -> bool: """Apply one inline GT edit from the review sheet to the per-track JSON. - Writes ``record[field] = new_value`` atomically (tmp file + rename). - Bumps ``record[f'{field}_confidence']`` to ``confidence`` (default "high") so describe-batch and the rest of the pipeline treat the human-edited value as authoritative. - When ``cascade_family`` is provided (only used when ``field == 'genre_subtype'`` and the picked subtype's parent family differs from the existing family), also updates ``record['genre_family']`` so the two fields stay consistent. - Snapshots the GT JSON to ``.json.bak.`` on the FIRST edit per track (subsequent edits skip the backup so the directory doesn't fill with .bak files). - Appends a structured entry to ``data/groundtruth/median/_audit.jsonl`` (append-only JSONL) for traceability. - Validates the record against the Pydantic schema BEFORE writing so a bad edit never lands on disk. Returns True on success, False on validation/write failure. In HF mode the edit is also pushed back to the private dataset in a single atomic commit (track JSON + audit log together); local edits skip the HF round-trip. """ in_hf = _data_source() == "hf" gt_dir = _groundtruth_dir() gt_path = gt_dir / f"{track_id}.json" if not gt_path.exists(): st.error(f"GT JSON not found for {track_id}") return False # First-edit-per-track backup (local only — we don't churn .bak files # on the ephemeral Space FS; the audit log captures the old value). if not in_hf: existing_baks = list(gt_dir.glob(f"{track_id}.json.bak.*")) if not existing_baks: ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") bak = gt_dir / f"{track_id}.json.bak.{ts}" bak.write_bytes(gt_path.read_bytes()) record = json.loads(gt_path.read_text()) record[field] = new_value if new_value != "" else None conf_field = f"{field}_confidence" if conf_field in record: record[conf_field] = confidence if cascade_family: record["genre_family"] = cascade_family if "genre_family_confidence" in record: record["genre_family_confidence"] = confidence # Validate before persisting. The Pydantic schema has extra='forbid', # so a typo'd field name would surface here instead of silently landing. from sync_pilot.groundtruth.schema import GroundTruthRecord try: GroundTruthRecord.model_validate(record) except Exception as e: # noqa: BLE001 st.error(f"GT edit rejected by schema for {track_id}.{field}: {e}") return False # Split audit log streams: ``_audit.jsonl`` is local-only and shipped # by ``sync_pilot publish``; ``_audit_hf.jsonl`` is Space-only and # mutated only by HF writebacks. Keeping them separate means a local # publish never clobbers Space-side history (and vice versa). audit_name = "_audit_hf.jsonl" if in_hf else "_audit.jsonl" audit_path = gt_dir / audit_name audit_path.parent.mkdir(parents=True, exist_ok=True) audit_entry = { "track_id": track_id, "field": field, "old_value": old_value, "new_value": new_value if new_value != "" else None, "cascade_family": cascade_family, "confidence_set_to": confidence, "timestamp": datetime.now(timezone.utc).isoformat(), "source": "gt-review-sheet" + ("-hf" if in_hf else ""), } with audit_path.open("a", encoding="utf-8") as f: f.write(json.dumps(audit_entry, ensure_ascii=False) + "\n") tmp = gt_path.with_suffix(".json.tmp") tmp.write_text( json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8" ) tmp.replace(gt_path) if in_hf: rel = f"sync_pilot/groundtruth/median/{track_id}.json" rel_audit = f"sync_pilot/groundtruth/median/{audit_name}" _hf_writeback( [(gt_path, rel), (audit_path, rel_audit)], commit_message=f"gt-review: {track_id}.{field}", ) load_groundtruth.clear() return True def save_description( track_id: str, new_value: str, *, old_value: str | None = None, ) -> bool: """Persist a human-edited description back to the track JSON. The description lives on ``TrackRecord`` (``data/outputs/median/.json``) — distinct from the GT JSON ``save_gt_edit`` writes to. We mirror the GT save path for safety: - Atomic write (tmp file + rename). - Pydantic validation against ``TrackRecord`` before persisting so a bad edit can't land on disk. - Audit-log entry appended to ``data/groundtruth/median/_audit.jsonl`` (same JSONL the GT edits use, ``source="gt-review-sheet-desc"`` to distinguish provenance on inspection). Unlike ``save_gt_edit`` we do *not* take a ``.bak`` snapshot — the track JSON is re-derivable by re-running ``describe-batch`` and the audit log captures the prior text. Returns True on success, False on validation/write failure. In HF mode the edit is also pushed back to the private dataset. """ in_hf = _data_source() == "hf" outputs_dir = _outputs_dir() track_path = outputs_dir / f"{track_id}.json" if not track_path.exists(): st.error(f"Track JSON not found for {track_id}") return False record = json.loads(track_path.read_text()) record["description"] = new_value if new_value else None from sync_pilot.schema import TrackRecord try: TrackRecord.model_validate(record) except Exception as e: # noqa: BLE001 st.error(f"Description edit rejected by schema for {track_id}: {e}") return False audit_name = "_audit_hf.jsonl" if in_hf else "_audit.jsonl" audit_path = _groundtruth_dir() / audit_name audit_path.parent.mkdir(parents=True, exist_ok=True) audit_entry = { "track_id": track_id, "field": "description", "old_value": old_value, "new_value": new_value if new_value else None, "timestamp": datetime.now(timezone.utc).isoformat(), "source": "gt-review-sheet-desc" + ("-hf" if in_hf else ""), } with audit_path.open("a", encoding="utf-8") as f: f.write(json.dumps(audit_entry, ensure_ascii=False) + "\n") tmp = track_path.with_suffix(".json.tmp") tmp.write_text( json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8" ) tmp.replace(track_path) if in_hf: rel = f"sync_pilot/outputs/median/{track_id}.json" rel_audit = f"sync_pilot/groundtruth/median/{audit_name}" _hf_writeback( [(track_path, rel), (audit_path, rel_audit)], commit_message=f"gt-review desc: {track_id}", ) load_tracks.clear() return True def save_expansion_gt_edit( track_id: str, field: str, new_value: Any, *, old_value: Any = None, confidence: str = "high", cascade_family: str | None = None, ) -> bool: """Apply one GT edit to the expansion ground-truth set. Mirrors ``save_gt_edit``: first-edit ``.bak`` snapshot (local only), atomic write, schema validation, and an audit-log append. In ``hf`` mode the edit is pushed back to the private dataset in a single atomic commit (GT JSON + Space-side ``_audit_hf.jsonl``) so curations made on the Space survive restarts; the Space audit stream stays isolated from the local ``_audit.jsonl`` so a re-publish never clobbers Space history. """ in_hf = _data_source() == "hf" gt_dir = _expansion_groundtruth_dir() gt_path = gt_dir / f"{track_id}.json" if not gt_path.exists(): st.error(f"Expansion GT JSON not found for {track_id}") return False if not in_hf: existing_baks = list(gt_dir.glob(f"{track_id}.json.bak.*")) if not existing_baks: ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") bak = gt_dir / f"{track_id}.json.bak.{ts}" bak.write_bytes(gt_path.read_bytes()) record = json.loads(gt_path.read_text()) record[field] = new_value if new_value != "" else None conf_field = f"{field}_confidence" if conf_field in record: record[conf_field] = confidence if cascade_family: record["genre_family"] = cascade_family if "genre_family_confidence" in record: record["genre_family_confidence"] = confidence from sync_pilot.groundtruth.schema import GroundTruthRecord try: GroundTruthRecord.model_validate(record) except Exception as e: # noqa: BLE001 st.error(f"Expansion GT edit rejected by schema for {track_id}.{field}: {e}") return False audit_name = "_audit_hf.jsonl" if in_hf else "_audit.jsonl" audit_path = gt_dir / audit_name audit_path.parent.mkdir(parents=True, exist_ok=True) audit_entry = { "track_id": track_id, "field": field, "old_value": old_value, "new_value": new_value if new_value != "" else None, "cascade_family": cascade_family, "confidence_set_to": confidence, "timestamp": datetime.now(timezone.utc).isoformat(), "source": "gt-review-sheet-ext" + ("-hf" if in_hf else ""), } with audit_path.open("a", encoding="utf-8") as f: f.write(json.dumps(audit_entry, ensure_ascii=False) + "\n") tmp = gt_path.with_suffix(".json.tmp") tmp.write_text(json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(gt_path) if in_hf: rel = f"sync_pilot/gt_expansion/median_adjacent_combined_500/groundtruth/{track_id}.json" rel_audit = f"sync_pilot/gt_expansion/median_adjacent_combined_500/groundtruth/{audit_name}" _hf_writeback( [(gt_path, rel), (audit_path, rel_audit)], commit_message=f"gt-review ext: {track_id}.{field}", ) load_expansion_groundtruth.clear() return True def save_expansion_description( track_id: str, new_value: str, *, old_value: str | None = None, ) -> bool: """Persist an edited description to the expansion TrackRecord. Mirrors ``save_description``: atomic write, schema validation, and an audit-log append. In ``hf`` mode the track JSON + Space-side ``_audit_hf.jsonl`` are pushed back to the private dataset so the edit survives Space restarts. """ in_hf = _data_source() == "hf" track_path = _expansion_outputs_dir() / f"{track_id}.json" if not track_path.exists(): st.error(f"Expansion track JSON not found for {track_id}") return False record = json.loads(track_path.read_text()) record["description"] = new_value if new_value else None try: TrackRecord.model_validate(record) except Exception as e: # noqa: BLE001 st.error(f"Expansion description edit rejected for {track_id}: {e}") return False audit_name = "_audit_hf.jsonl" if in_hf else "_audit.jsonl" audit_path = _expansion_groundtruth_dir() / audit_name audit_path.parent.mkdir(parents=True, exist_ok=True) audit_entry = { "track_id": track_id, "field": "description", "old_value": old_value, "new_value": new_value if new_value else None, "timestamp": datetime.now(timezone.utc).isoformat(), "source": "gt-review-sheet-desc-ext" + ("-hf" if in_hf else ""), } with audit_path.open("a", encoding="utf-8") as f: f.write(json.dumps(audit_entry, ensure_ascii=False) + "\n") tmp = track_path.with_suffix(".json.tmp") tmp.write_text(json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(track_path) if in_hf: rel = f"sync_pilot/outputs/gt_expansion/median_adjacent_combined_500/{track_id}.json" rel_audit = f"sync_pilot/gt_expansion/median_adjacent_combined_500/groundtruth/{audit_name}" _hf_writeback( [(track_path, rel), (audit_path, rel_audit)], commit_message=f"gt-review ext desc: {track_id}", ) load_expansion_tracks.clear() return True @st.cache_data(ttl=60, show_spinner=False) def load_review() -> dict[str, dict[str, Any]]: """Load the GT-review sidecar (``_review.json``). Structure is ``{track_id: {"categories": {: bool}, "updated_at": iso8601}}``. Returns ``{}`` when the file does not exist (fresh catalog, no reviews yet) so callers can use ``state.get(track_id, {})`` without further guards. Bad / unparseable files surface a Streamlit warning and return ``{}`` rather than crashing the page. """ p = _review_read_path() if not p.exists(): return {} try: out = json.loads(p.read_text()) except Exception as e: # noqa: BLE001 st.warning(f"Could not parse {p.name}: {e}") return {} if not isinstance(out, dict): st.warning(f"{p.name} is not a JSON object — ignoring") return {} return out @st.cache_data(ttl=60, show_spinner=False) def load_expansion_review() -> dict[str, dict[str, Any]]: """Load the expansion-only GT-review sidecar.""" p = _expansion_review_path() if not p.exists(): return {} try: out = json.loads(p.read_text()) except Exception as e: # noqa: BLE001 st.warning(f"Could not parse expansion {p.name}: {e}") return {} if not isinstance(out, dict): st.warning(f"Expansion {p.name} is not a JSON object — ignoring") return {} return out def save_review(state: dict[str, dict[str, Any]]) -> None: """Atomic write of the full review sidecar. In local mode writes to ``data/groundtruth/median/_review.json``; in HF mode writes to the snapshot dir and pushes the updated file back to the private dataset. ``load_review`` is invalidated on success so the next page read reflects the new state without a TTL wait. """ in_hf = _data_source() == "hf" p = _review_read_path() if in_hf else _review_local_path() p.parent.mkdir(parents=True, exist_ok=True) tmp = p.with_suffix(".json.tmp") tmp.write_text(json.dumps(state, indent=2, ensure_ascii=False), encoding="utf-8") tmp.replace(p) if in_hf: _hf_writeback( [(p, "sync_pilot/groundtruth/median/_review.json")], commit_message="gt-review: update _review.json", ) load_review.clear() def save_expansion_review(state: dict[str, dict[str, Any]]) -> None: """Atomic write of the expansion-only review sidecar. In ``hf`` mode the updated sidecar is pushed back to the private dataset (mirrors ``save_review``). ``_expansion_review_path`` already resolves to the snapshot dir in HF mode and the local data dir locally, so the same write target doubles as the push source. """ in_hf = _data_source() == "hf" p = _expansion_review_path() p.parent.mkdir(parents=True, exist_ok=True) tmp = p.with_suffix(".json.tmp") tmp.write_text(json.dumps(state, indent=2, ensure_ascii=False), encoding="utf-8") tmp.replace(p) if in_hf: _hf_writeback( [(p, "sync_pilot/gt_expansion/median_adjacent_combined_500/_review.json")], commit_message="gt-review ext: update _review.json", ) load_expansion_review.clear() @st.cache_data(ttl=300, show_spinner=False) def load_manifest() -> dict[str, dict[str, Any]]: """Return ``{track_id: manifest_row}`` for fast lookup of YouTube URLs.""" out: dict[str, dict[str, Any]] = {} p = _manifest_path() if not p.exists(): return out with p.open() as f: for line in f: line = line.strip() if not line: continue try: row = json.loads(line) except json.JSONDecodeError: continue tid = row.get("track_id") if tid: out[tid] = row return out @st.cache_data(ttl=300, show_spinner=False) def load_expansion_manifest() -> dict[str, dict[str, Any]]: """Return ``{track_id: manifest_row}`` for expansion-source display.""" out: dict[str, dict[str, Any]] = {} p = _expansion_manifest_path() if not p.exists(): return out with p.open() as f: for line in f: line = line.strip() if not line: continue try: row = json.loads(line) except json.JSONDecodeError: continue tid = row.get("track_id") if tid: out[tid] = row return out # --------------------------------------------------------------------------- # Lyrics subtitle-leak stripper (display-only) # --------------------------------------------------------------------------- # Whisper transcribes the burned-in subtitle credit ("Altyazı M.K." / "Çeviri # ve Altyazı M.K.") at the start and end of many YouTube rips. The raw lyrics # string in the JSON deliberately keeps the leak so the on-disk record is # faithful to what the model produced; we strip it here for display. _SUBTITLE_BOILERPLATE = re.compile( r"(?:^|\s)(?:Çeviri\s+ve\s+)?Altyazı(?:yan)?(?:\s+M\.K\.)?\s*", re.IGNORECASE, ) # Broader Whisper boilerplate hallucinations on YouTube/TRT rips: thanks-for- # watching and the TRT audio-description disclaimer. Explicit char classes (not # re.IGNORECASE) avoid the Turkish dotted-İ casefold pitfall. _ASR_HALLUCINATION = re.compile( r"\s*(?:" r"[İi]zlediğiniz için teşekkür(?:ler|\s+eder\w+)?" r"|[Bb]u dizinin betimlemesi[^.]*?yaptırılmıştır" r"|[Bb]u dizinin betimlemesi(?:\s+TRT\s+tarafından)?" r"|[Ss]esli [Bb]etimleme [Dd]erneği(?:'ne)?" r")\.?\s*" ) def strip_subtitle_leak(text: str | None) -> str: """Display-only cleanup of the burned-in subtitle credit in Turkish lyrics. Removes the ``Altyazı M.K.`` / ``Çeviri ve Altyazı M.K.`` boilerplate that Whisper picks up from baked-in YouTube subtitles, plus broader thanks-for-watching / TRT audio-description hallucinations. Does NOT touch the on-disk JSON — the raw string is preserved there for provenance. """ if not text: return "" cleaned = _SUBTITLE_BOILERPLATE.sub(" ", text) cleaned = _ASR_HALLUCINATION.sub(" ", cleaned) # Collapse the spaces we may have introduced. return re.sub(r"\s{2,}", " ", cleaned).strip() # --------------------------------------------------------------------------- # Taxonomy parsing # --------------------------------------------------------------------------- @dataclass class TaxonomyDimension: """One ``### Dimension N — Title`` block from the taxonomy markdown.""" number: int title: str tr_label: str = "" en_label: str = "" definition: str = "" multiplicity: str = "" default: str = "" source_authority: str = "" expert_review_required: bool = False controlled_vocab: list[dict[str, str]] = field(default_factory=list) example_tracks: list[dict[str, str]] = field(default_factory=list) notes: str = "" raw_markdown: str = "" @dataclass class TaxonomyOpenQuestion: """One ``Q.`` entry from the open-questions section.""" number: int question: str body: str reviewers: list[str] = field(default_factory=list) @dataclass class TaxonomySpec: """Structured representation of ``taxonomy.md`` for dashboard rendering.""" version: str = "" status: str = "" last_updated: str = "" catalog: str = "" authors: str = "" review_targets: str = "" purpose_md: str = "" schema_conventions_md: str = "" dimensions: list[TaxonomyDimension] = field(default_factory=list) open_questions: list[TaxonomyOpenQuestion] = field(default_factory=list) references_md: str = "" raw: str = "" _META_PATTERN = re.compile(r"", re.DOTALL) _BOLD_FIELD_PATTERN = re.compile(r"\*\*([^*:]+):\*\*\s*(.+)") def _parse_metadata_block(raw: str) -> dict[str, str]: m = _META_PATTERN.search(raw) out: dict[str, str] = {} if not m: return out block = m.group(1) # The header uses YAML-ish ``key: value`` lines; nested fields under # ``review_targets:`` are preserved as a multi-line string. current_key: str | None = None for line in block.splitlines(): line = line.rstrip() if not line.strip(): continue if line.startswith(" - ") and current_key: out[current_key] += "\n" + line.strip() continue if ":" in line: key, _, val = line.partition(":") key = key.strip() val = val.strip() if not key: continue current_key = key out[key] = val return out def _parse_dimension_block(num: int, body: str) -> TaxonomyDimension: """Parse a single ``### Dimension N — Title`` block into structured form. Best-effort: we extract the fields we know how to render and stash the whole body in ``raw_markdown`` so the renderer can fall back to a plain markdown dump for anything we didn't decompose. """ title_match = re.match(r"###\s+Dimension\s+\d+\s+—\s+(.+)", body.splitlines()[0]) title = title_match.group(1).strip() if title_match else f"Dimension {num}" dim = TaxonomyDimension(number=num, title=title, raw_markdown=body) # **TR / EN:** `tr` / `en` tr_en = re.search(r"\*\*TR\s*/\s*EN:\*\*\s*`([^`]+)`\s*/\s*`([^`]+)`", body) if tr_en: dim.tr_label = tr_en.group(1) dim.en_label = tr_en.group(2) defn = re.search(r"\*\*Definition:\*\*\s*(.+?)(?=\n\*\*|\Z)", body, re.DOTALL) if defn: dim.definition = defn.group(1).strip() mult = re.search(r"\*\*Multiplicity:\*\*\s*(.+)", body) if mult: dim.multiplicity = mult.group(1).strip() deflt = re.search(r"\*\*Default:\*\*\s*(.+)", body) if deflt: dim.default = deflt.group(1).strip() src = re.search(r"\*\*Source authority:\*\*\s*(.+?)(?=\n\*\*|\Z)", body, re.DOTALL) if src: dim.source_authority = src.group(1).strip() notes = re.search(r"\*\*Notes\s*/\s*ambiguities:\*\*\s*(.+?)(?=\n---|\Z)", body, re.DOTALL) if notes: dim.notes = notes.group(1).strip() # The makam dimension explicitly says it's expert_review_required. if "expert_review_required: true" in body or "⚠ CRITICAL" in body: dim.expert_review_required = True # Controlled-vocab extraction: top-level ``- `term` — definition`` bullets, # found after the ``**Controlled vocabulary`` heading and before the next # ``**`` heading. We don't try to disambiguate grouped sub-headings here — # the renderer falls back to raw markdown for that depth of structure. vocab_section = re.search( r"\*\*Controlled vocabulary[^*]*?:?\*\*\s*(.+?)(?=\n\*\*|\Z)", body, re.DOTALL, ) if vocab_section: for line in vocab_section.group(1).splitlines(): m = re.match(r"^\s*-\s*`([^`]+)`\s*(?:—\s*(.+))?$", line) if m: term = m.group(1).strip() definition = (m.group(2) or "").strip() dim.controlled_vocab.append({"term": term, "definition": definition}) # Example tracks: ``- `` → ``. ex_section = re.search( r"\*\*Example tracks[^*]*?:?\*\*\s*(.+?)(?=\n\*\*|\Z)", body, re.DOTALL, ) if ex_section: for line in ex_section.group(1).splitlines(): m = re.match(r"^\s*-\s*`([0-9_a-zA-ZşŞçÇğĞıİöÖüÜ]+)`\s*→\s*(.+)$", line) if m: dim.example_tracks.append( {"track_id": m.group(1).strip(), "note": m.group(2).strip()} ) return dim def _parse_open_questions(section_md: str) -> list[TaxonomyOpenQuestion]: """Parse the ``## 4. Open questions`` body into a list of cards. Each block starts ``**Q. ...?**`` and runs until the next ``**Q`` or the section terminator. The body may contain explicit reviewer call- outs like ``**Aran:`` / ``**Murat:`` / ``**Emre:`` which we lift into a ``reviewers`` list for badge rendering. """ out: list[TaxonomyOpenQuestion] = [] # Split on each Q-header. We keep the headers via a lookahead split. parts = re.split(r"\n(?=\*\*Q\d+\.)", section_md) for part in parts: m = re.match(r"\*\*Q(\d+)\.\s*(.+?)\*\*\s*(.*)", part, re.DOTALL) if not m: continue number = int(m.group(1)) question = m.group(2).strip().rstrip("*").strip() body = m.group(3).strip() reviewers: list[str] = [] for name in ("Aran", "Murat", "Emre"): if re.search(rf"\b{name}\b", body): reviewers.append(name) out.append( TaxonomyOpenQuestion( number=number, question=question, body=body, reviewers=reviewers ) ) out.sort(key=lambda q: q.number) return out @st.cache_data(ttl=600, show_spinner=False) def load_taxonomy() -> TaxonomySpec: """Parse ``taxonomy.md`` into a structured ``TaxonomySpec``. Any sections we fail to decompose are still available via ``.raw`` so page code can fall back to a plain markdown dump. We deliberately keep the parser narrow rather than pulling in a real markdown AST — the file's structure is stable and the parser is the only consumer. """ p = _taxonomy_path() if not p.exists(): return TaxonomySpec() raw = p.read_text() spec = TaxonomySpec(raw=raw) meta = _parse_metadata_block(raw) spec.version = meta.get("taxonomy_version", "") spec.status = meta.get("status", "") spec.last_updated = meta.get("last_updated", "") spec.catalog = meta.get("catalog", "") spec.authors = meta.get("authors", "") spec.review_targets = meta.get("review_targets", "") # The major section headers we slice on. We split conservatively so the # parser keeps working if new top-level sections are added. purpose_match = re.search( r"##\s*1\. Purpose & scope(.+?)(?=##\s*\d+\.)", raw, re.DOTALL ) if purpose_match: spec.purpose_md = purpose_match.group(1).strip() conv_match = re.search( r"##\s*2\. Schema conventions(.+?)(?=##\s*\d+\.)", raw, re.DOTALL ) if conv_match: spec.schema_conventions_md = conv_match.group(1).strip() dims_match = re.search( r"##\s*3\. Dimensions(.+?)(?=##\s*\d+\.\s*Open questions)", raw, re.DOTALL ) if dims_match: dims_body = dims_match.group(1) # Split on each ``### Dimension N`` header. chunks = re.split(r"\n(?=###\s+Dimension\s+\d+\s+—)", dims_body) for chunk in chunks: m = re.match(r"###\s+Dimension\s+(\d+)", chunk) if not m: continue n = int(m.group(1)) spec.dimensions.append(_parse_dimension_block(n, chunk.strip())) spec.dimensions.sort(key=lambda d: d.number) open_q_match = re.search( r"##\s*4\. Open questions[^\n]*\n(.+?)(?=##\s*\d+\.)", raw, re.DOTALL ) if open_q_match: spec.open_questions = _parse_open_questions(open_q_match.group(1)) refs_match = re.search(r"##\s*5\. References(.+)", raw, re.DOTALL) if refs_match: spec.references_md = refs_match.group(1).strip() return spec # --------------------------------------------------------------------------- # Convenience derived data # --------------------------------------------------------------------------- def total_audio_minutes(tracks: list[dict[str, Any]]) -> float: """Sum ``duration_sec`` across all tracks; convert to minutes.""" return sum(float(t.get("duration_sec", 0.0)) for t in tracks) / 60.0 def parse_iso(ts: str | None) -> datetime | None: """Best-effort ISO-8601 parse; returns None on failure or empty input.""" if not ts: return None try: return datetime.fromisoformat(ts.replace("Z", "+00:00")) except ValueError: return None