214 lines
9.6 KiB
Python
214 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Apply the versioned equipment companion artifact to a desktop database."""
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from trainlog_sqlite import connect_database
|
|
|
|
FORMAT = "trainlog-equipment-associations"
|
|
VERSION = 2
|
|
|
|
# CONTRACT: MACHINE_EXERCISE_MODEL_V1 split only these completed occurrences.
|
|
# The legacy exercise IDs are deliberately not aliases: other occurrences of
|
|
# the same source exercise remain valid unsplit history.
|
|
MACHINE_EXERCISE_SPLITS = {
|
|
("se_c0d07454-b58b-403e-8b79-744619815fc5", "sxe_draft_legacy_7"):
|
|
("ex_b432623f-bfe9-4daf-a653-60ec7fdffbde",
|
|
"ex_0e26c06f-a458-40a4-be20-4ed219ede30d"),
|
|
("se_c0d07454-b58b-403e-8b79-744619815fc5",
|
|
"sxe_093c1331-beaa-4b69-91b3-240292709be6"):
|
|
("ex_b1e6ffc6-75b5-45ff-a3c0-e7433c58013d",
|
|
"ex_f01d2a46-6984-4dec-8934-4d82fca6dfc2"),
|
|
("se_ac3908d6-8e3e-4ac6-81a6-62dc2c39075a",
|
|
"sxe_f25142c8-455e-4346-9bfc-31d0989e275d"):
|
|
("ex_b1e6ffc6-75b5-45ff-a3c0-e7433c58013d",
|
|
"ex_f01d2a46-6984-4dec-8934-4d82fca6dfc2"),
|
|
("se_8a7ac0cb-9e67-42b6-bac9-179bbedf0024",
|
|
"sxe_f2242691-ba6c-48a0-b938-a1f744375d75"):
|
|
("ex_b1e6ffc6-75b5-45ff-a3c0-e7433c58013d",
|
|
"ex_f01d2a46-6984-4dec-8934-4d82fca6dfc2"),
|
|
("se_8e6baa18-53ff-486b-b011-02a15291789e",
|
|
"sxe_9f8882f4-069e-49d5-8216-19d16b467e4a"):
|
|
("ex_b1e6ffc6-75b5-45ff-a3c0-e7433c58013d",
|
|
"ex_f01d2a46-6984-4dec-8934-4d82fca6dfc2"),
|
|
}
|
|
|
|
|
|
def fail(message):
|
|
raise ValueError(message)
|
|
|
|
|
|
def load_catalog(path):
|
|
root = json.loads(path.read_text(encoding="utf-8"))
|
|
if root.get("format") != "trainlog-equipment-catalog" or root.get("version") != 1:
|
|
fail("catalogue équipement non supporté")
|
|
return {item["id"] for item in root["equipment"]}
|
|
|
|
|
|
def validate_associations(associations, known):
|
|
"""Validate the complete v2 payload before any database mutation."""
|
|
validated = []
|
|
seen = set()
|
|
for item in associations:
|
|
if not isinstance(item, dict):
|
|
fail("association invalide")
|
|
session_id, entry_id, exercise_id, state = (
|
|
item.get(key) for key in ("session_id", "entry_id", "exercise_id", "state"))
|
|
key = (session_id, entry_id)
|
|
if (not isinstance(session_id, str) or not session_id or
|
|
not isinstance(entry_id, str) or not entry_id or
|
|
not isinstance(exercise_id, str) or not exercise_id or key in seen):
|
|
fail("identité association invalide ou dupliquée")
|
|
seen.add(key)
|
|
if state == "set":
|
|
if set(item) != {"session_id", "entry_id", "exercise_id", "state", "equipment_id"}:
|
|
fail("association set invalide")
|
|
equipment_id = item["equipment_id"]
|
|
# CONTRACT: the definitions-v1 companion is imported first, so a
|
|
# custom ID is as valid here as a supplied catalog ID.
|
|
if not isinstance(equipment_id, str) or equipment_id not in known:
|
|
fail("équipement inconnu: "
|
|
f"session_id={session_id} entry_id={entry_id} equipment_id={equipment_id}")
|
|
elif state == "cleared":
|
|
if set(item) != {"session_id", "entry_id", "exercise_id", "state"}:
|
|
fail("association cleared invalide")
|
|
equipment_id = None
|
|
else:
|
|
fail("état association invalide")
|
|
validated.append((session_id, entry_id, exercise_id, equipment_id))
|
|
return validated
|
|
|
|
|
|
def load_mobile_occurrences(path):
|
|
"""Return the current V2/V3 identity claimed for every stable occurrence."""
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
if (
|
|
payload.get("format") != "trainlog-mobile-export"
|
|
or payload.get("version") not in (2, 3, 4)
|
|
or not isinstance(payload.get("sessions"), list)
|
|
):
|
|
fail("snapshot mobile V2 de preuve invalide")
|
|
|
|
occurrences = {}
|
|
for session in payload["sessions"]:
|
|
if not isinstance(session, dict) or not isinstance(
|
|
session.get("exercises"), list
|
|
):
|
|
fail("snapshot mobile V2 de preuve invalide")
|
|
session_id = session.get("session_id")
|
|
for item in session["exercises"]:
|
|
if not isinstance(item, dict):
|
|
fail("snapshot mobile V2 de preuve invalide")
|
|
entry_id = item.get("entry_id")
|
|
exercise_id = item.get("exercise_id")
|
|
key = (session_id, entry_id)
|
|
if (
|
|
not isinstance(session_id, str)
|
|
or not session_id
|
|
or not isinstance(entry_id, str)
|
|
or not entry_id
|
|
or not isinstance(exercise_id, str)
|
|
or not exercise_id
|
|
or key in occurrences
|
|
):
|
|
fail("identité du snapshot mobile V2 de preuve invalide")
|
|
occurrences[key] = exercise_id
|
|
return occurrences
|
|
|
|
|
|
def apply_associations(connection, payload, known, mobile_occurrences):
|
|
if payload.get("format") != FORMAT or payload.get("version") != VERSION or set(payload) != {"format", "version", "generated_at", "associations"}:
|
|
fail("extension équipement non supportée")
|
|
known = set(known); known.update(row[0] for row in connection.execute("SELECT equipment_id FROM custom_equipment"))
|
|
associations = validate_associations(payload["associations"], known)
|
|
for session_id, entry_id, exercise_id, equipment_id in associations:
|
|
exists = connection.execute("SELECT e.exercise_id,se.equipment_id FROM session_exercises se JOIN sessions s ON s.id=se.session_row_id JOIN exercises e ON e.id=se.exercise_row_id WHERE s.session_id=? AND se.entry_id=?", (session_id, entry_id)).fetchone()
|
|
if exists is None: fail(f"entrée séance inconnue: {session_id}/{entry_id}")
|
|
stored_canonical = canonical_exercise_id(connection, exists[0]); incoming_canonical = canonical_exercise_id(connection, exercise_id)
|
|
if stored_canonical != incoming_canonical:
|
|
proof = mobile_occurrences.get((session_id, entry_id))
|
|
incoming_still_exists = connection.execute("SELECT 1 FROM exercises WHERE exercise_id=?", (exercise_id,)).fetchone() is not None
|
|
if not (proof == exercise_id and not incoming_still_exists) and not approved_machine_split(session_id, entry_id, incoming_canonical, stored_canonical, mobile_occurrences):
|
|
fail(f"conflit exercice association: {session_id}/{entry_id}")
|
|
if exists[1] != equipment_id: fail(f"conflit association équipement: {session_id}/{entry_id}")
|
|
return len(associations)
|
|
|
|
|
|
def approved_machine_split(session_id, entry_id, incoming_id, local_id,
|
|
mobile_occurrences):
|
|
"""Corroborate one frozen completed-history split with the current snapshot."""
|
|
expected = MACHINE_EXERCISE_SPLITS.get((session_id, entry_id))
|
|
if expected != (incoming_id, local_id):
|
|
return False
|
|
# INVARIANT: the stale companion alone is never authority. The mobile
|
|
# snapshot selected and imported in this same sync must already carry the
|
|
# exact v13 target identity for this stable completed occurrence.
|
|
return mobile_occurrences.get((session_id, entry_id)) == local_id
|
|
|
|
|
|
def canonical_exercise_id(connection, exercise_id):
|
|
"""Resolve one flattened durable alias without inventing identity."""
|
|
aliases_available = connection.execute(
|
|
"SELECT 1 FROM sqlite_master WHERE type='table' AND name='exercise_aliases';"
|
|
).fetchone() is not None
|
|
if not aliases_available:
|
|
return exercise_id
|
|
alias = connection.execute(
|
|
"SELECT canonical_exercise_id FROM exercise_aliases "
|
|
"WHERE source_exercise_id=?;",
|
|
(exercise_id,),
|
|
).fetchone()
|
|
if alias is None:
|
|
return exercise_id
|
|
target_exists = connection.execute(
|
|
"SELECT 1 FROM exercises WHERE exercise_id=?;", (alias[0],)
|
|
).fetchone() is not None
|
|
if not target_exists:
|
|
fail(f"alias exercice sans cible canonique: {exercise_id}")
|
|
return alias[0]
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("artifact", type=Path)
|
|
parser.add_argument("--database", type=Path, required=True)
|
|
parser.add_argument("--catalog", type=Path,
|
|
default=Path(__file__).resolve().parents[1] / "catalog/equipment-v1.json")
|
|
parser.add_argument(
|
|
"--mobile-export",
|
|
type=Path,
|
|
help=(
|
|
"snapshot mobile V2/V3 importé juste avant ce companion; requis "
|
|
"pour prouver un exercise_id réconcilié"
|
|
),
|
|
)
|
|
args = parser.parse_args()
|
|
payload = json.loads(args.artifact.read_text(encoding="utf-8"))
|
|
connection = connect_database(args.database)
|
|
try:
|
|
if connection.execute("PRAGMA user_version;").fetchone()[0] not in (8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23):
|
|
fail("schema desktop v8 à v21 requis")
|
|
proof_path = args.mobile_export
|
|
if proof_path is None:
|
|
candidate = args.artifact.with_name("trainlog-mobile-export-v2.json")
|
|
if candidate.exists():
|
|
proof_path = candidate
|
|
mobile_occurrences = (
|
|
load_mobile_occurrences(proof_path)
|
|
if proof_path is not None
|
|
else {}
|
|
)
|
|
association_count = apply_associations(connection, payload, load_catalog(args.catalog), mobile_occurrences)
|
|
print("EQUIPMENT_ASSOCIATIONS_IMPORT=PASS")
|
|
print(f"associations={association_count}")
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as error:
|
|
print(f"EQUIPMENT_ASSOCIATIONS_IMPORT=FAIL {error}")
|
|
raise SystemExit(1)
|