Spaces:
Running
Running
"""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()
"""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