fix(migration): encode proven Joomla table semantics

core/component_data_semantics.py

@@ -0,0 +1,175 @@
+"""Conservative Joomla table semantics discovered from the target runtime.
+
+This layer exists for tables that were being blocked only because the generic
+component-data classifier did not yet know what they mean.  It deliberately
+handles only cases we can prove without inventing a merge key:
+
+* Smart Search link/term joins and scheduler execution logs are derived state;
+* user profiles are client data keyed by the migrated user + profile key;
+* Guided Tours and Scheduler Tasks may be reused only when every source
+  semantic identity already exists on the target (no unsafe asset-id insert);
+* Joomla's stock one-stage/seven-transition workflow is target factory state and
+  article workflow associations are regenerated against the target default.
+
+Anything outside those proofs remains quarantined.  In particular this module
+does not guess identities for third-party Regular Labs tables or scheduler_jobs.
+"""
+from __future__ import annotations
+
+from . import component_data_guard as component
+from . import dbmesh as d
+from . import dbmesh_policy_patch as policy
+
+_INSTALLED = False
+_PRIOR_CLASSIFY = component.classify_table
+_PRIOR_PHASE_TABLES = component.phase_tables
+
+DERIVED = {
+    "finder_links_terms",
+    "scheduler_logs",
+}
+
+REUSE_ONLY_KEYS = {
+    # uid is indexed by Joomla and names the tour independently of install-local id.
+    "guidedtours": ("uid",),
+    # A task type may be instantiated more than once, so title is part of identity.
+    "scheduler_tasks": ("type", "title"),
+}
+
+BASIC_STAGE_TITLES = {"COM_WORKFLOW_BASIC_STAGE"}
+BASIC_TRANSITION_TITLES = {
+    "UNPUBLISH", "PUBLISH", "TRASH", "ARCHIVE",
+    "FEATURE", "UNFEATURE", "PUBLISH_AND_FEATURE",
+}
+
+
+def _rows(conn, table: str, cols: tuple[str, ...]) -> list[dict]:
+    available = d.columns(conn, table)
+    if any(col not in available for col in 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 _keys(conn, table: str, cols: tuple[str, ...]) -> set[tuple[str, ...]]:
+    return {
+        tuple("" if row.get(c) is None else str(row.get(c)) for c in cols)
+        for row in _rows(conn, table, cols)
+    }
+
+
+def _source_keys_are_target_owned(sconn, dconn, sp: str, dp: str,
+                                  table: str, cols: tuple[str, ...]) -> tuple[bool, list[tuple[str, ...]]]:
+    stabs, dtabs = d.tables(sconn, sp), d.tables(dconn, dp)
+    if table not in stabs or table not in dtabs:
+        return False, []
+    source = _keys(sconn, sp + table, cols)
+    target = _keys(dconn, dp + table, cols)
+    if not source:
+        return False, []
+    missing = sorted(source - target)
+    return not missing, missing[:20]
+
+
+def _basic_workflow_is_target_factory(sconn, dconn, sp: str, dp: str) -> bool:
+    needed = {"workflow_stages", "workflow_transitions"}
+    if not needed <= d.tables(sconn, sp) or not needed <= d.tables(dconn, dp):
+        return False
+    src_stages = {k[0] for k in _keys(sconn, sp + "workflow_stages", ("title",))}
+    src_transitions = {k[0] for k in _keys(sconn, sp + "workflow_transitions", ("title",))}
+    dst_stages = {k[0] for k in _keys(dconn, dp + "workflow_stages", ("title",))}
+    dst_transitions = {k[0] for k in _keys(dconn, dp + "workflow_transitions", ("title",))}
+    return (
+        src_stages == BASIC_STAGE_TITLES
+        and src_transitions == BASIC_TRANSITION_TITLES
+        and src_stages <= dst_stages
+        and src_transitions <= dst_transitions
+    )
+
+
+def classify_table(sconn, dconn, sp: str, dp: str, table: str, *, phase: str = "extensions") -> dict:
+    """Add only evidence-backed semantics to the component classifier."""
+    result = dict(_PRIOR_CLASSIFY(sconn, dconn, sp, dp, table, phase=phase))
+
+    if table in REUSE_ONLY_KEYS and phase == "extensions":
+        cols = REUSE_ONLY_KEYS[table]
+        safe, missing = _source_keys_are_target_owned(sconn, dconn, sp, dp, table, cols)
+        if safe:
+            result.update({
+                "class": "semantic_merge",
+                "identity": list(cols),
+                "reason": "every source semantic identity already exists on the target; reuse target-owned rows",
+            })
+        elif result.get("source_rows") and result.get("target_rows"):
+            result.update({
+                "class": "quarantine",
+                "reason": (
+                    "source contains semantic identities absent from the target; "
+                    "inserting this table safely needs its asset/relationship policy"
+                ),
+                "source_only_identities": [list(x) for x in missing],
+            })
+        return result
+
+    if table in {"workflow_stages", "workflow_transitions", "workflow_associations"} \
+            and phase == "extensions" and _basic_workflow_is_target_factory(sconn, dconn, sp, dp):
+        if table == "workflow_associations":
+            result.update({
+                "class": "regenerate",
+                "reason": "stock Joomla workflow associations are rebound to the target default stage after article migration",
+            })
+        else:
+            result.update({
+                "class": "target_factory",
+                "reason": "source contains Joomla's stock basic workflow definition already owned by the target runtime",
+            })
+        return result
+
+    return result
+
+
+def phase_tables(sconn, dconn, sp, dp, phase, exclude=()):
+    """Only classes that represent client rows may reach copy_table."""
+    rows = _PRIOR_PHASE_TABLES(sconn, dconn, sp, dp, phase, exclude)
+    if phase != "extensions":
+        return rows
+    out = []
+    for table in rows:
+        decision = classify_table(sconn, dconn, sp, dp, table, phase="extensions")
+        if decision.get("class") in {"client_entity", "semantic_merge"}:
+            out.append(table)
+    return out
+
+
+def install() -> None:
+    global _INSTALLED
+    if _INSTALLED:
+        return
+
+    # Derived tables are never copied. The target rebuilds them from canonical data.
+    policy.REGENERATE.update(DERIVED)
+    d.NEVER_COPY.update(DERIVED)
+
+    # Client profile values follow the migrated user id. No install-local asset id
+    # is involved, so this is a normal semantic merge rather than reuse-only.
+    d.NATURAL_KEY["user_profiles"] = ("user_id", "profile_key")
+    d.FKS.setdefault("user_profiles", {})["user_id"] = "users"
+
+    # These keys are used only after classify_table proves source ⊆ target. That
+    # guarantees copy_table can reuse them and cannot fall into an unsafe insert.
+    for table, cols in REUSE_ONLY_KEYS.items():
+        d.NATURAL_KEY[table] = cols
+
+    # Every existing planning/migration surface resolves these module globals at
+    # runtime, so one install covers Studio, MPort, plan and live execution.
+    component.classify_table = classify_table
+    component.phase_tables = phase_tables
+    policy.classify_table = classify_table
+    policy.phase_tables = phase_tables
+    d.classify_table = classify_table
+    d.phase_tables = phase_tables
+
+    _INSTALLED = True