RECORDGOD/migrate.py

#!/usr/bin/env python3
"""One-shot MariaDB (WowPlatter) -> Postgres (RecordGod) migration.

Migrates the shop's OWN data only — inventory (two tables unified into one), sales history,
and a slim crate table. The ~3.2M-row Discogs CATALOG is deliberately NOT copied: it lives in
the discogs_full mirror and is joined on release_id at read time. See RECORDGOD_PLAN.md.

    python migrate.py            # idempotent: create schema, upsert (ON CONFLICT DO NOTHING)
    python migrate.py --truncate # wipe the 4 tables first, then reload
"""
import json
import os
import pathlib
import re
import sys
from datetime import datetime, timezone

import pymysql
import psycopg
from psycopg.types.json import Jsonb

WP_CONFIG = os.getenv("WP_CONFIG", "/opt/homebrew/var/www/monsterrobot.localsite/wp-config.php")
PG_DSN = os.getenv("RECORDGOD_DSN", "postgresql://localhost/recordgod")
SOCK = os.getenv("MARIA_SOCK", "/opt/homebrew/var/mysql/mysql.sock")
HERE = pathlib.Path(__file__).parent


def wp_creds():
    t = pathlib.Path(WP_CONFIG).read_text()
    g = lambda k: re.search(rf"'{k}',\s*'([^']*)'", t).group(1)
    return g("DB_NAME"), g("DB_USER"), g("DB_PASSWORD")


def yn(v):
    return v == "y"


def js(v):
    if not v:
        return None
    try:
        return json.loads(v)
    except Exception:
        return None  # ponytail: skip a malformed json blob rather than abort the whole migration


def jb(v):
    return Jsonb(v) if v is not None else None


def insert_many(pg, table, cols, dicts):
    rows = [tuple(d.get(c) for c in cols) for d in dicts]
    if not rows:
        return 0
    ph = ",".join(["%s"] * len(cols))
    sql = f"INSERT INTO {table} ({','.join(cols)}) VALUES ({ph}) ON CONFLICT DO NOTHING"
    with pg.cursor() as cur:
        cur.executemany(sql, rows)
    return len(rows)


