#!/usr/bin/env python3 """Apply MACHINE_EXERCISE_MODEL_V1 to an explicit non-live SQLite database. The executor is intentionally manifest-driven and refuses the canonical live desktop path. Context splits match entry ID, canonical source exercise ID and the exact legacy equipment ID before any write begins. """ import argparse import json import os import sqlite3 from pathlib import Path from exercise_names import normalize_catalog_name from trainlog_sqlite import connect_database ROOT = Path(__file__).resolve().parents[1] DEFAULT_MANIFEST = Path("/tmp/trainlog-machine-exercise-manifest/migration-manifest-v1.json") LIVE_DB = Path(os.environ.get("XDG_DATA_HOME", str(Path.home() / ".local/share"))) / "trainlog/trainlog.db" COLUMNS = ( ("load_semantics", "TEXT CHECK(load_semantics IN ('none','external','assistance','bodyweight','cardio'))"), ("machine_variant", "TEXT"), ("machine_provenance", "TEXT"), ("scientific_profile_id", "TEXT"), ("science_state", "TEXT NOT NULL DEFAULT 'unresolved' CHECK(science_state IN ('resolved','unresolved'))"), ("legacy_equipment_id", "TEXT"), ) def fail(message): raise RuntimeError(message) def row_id(connection, exercise_id): row = connection.execute("SELECT id FROM exercises WHERE exercise_id=?", (exercise_id,)).fetchone() return None if row is None else row[0] def split_facts(connection, entry_ids): placeholders = ",".join("?" for _ in entry_ids) return connection.execute(f""" SELECT se.entry_id,e.exercise_id,se.equipment_id, (SELECT count(*) FROM performed_sets p WHERE p.session_exercise_row_id=se.id), (SELECT count(*) FROM max_results m WHERE m.session_exercise_row_id=se.id), (SELECT count(*) FROM continuous_activity c WHERE c.session_exercise_row_id=se.id) FROM session_exercises se JOIN exercises e ON e.id=se.exercise_row_id WHERE se.entry_id IN ({placeholders}) ORDER BY se.entry_id """, entry_ids).fetchall() def validate_preconditions(connection, manifest): version = connection.execute("PRAGMA user_version").fetchone()[0] if version not in (12, 13): fail(f"schema v12 or v13 required, found v{version}") split_entries = [] for rename in manifest["renames"]: row = connection.execute("SELECT name FROM exercises WHERE exercise_id=?", (rename["exercise_id"],)).fetchone() if row is None: fail(f"rename source missing: {rename['exercise_id']}") if version == 12 and row[0] != rename["old_display_name"]: fail(f"rename source name mismatch: {rename['exercise_id']}") if version == 13 and row[0] != rename["new_display_name"]: fail(f"migrated rename mismatch: {rename['exercise_id']}") for split in manifest["context_splits"]: split_entries.extend(split["entry_ids"]) facts = split_facts(connection, split["entry_ids"]) expected_exercise = split["target_exercise_id"] if version == 13 else split["source_exercise_id"] if len(facts) != split["expected_occurrence_count"] or {row[0] for row in facts} != set(split["entry_ids"]): fail(f"split occurrence mismatch: {split['target_exercise_id']}") if any(row[1] != expected_exercise or row[2] != split["source_equipment_id"] for row in facts): fail(f"split identity/context mismatch: {split['target_exercise_id']}") if sum(row[3] for row in facts) != split["expected_performed_set_count"] or sum(row[4] for row in facts) != split["expected_max_count"] or sum(row[5] for row in facts) != split["expected_continuous_count"]: fail(f"split child count mismatch: {split['target_exercise_id']}") if len(split_entries) != len(set(split_entries)): fail("one entry appears in multiple split targets") unresolved = set() for item in manifest["unresolved_contexts"]: unresolved.update([item["entry_id"]] if "entry_id" in item else item.get("historical_entry_ids", [])) if unresolved & set(split_entries): fail("unresolved entry included in automatic split") def profile_interpretations(): profiles = json.loads((ROOT / "catalog/scientific-profiles-v1.json").read_text(encoding="utf-8"))["profiles"] exercises = {row["exercise_id"]: row for row in json.loads((ROOT / "catalog/exercise-knowledge-v1.json").read_text(encoding="utf-8"))["exercises"]} equipment = {row["equipment_id"]: row for row in json.loads((ROOT / "catalog/equipment-knowledge-v1.json").read_text(encoding="utf-8"))["equipment"]} result = {} for profile in profiles: source = profile["knowledge_source"] if source["kind"] == "exercise": interpretation = exercises[source["exercise_id"]].get("interpretation") else: capability = next(cap for cap in equipment[source["equipment_id"]]["capabilities"] if cap["display_name"] == source["capability"]) interpretation = capability.get("interpretation") if interpretation is None: fail(f"resolved profile has no interpretation: {profile['scientific_profile_id']}") result[profile["scientific_profile_id"]] = interpretation return result def seed_zones(connection, exercise_id, interpretation): target = row_id(connection, exercise_id) if target is None: fail(f"cannot seed zones for missing exercise: {exercise_id}") existing = connection.execute("SELECT zone_id,role FROM exercise_body_zones WHERE exercise_row_id=? ORDER BY role,zone_id", (target,)).fetchall() expected = [(interpretation["primary_zone_id"], "primary")] + [(zone, "secondary") for zone in sorted(interpretation["secondary_zone_ids"])] if existing: if sorted(existing) != sorted(expected): fail(f"BODY ZONE conflict for {exercise_id}") else: connection.executemany("INSERT INTO exercise_body_zones(exercise_row_id,zone_id,role) VALUES(?,?,?)", [(target, zone, role) for zone, role in expected]) state = interpretation["primary_zone_id"] + "|" + ",".join(sorted(interpretation["secondary_zone_ids"])) connection.execute("INSERT OR IGNORE INTO exercise_body_zone_sync(exercise_row_id,synced_state) VALUES(?,?)", (target, state)) def apply(connection, manifest): validate_preconditions(connection, manifest) if connection.execute("PRAGMA user_version").fetchone()[0] == 13: return False interpretations = profile_interpretations() connection.execute("BEGIN IMMEDIATE") try: for name, definition in COLUMNS: connection.execute(f"ALTER TABLE exercises ADD COLUMN {name} {definition}") for rename in manifest["renames"]: changed = connection.execute("UPDATE exercises SET name=?,normalized_name=?,load_semantics=?,machine_variant=NULL,scientific_profile_id=?,science_state='resolved',legacy_equipment_id=? WHERE exercise_id=? AND name=?", (rename["new_display_name"], normalize_catalog_name(rename["new_display_name"]), rename["load_semantics"], rename["scientific_profile_id"], rename["legacy_equipment_id"], rename["exercise_id"], rename["old_display_name"])).rowcount if changed != 1: fail(f"rename update failed: {rename['exercise_id']}") machine_rows = {row["exercise_id"]: row for row in json.loads((ROOT / "catalog/machine-exercises-v1.json").read_text(encoding="utf-8"))["exercises"]} for rename in manifest["renames"]: machine = machine_rows[rename["exercise_id"]] connection.execute("UPDATE exercises SET machine_variant=? WHERE exercise_id=?", (machine["machine_variant"], rename["exercise_id"])) for item in manifest["new_exercises"]: machine = machine_rows[item["exercise_id"]] connection.execute("""INSERT INTO exercises(exercise_id,name,normalized_name,tracking_mode,recording_mode,data_fields,load_semantics,machine_variant,machine_provenance,scientific_profile_id,science_state,legacy_equipment_id) VALUES(?,?,?,?,?,?,?,?,NULL,?,?,?)""", (item["exercise_id"], item["display_name"], normalize_catalog_name(item["display_name"]), item["tracking_mode"], item["recording_mode"], item["data_fields"], item["load_semantics"], machine["machine_variant"], item["scientific_profile_id"], machine["science_state"], item["legacy_equipment_id"])) if item["scientific_profile_id"] is not None: seed_zones(connection, item["exercise_id"], interpretations[item["scientific_profile_id"]]) for split in manifest["context_splits"]: target = row_id(connection, split["target_exercise_id"]) source = row_id(connection, split["source_exercise_id"]) for entry in split["entry_ids"]: changed = connection.execute("UPDATE session_exercises SET exercise_row_id=? WHERE entry_id=? AND exercise_row_id=? AND equipment_id=?", (target, entry, source, split["source_equipment_id"])).rowcount if changed != 1: fail(f"exact split update failed: {entry}") seated = machine_rows["ex_617007f9-7420-4408-91b9-8ffb77900f13"] connection.execute("UPDATE exercises SET load_semantics=NULL,machine_variant=NULL,machine_provenance=NULL,scientific_profile_id=NULL,science_state='unresolved',legacy_equipment_id=NULL WHERE exercise_id=?", (seated["exercise_id"],)) connection.execute("PRAGMA user_version=13") validate_preconditions(connection, manifest) if connection.execute("PRAGMA foreign_key_check").fetchone() is not None: fail("foreign key check failed") connection.commit() except Exception: connection.rollback() raise return True def main(): parser = argparse.ArgumentParser() parser.add_argument("database", type=Path) parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST) parser.add_argument("--audit", type=Path) args = parser.parse_args() database = args.database.resolve() if database == LIVE_DB.resolve(): fail("refusing to migrate the canonical live desktop database") manifest = json.loads(args.manifest.read_text(encoding="utf-8")) connection = connect_database(database) connection.execute("PRAGMA foreign_keys=ON") changed = apply(connection, manifest) integrity = connection.execute("PRAGMA integrity_check").fetchone()[0] version = connection.execute("PRAGMA user_version").fetchone()[0] connection.close() audit = {"format":"trainlog-machine-exercise-migration-audit","version":1,"database":str(database),"changed":changed,"schema_version":version,"integrity_check":integrity} if args.audit: args.audit.write_text(json.dumps(audit, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps(audit, ensure_ascii=False, sort_keys=True)) if __name__ == "__main__": main()