#!/usr/bin/env python3 """Merge the Discord bot's old SQLite database into its live MariaDB WITHOUT losing either side. The bot rebuilds part of its state from Discord on every start, so after switching to MariaDB some tables hold fresh rows while the old SQLite file holds months of history. Neither is a superset. This merges them row by row and, where a row exists in both, column by column with an explicit rule per column. vgbot-merge-sqlite.py --source /data/migration-source.db # report only, touches nothing vgbot-merge-sqlite.py --source /data/migration-source.db --apply # one transaction, rolls back on error Run it with the gateway stopped (docker compose stop discord-bot) so nothing writes underneath it. Idempotent: running it twice changes nothing the second time. Rules, and why: min / max ISO-8601 text, so lexicographic order is chronological. "min" keeps the earliest known date, which is what makes joined-at, rank-since and tracking-start survive a rebuild that reset them to today. "max" keeps the most recent activity. dest the live row wins: it reflects the server as it is right now. source the old row wins. *_notnull prefer that side unless it is NULL, then take the other. source_unless_default take the old value unless it is the schema's placeholder. Every non-key column must appear in the rules; an unclassified column stops the run rather than being guessed. """ import argparse, sqlite3, sys from pathlib import Path from sqlalchemy import MetaData, create_engine, inspect, select, func # ---- how a row present on BOTH sides is resolved ------------------------------------------------------ ROW_RULE = { # whole row chosen by comparing one column (single-row marker tables) "activity_monitor": ("max", "last_check_at"), # the later check wins "membership_tracking": ("min", "started_at"), # keep the original tracking start, not tonight's } COL_RULE = { "member_activity": {"tracking_started_at": "min", "last_message_at": "max", "last_voice_at": "max"}, "member_profiles": {"first_seen_at": "min", "discord_joined_at": "min_notnull", "updated_at": "max", "username": "dest", "display_name": "dest", "current_rank": "dest", "is_present": "dest"}, "member_rank_tenure": {"started_at": "min", "joined_at": "min_notnull", "role_id": "dest", "active": "dest", "source": "dest"}, "membership_state": {"joined_at": "min", "observed_at": "max", "present": "dest"}, "recruits": {"recruit_started_at": "min", "discord_joined_at": "min_notnull", "recruited_by": "source_notnull", "invite_code": "source_notnull", "invite_source": ("source_unless_default", "unknown"), "activity_override": "source_notnull", "promoted_at": "dest_notnull", "username": "dest", "display_name": "dest", "status": "dest", "current_rank": "dest", "is_active": "dest"}, "voice_channel_departures": {"left_at": "max", "guild_id": "dest", "user_id": "dest", "user_name": "dest"}, } # Append-only logs: the id is a surrogate, so rows are matched on a natural key and inserted with a new id. APPEND_KEY = { "membership_events": ["user_id", "event_type", "joined_at"], # also a UNIQUE constraint in the schema "profile_history": ["user_id", "event", "created_at", "details"], "recruit_history": ["user_id", "event", "recorded_at"], } SKIP = {"schema_migrations"} # the bot owns this; never copied def pick(rule, source_value, dest_value): if isinstance(rule, tuple) and rule[0] == "source_unless_default": return dest_value if source_value in (None, rule[1]) else source_value if rule == "dest": return dest_value if rule == "source": return source_value if rule == "source_notnull": return dest_value if source_value is None else source_value if rule == "dest_notnull": return source_value if dest_value is None else dest_value if rule in ("min", "max"): if source_value is None: return dest_value if dest_value is None: return source_value return (min if rule == "min" else max)(source_value, dest_value) if rule == "min_notnull": if source_value is None: return dest_value if dest_value is None: return source_value return min(source_value, dest_value) raise ValueError(f"unknown rule {rule!r}") def source_tables(connection): names = [r[0] for r in connection.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name")] out = {} for name in names: if name in SKIP: continue info = list(connection.execute(f'PRAGMA table_info("{name}")')) out[name] = {"columns": [row[1] for row in info], "key": [row[1] for row in info if row[5]]} return out def plan(source_path, url, apply=False): source = sqlite3.connect(Path(source_path).resolve().as_uri() + "?mode=ro", uri=True) source.row_factory = sqlite3.Row tables = source_tables(source) engine = create_engine(url) metadata = MetaData() metadata.reflect(bind=engine, only=[n for n in tables if n in set(inspect(engine).get_table_names())]) report, problems = [], [] with engine.begin() as target: # one transaction; an exception rolls everything back for name, meta in tables.items(): if name not in metadata.tables: problems.append(f"{name}: not in the destination schema"); continue table = metadata.tables[name] key = meta["key"] or APPEND_KEY.get(name) rows = [dict(r) for r in source.execute(f'SELECT * FROM "{name}"')] if not rows: report.append((name, 0, 0, 0, "source empty")); continue # unclassified columns are a stop, not a guess if name in COL_RULE: managed = set(COL_RULE[name]) | set(meta["key"]) missing = [c for c in meta["columns"] if c not in managed] if missing: problems.append(f"{name}: no rule for {missing}"); continue if name in APPEND_KEY: natural = APPEND_KEY[name] existing = {tuple(r) for r in target.execute(select(*[table.c[c] for c in natural]))} fresh = [r for r in rows if tuple(r[c] for c in natural) not in existing] if apply and fresh: target.execute(table.insert(), [{k: v for k, v in r.items() if k != "id"} for r in fresh]) report.append((name, len(fresh), 0, len(rows) - len(fresh), f"appended on {'+'.join(natural)}")) continue dest = {tuple(r._mapping[c] for c in key): dict(r._mapping) for r in target.execute(select(table))} if key else {} inserted = updated = same = 0 for row in rows: ident = tuple(row[c] for c in key) current = dest.get(ident) if current is None: if apply: target.execute(table.insert(), [row]) inserted += 1 continue if name in ROW_RULE: how, column = ROW_RULE[name] winner = pick(how, row[column], current[column]) merged = row if winner == row[column] and row[column] != current[column] else None elif name in COL_RULE: merged = {c: pick(COL_RULE[name][c], row[c], current[c]) for c in COL_RULE[name]} merged = None if all(merged[c] == current[c] for c in merged) else merged else: merged = None # insert-only table and the row is already there: leave it alone if merged is None: same += 1 continue values = {c: v for c, v in merged.items() if c not in key} if apply: target.execute(table.update().where( *[table.c[c] == row[c] for c in key]).values(**values)) updated += 1 how = "merged per column" if name in COL_RULE else ("row rule" if name in ROW_RULE else "insert only") report.append((name, inserted, updated, same, how)) if problems: raise SystemExit("STOPPED, nothing changed:\n " + "\n ".join(problems)) source.close() return report if __name__ == "__main__": ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--source", required=True) ap.add_argument("--url", help="SQLAlchemy URL; default: built from the bot's DB_* environment") ap.add_argument("--apply", action="store_true", help="write; otherwise report only") args = ap.parse_args() url = args.url if not url: import os url = (f"mysql+pymysql://{os.environ['DB_USER']}:{os.environ['DB_PASSWORD']}" f"@{os.environ['DB_HOST']}:{os.environ.get('DB_PORT','3306')}/{os.environ['DB_NAME']}?charset=utf8mb4") rows = plan(args.source, url, apply=args.apply) print(f"\n{'table':28} {'insert':>7} {'update':>7} {'same':>6} how") for name, ins, upd, same, how in sorted(rows): print(f"{name:28} {ins:>7} {upd:>7} {same:>6} {how}") totals = [sum(r[i] for r in rows) for i in (1, 2, 3)] print(f"{'TOTAL':28} {totals[0]:>7} {totals[1]:>7} {totals[2]:>6}") print("\n" + ("applied." if args.apply else "report only; add --apply to write."))