RJBULK-APP/scripts/migrate_from_sqlite.py

#!/usr/bin/env python3
"""
One-shot + incremental migration: SQLite dump → PocketBase.

Order:
  1. Suppliers         (no deps)
  2. Customers         (no deps)
  3. Customer pricing  (FK to customers — uses the map from step 2)
  4. Fuel purchases    (FK to suppliers — uses the map from step 1)
  5. Delivery notes    (FK to customers — uses the map from step 2)

IDEMPOTENT: All record IDs are deterministic — derived from the SQLite primary
key. Re-running on a populated PB is safe: the existence check will match the
already-imported records and skip them. Use this for incremental sync: after
the legacy app adds new rows to SQLite, just re-run this script and it picks
up only the new ones.

Usage:
  python3 scripts/migrate_from_sqlite.py [path/to/db.sqlite3]

  Optional env:
    POCKETBASE_URL   default: http://pocketbase-zinlbpsext2ro0lgb2558vfy.169.239.180.130.sslip.io
    PB_ADMIN_EMAIL   default: David.els1@outlook.com
    PB_ADMIN_PASS    default: 1234567890
"""

import json
import os
import sqlite3
import sys
import time
import urllib.request
import urllib.error
from concurrent.futures import ThreadPoolExecutor, as_completed

SQLITE_PATH = sys.argv[1] if len(sys.argv) > 1 else "/Volumes/KIOXIA/Downloads/db.sqlite3"
PB_URL = os.environ.get("POCKETBASE_URL", "http://pocketbase-zinlbpsext2ro0lgb2558vfy.169.239.180.130.sslip.io")
ADMIN_EMAIL = os.environ.get("PB_ADMIN_EMAIL", "David.els1@outlook.com")
ADMIN_PASS = os.environ.get("PB_ADMIN_PASS", "1234567890")

# Concurrent requests during bulk inserts. PB + the ss-lip tunnel can
# flake under heavy concurrency, so we keep this low.
BATCH_WORKERS = 5
RETRY_ON_TIMEOUT = 3

# Legacy status values that map to a non-empty value in the new app.
# The new app's status enum is: Pending | EFT | CASH | CARD. Everything else
# (Statement Viewed, Report Viewed, Invoiced) becomes "" (treated as "done")
# but the original value is preserved in `old_status` for reference.
LEGACY_STATUS_PASSTHROUGH = {"Pending", "EFT", "CASH", "CARD"}


def log(msg):
    print(f"[migrate] {msg}", flush=True)


def deterministic_id(prefix, pk):
    """Deterministic 15-char base36 ID — same input always produces the same ID."""
    body = f"{prefix}{pk}".encode("utf-8")
    h = 0
    for b in body:
        h = (h * 31 + b) & 0xFFFFFFFFFFFFFFFF
    chars = "abcdefghijklmnopqrstuvwxyz0123456789"
    out = prefix
    for _ in range(15 - len(prefix)):
        h = (h * 1103515245 + 12345) & 0x7FFFFFFF
        out += chars[h % 36]
    return out


def clean(obj):
    out = {}
    for k, v in obj.items():
        if v == "" or v is None:
            continue
        out[k] = v
    return out


def pb_request(method, path, token=None, body=None):
    url = f"{PB_URL}{path}"
    data = json.dumps(body).encode("utf-8") if body is not None else None
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Content-Type", "application/json")
    if token:
        req.add_header("Authorization", token)
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            raw = resp.read()
            return resp.status, (json.loads(raw.decode("utf-8")) if raw else None)
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")
        return e.code, body


def get_auth_token():
    code, data = pb_request("POST", "/api/collections/_superusers/auth-with-password", body={
        "identity": ADMIN_EMAIL,
        "password": ADMIN_PASS,
    })
    if code != 200:
        raise SystemExit(f"Auth failed: {code} {data}")
    return data["token"]


def get_existing_ids(token, collection, prefix=None):
    """Return set of IDs that exist. If prefix is set, filter to that prefix for speed."""
    ids = set()
    page = 1
    while True:
        path = f"/api/collections/{collection}/records?page={page}&perPage=500&fields=id"
        code, data = pb_request("GET", path, token=token)
        if code != 200 or not isinstance(data, dict) or not data.get("items"):
            break
        for r in data["items"]:
            if prefix is None or r["id"].startswith(prefix):
                ids.add(r["id"])
        if page >= data.get("totalPages", 1):
            break
        page += 1
    return ids


