"""Site -> site migration: rehydrate a legacy Joomla into a fresh target.

The target installation is the contract.  Client-owned entities/configuration
move; target-installed factory/catalog rows do not.  Where a source entity
points at an install-assigned id, the id is translated by semantic identity.

This is intentionally not a dump importer.  The new Joomla/extension schema and
factory data win, while the client's meaning, relationships and visual state are
reconstructed on top of them.
"""

import argparse
import fnmatch
import json
import re

import extension_compat
import mesh
import migration_policy


# Dependency order: a thing is moved only after what it points at.
PHASES = ("users", "styles", "categories", "articles", "menus", "modules",
          "fields", "tags", "extensions")


# Foreign keys we understand.  @ lookups are install-time registries resolved by
# name; normal names point at an id map built by an earlier migration/factory pass.
FKS = {
    "usergroups":          {"parent_id": "usergroups"},
    "categories":          {"parent_id": "categories", "created_user_id": "users",
                            "modified_user_id": "users", "access": "viewlevels"},
    "content":             {"catid": "categories", "created_by": "users",
                            "modified_by": "users", "access": "viewlevels"},
    "menu":                {"parent_id": "menu", "template_style_id": "template_styles",
                            "component_id": "@component", "access": "viewlevels"},
    "modules":             {"access": "viewlevels"},
    "modules_menu":        {"moduleid": "modules", "menuid": "menu"},
    "user_usergroup_map":  {"user_id": "users", "group_id": "usergroups"},
    "content_frontpage":   {"content_id": "content"},
    "fields":              {"group_id": "fields_groups", "catid": "categories",
                            "access": "viewlevels"},
    "fields_groups":       {"access": "viewlevels"},
    "tags":                {"parent_id": "tags", "created_user_id": "users",
                            "access": "viewlevels"},
    "contentitem_tag_map": {"tag_id": "tags", "content_item_id": "content",
                            "type_id": "@content_type"},

    # DPCalendar stores a numeric FK into its factory-seeded ISO catalogue.
    # The country row itself never migrates; this value is translated by code.
    "dpcalendar_locations": {"country": "dpcalendar_countries",
                             "created_by": "users", "modified_by": "users",
                             "access": "viewlevels"},
}


# Semantic identity for client-owned rows.  Re-running is safe: a row already on
# the target under the same key is reused instead of duplicated.
NATURAL_KEY = {
    "users":           ("username",),
    "usergroups":      ("title",),
    "viewlevels":      ("title",),
    "categories":      ("extension", "path"),
    "content":         ("catid", "alias"),
    "menu":            ("menutype", "path"),
    "menu_types":      ("menutype",),
    "template_styles": ("template", "title"),
    "modules":         ("module", "title", "position"),
    "fields":          ("context", "name"),
    "fields_groups":   ("context", "title"),
    "tags":            ("path",),
    "user_usergroup_map":  ("user_id", "group_id"),
    "content_frontpage":   ("content_id",),
    "modules_menu":        ("moduleid", "menuid"),
    "contentitem_tag_map": ("content_item_id", "tag_id"),
}


# Joomla bookkeeping/factory registries handled elsewhere or regenerated.
# content_types/extensions are synchronised by the factory pass before phases;
# they stay here so the normal copy loop never treats them as entity tables.
NEVER_COPY = {"assets", "session", "updates", "update_sites",
              "update_sites_extensions", "extensions", "action_logs",
              "user_keys", "schemas", "content_types"}

WIDGETKIT_PREFIX = "widgetkit"
UPDATE_ON_MATCH = {"template_styles"}
ROW_FILTER = {"template_styles": "client_id = 0"}
_FILTER_COLS = {"template_styles": ("client_id",)}
TREE_TABLES = {"categories", "menu", "tags", "usergroups"}

LOOKUPS = {
    "@component":    ("extensions", "extension_id", "element", "element",
                      "type = 'component'"),
    "@content_type": ("content_types", "type_id", "type_alias", "type_alias", ""),
}


# ---------------------------------------------------------------------------
# schema helpers

def tables(conn, prefix):
    with conn.cursor() as cur:
        cur.execute("SHOW TABLES")
        out = set()
        for row in cur.fetchall():
            name = next(iter(row.values()))
            if name.startswith(prefix):
                out.add(name[len(prefix):])
        return out


def columns(conn, table):
    with conn.cursor() as cur:
        cur.execute(f"SHOW COLUMNS FROM `{table}`")
        return {r["Field"]: r for r in cur.fetchall()}


def shared_columns(scols, dcols):
    return [c for c in scols if c in dcols]


def row_count(conn, table):
    with conn.cursor() as cur:
        try:
            cur.execute(f"SELECT COUNT(*) AS n FROM `{table}`")
            return cur.fetchone()["n"]
        except Exception:
            return 0


def is_tree(cols):
    return {"lft", "rgt", "parent_id"} <= set(cols)


