gigafibre-fsm/scripts/migration/migrate_phase3.py

#!/usr/bin/env python3
"""
Phase 3b: Create Subscription Plans from active products.
Phase 4: Create Subscriptions from active services.

Direct PostgreSQL — runs detached inside erpnext-backend-1.
Log: /tmp/migrate_phase3.log
"""
import pymysql
import psycopg2
import uuid
from datetime import datetime, timezone

LEGACY = {"host": "legacy-db", "user": "facturation", "password": "VD67owoj",
          "database": "gestionclient", "connect_timeout": 30, "read_timeout": 300}
PG = {"host": "db", "port": 5432, "user": "postgres", "password": "123",
      "dbname": "_eb65bdc0c4b1b2d6"}

ADMIN = "Administrator"
COMPANY = "TARGO"
TAX_TEMPLATE = "QC TPS 5% + TVQ 9.975% - T"

# Recurring categories (monthly billing)
RECURRING_CATS = {4, 9, 17, 21, 32, 33}  # sans fil, téléphonie, IP fixe, P2P, fibre, TV

def uid(prefix=""):
    return prefix + uuid.uuid4().hex[:10]

def now():
    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f")

def log(msg):
    print(msg, flush=True)

def main():
    ts = now()
    log("=== Phase 3b+4: Subscription Plans + Subscriptions ===")

    # 1. Read legacy data
    log("Reading legacy data...")
    mc = pymysql.connect(**LEGACY)
    cur = mc.cursor(pymysql.cursors.DictCursor)

    # Products that have active services
    cur.execute("""
        SELECT p.id, p.sku, p.price, p.category,
               COUNT(s.id) as active_count
        FROM product p
        JOIN service s ON s.product_id = p.id AND s.status = 1
        WHERE p.category IN (4,9,17,21,32,33)
        GROUP BY p.id
        HAVING active_count > 0
        ORDER BY active_count DESC
    """)
    products = cur.fetchall()
    log("  {} recurring products with active services".format(len(products)))

    # Active services with customer mapping
    cur.execute("""
        SELECT s.id as service_id, s.product_id, s.delivery_id, s.device_id,
               s.date_orig, s.date_next_invoice, s.payment_recurrence,
               s.hijack, s.hijack_price, s.hijack_desc,
               s.radius_user, s.radius_pwd, s.date_end_contract,
               d.account_id
        FROM service s
        JOIN delivery d ON s.delivery_id = d.id
        WHERE s.status = 1
        AND s.product_id IN (SELECT id FROM product WHERE category IN (4,9,17,21,32,33))
        ORDER BY s.id
    """)
    services = cur.fetchall()
    log("  {} active recurring services".format(len(services)))

    # Product SKU lookup
    cur.execute("SELECT id, sku FROM product")
    sku_map = {r["id"]: r["sku"] for r in cur.fetchall()}
    mc.close()
    log("  Legacy DB closed.")

    # 2. Connect PostgreSQL
    log("Connecting to ERPNext PostgreSQL...")
    pg = psycopg2.connect(**PG)
    pgc = pg.cursor()

    # Get customer mapping: legacy_account_id → ERPNext customer name
    pgc.execute('SELECT legacy_account_id, name FROM "tabCustomer" WHERE legacy_account_id > 0')
    cust_map = {r[0]: r[1] for r in pgc.fetchall()}
    log("  {} customers mapped".format(len(cust_map)))

    # Check existing subscription plans
    pgc.execute('SELECT plan_name FROM "tabSubscription Plan"')
    existing_plans = set(r[0] for r in pgc.fetchall())

    # Check existing subscriptions by legacy_service_id
    pgc.execute('SELECT legacy_service_id FROM "tabSubscription" WHERE legacy_service_id IS NOT NULL AND legacy_service_id > 0')
    existing_subs = set(r[0] for r in pgc.fetchall())
    log("  {} plans exist, {} subscriptions exist".format(len(existing_plans), len(existing_subs)))

    # ============================
    # Phase 3b: Subscription Plans
    # ============================
    log("")
    log("--- Phase 3b: Creating Subscription Plans ---")
    plans_created = 0

    for p in products:
        sku = sku_map.get(p["id"], "UNKNOWN")
        plan_name = "PLAN-{}".format(sku)

        if plan_name in existing_plans:
            continue

        price = float(p["price"] or 0)
        plan_id = uid("SP-")

        try:
            pgc.execute("""
                INSERT INTO "tabSubscription Plan" (
                    name, creation, modified, modified_by, owner, docstatus, idx,
                    plan_name, item, currency, price_determination, cost,
                    billing_interval, billing_interval_count
                ) VALUES (%s, %s, %s, %s, %s, 0, 0,
                          %s, %s, 'CAD', 'Fixed Rate', %s, 'Month', 1)
            """, (plan_id, ts, ts, ADMIN, ADMIN, plan_name, sku, price))
            plans_created += 1
            existing_plans.add(plan_name)
        except Exception as e:
            pg.rollback()
            log("  ERR plan {} -> {}".format(sku, str(e)[:80]))

    pg.commit()
    log("  {} Subscription Plans created".format(plans_created))

    # Build plan lookup: product_id → plan_name
    plan_by_product = {}
    for p in products:
        sku = sku_map.get(p["id"], "UNKNOWN")
        plan_by_product[p["id"]] = "PLAN-{}".format(sku)

    # ============================
    # Phase 4: Subscriptions
    # ============================
    log("")
    log("--- Phase 4: Creating Subscriptions ---")
    sub_ok = sub_skip = sub_err = 0

    for i, s in enumerate(services):
        sid = s["service_id"]

        if sid in existing_subs:
            sub_skip += 1
            continue

        acct_id = s["account_id"]
        cust_name = cust_map.get(acct_id)
        if not cust_name:
            sub_err += 1
            continue

        product_id = s["product_id"]
        plan_name = plan_by_product.get(product_id)
        if not plan_name:
            sub_err += 1
            continue

        # Convert unix timestamps to dates
        start_date = None
        if s["date_orig"] and s["date_orig"] > 0:
            try:
                start_date = datetime.fromtimestamp(s["date_orig"], tz=timezone.utc).strftime("%Y-%m-%d")
            except (ValueError, OSError):
                start_date = "2020-01-01"
        if not start_date:
            start_date = "2020-01-01"

        sub_id = uid("SUB-")

        # Discount from hijack
        discount_amt = 0
        if s["hijack"] and s["hijack_price"]:
            discount_amt = abs(float(s["hijack_price"]))

        try:
            pgc.execute("""
                INSERT INTO "tabSubscription" (
                    name, creation, modified, modified_by, owner, docstatus, idx,
                    party_type, party, company, status,
                    start_date, generate_invoice_at, days_until_due,
                    follow_calendar_months, generate_new_invoices_past_due_date,
                    submit_invoice, cancel_at_period_end,
                    sales_tax_template,
                    additional_discount_amount,
                    radius_user, radius_pwd, legacy_service_id
                ) VALUES (
                    %s, %s, %s, %s, %s, 0, 0,
                    'Customer', %s, %s, 'Active',
                    %s, 'Beginning of the current subscription period', 30,
                    1, 1, 0, 0,
                    %s, %s,
                    %s, %s, %s
                )
            """, (sub_id, ts, ts, ADMIN, ADMIN,
                  cust_name, COMPANY,
                  start_date,
                  TAX_TEMPLATE,
                  discount_amt,
                  s.get("radius_user") or None,
                  s.get("radius_pwd") or None,
                  sid))

            # Subscription Plan Detail (child table linking plan to subscription)
            pgc.execute("""
                INSERT INTO "tabSubscription Plan Detail" (
                    name, creation, modified, modified_by, owner, docstatus, idx,
                    plan, qty,
                    parent, parentfield, parenttype
                ) VALUES (%s, %s, %s, %s, %s, 0, 1,
                          %s, 1,
                          %s, 'plans', 'Subscription')
            """, (uid("SPD-"), ts, ts, ADMIN, ADMIN, plan_name, sub_id))

            sub_ok += 1

        except Exception as e:
            sub_err += 1
            pg.rollback()
            if sub_err <= 20:
                log("  ERR svc#{} -> {}".format(sid, str(e)[:100]))
            continue

        if sub_ok % 500 == 0:
            pg.commit()
            log("  [{}/{}] created={} skip={} err={}".format(
                i+1, len(services), sub_ok, sub_skip, sub_err))

    pg.commit()
    pg.close()

    log("")
    log("=" * 60)
    log("Subscription Plans: {} created".format(plans_created))
    log("Subscriptions: {} created, {} skipped, {} errors".format(sub_ok, sub_skip, sub_err))
    log("=" * 60)
    log("")
    log("Next: bench --site erp.gigafibre.ca clear-cache")

if __name__ == "__main__":
    main()