Add vgbot-merge-sqlite.py: merge the bot's old SQLite history into the live MariaDB
The bot's own importer requires an empty destination, but the bot rebuilds part of its state from Discord on every start, so after the backend switch neither side is a superset. This merges them: log tables append on a natural key, and the profile/state tables resolve column by column with one explicit rule per column, so the original joined_at, first_seen_at, rank started_at and tracking start survive a rebuild that stamped today, while current rank, presence and latest activity stay fresh. One transaction, idempotent, and it stops rather than guessing when a column has no rule or a table is absent from the destination. Tested against a fixture built from the real 1041-row file plus a simulated rebuild: history restored, fresh values kept, 30 membership events (29 historical + tonight's), re-run a no-op, and both guards verified. A KeyError on insert-only tables during the second run was found and fixed that way. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
174
bin/vgbot-merge-sqlite.py
Executable file
174
bin/vgbot-merge-sqlite.py
Executable file
@@ -0,0 +1,174 @@
|
||||
#!/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."))
|
||||
Reference in New Issue
Block a user