def _first_col(cols, *names):
    for name in names:
        if name in cols:
            return name
    return ""


def _key(row, cols):
    return tuple("" if row.get(c) is None else str(row.get(c)) for c in cols)


def _select_rows(conn, table, wanted):
    wanted = [c for c in wanted if c]
    if not wanted:
        return []
    with conn.cursor() as cur:
        cur.execute("SELECT " + ", ".join(f"`{c}`" for c in wanted) + f" FROM `{table}`")
        return list(cur.fetchall())


# ---------------------------------------------------------------------------
# target-owned factory/catalog data

def sync_factory_table(sconn, dconn, sp, dp, table, idmaps, *, live):
    """Map source factory identities onto target rows without duplicating them.

    A policy can also expose narrowly defined client overlays.  Joomla's
    #__extensions row, for example, belongs to the target install, but params /
    enabled state are site configuration.  Mail master templates remain target
    owned while language-specific override rows are migrated.
    """
    policy = migration_policy.policy(table)
    if not policy:
        return {"policy": "none", "matched": 0, "inserted": 0, "updated": 0,
                "skipped": 0}

    stabs, dtabs = tables(sconn, sp), tables(dconn, dp)
    if table not in stabs or table not in dtabs:
        return {"policy": "target_factory", "available": False,
                "reason": "table is not present on both source and target",
                "matched": 0, "inserted": 0, "updated": 0, "skipped": 0}

    scols, dcols = columns(sconn, sp + table), columns(dconn, dp + table)
    identity = tuple(policy.get("identity") or ())
    missing = [c for c in identity if c not in scols or c not in dcols]
    if missing:
        return {"policy": "target_factory", "available": False,
                "reason": "identity columns missing: " + ", ".join(missing),
                "matched": 0, "inserted": 0, "updated": 0, "skipped": 0}

    pk = policy.get("pk")
    if pk and (pk not in scols or pk not in dcols):
        return {"policy": "target_factory", "available": False,
                "reason": f"factory primary key {pk!r} is missing",
                "matched": 0, "inserted": 0, "updated": 0, "skipped": 0}

    shared = shared_columns(scols, dcols)
    overlays = migration_policy.overlay_columns(table, shared)
    src_cols = list(dict.fromkeys(([pk] if pk else []) + list(identity) + overlays))
    dst_cols = list(dict.fromkeys(([pk] if pk else []) + list(identity) + overlays))
    source = _select_rows(sconn, sp + table, src_cols)
    target = _select_rows(dconn, dp + table, dst_cols)
    target_by_key = {_key(row, identity): row for row in target}
    idmap = idmaps.setdefault(table, {})

    matched = inserted = updated = skipped = 0
    source_only = []
    with dconn.cursor() as cur:
        for row in source:
            key = _key(row, identity)
            existing = target_by_key.get(key)
            mode = migration_policy.row_mode(table, row)

            if existing is not None:
                matched += 1
                if pk and row.get(pk) is not None and existing.get(pk) is not None:
                    idmap[row[pk]] = existing[pk]
                if mode in ("overlay", "overlay_insert") and overlays:
                    # The target keeps all structural/version fields; only the
                    # explicitly client-owned overlay is copied.
                    if live:
                        sets = ", ".join(f"`{c}` = %s" for c in overlays)
                        where = " AND ".join(f"`{c}` = %s" for c in identity)
                        cur.execute(f"UPDATE `{dp + table}` SET {sets} WHERE {where}",
                                    [row.get(c) for c in overlays] + [row.get(c) for c in identity])
                    updated += 1
                continue

            if mode == "overlay_insert":
                # A language-specific Joomla mail template override is real
                # client state.  The master row is factory-owned; this row isn't.
                write_cols = list(dict.fromkeys(list(identity) + overlays))
                if live:
                    cols_sql = ", ".join(f"`{c}`" for c in write_cols)
                    marks = ", ".join(["%s"] * len(write_cols))
                    cur.execute(f"INSERT INTO `{dp + table}` ({cols_sql}) VALUES ({marks})",
                                [row.get(c) for c in write_cols])
                inserted += 1
                continue

            # A source factory identity absent from the new version is not
            # resurrected.  Any FK into it becomes unresolved/0 rather than
            # accidentally pointing at another target row with the same number.
            skipped += 1
            source_only.append(dict(zip(identity, key)))
            if pk and row.get(pk) is not None:
                idmap[row[pk]] = 0

    if live and (updated or inserted):
        dconn.commit()
    target_keys = set(target_by_key)
    source_keys = {_key(row, identity) for row in source}
    return {
        "policy": "target_factory",
        "available": True,
        "reason": policy.get("reason", "target install owns these rows"),
        "identity": list(identity),
        "source_rows": len(source),
        "target_rows": len(target),
        "matched": matched,
        "inserted": inserted,
        "updated": updated,
        "skipped": skipped,
        "source_only": source_only[:20],
        "target_only": len(target_keys - source_keys),
    }


