#!/usr/bin/env python3 # ═══════════════════════════════════════════════════════════════════════ # iris-db-repair.py — reconstruit une iris.db saine depuis une copie # corrompue (2026-10-17, lot gates v1). # # Constat (snapshot 2026-10-17) : # PRAGMA integrity_check → "Tree 10 page 12 cell 6: Rowid 7 out of order" # table vms : 7 lignes fantômes (name NULL, colonnes décalées) qui # dupliquent les rowids 1-7 et sont absentes des index. # iris-atlas-fr absent de vms depuis l'incident du 2026-07-23. # # Ce que fait le script (stdlib Python uniquement, aucune dépendance) : # 1. ouvre la source en lecture seule (immutable, aucun verrou posé) # 2. recrée le schéma à l'identique dans une NOUVELLE base # 3. copie chaque table par scan complet de la table (NOT INDEXED : # on ne fait pas confiance aux index corrompus) # 4. vms : écarte les lignes sans nom, dédoublonne par nom # 5. --restore-vm NOM=BASE : réinjecte une VM absente depuis une # autre copie (ex. iris-atlas-fr depuis iris.db.corrupt-20260723) # 6. recale sqlite_sequence, puis integrity_check + foreign_key_check # # Il n'écrit JAMAIS la source et refuse d'écraser une sortie existante. # # Usage : # python3 -I iris-db-repair.py iris.db iris.db.repaired \ # --restore-vm iris-atlas-fr=iris.db.corrupt-20260723 # ═══════════════════════════════════════════════════════════════════════ import argparse import os import sqlite3 import sys def ro_connect(path): uri = 'file:' + os.path.abspath(path) + '?mode=ro&immutable=1' con = sqlite3.connect(uri, uri=True) con.row_factory = sqlite3.Row return con def q(ident): return '"' + ident.replace('"', '""') + '"' def main(): ap = argparse.ArgumentParser() ap.add_argument('src') ap.add_argument('dst') ap.add_argument('--restore-vm', action='append', default=[], metavar='NAME=SOURCE_DB') args = ap.parse_args() if os.path.exists(args.dst): sys.exit(f'refus : {args.dst} existe déjà') src = ro_connect(args.src) dst = sqlite3.connect(args.dst) dst.execute('PRAGMA foreign_keys = OFF') dst.execute('PRAGMA journal_mode = DELETE') objs = src.execute( "SELECT type, name, tbl_name, sql FROM sqlite_master " "WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%' " "ORDER BY CASE type WHEN 'table' THEN 0 WHEN 'index' THEN 1 " "WHEN 'view' THEN 2 ELSE 3 END, rowid").fetchall() tables = [o for o in objs if o['type'] == 'table'] others = [o for o in objs if o['type'] != 'table'] for t in tables: dst.execute(t['sql']) report = [] rejected_rows = [] for t in tables: name = t['name'] cols = [r['name'] for r in src.execute(f'PRAGMA table_info({q(name)})')] col_list = ','.join(q(c) for c in cols) ph = ','.join('?' * len(cols)) rows = src.execute(f'SELECT {col_list} FROM {q(name)} NOT INDEXED').fetchall() kept = dropped_null = dropped_dup = rejected = 0 seen = set() for r in rows: vals = tuple(r[c] for c in cols) if name == 'vms': vm = r['name'] if vm is None or str(vm).strip() == '': dropped_null += 1 continue if vm in seen: dropped_dup += 1 continue seen.add(vm) try: dst.execute(f'INSERT INTO {q(name)} ({col_list}) VALUES ({ph})', vals) kept += 1 except sqlite3.IntegrityError as e: rejected += 1 rejected_rows.append((name, str(e), vals[:3])) report.append((name, len(rows), kept, dropped_null, dropped_dup, rejected)) # ── Réinjection de VMs absentes ────────────────────────────────── restored = [] dst_cols = [r[1] for r in dst.execute('PRAGMA table_info(vms)')] for spec in args.restore_vm: if '=' not in spec: sys.exit(f'--restore-vm mal formé : {spec}') vm, other = spec.split('=', 1) if dst.execute('SELECT 1 FROM vms WHERE name = ?', (vm,)).fetchone(): restored.append((vm, 'déjà présente, rien fait')) continue o = ro_connect(other) o_cols = [r['name'] for r in o.execute('PRAGMA table_info(vms)')] common = [c for c in dst_cols if c in o_cols] row = o.execute( f"SELECT {','.join(q(c) for c in common)} FROM vms NOT INDEXED " "WHERE name = ? LIMIT 1", (vm,)).fetchone() if not row: restored.append((vm, f'introuvable dans {other}')) continue dst.execute( f"INSERT INTO vms ({','.join(q(c) for c in common)}) " f"VALUES ({','.join('?' * len(common))})", tuple(row[c] for c in common)) missing = [c for c in dst_cols if c not in o_cols] restored.append((vm, f'restaurée depuis {other}' + (f' (colonnes absentes, laissées au défaut : {missing})' if missing else ''))) # ── Index, vues, triggers ──────────────────────────────────────── for o in others: dst.execute(o['sql']) # ── sqlite_sequence ────────────────────────────────────────────── has_seq = src.execute( "SELECT 1 FROM sqlite_master WHERE name='sqlite_sequence'").fetchone() if has_seq: for s in src.execute('SELECT name, seq FROM sqlite_sequence NOT INDEXED'): cur = dst.execute('SELECT seq FROM sqlite_sequence WHERE name = ?', (s['name'],)).fetchone() seq = max(s['seq'] or 0, (cur[0] if cur else 0)) if cur: dst.execute('UPDATE sqlite_sequence SET seq = ? WHERE name = ?', (seq, s['name'])) else: dst.execute('INSERT INTO sqlite_sequence(name, seq) VALUES (?, ?)', (s['name'], seq)) uv = src.execute('PRAGMA user_version').fetchone()[0] dst.execute(f'PRAGMA user_version = {int(uv)}') dst.commit() integ = [r[0] for r in dst.execute('PRAGMA integrity_check')] fk = dst.execute('PRAGMA foreign_key_check').fetchall() dst.execute('VACUUM') dst.close() print(f'{"table":32} {"lues":>7} {"copiées":>8} {"sans nom":>9} {"doublons":>9} {"rejetées":>9}') for name, n, k, dn, dd, rj in report: flag = '' if (dn == 0 and dd == 0 and rj == 0) else ' <-- nettoyée' print(f'{name:32} {n:7} {k:8} {dn:9} {dd:9} {rj:9}{flag}') for name, err, head in rejected_rows[:20]: print(f' rejet {name} : {err} : {head}') for vm, msg in restored: print(f'restore {vm} : {msg}') print('integrity_check :', ' | '.join(integ)) print('foreign_key_check :', 'ok' if not fk else f'{len(fk)} violations : {fk[:10]}') if integ != ['ok']: sys.exit(2) if fk or rejected_rows: print('ATTENTION : violations FK ou lignes rejetées — vérifier avant de swapper') sys.exit(3) if __name__ == '__main__': main()