def list_existing_numbers(token, collection):
    """For delivery_notes: we need to know which `number` values are taken so we
    avoid the unique-index collision when a legacy number is duplicated."""
    nums = set()
    page = 1
    while True:
        path = f"/api/collections/{collection}/records?page={page}&perPage=500&fields=number"
        code, data = pb_request("GET", path, token=token)
        if code != 200 or not isinstance(data, dict) or not data.get("items"):
            break
        for r in data["items"]:
            if r.get("number"):
                nums.add(str(r["number"]))
        if page >= data.get("totalPages", 1):
            break
        page += 1
    return nums


def create_record(token, collection, body, retries=RETRY_ON_TIMEOUT):
    """Create a record, retrying on connection/timeout errors."""
    last_err = None
    for attempt in range(retries):
        try:
            code, data = pb_request("POST", f"/api/collections/{collection}/records", token=token, body=body)
            return code, data
        except (urllib.error.URLError, TimeoutError) as e:
            last_err = e
            if attempt < retries - 1:
                time.sleep(0.5 * (attempt + 1))
    return 0, f"URLError after {retries} retries: {last_err}"


def fetch_rows(sql, table, where=None, order_by=None):
    cur = sql.execute(f"SELECT * FROM {table}" + (f" WHERE {where}" if where else "") + (f" ORDER BY {order_by}" if order_by else ""))
    cols = [d[0] for d in cur.description]
    return [dict(zip(cols, row)) for row in cur.fetchall()]


def build_customer_index(name_to_id):
    """Return (lower_name, id) sorted by name length DESC so longest match wins."""
    items = [(n.lower(), i) for n, i in name_to_id.items() if n and i]
    items.sort(key=lambda x: (-len(x[0]), x[0]))
    return items


def match_customer(legacy_name, index):
    """Best-effort match: exact (case-insensitive) → legacy is a prefix of a PB name."""
    if not legacy_name:
        return None
    key = legacy_name.strip().lower()
    for n, cid in index:
        if n == key:
            return cid
    for n, cid in index:
        if len(key) >= 3 and (n.startswith(key + " ") or n.startswith(key)) and len(n) > len(key):
            return cid
    return None


def migrate_suppliers(sql, token):
    log("Suppliers…")
    rows = fetch_rows(sql, "supplier")
    existing = get_existing_ids(token, "suppliers")
    name_to_id = {}
    created, skipped = 0, 0
    for row in rows:
        name = (row.get("supplier_name") or "").strip()
        new_id = deterministic_id("sup", row["id"])
        if new_id in existing:
            skipped += 1
        else:
            body = clean({
                "id": new_id,
                "name": name or "(unnamed)",
                "contact_number": row.get("contact_number") or "",
                "email": row.get("email") or "",
                "address": row.get("address") or "",
                "vat_number": row.get("vat_number") or "",
            })
            code, _ = create_record(token, "suppliers", body)
            if code == 200:
                created += 1
        if name:
            name_to_id[name] = new_id
    # Augment with pre-existing suppliers in PB (so we can map legacy names that were already created outside this script)
    page = 1
    while True:
        code, data = pb_request("GET", f"/api/collections/suppliers/records?page={page}&perPage=500&fields=id,name", token=token)
        if code != 200 or not isinstance(data, dict) or not data.get("items"):
            break
        for r in data["items"]:
            n = (r.get("name") or "").strip()
            if n and n not in name_to_id:
                name_to_id[n] = r["id"]
        if page >= data.get("totalPages", 1):
            break
        page += 1
    log(f"  suppliers: created={created} skipped={skipped} (of {len(rows)} in sqlite)")
    return created, skipped, name_to_id