def sync_factory_catalogs(sconn, dconn, sp, dp, idmaps, *, live):
    out = {}
    for table in migration_policy.TARGET_FACTORY:
        stat = sync_factory_table(sconn, dconn, sp, dp, table, idmaps, live=live)
        # Only surface policies that exist somewhere or failed for a real reason.
        if stat.get("available") or table in (tables(sconn, sp) | tables(dconn, dp)):
            out[table] = stat
    return out


# ---------------------------------------------------------------------------
# Widgetkit historical compatibility gate (kept until generic package scanning
# can safely perform its opaque JSON upgrade itself).

def is_widgetkit_table(table):
    return table == WIDGETKIT_PREFIX or table.startswith(WIDGETKIT_PREFIX + "_")


def _manifest_version(value):
    if not value:
        return ""
    if isinstance(value, str):
        try:
            value = json.loads(value)
        except Exception:
            return ""
    return str((value or {}).get("version") or "").strip() if isinstance(value, dict) else ""


def _version_key(version):
    parts = re.findall(r"\d+|[A-Za-z]+", version or "")
    return [(0, int(p)) if p.isdigit() else (1, p.lower()) for p in parts]


def newest_version(versions):
    return max((v for v in versions if v), key=_version_key, default="")


def widgetkit_extension_rows(conn, prefix):
    try:
        ecols = columns(conn, prefix + "extensions")
        id_col = _first_col(ecols, "extension_id", "id")
        wanted = [c for c in (id_col, "element", "name", "manifest_cache", "version")
                  if c and c in ecols]
        if not wanted:
            return []
        rows = []
        for row in _select_rows(conn, prefix + "extensions", wanted):
            text = " ".join(str(row.get(c) or "") for c in ("element", "name")).lower()
            if WIDGETKIT_PREFIX not in text:
                continue
            rows.append({
                "extension_id": row.get(id_col) if id_col else None,
                "element": row.get("element"),
                "name": row.get("name"),
                "version": _manifest_version(row.get("manifest_cache")) or str(row.get("version") or "").strip(),
            })
        return rows
    except Exception:
        return []


def widgetkit_versions(conn, prefix):
    return {r["version"] for r in widgetkit_extension_rows(conn, prefix) if r["version"]}


def widgetkit_update_sources(conn, prefix):
    try:
        present = tables(conn, prefix)
        if not {"extensions", "update_sites", "update_sites_extensions"} <= present:
            return []
        widget_ids = {r["extension_id"] for r in widgetkit_extension_rows(conn, prefix)
                      if r["extension_id"] is not None}
        if not widget_ids:
            return []
        bcols = columns(conn, prefix + "update_sites_extensions")
        ucols = columns(conn, prefix + "update_sites")
        bridge_ext = _first_col(bcols, "extension_id", "extensionid")
        bridge_site = _first_col(bcols, "update_site_id", "update_siteid", "id")
        site_id = _first_col(ucols, "update_site_id", "id")
        if not (bridge_ext and bridge_site and site_id):
            return []
        bridge = _select_rows(conn, prefix + "update_sites_extensions", [bridge_ext, bridge_site])
        site_ids = {r[bridge_site] for r in bridge if r.get(bridge_ext) in widget_ids}
        if not site_ids:
            return []
        fields = [c for c in (site_id, "name", "type", "location", "enabled") if c in ucols]
        return [{k: row.get(k) for k in fields if k != site_id}
                for row in _select_rows(conn, prefix + "update_sites", fields)
                if row.get(site_id) in site_ids]
    except Exception:
        return []


def widgetkit_compatible(sconn, dconn, sp, dp):
    src, dst = widgetkit_versions(sconn, sp), widgetkit_versions(dconn, dp)
    return bool(src) and src == dst


def widgetkit_status(sconn, dconn, sp, dp):
    src, dst = widgetkit_versions(sconn, sp), widgetkit_versions(dconn, dp)
    latest = newest_version(src | dst)
    status = {
        "source_versions": sorted(src),
        "target_versions": sorted(dst),
        "source_update_sites": widgetkit_update_sources(sconn, sp),
        "target_update_sites": widgetkit_update_sources(dconn, dp),
        "newest_version": latest,
        "compatible": bool(src) and src == dst,
    }
    if status["compatible"]:
        status["action"] = "copy"
    elif not src:
        status["action"] = "source_missing_widgetkit"
    elif not dst:
        status["action"] = "target_missing_widgetkit"
    else:
        needs = []
        if latest and latest not in src:
            needs.append("source")
        if latest and latest not in dst:
            needs.append("target")
        status["action"] = "bring_to_newest_version"
        status["upgrade_required"] = needs
    return status


# ---------------------------------------------------------------------------
# planning / extension ownership

def site_resolver(name):
    return mesh.connect(name), mesh.prefix(name), mesh.env(name).get("JOOMLA_URL", "")


