#!/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()