Media-Server-Setup/bin/vgbot-merge-sqlite.py

#!/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."))