def installed_components(conn, prefix):
    try:
        with conn.cursor() as cur:
            cur.execute(f"SELECT element, name FROM `{prefix}extensions` "
                        "WHERE type = 'component' AND element IS NOT NULL")
            return {r["element"]: (r["name"] or r["element"]) for r in cur.fetchall()}
    except Exception:
        return {}


def owning_component(table, components):
    best = None
    for element in components:
        stem = element[4:] if element.startswith("com_") else element
        if not stem:
            continue
        if table == stem or table.startswith(stem + "_"):
            if best is None or len(stem) > len(best[0]):
                best = (stem, element)
    return best[1] if best else None


def left_behind(sconn, dconn, sp, dp):
    stabs, dtabs = tables(sconn, sp), tables(dconn, dp)
    orphans = sorted(stabs - dtabs - NEVER_COPY)
    if not orphans:
        return []
    src_components = installed_components(sconn, sp)
    dst_components = installed_components(dconn, dp)
    grouped = {}
    for table in orphans:
        rows = row_count(sconn, sp + table)
        element = owning_component(table, src_components)
        key = element or "(no component)"
        entry = grouped.setdefault(key, {
            "component": key,
            "name": src_components.get(element, element or "unrecognised tables"),
            "installed_on_target": element in dst_components if element else False,
            "tables": [], "rows": 0})
        entry["tables"].append(table)
        entry["rows"] += rows
    return sorted(grouped.values(), key=lambda g: -g["rows"])


def plan(src, dst, resolve=site_resolver):
    sconn, sp, _ = resolve(src)
    dconn, dp, _ = resolve(dst)
    try:
        stabs, dtabs = tables(sconn, sp), tables(dconn, dp)
        widgetkit = widgetkit_status(sconn, dconn, sp, dp)
        idmaps = {}
        factory = sync_factory_catalogs(sconn, dconn, sp, dp, idmaps, live=False)
        compatibility = extension_compat.compatibility_profile(sconn, dconn, sp, dp)
        moving, dropped = [], []
        for t in sorted(stabs):
            if t in NEVER_COPY or migration_policy.is_factory(t):
                continue
            n = row_count(sconn, sp + t)
            if t not in dtabs:
                if n:
                    dropped.append({"table": t, "rows": n,
                                    "reason": "no such table on the target site"})
                continue
            if is_widgetkit_table(t) and not widgetkit["compatible"]:
                if n:
                    dropped.append({"table": t, "rows": n,
                                    "reason": "Widgetkit must be at version parity before copying",
                                    "widgetkit": widgetkit})
                continue
            scols, dcols = columns(sconn, sp + t), columns(dconn, dp + t)
            shared = shared_columns(scols, dcols)
            moving.append({"table": t, "rows": n, "columns": len(shared),
                           "columns_dropped": sorted(set(scols) - set(dcols))})
        waiting = left_behind(sconn, dconn, sp, dp)
        return {
            "source": src, "target": dst,
            "moving": [m for m in moving if m["rows"]],
            "not_moving": dropped,
            "needs_installing": [w for w in waiting if w["rows"]],
            "target_factory": factory,
            "extension_compatibility": compatibility,
            "target_only": sorted(dtabs - stabs - NEVER_COPY - set(migration_policy.TARGET_FACTORY)),
            "totals": {
                "rows_to_move": sum(m["rows"] for m in moving),
                "rows_left_behind": sum(d["rows"] for d in dropped),
                "factory_rows_matched": sum(s.get("matched", 0) for s in factory.values()),
                "factory_rows_not_resurrected": sum(s.get("skipped", 0) for s in factory.values()),
            },
        }
    finally:
        sconn.close(); dconn.close()


# ---------------------------------------------------------------------------
# value transforms

def rewrite_urls(value, old_base, new_base):
    if not isinstance(value, str) or not old_base:
        return value
    host = re.sub(r"^https?://", "", old_base.rstrip("/")).lstrip("/")
    host = re.sub(r"^www\.", "", host)
    new = new_base.rstrip("/")
    for pat in (f"https://www.{host}", f"http://www.{host}",
                f"https://{host}", f"http://{host}",
                f"//www.{host}", f"//{host}"):
        value = value.replace(pat, new)
    return value


def remap(value, idmap, strict=False):
    try:
        old = int(value)
    except (TypeError, ValueError):
        return value
    if old in idmap:
        return idmap[old]
    return 0 if (strict and old) else value


ROUTE_IDMAP = {
    ("com_content", "article"):  "content",
    ("com_content", "category"): "categories",
    ("com_content", "featured"): "categories",
    ("com_contact", "contact"):  "contact_details",
    ("com_tags", "tag"):         "tags",
}
_ROUTE_ID_RE = re.compile(r"(?P<pre>[?&]id=)(?P<id>\d+)")


