#!/usr/bin/env python3 """Atomically reconcile exercise/body-zone companion v1 with a sync baseline.""" import argparse import json import os import re import sqlite3 import unicodedata from datetime import datetime from pathlib import Path from trainlog_sqlite import connect_database ROOT = Path(__file__).resolve().parents[1] CATALOG = ROOT / "catalog/body-zones-v1.json" class ImportFailure(RuntimeError): pass EXERCISE_ID_PATTERN = re.compile( r"ex_[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}" ) def default_database(): return Path(os.environ.get("XDG_DATA_HOME", str(Path.home() / ".local/share"))) / "trainlog/trainlog.db" def exact(value, keys, label): if not isinstance(value, dict) or set(value) != keys: raise ImportFailure(f"{label}: forme invalide") def load_catalog(): root = json.loads(CATALOG.read_text(encoding="utf-8")) if root.get("format") != "trainlog-body-zone-catalog" or root.get("version") != 1: raise ImportFailure("catalogue zones v1 invalide") return {item["zone_id"]: item for item in root["zones"]} def state(primary, secondary): return (primary or "") + "|" + ",".join(sorted(secondary)) def current(connection, row_id): primary = connection.execute( "SELECT zone_id FROM exercise_body_zones WHERE exercise_row_id=? AND role='primary'", (row_id,), ).fetchone() secondary = [row[0] for row in connection.execute( "SELECT zone_id FROM exercise_body_zones WHERE exercise_row_id=? AND role='secondary' ORDER BY zone_id", (row_id,), )] return (None if primary is None else primary[0], secondary) def normalize_name(value): """Match the desktop importer's stable normalized-name comparison.""" folded = unicodedata.normalize("NFC", value.casefold()) output = [] pending_space = False wrote_content = False for char in folded: if char.isspace(): if wrote_content: pending_space = True continue if pending_space: output.append(" ") pending_space = False output.append(char) wrote_content = True normalized = "".join(output) if not normalized: raise ImportFailure("preuve mobile: nom d'exercice vide") return normalized def load_mobile_exercise_proof(path): """Load only the already-imported V2 definitions needed to prove aliases.""" payload = json.loads(path.read_text(encoding="utf-8")) if payload.get("format") != "trainlog-mobile-export" or payload.get("version") != 2 or \ not isinstance(payload.get("exercises"), list): raise ImportFailure("snapshot mobile V2 de preuve invalide") proof = {} for index, item in enumerate(payload["exercises"]): if not isinstance(item, dict): raise ImportFailure(f"preuve mobile exercises[{index}] invalide") exercise_id = item.get("exercise_id") name = item.get("name") recording = item.get("recording_mode") tracking = item.get("tracking_mode") data_fields = item.get("data_fields") if not isinstance(exercise_id, str) or not exercise_id or exercise_id in proof or \ not isinstance(name, str) or not name or recording not in {"sets", "continuous"} or \ tracking not in {"reps", "duration"} or isinstance(data_fields, bool) or \ not isinstance(data_fields, int) or data_fields < 0: raise ImportFailure(f"preuve mobile exercises[{index}] invalide") proof[exercise_id] = (normalize_name(name), recording, tracking, data_fields) return proof def resolve_exercise_row(connection, exercise_id, mobile_proof): row = connection.execute( "SELECT id FROM exercises WHERE exercise_id=?", (exercise_id,), ).fetchone() if row is not None: return row[0] aliases_available = connection.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name='exercise_aliases'", ).fetchone() is not None if aliases_available: row = connection.execute( "SELECT e.id FROM exercise_aliases a " "JOIN exercises e ON e.exercise_id=a.canonical_exercise_id " "WHERE a.source_exercise_id=?", (exercise_id,), ).fetchone() if row is not None: return row[0] proof = mobile_proof.get(exercise_id) if proof is None: raise ImportFailure(f"exercice inconnu: {exercise_id}") normalized, recording, tracking, data_fields = proof row = connection.execute( "SELECT id,recording_mode,tracking_mode,data_fields FROM exercises " "WHERE normalized_name=?", (normalized,), ).fetchone() if row is None or row[1] != recording or row[2] != tracking or \ not ((row[3] & ~data_fields) == 0 or (data_fields & ~row[3]) == 0): raise ImportFailure(f"preuve de réconciliation invalide: {exercise_id}") # CONTRACT: import_mobile_export.py ran first in the same sync and removed # only a profile-compatible duplicate. The source V2 definition proves that # this now-absent creator ID resolves to the one normalized desktop row. return row[0] def replace(connection, row_id, primary, secondary): connection.execute("DELETE FROM exercise_body_zones WHERE exercise_row_id=?", (row_id,)) if primary is not None: connection.execute( "INSERT INTO exercise_body_zones VALUES(?,?,'primary')", (row_id, primary), ) connection.executemany( "INSERT INTO exercise_body_zones VALUES(?,?,'secondary')", [(row_id, zone_id) for zone_id in secondary], ) def apply_body_zones(connection, parsed, mobile_proof, complete_causal_envelope=False): """Apply validated zone rows without owning transaction boundaries.""" supported_versions = range(11, 26) if connection.execute("PRAGMA user_version").fetchone()[0] not in supported_versions: raise ImportFailure("schema desktop v11-v25 requis") if not complete_causal_envelope and connection.execute("PRAGMA user_version").fetchone()[0]>=20 and connection.execute("SELECT 1 FROM sync_causal_state WHERE target_kind='body_zone_relation' AND deleted=1 LIMIT 1").fetchone(): raise ImportFailure("causal BODY ZONES protection requires the staged artifact") connection.execute("PRAGMA foreign_keys=ON");grouped={} # CONTRACT: resolve complete alias groups before the first write; direct # primary/secondary claims may be equal but are never unioned. for exercise_id,primary,secondary in parsed: row_id=resolve_exercise_row(connection,exercise_id,mobile_proof);incoming=(primary,secondary);group=grouped.get(row_id) if group is None:grouped[row_id]=(exercise_id,incoming) elif group[1]!=incoming:raise ImportFailure(f"données de zones incompatibles après résolution d'alias: {group[0]} / {exercise_id}") planned=[] for row_id,(exercise_id,incoming) in grouped.items(): primary,secondary=incoming;local_primary,local_secondary=current(connection,row_id);local_state=state(local_primary,local_secondary);incoming_state=state(primary,secondary) baseline_row=connection.execute("SELECT synced_state FROM exercise_body_zone_sync WHERE exercise_row_id=?",(row_id,)).fetchone();baseline=None if baseline_row is None else baseline_row[0] if local_state==incoming_state:action="skip" elif baseline is not None and local_state==baseline:action="update" elif baseline is not None and incoming_state==baseline:action="keep" elif baseline is None and local_state=="|":action="update" else:raise ImportFailure(f"conflit zones simultané: {exercise_id}") planned.append((action,row_id,primary,secondary,incoming_state)) updated=skipped=kept_local=0 # WHY: aliases are compatibility identities, not extra exercises. Apply # once per canonical row so input order cannot change the chosen payload. for action,row_id,primary,secondary,incoming_state in planned: if action=="keep":kept_local+=1;continue if action=="update":replace(connection,row_id,primary,secondary);updated+=1 else:skipped+=1 connection.execute("INSERT OR REPLACE INTO exercise_body_zone_sync VALUES(?,?)",(row_id,incoming_state)) return updated,skipped,kept_local def parse_payload(payload): zones=load_catalog();exact(payload,{"format","version","generated_at","exercises"},"racine") if payload["format"]!="trainlog-exercise-body-zones" or payload["version"]!=1 or isinstance(payload["version"],bool) or not isinstance(payload["generated_at"],str) or not payload["generated_at"] or not isinstance(payload["exercises"],list):raise ImportFailure("companion zones v1 invalide") try: generated_at=datetime.fromisoformat(payload["generated_at"].replace("Z","+00:00")) except ValueError as error:raise ImportFailure("generated_at invalide") from error if generated_at.utcoffset() is None:raise ImportFailure("generated_at sans offset") parsed=[];seen=set() for index,item in enumerate(payload["exercises"]): exact(item,{"exercise_id","primary_zone_id","secondary_zone_ids"},f"exercises[{index}]");exercise_id=item["exercise_id"];primary=item["primary_zone_id"];secondary=item["secondary_zone_ids"] if not isinstance(exercise_id,str) or EXERCISE_ID_PATTERN.fullmatch(exercise_id) is None or exercise_id in seen:raise ImportFailure(f"exercises[{index}].exercise_id invalide") if primary is not None and (not isinstance(primary,str) or primary not in zones):raise ImportFailure(f"exercises[{index}].primary_zone_id invalide") if not isinstance(secondary,list) or any(not isinstance(zone,str) or zone not in zones for zone in secondary):raise ImportFailure(f"exercises[{index}].secondary_zone_ids invalide") if len(secondary)!=len(set(secondary)) or primary in secondary:raise ImportFailure(f"exercises[{index}]: zones dupliquées") if primary is None and secondary:raise ImportFailure(f"exercises[{index}]: secondaires sans zone principale") if any(zones[zone]["kind"]=="group" for zone in ([primary] if primary else [])+secondary):raise ImportFailure(f"exercises[{index}]: relation parent dérivable interdite") seen.add(exercise_id);parsed.append((exercise_id,primary,sorted(secondary))) return parsed def main(): parser = argparse.ArgumentParser() parser.add_argument("input", type=Path) parser.add_argument("--database", type=Path, default=default_database()) parser.add_argument( "--mobile-export", type=Path, help="snapshot V2 importé juste avant, utilisé seulement pour prouver un ID réconcilié", ) args = parser.parse_args() payload = json.loads(args.input.read_text(encoding="utf-8")) parsed = parse_payload(payload) # A V2 snapshot is reconciliation evidence only when the caller selected # and supplied that exact snapshot. Never discover a sibling implicitly: # a stale V2 file must not prove IDs for a separately selected V3/V1 run. mobile_proof = (load_mobile_exercise_proof(args.mobile_export) if args.mobile_export is not None else {}) connection = connect_database(args.database) try: connection.execute("BEGIN IMMEDIATE") updated,skipped,kept_local=apply_body_zones(connection,parsed,mobile_proof) connection.commit() print("EXERCISE_BODY_ZONES_IMPORT=PASS") print(f"zones_updated={updated}") print(f"zones_skipped={skipped}") print(f"zones_kept_local={kept_local}") except Exception: connection.rollback() raise finally: connection.close() if __name__ == "__main__": try: main() except Exception as error: print(f"EXERCISE_BODY_ZONES_IMPORT=FAIL {error}") raise SystemExit(1)