def migrate_customers(sql, token):
    log("Customers…")
    rows = fetch_rows(sql, "customer", where="customer_name IS NOT NULL AND customer_name != ''")
    existing = get_existing_ids(token, "customers")
    created, skipped, failed = 0, 0, 0
    old_id_to_new = {}
    name_to_id = {}
    for row in rows:
        name = (row.get("customer_name") or "").strip()
        if not name:
            continue
        new_id = deterministic_id("cu", row["id"])
        if new_id in existing:
            skipped += 1
        else:
            body = clean({
                "id": new_id,
                "name": name,
                "contact_number": row.get("contact_number") or "",
                "email": row.get("email") or "",
                "address": row.get("address") or "",
                "vat_number": row.get("vat_number") or "",
            })
            code, _ = create_record(token, "customers", body)
            if code == 200:
                created += 1
            else:
                failed += 1
        old_id_to_new[str(row["id"])] = new_id
        name_to_id[name] = new_id
    page = 1
    while True:
        code, data = pb_request("GET", f"/api/collections/customers/records?page={page}&perPage=500&fields=id,name", token=token)
        if code != 200 or not isinstance(data, dict) or not data.get("items"):
            break
        for r in data["items"]:
            n = (r.get("name") or "").strip()
            if n and n not in name_to_id:
                name_to_id[n] = r["id"]
        if page >= data.get("totalPages", 1):
            break
        page += 1
    log(f"  customers: created={created} skipped={skipped} failed={failed} (of {len(rows)} valid in sqlite)")
    return created, skipped, old_id_to_new, name_to_id


def migrate_pricing(sql, token, old_id_to_new):
    log("Customer pricing…")
    rows = fetch_rows(sql, "customer_pricing")
    existing = get_existing_ids(token, "customer_pricing")
    created, skipped, missing = 0, 0, 0
    for row in rows:
        new_cust = old_id_to_new.get(str(row["customer_id"]))
        if not new_cust:
            missing += 1
            continue
        new_id = deterministic_id("pr", row["id"])
        if new_id in existing:
            skipped += 1
            continue
        body = clean({
            "id": new_id,
            "customer": new_cust,
            "year": int(row["year"]),
            "month": int(row["month"]),
            "price_per_litre": float(row["price_per_litre"]),
        })
        if row.get("special_price") is not None:
            body["special_price"] = float(row["special_price"])
        code, _ = create_record(token, "customer_pricing", body)
        if code == 200:
            created += 1
    log(f"  customer_pricing: created={created} skipped={skipped} dropped_no_customer={missing} (of {len(rows)} in sqlite)")
    return created, skipped, missing


def migrate_fuel_purchases(sql, token, supplier_name_to_id):
    log("Fuel purchases…")
    rows = fetch_rows(sql, "fuel_purchase")
    existing = get_existing_ids(token, "fuel_purchases")
    created, skipped, missing = 0, 0, 0
    lc_lookup = {k.lower(): v for k, v in supplier_name_to_id.items() if v}
    for row in rows:
        legacy_name = (row.get("supplier_name") or "").strip()
        match = lc_lookup.get(legacy_name.lower())
        if not match:
            missing += 1
            log(f"    !! no supplier named '{legacy_name}' — skipping fuel_purchase id={row.get('id')}")
            continue
        new_id = deterministic_id("fp", row["id"])
        if new_id in existing:
            skipped += 1
            continue
        body = clean({
            "id": new_id,
            "supplier": match,
            "purchase_date": row.get("purchase_date") or "",
            "purchase_qty": float(row.get("purchase_qty") or 0),
            "purchase_price": float(row.get("purchase_price") or 0),
        })
        code, _ = create_record(token, "fuel_purchases", body)
        if code == 200:
            created += 1
    log(f"  fuel_purchases: created={created} skipped={skipped} dropped_no_supplier={missing} (of {len(rows)} in sqlite)")
    return created, skipped, missing