def rewrite_route_ids(value, idmaps):
    if not isinstance(value, str) or "option=com_" not in value:
        return value
    option = re.search(r"option=(com_[\w]+)", value)
    view = re.search(r"view=(\w+)", value)
    if not option:
        return value
    idmap_name = ROUTE_IDMAP.get((option.group(1), view.group(1) if view else ""))
    if not idmap_name:
        return value
    idmap = idmaps.get(idmap_name) or {}
    if not idmap:
        return value

    def swap(match):
        new = idmap.get(int(match.group("id")))
        return match.group("pre") + str(new if new and new > 0 else match.group("id"))
    return _ROUTE_ID_RE.sub(swap, value)


def name_lookup(sconn, dconn, sp, dp, kind):
    table, pk, scol, dcol, where = LOOKUPS[kind]
    cond = f" WHERE {where}" if where else ""
    with sconn.cursor() as scur, dconn.cursor() as dcur:
        scur.execute(f"SELECT `{pk}` AS id, `{scol}` AS k FROM `{sp + table}`{cond}")
        src = {r["id"]: r["k"] for r in scur.fetchall()}
        dcur.execute(f"SELECT `{pk}` AS id, `{dcol}` AS k FROM `{dp + table}`{cond}")
        dst = {r["k"]: r["id"] for r in dcur.fetchall()}
    return {old: dst[key] for old, key in src.items() if key in dst}


# ---------------------------------------------------------------------------
# row copying

def existing_keys(conn, table, key_cols, has_id=True):
    if not key_cols:
        return {}
    cols = ", ".join(f"`{c}`" for c in key_cols)
    sel = f"id, {cols}" if has_id else cols
    with conn.cursor() as cur:
        cur.execute(f"SELECT {sel} FROM `{table}`")
        return {tuple(str(r[c]) for c in key_cols): (r["id"] if has_id else True)
                for r in cur.fetchall()}


def row_savepoint(cur, name="mport_row"):
    cur.execute(f"SAVEPOINT {name}")


def rollback_row(cur, name="mport_row"):
    cur.execute(f"ROLLBACK TO SAVEPOINT {name}")
    cur.execute(f"RELEASE SAVEPOINT {name}")


def release_row(cur, name="mport_row"):
    cur.execute(f"RELEASE SAVEPOINT {name}")


def copy_table(sconn, dconn, sp, dp, table, idmaps, *, live, urls=None, order_by=None):
    scols, dcols = columns(sconn, sp + table), columns(dconn, dp + table)
    shared = [c for c in shared_columns(scols, dcols) if c != "id"]
    if not shared:
        return 0, 0, row_count(sconn, sp + table), []

    has_id = "id" in scols and "id" in dcols
    key_cols = [c for c in NATURAL_KEY.get(table, ()) if c in dcols]
    present = existing_keys(dconn, dp + table, key_cols, has_id)
    taken = set()
    if has_id:
        with dconn.cursor() as c:
            c.execute(f"SELECT id FROM `{dp + table}`")
            taken = {r["id"] for r in c.fetchall()}

    fks = {c: m for c, m in FKS.get(table, {}).items() if c in shared}
    for kind in {m for m in fks.values() if m.startswith("@")}:
        idmaps.setdefault(kind, name_lookup(sconn, dconn, sp, dp, kind))
    idmap = idmaps.setdefault(table, {})

    sel = ", ".join(f"`{c}`" for c in ((["id"] if has_id else []) + shared))
    sql = f"SELECT {sel} FROM `{sp + table}`"
    where = ROW_FILTER.get(table)
    if where and all(c in scols for c in _FILTER_COLS.get(table, ())):
        sql += f" WHERE {where}"
    if order_by and order_by in scols:
        sql += f" ORDER BY `{order_by}`"

    inserted = reused = 0
    failures = []
    unresolved = {}
    with sconn.cursor() as scur, dconn.cursor() as dcur:
        scur.execute(sql)
        for row in scur.fetchall():
            old_id = row.get("id")
            values = {}
            for c in shared:
                v = row.get(c)
                if c in fks:
                    ref = fks[c]
                    was = v
                    strict = ref.startswith("@") or migration_policy.is_factory_reference(ref)
                    v = remap(v, idmaps.setdefault(ref, {}), strict=strict)
                    if strict and v == 0 and was not in (0, None, "0"):
                        unresolved.setdefault(c, set()).add(was)
                if urls:
                    v = rewrite_urls(v, urls[0], urls[1])
                v = rewrite_route_ids(v, idmaps)
                values[c] = v

            key = tuple(str(values.get(c, row.get(c))) for c in key_cols) if key_cols else None
            if key is not None and key in present:
                if has_id:
                    idmap[old_id] = present[key]
                if table in UPDATE_ON_MATCH and live:
                    sets = ", ".join(f"`{c}` = %s" for c in values)
                    try:
                        row_savepoint(dcur)
                        dcur.execute(f"UPDATE `{dp + table}` SET {sets} WHERE id = %s",
                                     list(values.values()) + [present[key]])
                        release_row(dcur)
                    except Exception as exc:
                        rollback_row(dcur)
                        failures.append({"table": table, "source_id": old_id,
                                         "error": f"{type(exc).__name__}: {str(exc)[:180]}"})
                        continue
                reused += 1
                continue

            if not live:
                if has_id:
                    idmap[old_id] = old_id if old_id not in taken else -1
                inserted += 1
                continue

            keep_id = has_id and old_id is not None and old_id not in taken
            write = {"id": old_id, **values} if keep_id else dict(values)
            cols_sql = ", ".join(f"`{c}`" for c in write)
            marks = ", ".join(["%s"] * len(write))
            try:
                row_savepoint(dcur)
                dcur.execute(f"INSERT INTO `{dp + table}` ({cols_sql}) VALUES ({marks})",
                             list(write.values()))
                release_row(dcur)
            except Exception as exc:
                rollback_row(dcur)
                failures.append({"table": table, "source_id": old_id,
                                 "error": f"{type(exc).__name__}: {str(exc)[:180]}"})
                continue
            new_id = (old_id if keep_id else dcur.lastrowid) if has_id else True
            if has_id:
                idmap[old_id] = new_id
                taken.add(new_id)
            if key is not None:
                present[key] = new_id
            inserted += 1

    if live:
        dconn.commit()
    if unresolved and failures:
        failures[0].setdefault("unresolved", {c: sorted(v)[:20] for c, v in unresolved.items()})
    return inserted, reused, 0, failures


