Spaces:
Running
Running
File size: 8,660 Bytes
3f558da 4e99f09 3f558da 4e99f09 bbc443a 4e99f09 bbc443a 4e99f09 3f558da 4e99f09 3f558da 4e99f09 3f558da 4e99f09 bbc443a 4e99f09 bbc443a 4e99f09 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 | """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
|