Spaces:
Running
Running
| """Evidence-backed Joomla component table semantics. | |
| The migration must never merge populated third-party tables by guesswork, but it | |
| also must not stop merely because a vendor-specific table name is absent from a | |
| hard-coded dictionary. The target schema is itself Top-Down evidence. | |
| This layer therefore combines known Joomla semantics with one universal rule: | |
| when source and target expose the same full, non-PK UNIQUE identity and current | |
| rows prove that key contains no NULLs, the identity is safe to use as the | |
| semantic merge key for this exact migration pair. The key is registered in | |
| ``dbmesh.NATURAL_KEY`` so planning and the eventual copy use the same evidence. | |
| Anything not proven remains quarantined. | |
| """ | |
| from __future__ import annotations | |
| from . import component_data_guard as component | |
| from . import dbmesh as d | |
| from . import dbmesh_policy_patch as policy | |
| from . import schema_semantic_identity | |
| _INSTALLED = False | |
| _PRIOR_CLASSIFY = component.classify_table | |
| _PRIOR_PHASE_TABLES = component.phase_tables | |
| FINDER_TERM_SHARDS = {f"finder_links_terms{x}" for x in "0123456789abcdef"} | |
| DERIVED = { | |
| "finder_links_terms", | |
| "scheduler_logs", | |
| *FINDER_TERM_SHARDS, | |
| } | |
| REUSE_ONLY_KEYS = { | |
| # 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 _schema_identity(sconn, dconn, sp: str, dp: str, table: str) -> dict: | |
| try: | |
| return schema_semantic_identity.discover(sconn, dconn, sp + table, dp + table) | |
| except Exception as exc: | |
| return { | |
| "proven": False, | |
| "identity": [], | |
| "evidence": f"schema identity proof failed: {type(exc).__name__}: {str(exc)[:180]}", | |
| "rejected": [], | |
| } | |
| 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 | |
| # Universal TD proof: if a component-owned table is populated on both sides | |
| # and the lower classifier has no semantic key, ask the two live schemas. | |
| # This is specifically the path that turns extension-specific tables such as | |
| # Advanced Modules / Conditions from hard-coded chores into evidence-driven | |
| # migration semantics when their vendor schema declares a real unique key. | |
| if phase == "extensions" and result.get("class") == "quarantine" \ | |
| and result.get("source_rows") and result.get("target_rows"): | |
| proof = _schema_identity(sconn, dconn, sp, dp, table) | |
| if proof.get("proven") and proof.get("identity"): | |
| cols = tuple(str(c) for c in proof["identity"]) | |
| d.NATURAL_KEY[table] = cols | |
| result.update({ | |
| "class": "semantic_merge", | |
| "identity": list(cols), | |
| "identity_source": "schema_unique_index", | |
| "reason": proof.get("evidence") or "shared schema-declared unique identity", | |
| "identity_evidence": proof, | |
| }) | |
| else: | |
| result["identity_evidence"] = proof | |
| result["reason"] = ( | |
| result.get("reason") | |
| or "target already contains rows but no semantic identity policy is known" | |
| ) | |
| 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) | |
| priority = {"guidedtours": 0, "guidedtour_steps": 1} | |
| return sorted(out, key=lambda table: (priority.get(table, 10), table)) | |
| def install() -> None: | |
| global _INSTALLED | |
| if _INSTALLED: | |
| return | |
| policy.REGENERATE.update(DERIVED) | |
| d.NEVER_COPY.update(DERIVED) | |
| d.NATURAL_KEY["user_profiles"] = ("user_id", "profile_key") | |
| d.FKS.setdefault("user_profiles", {})["user_id"] = "users" | |
| d.NATURAL_KEY["guidedtours"] = ("uid",) | |
| d.NATURAL_KEY["guidedtour_steps"] = ("tour_id", "ordering", "title") | |
| d.FKS.setdefault("guidedtour_steps", {})["tour_id"] = "guidedtours" | |
| d.FKS.setdefault("guidedtours", {}).update({ | |
| "created_by": "users", "modified_by": "users", "access": "viewlevels", | |
| }) | |
| d.FKS.setdefault("guidedtour_steps", {}).update({ | |
| "created_by": "users", "modified_by": "users", | |
| }) | |
| for table, cols in REUSE_ONLY_KEYS.items(): | |
| d.NATURAL_KEY[table] = cols | |
| 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 | |