# ---------------------------------------------------------------------------
# trees, ACL, workflows

def rebuild_tree(conn, table, *, live):
    with conn.cursor() as cur:
        cur.execute(f"SELECT id, parent_id FROM `{table}` ORDER BY id")
        rows = cur.fetchall()
        ids = {r["id"] for r in rows}
        kids = {}
        for row in rows:
            kids.setdefault(row["parent_id"], []).append(row["id"])
        n, out = [0], {}

        def walk(node, level):
            n[0] += 1
            left = n[0]
            for child in kids.get(node, []):
                walk(child, level + 1)
            n[0] += 1
            out[node] = (left, n[0], level)

        roots = [r["id"] for r in rows if r["parent_id"] in (0, None) or r["parent_id"] not in ids]
        for root in roots:
            walk(root, 0)
        if live:
            for node, (left, right, level) in out.items():
                cur.execute(f"UPDATE `{table}` SET lft=%s, rgt=%s, level=%s WHERE id=%s",
                            (left, right, level, node))
            conn.commit()
        return len(out)


_ASSET_NAME = re.compile(r"^(com_content\.(?:article|category))\.(\d+)$")


def translate_asset_name(name, idmaps):
    match = _ASSET_NAME.match(name or "")
    if not match:
        return name
    kind, old = match.group(1), int(match.group(2))
    table = "content" if kind.endswith("article") else "categories"
    return f"{kind}.{idmaps.get(table, {}).get(old, old)}"


def sync_assets(sconn, sp, dconn, dp, kind, idmap, idmaps, *, live, table):
    if not live or not idmap:
        return {"created": 0, "rules": 0}
    names = {f"{kind}.{old}": f"{kind}.{new}" for old, new in idmap.items()}
    created = ruled = 0
    with sconn.cursor() as scur, dconn.cursor() as dcur:
        scur.execute(
            f"SELECT a.name, a.title, a.rules, p.name AS parent_name "
            f"FROM `{sp}assets` a LEFT JOIN `{sp}assets` p ON p.id = a.parent_id "
            "WHERE a.name IN %s", (tuple(names),))
        source = {r["name"]: r for r in scur.fetchall()}
        dcur.execute(f"SELECT id, name FROM `{dp}assets`")
        target = {r["name"]: r["id"] for r in dcur.fetchall()}
        for old_name, new_name in names.items():
            src = source.get(old_name) or {}
            rules = src.get("rules") or "{}"
            title = src.get("title") or new_name
            parent = target.get(translate_asset_name(src.get("parent_name"), idmaps), 1)
            if new_name in target:
                dcur.execute(f"UPDATE `{dp}assets` SET rules=%s, parent_id=%s WHERE id=%s",
                             (rules, parent, target[new_name]))
            else:
                dcur.execute(f"INSERT INTO `{dp}assets` "
                             "(parent_id, lft, rgt, level, name, title, rules) "
                             "VALUES (%s,%s,%s,%s,%s,%s,%s)",
                             (parent, 0, 0, 2, new_name, title, rules))
                target[new_name] = dcur.lastrowid
                created += 1
            if rules != "{}":
                ruled += 1
        dconn.commit()
        rebuild_tree(dconn, dp + "assets", live=True)
        dcur.execute(f"UPDATE `{dp}{table}` t JOIN `{dp}assets` a "
                     "ON a.name = CONCAT(%s, '.', t.id) SET t.asset_id = a.id "
                     "WHERE t.asset_id <> a.id", (kind,))
        dconn.commit()
    return {"created": created, "rules": ruled}


