389 lines
14 KiB
Python
Executable File
389 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Migrate the legacy SQLite database into MySQL.
|
|
|
|
The app stored everything in `backend/data/ar_aging.db` before the MySQL move. That file
|
|
still holds the real month-end closings (Jan-2026: 3,399,517 transaction rows), and MySQL
|
|
starts empty, so the closings have to be copied across once.
|
|
|
|
python3 migrate_sqlite_to_mysql.py --dry-run # inspect the source, touch nothing
|
|
python3 migrate_sqlite_to_mysql.py # migrate
|
|
python3 migrate_sqlite_to_mysql.py --force # migrate into a non-empty MySQL
|
|
|
|
What it does
|
|
* copies every table in foreign-key order, so a child row never precedes its session
|
|
* converts SQLite's text dates / 0-1 booleans to real MySQL DATE, DATETIME and BOOLEAN
|
|
* copies only the columns both schemas share, and reports any it had to skip
|
|
* streams in batches, so 3.4M rows never sit in memory
|
|
* verifies afterwards: row counts per table AND financial checksums (Σ transaction
|
|
totals, per-marketplace receivable) must match the source exactly
|
|
|
|
It never deletes anything from SQLite — the file is opened read-only.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
BACKEND = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(BACKEND))
|
|
|
|
BATCH = 5000
|
|
|
|
# Tables whose contents are re-derivable by re-processing, but copied anyway so the
|
|
# migrated database is byte-identical in what the dashboard shows.
|
|
SKIP_TABLES: set[str] = set()
|
|
|
|
|
|
def log(msg: str = "") -> None:
|
|
print(msg, flush=True)
|
|
|
|
|
|
# --------------------------------------------------------------------------- source
|
|
def sqlite_tables(conn: sqlite3.Connection) -> set[str]:
|
|
return {r[0] for r in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")}
|
|
|
|
|
|
def sqlite_columns(conn: sqlite3.Connection, table: str) -> list[str]:
|
|
return [r[1] for r in conn.execute(f"PRAGMA table_info({table})")]
|
|
|
|
|
|
def sqlite_count(conn: sqlite3.Connection, table: str) -> int:
|
|
return conn.execute(f"SELECT COUNT(*) FROM `{table}`").fetchone()[0]
|
|
|
|
|
|
def open_sqlite(path: Path) -> sqlite3.Connection:
|
|
"""Open read-only. A stale -wal is checkpointed into a COPY, never the original."""
|
|
if not path.exists():
|
|
raise SystemExit(f"SQLite file not found: {path}")
|
|
# immutable=0 so an existing -wal is still applied; mode=ro keeps us from writing.
|
|
conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
# --------------------------------------------------------------------------- convert
|
|
def make_converter(col_type) -> "callable":
|
|
"""Return a function turning a SQLite value into something MySQL accepts."""
|
|
name = col_type.__class__.__name__
|
|
|
|
if name == "Date":
|
|
def conv(v):
|
|
if v in (None, ""):
|
|
return None
|
|
if isinstance(v, dt.date) and not isinstance(v, dt.datetime):
|
|
return v
|
|
if isinstance(v, dt.datetime):
|
|
return v.date()
|
|
try:
|
|
return dt.date.fromisoformat(str(v)[:10])
|
|
except ValueError:
|
|
return None
|
|
return conv
|
|
|
|
if name == "DateTime":
|
|
def conv(v):
|
|
if v in (None, ""):
|
|
return None
|
|
if isinstance(v, dt.datetime):
|
|
return v
|
|
s = str(v).replace("T", " ")
|
|
for fmt in ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
|
|
try:
|
|
return dt.datetime.strptime(s[:26], fmt)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
return conv
|
|
|
|
if name == "Boolean":
|
|
def conv(v):
|
|
if v is None:
|
|
return None
|
|
if isinstance(v, bool):
|
|
return v
|
|
return bool(int(v)) if str(v).strip() in ("0", "1") else bool(v)
|
|
return conv
|
|
|
|
if name in ("Integer", "BigInteger", "SmallInteger"):
|
|
def conv(v):
|
|
if v in (None, ""):
|
|
return None
|
|
try:
|
|
return int(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return conv
|
|
|
|
if name in ("Float", "Numeric"):
|
|
def conv(v):
|
|
if v in (None, ""):
|
|
return None
|
|
try:
|
|
return float(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return conv
|
|
|
|
# String / Text: MySQL columns are sized, so over-long values would be truncated or
|
|
# rejected. Trim to the declared length and report it rather than failing the batch.
|
|
length = getattr(col_type, "length", None)
|
|
|
|
def conv(v):
|
|
if v is None:
|
|
return None
|
|
s = v if isinstance(v, str) else str(v)
|
|
return s[:length] if length and len(s) > length else s
|
|
return conv
|
|
|
|
|
|
# --------------------------------------------------------------------------- migrate
|
|
def migrate() -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("--sqlite", default=str(BACKEND / "data" / "ar_aging.db"),
|
|
help="path to the legacy SQLite file")
|
|
ap.add_argument("--dry-run", action="store_true",
|
|
help="inspect the source and print the plan; do not connect to MySQL")
|
|
ap.add_argument("--force", action="store_true",
|
|
help="migrate even if the MySQL tables already contain rows")
|
|
args = ap.parse_args()
|
|
|
|
src_path = Path(args.sqlite)
|
|
src = open_sqlite(src_path)
|
|
have = sqlite_tables(src)
|
|
|
|
log("")
|
|
log(" Migrate SQLite → MySQL")
|
|
log(f" source: {src_path} ({src_path.stat().st_size / 1e9:.2f} GB)")
|
|
wal = src_path.with_name(src_path.name + "-wal")
|
|
if wal.exists() and wal.stat().st_size > 0:
|
|
log(f" note: a {wal.stat().st_size / 1e6:.0f} MB write-ahead log is present and "
|
|
f"will be read as part of the database")
|
|
log("")
|
|
|
|
# ---- source inventory (works with no MySQL at all) ----
|
|
log(" Source contents")
|
|
src_counts: dict[str, int] = {}
|
|
for t in sorted(have):
|
|
src_counts[t] = sqlite_count(src, t)
|
|
for t, n in sorted(src_counts.items(), key=lambda kv: -kv[1]):
|
|
if n:
|
|
log(f" {t:22} {n:>10,}")
|
|
empty = [t for t, n in src_counts.items() if not n]
|
|
if empty:
|
|
log(f" ({len(empty)} empty: {', '.join(sorted(empty))})")
|
|
|
|
src_checks = financial_checksums_sqlite(src)
|
|
log("")
|
|
log(" Financial checksums to preserve")
|
|
for k, v in src_checks.items():
|
|
log(f" {k:34} {v}")
|
|
|
|
if args.dry_run:
|
|
log("")
|
|
log(" Dry run — MySQL was not contacted and nothing was written.")
|
|
log(" Fill in ar-aging-app/.env, then re-run without --dry-run.")
|
|
return 0
|
|
|
|
# ---- target ----
|
|
try:
|
|
from app.db.database import ENGINE, init_db
|
|
from app.db import models # noqa: F401
|
|
except Exception as e: # noqa: BLE001
|
|
log("")
|
|
log(f" Could not connect to MySQL: {e}")
|
|
log(" Check MYSQL_* in ar-aging-app/.env and that the server is reachable.")
|
|
return 1
|
|
|
|
log("")
|
|
log(" Creating the MySQL schema (safe if it already exists)…")
|
|
init_db()
|
|
|
|
meta = models.Base.metadata
|
|
ordered = [t for t in meta.sorted_tables if t.name in have and t.name not in SKIP_TABLES]
|
|
missing_in_sqlite = [t.name for t in meta.sorted_tables if t.name not in have]
|
|
if missing_in_sqlite:
|
|
log(f" tables absent from the SQLite file (created empty): "
|
|
f"{', '.join(missing_in_sqlite)}")
|
|
|
|
raw = ENGINE.raw_connection()
|
|
cur = raw.cursor()
|
|
# Existing rows?
|
|
non_empty = []
|
|
for t in ordered:
|
|
cur.execute(f"SELECT COUNT(*) FROM `{t.name}`")
|
|
n = cur.fetchone()[0]
|
|
if n:
|
|
non_empty.append((t.name, n))
|
|
if non_empty and not args.force:
|
|
log("")
|
|
log(" MySQL already contains data — refusing to migrate on top of it:")
|
|
for name, n in non_empty:
|
|
log(f" {name:22} {n:>10,} rows")
|
|
log("")
|
|
log(" Re-run with --force to add these rows anyway (duplicates are possible),")
|
|
log(" or empty the MySQL database first.")
|
|
return 1
|
|
|
|
log("")
|
|
log(" Copying tables (foreign-key order)")
|
|
cur.execute("SET FOREIGN_KEY_CHECKS=0")
|
|
cur.execute("SET UNIQUE_CHECKS=0")
|
|
truncated: list[str] = []
|
|
copied: dict[str, int] = {}
|
|
try:
|
|
for table in ordered:
|
|
name = table.name
|
|
total = src_counts.get(name, 0)
|
|
if not total:
|
|
copied[name] = 0
|
|
continue
|
|
sq_cols = set(sqlite_columns(src, name))
|
|
cols = [c for c in table.columns if c.name in sq_cols]
|
|
dropped = [c.name for c in table.columns if c.name not in sq_cols]
|
|
extra = sq_cols - {c.name for c in table.columns}
|
|
convs = [make_converter(c.type) for c in cols]
|
|
names = [c.name for c in cols]
|
|
placeholders = ", ".join(["%s"] * len(names))
|
|
collist = ", ".join(f"`{n}`" for n in names)
|
|
sql = f"INSERT INTO `{name}` ({collist}) VALUES ({placeholders})"
|
|
|
|
done = 0
|
|
batch: list[tuple] = []
|
|
for row in src.execute(f"SELECT {', '.join(f'`{n}`' for n in names)} "
|
|
f"FROM `{name}`"):
|
|
vals = []
|
|
for i, conv in enumerate(convs):
|
|
v = conv(row[i])
|
|
vals.append(v)
|
|
batch.append(tuple(vals))
|
|
if len(batch) >= BATCH:
|
|
cur.executemany(sql, batch)
|
|
raw.commit()
|
|
done += len(batch)
|
|
batch.clear()
|
|
if total > 50000:
|
|
pct = 100.0 * done / total
|
|
print(f" {name:22} {done:>10,} / {total:,} ({pct:5.1f}%)",
|
|
end="\r", flush=True)
|
|
if batch:
|
|
cur.executemany(sql, batch)
|
|
raw.commit()
|
|
done += len(batch)
|
|
copied[name] = done
|
|
note = ""
|
|
if dropped:
|
|
note += f" [not in source: {', '.join(dropped)}]"
|
|
if extra:
|
|
note += f" [source-only, skipped: {', '.join(sorted(extra))}]"
|
|
truncated.append(name)
|
|
print(" " * 78, end="\r")
|
|
log(f" {name:22} {done:>10,}{note}")
|
|
finally:
|
|
cur.execute("SET FOREIGN_KEY_CHECKS=1")
|
|
cur.execute("SET UNIQUE_CHECKS=1")
|
|
raw.commit()
|
|
|
|
# ---- verify ----
|
|
log("")
|
|
log(" Verifying")
|
|
ok = True
|
|
for name, n in sorted(copied.items()):
|
|
cur.execute(f"SELECT COUNT(*) FROM `{name}`")
|
|
got = cur.fetchone()[0]
|
|
want = src_counts.get(name, 0)
|
|
if got != want:
|
|
ok = False
|
|
log(f" ✗ {name:22} MySQL {got:,} != SQLite {want:,}")
|
|
if ok:
|
|
log(f" ✓ row counts match on all {len(copied)} tables")
|
|
|
|
dst_checks = financial_checksums_mysql(cur)
|
|
for k, want in src_checks.items():
|
|
got = dst_checks.get(k)
|
|
if str(got) != str(want):
|
|
ok = False
|
|
log(f" ✗ {k}: MySQL {got} != SQLite {want}")
|
|
if ok:
|
|
log(" ✓ financial checksums match")
|
|
|
|
cur.close()
|
|
raw.close()
|
|
src.close()
|
|
|
|
log("")
|
|
if ok:
|
|
log(" Migration complete. Start the dashboard with start.command.")
|
|
return 0
|
|
log(" Migration finished with MISMATCHES — do not rely on the MySQL data until")
|
|
log(" they are explained. The SQLite file is untouched.")
|
|
return 1
|
|
|
|
|
|
# --------------------------------------------------------------------------- checksums
|
|
def financial_checksums_sqlite(conn: sqlite3.Connection) -> dict[str, str]:
|
|
out: dict[str, str] = {}
|
|
tables = sqlite_tables(conn)
|
|
|
|
def one(sql: str, default="—") -> str:
|
|
try:
|
|
r = conn.execute(sql).fetchone()
|
|
return "—" if r is None or r[0] is None else str(r[0])
|
|
except sqlite3.Error:
|
|
return default
|
|
|
|
if "sessions" in tables:
|
|
out["sessions"] = one("SELECT COUNT(*) FROM sessions")
|
|
if "transactions" in tables:
|
|
out["transaction rows"] = one("SELECT COUNT(*) FROM transactions")
|
|
out["Σ transactions.total"] = one("SELECT ROUND(SUM(total),2) FROM transactions")
|
|
out["receivable-flagged rows"] = one(
|
|
"SELECT COUNT(*) FROM transactions WHERE receivable_flag=1")
|
|
if "receivable_results" in tables:
|
|
out["USA receivable_local (TOTAL)"] = one(
|
|
"SELECT ROUND(receivable_local) FROM receivable_results "
|
|
"WHERE marketplace='USA' AND account_type='TOTAL'")
|
|
out["Σ receivable_usd (TOTAL rows)"] = one(
|
|
"SELECT ROUND(SUM(receivable_usd),2) FROM receivable_results "
|
|
"WHERE account_type='TOTAL'")
|
|
return out
|
|
|
|
|
|
def financial_checksums_mysql(cur) -> dict[str, str]:
|
|
out: dict[str, str] = {}
|
|
|
|
def one(sql: str) -> str:
|
|
try:
|
|
cur.execute(sql)
|
|
r = cur.fetchone()
|
|
return "—" if r is None or r[0] is None else str(r[0])
|
|
except Exception: # noqa: BLE001
|
|
return "—"
|
|
|
|
out["sessions"] = one("SELECT COUNT(*) FROM sessions")
|
|
out["transaction rows"] = one("SELECT COUNT(*) FROM transactions")
|
|
out["Σ transactions.total"] = one("SELECT ROUND(SUM(total),2) FROM transactions")
|
|
out["receivable-flagged rows"] = one(
|
|
"SELECT COUNT(*) FROM transactions WHERE receivable_flag=1")
|
|
out["USA receivable_local (TOTAL)"] = one(
|
|
"SELECT ROUND(receivable_local) FROM receivable_results "
|
|
"WHERE marketplace='USA' AND account_type='TOTAL'")
|
|
out["Σ receivable_usd (TOTAL rows)"] = one(
|
|
"SELECT ROUND(SUM(receivable_usd),2) FROM receivable_results "
|
|
"WHERE account_type='TOTAL'")
|
|
return out
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(migrate())
|
|
except KeyboardInterrupt:
|
|
log("\n Interrupted. The SQLite source is unchanged.")
|
|
raise SystemExit(130)
|