#!/usr/bin/env python3 """Diff objet-par-objet entre le schéma de référence (migrate:fresh) et la prod.""" import json import re import sys from collections import defaultdict def parse(path): sections = defaultdict(list) cur = None with open(path) as f: for line in f: line = line.rstrip("\n") if line.startswith("###"): cur = line.strip("#") continue if not line.strip() or cur is None: continue sections[cur].append(line.split("\t")) return sections def norm_type(t): """Neutralise les écarts de représentation entre MySQL 8.0 et 9.6.""" t = t.strip().lower() # largeur d'affichage des entiers (supprimée en 8.0.19+, mais restes possibles) t = re.sub(r"\b(tinyint|smallint|mediumint|int|integer|bigint)\(\d+\)", r"\1", t) t = t.replace("integer", "int") # int(1) unsigned etc : normaliser les espaces t = re.sub(r"\s+", " ", t) return t def norm_default(d, coltype): if d in ("__NULL__", None): return None d = d.strip() # MySQL renvoie parfois les défauts entre quotes simples selon la version if len(d) >= 2 and d[0] == "'" and d[-1] == "'": d = d[1:-1] dl = d.lower() if dl in ("current_timestamp", "current_timestamp()", "now()"): return "CURRENT_TIMESTAMP" # défauts numériques : 0 vs 0.00 vs '0' try: f = float(d) if f == int(f): return str(int(f)) return str(f) except ValueError: pass return d def build(sections): tables = {r[0]: {"engine": r[1], "collation": r[2]} for r in sections["TABLES"]} cols = {} for r in sections["COLUMNS"]: tn, cn, ctype, nullable, default, extra, coll = r[0], r[1], r[2], r[3], r[4], r[5], r[6] if len(r) > 6 else "" cols[(tn, cn)] = { "type": norm_type(ctype), "raw_type": ctype, "nullable": nullable, "default": norm_default(default, ctype), "raw_default": default, "extra": (extra or "").lower().replace("default_generated", "").strip(), "collation": coll, } idx = defaultdict(list) for r in sections["INDEXES"]: tn, iname, non_unique, seq, cname, itype = r[0], r[1], r[2], r[3], r[4], r[5] idx[(tn, iname)].append((int(seq), cname, non_unique, itype)) indexes = {} for k, v in idx.items(): v.sort() indexes[k] = { "columns": [c for _, c, _, _ in v], "unique": v[0][2] == "0", "type": v[0][3], } fks = {} for r in sections["FKS"]: tn, cname_, col, rt, rc, dr, ur = r[0], r[1], r[2], r[3], r[4], r[5], r[6] fks.setdefault((tn, cname_), {"columns": [], "ref_table": rt, "ref_columns": [], "on_delete": dr, "on_update": ur}) fks[(tn, cname_)]["columns"].append(col) fks[(tn, cname_)]["ref_columns"].append(rc) return {"tables": tables, "columns": cols, "indexes": indexes, "fks": fks} ref = build(parse(sys.argv[1])) prod = build(parse(sys.argv[2])) out = { "tables_missing_in_prod": sorted(set(ref["tables"]) - set(prod["tables"])), "tables_extra_in_prod": sorted(set(prod["tables"]) - set(ref["tables"])), "columns_missing_in_prod": [], "columns_extra_in_prod": [], "columns_divergent": [], "indexes_missing_in_prod": [], "indexes_extra_in_prod": [], "fks_missing_in_prod": [], "engines": {}, } common_tables = set(ref["tables"]) & set(prod["tables"]) for (tn, cn), rc in sorted(ref["columns"].items()): if tn not in common_tables: continue pc = prod["columns"].get((tn, cn)) if pc is None: out["columns_missing_in_prod"].append( {"table": tn, "column": cn, "type": rc["raw_type"], "nullable": rc["nullable"], "default": rc["raw_default"]}) continue diffs = {} if rc["type"] != pc["type"]: diffs["type"] = {"ref": rc["raw_type"], "prod": pc["raw_type"]} if rc["nullable"] != pc["nullable"]: diffs["nullable"] = {"ref": rc["nullable"], "prod": pc["nullable"]} if rc["default"] != pc["default"]: diffs["default"] = {"ref": rc["raw_default"], "prod": pc["raw_default"]} if rc["extra"] != pc["extra"]: diffs["extra"] = {"ref": rc["extra"], "prod": pc["extra"]} if rc["collation"] != pc["collation"]: diffs["collation"] = {"ref": rc["collation"], "prod": pc["collation"]} if diffs: out["columns_divergent"].append({"table": tn, "column": cn, "diffs": diffs}) for (tn, cn), pc in sorted(prod["columns"].items()): if tn not in common_tables: continue if (tn, cn) not in ref["columns"]: out["columns_extra_in_prod"].append( {"table": tn, "column": cn, "type": pc["raw_type"], "nullable": pc["nullable"], "default": pc["raw_default"]}) # Index : comparer par (table, ensemble ordonné de colonnes, unicité) plutôt que par nom, # les noms auto-générés pouvant différer. def idx_sig(d): return (tuple(d["columns"]), d["unique"]) for tn in sorted(common_tables): ref_idx = {k[1]: v for k, v in ref["indexes"].items() if k[0] == tn} prod_idx = {k[1]: v for k, v in prod["indexes"].items() if k[0] == tn} ref_sigs = {idx_sig(v): n for n, v in ref_idx.items()} prod_sigs = {idx_sig(v): n for n, v in prod_idx.items()} for sig, name in ref_sigs.items(): if sig not in prod_sigs: out["indexes_missing_in_prod"].append( {"table": tn, "name": name, "columns": list(sig[0]), "unique": sig[1]}) for sig, name in prod_sigs.items(): if sig not in ref_sigs: out["indexes_extra_in_prod"].append( {"table": tn, "name": name, "columns": list(sig[0]), "unique": sig[1]}) for (tn, cname_), rf in sorted(ref["fks"].items()): if tn not in common_tables: continue match = [p for (ptn, _), p in prod["fks"].items() if ptn == tn and p["columns"] == rf["columns"] and p["ref_table"] == rf["ref_table"]] if not match: out["fks_missing_in_prod"].append( {"table": tn, "name": cname_, "columns": rf["columns"], "ref_table": rf["ref_table"], "ref_columns": rf["ref_columns"], "on_delete": rf["on_delete"]}) out["engines"] = { "ref": sorted({v["engine"] for v in ref["tables"].values()}), "prod": sorted({v["engine"] for v in prod["tables"].values()}), } out["counts"] = { "ref_tables": len(ref["tables"]), "prod_tables": len(prod["tables"]), "ref_columns": len(ref["columns"]), "prod_columns": len(prod["columns"]), "ref_fks": len(ref["fks"]), "prod_fks": len(prod["fks"]), } print(json.dumps(out, indent=2, ensure_ascii=False))