def ensure_workflow(conn, prefix, ids, *, live, extension="com_content.article"):
    if not live or not ids:
        return 0
    if "workflow_associations" not in tables(conn, prefix):
        return 0
    with conn.cursor() as cur:
        cur.execute(f"SELECT s.id FROM `{prefix}workflow_stages` s "
                    f"JOIN `{prefix}workflows` w ON w.id = s.workflow_id "
                    "WHERE w.extension = %s AND s.published = 1 "
                    "ORDER BY w.`default` DESC, s.`default` DESC, s.id LIMIT 1",
                    (extension,))
        row = cur.fetchone()
        if not row:
            return 0
        stage, made = row["id"], 0
        for item in ids:
            cur.execute(f"SELECT item_id FROM `{prefix}workflow_associations` "
                        "WHERE item_id=%s AND extension=%s", (item, extension))
            if cur.fetchone():
                continue
            cur.execute(f"INSERT INTO `{prefix}workflow_associations` "
                        "(item_id, stage_id, extension) VALUES (%s,%s,%s)",
                        (item, stage, extension))
            made += 1
        conn.commit()
        return made


# ---------------------------------------------------------------------------
# run

def excluded(table, exclude):
    return any(fnmatch.fnmatch(table, p.strip()) for p in exclude if p.strip())


def phase_tables(sconn, dconn, sp, dp, phase, exclude=()):
    stabs, dtabs = tables(sconn, sp), tables(dconn, dp)
    both = {t for t in (stabs & dtabs) - NEVER_COPY
            if not migration_policy.is_factory(t) and not excluded(t, exclude)}
    if not widgetkit_status(sconn, dconn, sp, dp)["compatible"]:
        both = {t for t in both if not is_widgetkit_table(t)}
    named = {
        # ACL reference ids must exist before any entity's access column moves.
        "users":      ["usergroups", "viewlevels", "users", "user_usergroup_map"],
        "styles":     ["template_styles"],
        "categories": ["categories"],
        "articles":   ["content", "content_frontpage"],
        "menus":      ["menu_types", "menu"],
        "modules":    ["modules", "modules_menu"],
        "fields":     ["fields_groups", "fields", "fields_values"],
        "tags":       ["tags", "contentitem_tag_map"],
    }
    if phase != "extensions":
        return [t for t in named[phase] if t in both]
    claimed = {t for group in named.values() for t in group}
    return sorted(both - claimed)


