gigafibre-fsm/scripts/migration/migrate_users.py

#!/usr/bin/env python3
"""
Create ERPNext Users from legacy staff + link to Authentik via email.
Authentik forwardAuth sends X-authentik-email header → ERPNext matches by email.
Direct PG. Detached.
"""
import pymysql
import psycopg2
import uuid
from datetime import datetime, timezone
from html import unescape

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

ADMIN = "Administrator"

# Roles for ISP staff
ROLES_ALL = ["System Manager", "Support Team", "Sales User", "Accounts User"]
ROLES_TECH = ["Support Team"]
ROLES_ADMIN = ["System Manager", "Support Team", "Sales User", "Accounts User", "Sales Manager", "Accounts Manager"]

# Staff IDs that are admin/sysadmin
ADMIN_IDS = {2, 4, 11, 12, 3293, 4661, 4664, 4671, 4749}
# Staff IDs that are service accounts (skip)
SKIP_IDS = {1, 3301, 3393, 4663, 4682}

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 clean(val):
    if not val: return ""
    return unescape(str(val)).strip()

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

def main():
    ts = now()
    log("=== User Migration ===")

    mc = pymysql.connect(**LEGACY)
    cur = mc.cursor(pymysql.cursors.DictCursor)
    cur.execute("SELECT * FROM staff WHERE status = 1 ORDER BY id")
    staff = cur.fetchall()
    mc.close()
    log("{} active staff loaded".format(len(staff)))

    pg = psycopg2.connect(**PG)
    pgc = pg.cursor()

    # Existing users
    pgc.execute('SELECT name FROM "tabUser"')
    existing = set(r[0] for r in pgc.fetchall())

    created = skipped = errors = 0

    for s in staff:
        sid = s["id"]
        if sid in SKIP_IDS:
            skipped += 1
            continue

        email = clean(s.get("email"))
        if not email or "@" not in email:
            skipped += 1
            continue

        if email in existing or email.lower() in {e.lower() for e in existing}:
            skipped += 1
            continue

        first = clean(s["first_name"])
        last = clean(s["last_name"])
        full = "{} {}".format(first, last).strip()
        username = clean(s.get("username"))
        cell = clean(s.get("cell"))

        # Determine roles
        if sid in ADMIN_IDS:
            roles = ROLES_ADMIN
        else:
            roles = ROLES_ALL

        try:
            # Create User
            pgc.execute("""
                INSERT INTO "tabUser" (
                    name, creation, modified, modified_by, owner, docstatus, idx,
                    email, first_name, last_name, full_name, username,
                    mobile_no, language, time_zone, enabled,
                    user_type, role_profile_name,
                    new_password
                ) VALUES (
                    %s, %s, %s, %s, %s, 0, 0,
                    %s, %s, %s, %s, %s,
                    %s, 'fr', 'America/Toronto', 1,
                    'System User', NULL,
                    ''
                )
            """, (email, ts, ts, ADMIN, ADMIN,
                  email, first, last or None, full, username,
                  cell or None))

            # Create Role assignments
            for i, role in enumerate(roles):
                pgc.execute("""
                    INSERT INTO "tabHas Role" (
                        name, creation, modified, modified_by, owner, docstatus, idx,
                        role, parent, parentfield, parenttype
                    ) VALUES (%s, %s, %s, %s, %s, 0, %s, %s, %s, 'roles', 'User')
                """, (uid("HR-"), ts, ts, ADMIN, ADMIN, i+1, role, email))

            # Create UserRole for "All"
            pgc.execute("""
                INSERT INTO "tabHas Role" (
                    name, creation, modified, modified_by, owner, docstatus, idx,
                    role, parent, parentfield, parenttype
                ) VALUES (%s, %s, %s, %s, %s, 0, %s, 'All', %s, 'roles', 'User')
            """, (uid("HR-"), ts, ts, ADMIN, ADMIN, len(roles)+1, email))

            created += 1
            log("  OK  {} ({})".format(full, email))

        except Exception as e:
            errors += 1
            pg.rollback()
            log("  ERR {} -> {}".format(email, str(e)[:80]))
            continue

    pg.commit()
    pg.close()

    log("")
    log("=" * 50)
    log("Users: {} created, {} skipped, {} errors".format(created, skipped, errors))
    log("=" * 50)
    log("")
    log("Users are created with NO password (Authentik SSO handles login).")
    log("Next: bench --site erp.gigafibre.ca clear-cache")

if __name__ == "__main__":
    main()