def migrate_delivery_notes(sql, token, customer_name_to_id):
    log("Delivery notes… (this is the big one — ~23k records, will take a few minutes)")
    rows = fetch_rows(sql, "delivery_note", order_by="id")
    existing = get_existing_ids(token, "delivery_notes", prefix="dn")
    taken_numbers = list_existing_numbers(token, "delivery_notes")
    cust_index = build_customer_index(customer_name_to_id)
    log(f"  preloaded: {len(existing)} already-imported notes, {len(taken_numbers)} taken numbers, {len(cust_index)} PB customers for matching")

    created, skipped, missing_cust, dupes = 0, 0, 0, 0
    to_create = []

    for row in rows:
        legacy_pk = row["id"]
        new_id = deterministic_id("dn", legacy_pk)
        if new_id in existing:
            skipped += 1
            continue
        customer_id = match_customer(row.get("customer_name") or "", cust_index)
        if not customer_id:
            missing_cust += 1
            continue
        old_status = (row.get("status") or "").strip()
        new_status = old_status if old_status in LEGACY_STATUS_PASSTHROUGH else ""
        raw_number = row.get("delivery_note_number")
        number = str(raw_number) if raw_number is not None else f"LEGACY-{legacy_pk}"
        if number in taken_numbers:
            number = f"{number}-{legacy_pk}"
            dupes += 1
        taken_numbers.add(number)
        qty = row.get("qty_litres_supplied")
        if qty is None or qty == 0:
            qty = 1
        date_str = (row.get("delivery_date") or "").split(" ")[0]
        body = {
            "id": new_id,
            "number": number,
            "date": date_str,
            "customer": customer_id,
            "vehicle_registration": row.get("vehicle_registration") or "",
            "qty_litres_supplied": float(qty),
            "plant_dispensing": row.get("plant_dispensing") or "",
            "status": new_status,
            "old_status": old_status,
        }
        if row.get("price_per_litre") is not None:
            body["price_per_litre"] = float(row["price_per_litre"])
        if row.get("total_value") is not None:
            body["total_value"] = float(row["total_value"])
        if row.get("pricing_source"):
            body["pricing_source"] = row["pricing_source"]
        to_create.append((new_id, body))

    log(f"  prepared: {len(to_create)} to create, {skipped} already imported, {missing_cust} skipped (no customer match), {dupes} duplicate numbers suffixed")

    if not to_create:
        return created, skipped, missing_cust, dupes

    def _do_create(args):
        new_id, body = args
        code, _ = create_record(token, "delivery_notes", body)
        return new_id, code

    with ThreadPoolExecutor(max_workers=BATCH_WORKERS) as ex:
        futures = [ex.submit(_do_create, item) for item in to_create]
        done = 0
        ok = 0
        fail = 0
        for f in as_completed(futures):
            done += 1
            new_id, code = f.result()
            if code == 200:
                ok += 1
            else:
                fail += 1
                if fail <= 5:
                    log(f"    !! create failed for {new_id}: {code}")
            if done % 500 == 0:
                log(f"  … progress: {done}/{len(to_create)} processed (ok={ok} fail={fail})")
    created = ok
    log(f"  delivery_notes: created={created} skipped={skipped} dropped_no_customer={missing_cust} duplicates_suffixed={dupes} (of {len(rows)} in sqlite)")
    return created, skipped, missing_cust, dupes


def main():
    if not os.path.exists(SQLITE_PATH):
        raise SystemExit(f"SQLite file not found: {SQLITE_PATH}")
    log(f"Source: {SQLITE_PATH}")
    log(f"Target: {PB_URL}")
    log("Authenticating as superuser…")
    token = get_auth_token()
    log("Authenticated.")

    sql = sqlite3.connect(SQLITE_PATH)
    sql.row_factory = None

    sup = migrate_suppliers(sql, token)
    cust = migrate_customers(sql, token)
    pricing = migrate_pricing(sql, token, cust[2])
    fuel = migrate_fuel_purchases(sql, token, sup[2])
    dn = migrate_delivery_notes(sql, token, cust[3])

    sql.close()
    log("")
    log("=== SUMMARY ===")
    log(f"  Suppliers:        +{sup[0]} new, {sup[1]} skipped")
    log(f"  Customers:        +{cust[0]} new, {cust[1]} skipped, {cust[2]} failed")
    log(f"  Customer pricing: +{pricing[0]} new, {pricing[1]} skipped, {pricing[2]} dropped (no matching customer)")
    log(f"  Fuel purchases:   +{fuel[0]} new, {fuel[1]} skipped, {fuel[2]} dropped (no matching supplier)")
    log(f"  Delivery notes:   +{dn[0]} new, {dn[1]} skipped, {dn[2]} dropped (no customer match), {dn[3]} numbers suffixed (dup collision)")
    log("✅ Migration done.")


if __name__ == "__main__":
    main()