def migrate(src, dst, phases=PHASES, *, live=False, urls=True, resolve=site_resolver,
            url_override=None, on_progress=None, exclude=()):
    def report_progress(**event):
        if on_progress:
            on_progress(event)

    sconn, sp, surl = resolve(src)
    dconn, dp, durl = resolve(dst)
    url_pair = url_override or ((surl, durl) if urls else None)
    widgetkit = widgetkit_status(sconn, dconn, sp, dp)
    idmaps = {}
    report = {"source": src, "target": dst, "live": live,
              "widgetkit": widgetkit, "phases": {}}
    try:
        # First map install-owned coordinates.  This is what prevents the second
        # DPCalendar country catalogue and gives later rows the target country id.
        factory = sync_factory_catalogs(sconn, dconn, sp, dp, idmaps, live=live)
        report["target_factory"] = factory
        report["extension_compatibility"] = extension_compat.compatibility_profile(sconn, dconn, sp, dp)

        total_steps = len(phases) + (1 if factory else 0)
        offset = 0
        if factory:
            offset = 1
            report_progress(event="phase", phase="factory maps", step=1, of=total_steps,
                            tables=len(factory))
            for table, stats in factory.items():
                report_progress(event="table", phase="factory maps", table=table,
                                inserted=stats.get("inserted", 0),
                                reused=stats.get("matched", 0),
                                skipped=stats.get("skipped", 0),
                                policy="target_factory")

        waiting = [w for w in left_behind(sconn, dconn, sp, dp) if w["rows"]]
        report["needs_installing"] = waiting
        for group in waiting:
            report_progress(event="skipped", component=group["component"],
                            name=group["name"], tables=len(group["tables"]), rows=group["rows"])

        for n, phase in enumerate(phases, 1):
            stats = {}
            todo = phase_tables(sconn, dconn, sp, dp, phase, exclude=exclude)
            if phase == "extensions" and not widgetkit["compatible"]:
                skipped = []
                for table in sorted((tables(sconn, sp) & tables(dconn, dp)) - NEVER_COPY):
                    if is_widgetkit_table(table):
                        rows = row_count(sconn, sp + table)
                        if rows:
                            skipped.append({"table": table, "rows": rows})
                if skipped:
                    stats["widgetkit"] = {"skipped": skipped,
                                          "reason": "Widgetkit must be at version parity before copying",
                                          "widgetkit": widgetkit}
            report_progress(event="phase", phase=phase, step=n + offset, of=total_steps,
                            tables=len(todo))
            for table in todo:
                tree = table in TREE_TABLES or is_tree(columns(dconn, dp + table))
                try:
                    ins, reused, skipped, failed = copy_table(
                        sconn, dconn, sp, dp, table, idmaps, live=live, urls=url_pair,
                        order_by="parent_id" if tree else "id")
                except Exception as exc:
                    stats[table] = {"error": f"{type(exc).__name__}: {str(exc)[:200]}"}
                    report_progress(event="table", phase=phase, table=table,
                                    error=stats[table]["error"])
                    continue
                stats[table] = {"inserted": ins, "reused": reused, "skipped": skipped}
                if failed:
                    stats[table]["failed"] = len(failed)
                    stats[table]["errors"] = failed[:5]
                if tree and live:
                    stats[table]["tree_nodes"] = rebuild_tree(dconn, dp + table, live=True)
                report_progress(event="table", phase=phase, table=table, **stats[table])

            if phase == "categories" and live:
                stats["assets"] = sync_assets(sconn, sp, dconn, dp,
                                              "com_content.category",
                                              idmaps.get("categories", {}), idmaps,
                                              live=live, table="categories")
            if phase == "articles" and live:
                article_ids = list(idmaps.get("content", {}).values())
                stats["assets"] = sync_assets(sconn, sp, dconn, dp,
                                              "com_content.article",
                                              idmaps.get("content", {}), idmaps,
                                              live=live, table="content")
                stats["workflow"] = ensure_workflow(dconn, dp, article_ids, live=live)
                report_progress(event="repair", phase=phase,
                                assets=stats["assets"], workflow=stats["workflow"])
            report["phases"][phase] = stats

        report["id_maps"] = {k: len(v) for k, v in idmaps.items()}
        users = idmaps.get("users", {})
        report["users_kept_id"] = sum(1 for old, new in users.items() if old == new)
        report["users_remapped"] = len(users) - report["users_kept_id"]
        report["factory_ids_mapped"] = sum(
            len(idmaps.get(t, {})) for t, p in migration_policy.TARGET_FACTORY.items() if p.get("pk"))
        report_progress(event="done", users_kept_id=report["users_kept_id"],
                        users_remapped=report["users_remapped"],
                        factory_ids_mapped=report["factory_ids_mapped"])
        return report
    finally:
        sconn.close(); dconn.close()


# ---------------------------------------------------------------------------
# media
_MEDIA_RE = re.compile(r'''(?:images|media)/[^"'<>\\]*?\.\w{2,5}''', re.I)


def media_manifest(site, tables_to_scan=("content", "modules", "template_styles"),
                   resolve=site_resolver):
    conn, prefix, _ = resolve(site)
    found = set()
    try:
        present = tables(conn, prefix)
        with conn.cursor() as cur:
            for table in tables_to_scan:
                if table not in present:
                    continue
                for col in columns(conn, prefix + table):
                    if col in ("id", "checked_out"):
                        continue
                    try:
                        cur.execute(f"SELECT `{col}` AS v FROM `{prefix + table}`")
                    except Exception:
                        continue
                    for row in cur.fetchall():
                        value = row["v"]
                        if isinstance(value, str) and ("images/" in value or "media/" in value):
                            found.update(m.group(0).replace("\\/", "/") for m in _MEDIA_RE.finditer(value))
    finally:
        conn.close()
    return sorted(found)


def copy_media(paths, src_dir, dst_dir, *, live):
    import shutil
    from pathlib import Path
    src_root, dst_root = Path(src_dir), Path(dst_dir)
    copied, missing = 0, []
    for rel in paths:
        source = src_root / rel
        if not source.is_file():
            missing.append(rel)
            continue
        if live:
            target = dst_root / rel
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(source, target)
        copied += 1
    return {"copied": copied, "missing": missing[:50], "missing_count": len(missing)}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--from", dest="src", required=True)
    parser.add_argument("--to", dest="dst", required=True)
    parser.add_argument("--phases", default=",".join(PHASES))
    parser.add_argument("--live", action="store_true")
    parser.add_argument("--keep-urls", action="store_true")
    parser.add_argument("--media-from")
    parser.add_argument("--media-to")
    args = parser.parse_args()
    if args.media_from and args.media_to:
        paths = media_manifest(args.dst if args.live else args.src)
        print(json.dumps({"referenced_media": len(paths),
                          **copy_media(paths, args.media_from, args.media_to, live=args.live)}, indent=2))
        return
    if not args.live:
        print(json.dumps(plan(args.src, args.dst), indent=2))
        return
    print(json.dumps(migrate(args.src, args.dst, tuple(args.phases.split(",")),
                             live=True, urls=not args.keep_urls), indent=2))


if __name__ == "__main__":
    main()