htpc/migrate_lidarr_sqlite_to_postgres.py

#!/usr/bin/env python3
import sqlite3
import csv
import sys
from collections import defaultdict

def escape_pg_string(s):
    if s is None:
        return "NULL"
    if isinstance(s, bytes):
        s = s.decode('utf-8', errors='replace')
    s = str(s).replace("'", "''")
    return f"'{s}'"

def main():
    sqlite_db_path = "/tmp/lidarr.db"
    pg_columns_path = "/tmp/pg_lidarr_columns.csv"
    output_sql_path = "/tmp/migrate_lidarr_data.sql"

    # 1. Lees PG tabellen en kolommen
    pg_schema = defaultdict(dict)
    with open(pg_columns_path, mode='r', encoding='utf-8') as f:
        reader = csv.reader(f)
        for row in reader:
            if not row or len(row) < 3:
                continue
            table, col, dtype = row[0].strip(), row[1].strip(), row[2].strip()
            pg_schema[table][col] = dtype

    # 2. Open SQLite
    conn = sqlite3.connect(f"file:{sqlite_db_path}?mode=ro&immutable=1", uri=True)
    cur = conn.cursor()

    cur.execute("SELECT name FROM sqlite_master WHERE type='table';")
    sqlite_tables = set(r[0] for r in cur.fetchall())

    with open(output_sql_path, mode='w', encoding='utf-8') as out:
        out.write("-- Lidarr SQLite naar PostgreSQL Data Migratie\n")
        out.write("BEGIN;\n")
        out.write("SET session_replication_role = 'replica';\n\n")

        # Tabellen die we NIET moeten overschrijven
        skip_tables = {"VersionInfo"}

        total_rows_imported = 0

        for table, pg_cols in pg_schema.items():
            if table in skip_tables:
                continue

            out.write(f"-- Tabel: {table}\n")
            out.write(f'TRUNCATE TABLE "{table}" CASCADE;\n')

            if table == "DelayProfiles":
                out.write('INSERT INTO "DelayProfiles" ("Id", "EnableUsenet", "EnableTorrent", "PreferredProtocol", "UsenetDelay", "TorrentDelay", "Order", "Tags", "BypassIfHighestQuality", "BypassIfAboveCustomFormatScore", "MinimumCustomFormatScore") VALUES (1, true, true, 1, 0, 0, 2147483647, \'[]\', true, false, 0);\n\n')
                total_rows_imported += 1
                continue

            if table not in sqlite_tables:
                out.write(f"-- Tabel {table} bestaat niet in SQLite, leeggemaakt en overgeslagen.\n\n")
                continue

            # Haal SQLite kolommen op
            cur.execute(f'PRAGMA table_info("{table}");')
            sqlite_col_names = set(r[1] for r in cur.fetchall())

            # Gemeenschappelijke kolommen
            common_cols = [c for c in pg_cols.keys() if c in sqlite_col_names]
            if not common_cols:
                out.write(f"-- Geen gemeenschappelijke kolommen voor {table}\n\n")
                continue

            # Query data uit SQLite
            cols_quoted = [f'"{c}"' for c in common_cols]
            cur.execute(f'SELECT {", ".join(cols_quoted)} FROM "{table}";')
            rows = cur.fetchall()

            if not rows:
                out.write(f"-- Geen rijen in SQLite voor {table}\n\n")
                continue

            out.write(f'-- {len(rows)} rijen migreren voor {table}\n')
            total_rows_imported += len(rows)

            chunk_size = 500
            for i in range(0, len(rows), chunk_size):
                chunk = rows[i:i + chunk_size]
                out.write(f'INSERT INTO "{table}" ({", ".join([f"""\"{c}\"""" for c in common_cols])}) VALUES\n')
                val_lines = []
                for row_data in chunk:
                    val_parts = []
                    for val, col_name in zip(row_data, common_cols):
                        dtype = pg_cols[col_name].lower()
                        if val is None:
                            val_parts.append("NULL")
                        elif "bool" in dtype:
                            val_parts.append("TRUE" if str(val) in ("1", "True", "true") else "FALSE")
                        elif dtype in ("integer", "bigint", "smallint", "numeric", "double precision", "real"):
                            s = str(val).strip()
                            if s == "TorrentDownloadProtocol":
                                val_parts.append("2")
                            elif s == "SoulseekDownloadProtocol":
                                val_parts.append("3")
                            elif s == "UsenetDownloadProtocol":
                                val_parts.append("1")
                            else:
                                try:
                                    val_parts.append(str(int(s)))
                                except ValueError:
                                    try:
                                        float(s)
                                        val_parts.append(s)
                                    except ValueError:
                                        val_parts.append("0")
                        else:
                            val_parts.append(escape_pg_string(val))
                    val_lines.append(f'  ({", ".join(val_parts)})')
                out.write(",\n".join(val_lines) + ";\n")
            out.write("\n")

        # Sequences resetten
        out.write("-- Sequences bijwerken naar MAX(Id)\n")
        out.write("""DO $$
DECLARE
    r RECORD;
    has_seq text;
BEGIN
    FOR r IN SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name != 'VersionInfo' LOOP
        IF EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema = 'public' AND table_name = r.table_name AND column_name = 'Id') THEN
            has_seq := pg_get_serial_sequence('"' || r.table_name || '"', 'Id');
            IF has_seq IS NOT NULL THEN
                EXECUTE format('SELECT setval(%L, coalesce(max("Id"), 1)) FROM %I', has_seq, r.table_name);
            END IF;
        END IF;

    END LOOP;
END $$;\n\n""")

        out.write("SET session_replication_role = 'origin';\n")
        out.write("COMMIT;\n")

    print(f"SQL dump succesvol gegenereerd: {output_sql_path}")
    print(f"Totaal aantal rijen klaargezet voor import: {total_rows_imported}")

if __name__ == "__main__":
    main()