def main():
    truncate = "--truncate" in sys.argv
    now = datetime.now(timezone.utc)
    name, user, pw = wp_creds()
    my = pymysql.connect(unix_socket=SOCK, user=user, password=pw, database=name,
                         cursorclass=pymysql.cursors.DictCursor)
    pg = psycopg.connect(PG_DSN, autocommit=False)
    pg.execute(open(HERE / "schema.sql").read())
    pg.commit()
    if truncate:
        pg.execute("TRUNCATE inventory, sale_items, sales, crate, customer, sales_discount, sales_setting")

    c = my.cursor()

    # --- inventory: disc_inventory (vinyl) -----------------------------------------------------
    c.execute("SELECT * FROM wp_rmp_disc_inventory")
    vinyl = [dict(
        sku=r["sku"], store_id=1, kind="vinyl", release_id=r["release_id"],
        price=r["price"], qty=1, in_stock=yn(r["instock"]),
        condition=r["media_condition"], sleeve_cond=r["sleeve_condition"],
        location=r["location"], crate_id=r["crate_id"], slot_number=r["slot_number"],
        status="publish" if yn(r["instock"]) else "sold", notes=r["comment"],
        square_item_id=r["square_item_id"], square_catalog_object_id=r["square_catalog_object_id"],
        square_sku=r["square_sku"], square_synced_at=r["square_synced_at"],
        staged=False, sold_date=r["sold_date"], created_at=now, updated_at=now,
    ) for r in c.fetchall()]

    # --- inventory: disc_general_inventory (books/mags/etc.) -----------------------------------
    c.execute("SELECT * FROM wp_rmp_disc_general_inventory")
    general = []
    for g in c.fetchall():
        attrs = js(g["attributes"]) or {}
        dims = js(g["dimensions"])
        if dims:
            attrs["dimensions"] = dims
        general.append(dict(
            sku=g["sku"], store_id=1, kind=(g["type"] or g["category"] or "other").lower(),
            identifier=g["identifier"] or g["upc"], title=g["title"], price=g["price"],
            qty=g["quantity"] or 1, in_stock=yn(g["instock"]), condition=g["condition_desc"],
            brand=g["brand"], model=g["model"], weight_g=g["weight"], location=g["location"],
            crate_id=g["crate_id"], slot_number=g["slot_number"],
            status="publish" if yn(g["instock"]) else "sold",
            square_item_id=g["square_item_id"], square_catalog_object_id=g["square_catalog_object_id"],
            square_sku=g["square_sku"], square_synced_at=g["square_synced_at"],
            est_market_value=g["est_market_value"], lowest_competitor=g["lowest_competitor"],
            target_price=g["target_price"], attributes=jb(attrs or None),
            images=jb(js(g["images"])), raw_json=jb(js(g["raw_json"])), staged=False,
            created_at=g["created_at"] or now, updated_at=g["updated_at"] or g["created_at"] or now,
        ))

    inv_cols = ["sku", "store_id", "kind", "release_id", "identifier", "title", "price", "qty",
                "in_stock", "condition", "sleeve_cond", "brand", "model", "weight_g", "location",
                "crate_id", "slot_number", "status", "notes", "est_market_value",
                "lowest_competitor", "target_price", "square_item_id", "square_catalog_object_id",
                "square_sku", "square_synced_at", "staged", "attributes", "images", "raw_json",
                "sold_date", "created_at", "updated_at"]

    # --- crate (slim) --------------------------------------------------------------------------
    c.execute("SELECT id, name, label_text, crate_purpose, notes FROM wp_rmp_disc_virtual_crate")
    crates = [dict(id=v["id"], name=v["name"], label_text=v["label_text"],
                   purpose=v["crate_purpose"], notes=v["notes"]) for v in c.fetchall()]

    # --- sales + sale_items (reporting copy; WP owns live sale-state) ---------------------------
    c.execute("SELECT * FROM wp_rmp_disc_sales")
    sales = [dict(id=s["id"], sale_number=s["sale_number"], customer_id=s["customer_id"],
                  total=s["total_amount"] or s["total"], tax_amount=s["tax_amount"],
                  status=s["payment_status"] or s["status"], payment_method=s["payment_method"],
                  sale_date=s["sale_date"], created_at=s["created_at"]) for s in c.fetchall()]
    c.execute("SELECT * FROM wp_rmp_disc_sales_items")
    items = [dict(id=i["id"], sale_id=i["sale_id"], sku=(i["sku"] or None),
                  release_id=i["release_id"], item_name=i["item_name"], qty=i["quantity"],
                  unit_price=i["unit_price"], line_total=i["line_total"]) for i in c.fetchall()]
    # drop orphan line-items that point at a sale that no longer exists (real junk in the source)
    sale_ids = {s["id"] for s in sales}
    n_orphan = sum(1 for i in items if i["sale_id"] not in sale_ids)
    items = [i for i in items if i["sale_id"] in sale_ids]

    # --- customers (the POS customer picker) ---------------------------------------------------
    c.execute("SELECT * FROM wp_rmp_disc_sales_customers")
    customers = [dict(
        id=r["id"], wp_user_id=r["wp_user_id"], first_name=r["first_name"] or "",
        last_name=r["last_name"] or "", email=r["email"], phone=r["phone"],
        address=r.get("address") or r.get("address_line1"), is_guest=bool(r["is_guest"]),
        created_at=r["created_at"] or now,
    ) for r in c.fetchall()]

    # --- named discounts / promotions ----------------------------------------------------------
    c.execute("SELECT * FROM wp_rmp_disc_sales_discounts")
    discounts = [dict(
        id=r["id"], name=r["name"], percentage=r["percentage"], is_active=bool(r["is_active"]),
        manual_active=bool(r["manual_active"]), discount_type=r["discount_type"],
        description=r["description"], bubble_text=r["bubble_text"], bubble_color=r["bubble_color"],
        start_at=r["start_at"], end_at=r["end_at"], created_at=r["created_at"] or now,
    ) for r in c.fetchall()]

    # --- sales settings (currency / tax / default discount) ------------------------------------
    c.execute("SELECT * FROM wp_rmp_disc_sales_settings")
    settings = [dict(setting_key=r["setting_key"], setting_value=r["setting_value"],
                     is_active=bool(r["is_active"]), updated_at=r["updated_at"] or now)
                for r in c.fetchall()]

    # order matters: crate + sales before things that could reference them
    n = {}
    n["crate"] = insert_many(pg, "crate", ["id", "name", "label_text", "purpose", "notes"], crates)
    n["inventory(vinyl)"] = insert_many(pg, "inventory", inv_cols, vinyl)
    n["inventory(general)"] = insert_many(pg, "inventory", inv_cols, general)
    n["sales"] = insert_many(pg, "sales",
                             ["id", "sale_number", "customer_id", "total", "tax_amount", "status",
                              "payment_method", "sale_date", "created_at"], sales)
    n["sale_items"] = insert_many(pg, "sale_items",
                                  ["id", "sale_id", "sku", "release_id", "item_name", "qty",
                                   "unit_price", "line_total"], items)
    n["customer"] = insert_many(pg, "customer",
                                ["id", "wp_user_id", "first_name", "last_name", "email", "phone",
                                 "address", "is_guest", "created_at"], customers)
    n["sales_discount"] = insert_many(pg, "sales_discount",
                                      ["id", "name", "percentage", "is_active", "manual_active",
                                       "discount_type", "description", "bubble_text", "bubble_color",
                                       "start_at", "end_at", "created_at"], discounts)
    n["sales_setting"] = insert_many(pg, "sales_setting",
                                     ["setting_key", "setting_value", "is_active", "updated_at"], settings)
    pg.commit()
    for k, v in n.items():
        print(f"  {k:22} {v:>6}")
    if n_orphan:
        print(f"  (skipped {n_orphan} orphan sale_items pointing at deleted sales)")
    print("done.")


if __name__ == "__main__":
    main()