"""Safety/classification layer for the MPort DB-mesh migrator. Studio -> Meshes uses :mod:`core.dbmesh`, while the editor has its own migration surface. This layer keeps MPort conservative without replacing the registry and transport code in dbmesh: * target-installed factory/catalog rows are identities, never migration cargo; * derived/search/cache state is rebuilt by the target, never copied; * populated extension tables without a semantic merge key are quarantined instead of "try every INSERT and count duplicate failures"; * install-assigned foreign keys are translated onto target identities; and * reassigned numeric ids are allocated outside the *entire* source+target id space so one conflict cannot steal an id needed by a later source row. install() mutates the already-imported core.dbmesh module in place. Existing callers keep the same API and registry. """ from __future__ import annotations from . import dbmesh as d FACTORY = { "content_types": { "pk": "type_id", "identity": ("type_alias",), "mode": "map", "reason": "Joomla/installed extensions own the content-type registry", }, "extensions": { "pk": "extension_id", "identity": ("type", "element", "folder", "client_id"), "mode": "overlay", "overlay": ("enabled", "access", "ordering", "params", "custom_data", "note"), "reason": "target extension install owns code/version metadata; source owns site configuration", }, "dpcalendar_countries": { "pk": "id", "identity": ("short_code",), "mode": "map", "reason": "DPCalendar seeds the country catalogue on every install", }, "files_mimetypes": { "pk": None, "identity": ("mimetype", "extension"), "mode": "map", "reason": "Joomlatools Files seeds the MIME catalogue on every install", }, "mail_templates": { "pk": None, "identity": ("template_id", "language"), "mode": "row", "overlay": ("extension", "subject", "body", "htmlbody", "attachments", "params"), "reason": "Joomla owns master mail templates; language overrides are client customisations", }, "postinstall_messages": { "pk": None, "identity": (), "mode": "skip", "reason": "Joomla/extension install creates post-install notices for the target version", }, "update_sites_extensions": { "pk": None, "identity": (), "mode": "skip", "reason": "extension update-site bindings belong to the target installation", }, } REGENERATE = { "finder_links", "finder_taxonomy", "finder_taxonomy_map", "finder_terms", "finder_terms_common", "finder_tokens", "finder_tokens_aggregate", "scheduler_metadata", "ucm_base", "ucm_content", } SEMANTIC_KEYS = { "viewlevels": ("title",), "languages": ("lang_code",), "finder_types": ("title",), "dpcalendar_caldav_principals": ("uri",), "scheduler_jobs": ("name",), } ACCESS_TABLES = ( "categories", "content", "menu", "modules", "fields", "fields_groups", "tags", "dpcalendar_locations", ) def _key(row: dict, cols) -> tuple: return tuple("" if row.get(c) is None else str(row.get(c)) for c in cols) def _select(conn, table: str, cols) -> list[dict]: cols = [c for c in cols if c] if not cols: return [] with conn.cursor() as cur: cur.execute("SELECT " + ", ".join(f"`{c}`" for c in cols) + f" FROM `{table}`") return list(cur.fetchall()) def _row_mode(table: str, row: dict) -> str: p = FACTORY.get(table) or {} mode = p.get("mode", "map") if mode != "row": return mode if table == "mail_templates": return "overlay_insert" if str(row.get("language") or "").strip() else "map" return "map" def _overlay_columns(table: str, shared) -> list[str]: allowed = set((FACTORY.get(table) or {}).get("overlay") or ()) return [c for c in shared if c in allowed] def sync_factory_table(sconn, dconn, sp: str, dp: str, table: str, idmaps: dict, *, live: bool): """Map a target-owned table without copying its factory definitions.""" policy = FACTORY[table] mode = policy.get("mode") if mode == "skip": return { "policy": "target_factory", "available": table in (d.tables(sconn, sp) | d.tables(dconn, dp)), "reason": policy["reason"], "matched": 0, "inserted": 0, "updated": 0, "skipped": d.row_count(sconn, sp + table) if table in d.tables(sconn, sp) else 0, } stabs, dtabs = d.tables(sconn, sp), d.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 = d.columns(sconn, sp + table), d.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 = d.shared_columns(scols, dcols) overlays = _overlay_columns(table, shared) wanted = list(dict.fromkeys(([pk] if pk else []) + list(identity) + overlays)) source = _select(sconn, sp + table, wanted) target = _select(dconn, dp + table, wanted) 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) row_mode = _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 row_mode in ("overlay", "overlay_insert") and overlays: 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 row_mode == "overlay_insert": write_cols = list(dict.fromkeys(list(identity) + overlays)) if live and write_cols: 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 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() return { "policy": "target_factory", "available": True, "reason": policy["reason"], "identity": list(identity), "source_rows": len(source), "target_rows": len(target), "matched": matched, "inserted": inserted, "updated": updated, "skipped": skipped, "source_only": source_only[:20], } def sync_factory_catalogs(sconn, dconn, sp: str, dp: str, idmaps: dict, *, live: bool): out = {} present = d.tables(sconn, sp) | d.tables(dconn, dp) for table in FACTORY: if table not in present: continue out[table] = sync_factory_table(sconn, dconn, sp, dp, table, idmaps, live=live) return out def _extension_compatibility(sconn, dconn, sp: str, dp: str) -> dict: try: from editor import extension_compat return extension_compat.compatibility_profile(sconn, dconn, sp, dp) except Exception as exc: return { "available": False, "reason": f"{type(exc).__name__}: {str(exc)[:180]}", "extensions": [], } def classify_table(sconn, dconn, sp: str, dp: str, table: str, *, phase: str = "extensions") -> dict: """Ownership decision for one physical table.""" stabs, dtabs = d.tables(sconn, sp), d.tables(dconn, dp) src_rows = d.row_count(sconn, sp + table) if table in stabs else 0 dst_rows = d.row_count(dconn, dp + table) if table in dtabs else 0 if table in FACTORY or table in d.NEVER_COPY: return {"class": "target_factory", "source_rows": src_rows, "target_rows": dst_rows} if table in REGENERATE: return {"class": "regenerate", "source_rows": src_rows, "target_rows": dst_rows} if table not in dtabs: return {"class": "missing_target", "source_rows": src_rows, "target_rows": 0} if table not in stabs: return {"class": "target_only", "source_rows": 0, "target_rows": dst_rows} key_cols = [c for c in d.NATURAL_KEY.get(table, ()) if c in d.columns(dconn, dp + table)] if key_cols: return { "class": "semantic_merge", "identity": key_cols, "source_rows": src_rows, "target_rows": dst_rows, } if phase == "extensions" and dst_rows: return { "class": "quarantine", "source_rows": src_rows, "target_rows": dst_rows, "reason": "target already contains rows but no semantic identity policy is known", } return {"class": "client_entity", "source_rows": src_rows, "target_rows": dst_rows} def phase_tables(sconn, dconn, sp, dp, phase, exclude=()): stabs, dtabs = d.tables(sconn, sp), d.tables(dconn, dp) both = { t for t in (stabs & dtabs) - d.NEVER_COPY if t not in FACTORY and t not in REGENERATE and not d.excluded(t, exclude) } named = { "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 ts in named.values() for t in ts} out = [] for table in sorted(both - claimed): decision = classify_table(sconn, dconn, sp, dp, table, phase="extensions") if decision["class"] != "quarantine": out.append(table) return out def copy_table(sconn, dconn, sp, dp, table, idmaps, *, live, urls=None, order_by=None, on_conflict="reassign"): """dbmesh.copy_table with collision-safe reassignment and strict factory FKs.""" scols = d.columns(sconn, sp + table) dcols = d.columns(dconn, dp + table) shared = [c for c in d.shared_columns(scols, dcols) if c != "id"] if not shared: return 0, 0, d.row_count(sconn, sp + table), 0, [] has_id = "id" in scols and "id" in dcols key_cols = [c for c in d.NATURAL_KEY.get(table, ()) if c in dcols] present = d.existing_keys(dconn, dp + table, key_cols, has_id) used = d.taken_ids(dconn, dp + table) if has_id else set() fks = {c: m for c, m in d.FKS.get(table, {}).items() if c in shared} 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 = d.ROW_FILTER.get(table) if where and all(c in scols for c in d._FILTER_COLS.get(table, ())): sql += f" WHERE {where}" if order_by and order_by in scols: sql += f" ORDER BY `{order_by}`" with sconn.cursor() as scur: scur.execute(sql) source_rows = list(scur.fetchall()) source_ids = { int(row["id"]) for row in source_rows if has_id and row.get("id") is not None } reserved = set(used) | source_ids next_id = max(reserved or {0}) + 1 def allocate_id(): nonlocal next_id while next_id in reserved or next_id in used: next_id += 1 out = next_id reserved.add(out) used.add(out) next_id += 1 return out inserted = reused = remapped = overwritten = 0 failures = [] conflicts = [] unresolved = {} with dconn.cursor() as dcur: for row in source_rows: 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 ref in FACTORY v = d.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 = d.rewrite_urls(v, urls[0], urls[1]) 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] reused += 1 continue keep = has_id and old_id is not None and old_id not in used clash = has_id and old_id is not None and not keep assigned_id = old_id if keep else None if clash: conflicts.append({"table": table, "id": old_id, "resolved": on_conflict}) if on_conflict == "reassign": assigned_id = allocate_id() if not live: if has_id: idmap[old_id] = ( old_id if (keep or on_conflict == "overwrite") else assigned_id ) if clash: remapped += on_conflict != "overwrite" overwritten += on_conflict == "overwrite" inserted += 1 if key is not None: present[key] = idmap.get(old_id, True) if has_id else True continue if clash and on_conflict == "overwrite": sets = ", ".join(f"`{c}` = %s" for c in values) try: d.row_savepoint(dcur) dcur.execute( f"UPDATE `{dp + table}` SET {sets} WHERE id = %s", list(values.values()) + [old_id], ) d.release_row(dcur) except Exception as exc: d.rollback_row(dcur) failures.append({ "table": table, "source_id": old_id, "error": f"{type(exc).__name__}: {str(exc)[:180]}", }) continue idmap[old_id] = old_id if key is not None: present[key] = old_id overwritten += 1 continue write = dict(values) if has_id and assigned_id is not None: write["id"] = assigned_id cols = ", ".join(f"`{c}`" for c in write) marks = ", ".join(["%s"] * len(write)) try: d.row_savepoint(dcur) dcur.execute( f"INSERT INTO `{dp + table}` ({cols}) VALUES ({marks})", list(write.values()), ) d.release_row(dcur) except Exception as exc: d.rollback_row(dcur) failures.append({ "table": table, "source_id": old_id, "error": f"{type(exc).__name__}: {str(exc)[:180]}", }) continue new_id = (assigned_id if has_id else True) if has_id: if new_id is None: new_id = dcur.lastrowid idmap[old_id] = new_id used.add(new_id) if not keep: remapped += 1 if key is not None: present[key] = new_id inserted += 1 if live: dconn.commit() return inserted, reused, 0, { "remapped": remapped, "overwritten": overwritten, "conflicts": conflicts[:20], "unresolved": {c: sorted(v)[:20] for c, v in unresolved.items()}, }, failures def plan(src, dst, resolve, exclude=()): """Classified preflight: tells the operator what is cargo vs identity.""" sconn, sp, _ = resolve(src) dconn, dp, _ = resolve(dst) try: stabs, dtabs = d.tables(sconn, sp), d.tables(dconn, dp) buckets = { "moving": [], "factory": [], "regenerate": [], "quarantined": [], "not_moving": [], "target_only": [], "excluded": [], } claimed_core = { "usergroups", "viewlevels", "users", "user_usergroup_map", "template_styles", "categories", "content", "content_frontpage", "menu_types", "menu", "modules", "modules_menu", "fields_groups", "fields", "fields_values", "tags", "contentitem_tag_map", } for table in sorted(stabs | dtabs): if d.excluded(table, exclude): buckets["excluded"].append(table) continue phase = "core" if table in claimed_core else "extensions" decision = classify_table(sconn, dconn, sp, dp, table, phase=phase) entry = {"table": table, **decision} cls = decision["class"] if cls == "target_factory": buckets["factory"].append(entry) elif cls == "regenerate": buckets["regenerate"].append(entry) elif cls == "quarantine": buckets["quarantined"].append(entry) elif cls == "missing_target": if decision["source_rows"]: entry["reason"] = "no such table on the target site" buckets["not_moving"].append(entry) elif cls == "target_only": buckets["target_only"].append(entry) else: buckets["moving"].append(entry) return { "source": src, "target": dst, **buckets, "totals": { "rows_to_move": sum(x.get("source_rows", 0) for x in buckets["moving"]), "factory_rows_mapped": sum(x.get("source_rows", 0) for x in buckets["factory"]), "rows_regenerated": sum(x.get("source_rows", 0) for x in buckets["regenerate"]), "rows_quarantined": sum(x.get("source_rows", 0) for x in buckets["quarantined"]), "rows_left_behind": sum(x.get("source_rows", 0) for x in buckets["not_moving"]), }, } finally: sconn.close() dconn.close() def migrate(src, dst, phases=d.PHASES, *, live=False, urls=True, resolve, url_override=None, on_progress=None, on_conflict="reassign", exclude=()): """Classified migration used by Studio -> Meshes.""" def say(**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) idmaps, conflicts = {}, [] report = { "source": src, "target": dst, "live": live, "on_conflict": on_conflict, "excluded": list(exclude), "phases": {}, } try: factory = sync_factory_catalogs(sconn, dconn, sp, dp, idmaps, live=live) report["target_factory"] = factory report["extension_compatibility"] = _extension_compatibility(sconn, dconn, sp, dp) stabs, dtabs = d.tables(sconn, sp), d.tables(dconn, dp) core_claimed = { "usergroups", "viewlevels", "users", "user_usergroup_map", "template_styles", "categories", "content", "content_frontpage", "menu_types", "menu", "modules", "modules_menu", "fields_groups", "fields", "fields_values", "tags", "contentitem_tag_map", } quarantined = [] for table in sorted((stabs & dtabs) - core_claimed): if table in d.NEVER_COPY or table in FACTORY or table in REGENERATE or d.excluded(table, exclude): continue decision = classify_table(sconn, dconn, sp, dp, table, phase="extensions") if decision["class"] == "quarantine" and decision["source_rows"]: quarantined.append({"table": table, **decision}) report["quarantined"] = quarantined report["regenerate"] = [ {"table": t, "rows": d.row_count(sconn, sp + t)} for t in sorted(REGENERATE & stabs) if d.row_count(sconn, sp + t) ] total_steps = len(phases) + (1 if factory else 0) offset = 0 if factory: offset = 1 say(event="phase", phase="factory maps", step=1, of=total_steps, tables=len(factory)) for table, stats in factory.items(): say( 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", ) for n, phase in enumerate(phases, 1): stats = {} todo = phase_tables(sconn, dconn, sp, dp, phase, exclude) say(event="phase", phase=phase, step=n + offset, of=total_steps, tables=len(todo)) for table in todo: say(event="start", phase=phase, table=table) try: ins, reused, skipped, ident, failed = copy_table( sconn, dconn, sp, dp, table, idmaps, live=live, urls=url_pair, order_by="parent_id" if table in d.TREE_TABLES else "id", on_conflict=on_conflict, ) except Exception as exc: stats[table] = {"error": f"{type(exc).__name__}: {str(exc)[:200]}"} say(event="table", phase=phase, table=table, **stats[table]) continue stats[table] = {"inserted": ins, "reused": reused, "skipped": skipped} stats[table].update({k: v for k, v in ident.items() if v}) conflicts.extend(ident.get("conflicts", ())) if failed: stats[table]["failed"] = len(failed) stats[table]["errors"] = failed[:5] if table in d.TREE_TABLES and live: stats[table]["tree_nodes"] = d.rebuild_tree(dconn, dp + table, live=True) say(event="table", phase=phase, table=table, **stats[table]) if phase == "categories" and live: stats["assets"] = d.sync_assets( sconn, sp, dconn, dp, "com_content.category", idmaps.get("categories", {}), idmaps, live=live, table="categories", ) say(event="repair", phase=phase, assets=stats["assets"]) if phase == "articles" and live: stats["assets"] = d.sync_assets( sconn, sp, dconn, dp, "com_content.article", idmaps.get("content", {}), idmaps, live=live, table="content", ) stats["workflow"] = d.ensure_workflow(dconn, dp, live=live) say( 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()} report["factory_ids_mapped"] = sum( len(idmaps.get(t, {})) for t, p in FACTORY.items() if p.get("pk") ) report["id_conflicts"] = conflicts[:100] report["id_conflict_count"] = len(conflicts) say( event="done", conflicts=len(conflicts), factory_ids_mapped=report["factory_ids_mapped"], **report["id_maps"], ) return report finally: sconn.close() dconn.close() def install(): """Patch the public dbmesh migration surface in place.""" d.NATURAL_KEY.update(SEMANTIC_KEYS) d.FKS.setdefault("usergroups", {})["parent_id"] = "usergroups" for table in ACCESS_TABLES: d.FKS.setdefault(table, {})["access"] = "viewlevels" d.FKS.setdefault("dpcalendar_locations", {}).update({ "country": "dpcalendar_countries", "created_by": "users", "modified_by": "users", }) d.FKS.setdefault("menu", {})["component_id"] = "extensions" d.FKS.setdefault("contentitem_tag_map", {})["type_id"] = "content_types" d.TREE_TABLES.add("usergroups") d.NEVER_COPY.update(FACTORY) d.NEVER_COPY.update(REGENERATE) d.copy_table = copy_table d.phase_tables = phase_tables d.plan = plan d.migrate = migrate d.classify_table = classify_table d.sync_factory_catalogs = sync_factory_catalogs d.FACTORY_TABLES = FACTORY d.REGENERATE_TABLES = REGENERATE return d