CRM 表结构差异报告

#!/usr/bin/env python3 # -*- coding: utf-8 -*- from __future__ import annotations import argparse import dataclasses import html import re from pathlib import Path from typing import Dict, List, Tuple @dataclasses.dataclass(frozen=True) class Column: name: str raw_def: str # definition text after column name, normalized @dataclasses.dataclass(frozen=True) class Table: name: str columns: Dict[str, Column] CREATE_TABLE_RE = re.compile(r"^\s*CREATE\s+TABLE\s+`(?Pcrm_[^`]+)`\s*\(\s*$", re.IGNORECASE) COL_RE = re.compile(r"^\s*`(?P[^`]+)`\s+(?P.+)$") def _normalize_ws(s: str) -> str: s = s.strip() s = re.sub(r"\s+", " ", s) return s def parse_crm_tables(sql_text: str) -> Dict[str, Table]: """ Parse Navicat-style MySQL DDL and extract CREATE TABLE blocks for crm_*. Only captures column lines beginning with backticked column names. """ lines = sql_text.splitlines() tables: Dict[str, Table] = {} i = 0 while i < len(lines): m = CREATE_TABLE_RE.match(lines[i]) if not m: i += 1 continue table_name = m.group("name") i += 1 cols: Dict[str, Column] = {} while i < len(lines): line = lines[i].rstrip("\n") # End of CREATE TABLE block if re.search(r"^\s*\)\s*ENGINE=.*;\s*$", line, flags=re.IGNORECASE): break cm = COL_RE.match(line) if cm: col_name = cm.group("col").strip() rest = cm.group("rest").strip() # strip trailing comma if rest.endswith(","): rest = rest[:-1].rstrip() cols[col_name] = Column(name=col_name, raw_def=_normalize_ws(rest)) i += 1 tables[table_name] = Table(name=table_name, columns=cols) # move past ENGINE line if present while i < len(lines) and not lines[i].strip().endswith(";"): i += 1 if i < len(lines): i += 1 return tables def diff_tables( left: Dict[str, Table], right: Dict[str, Table], ) -> Tuple[List[str], List[str], Dict[str, Dict[str, Tuple[str | None, str | None]]]]: """ Returns: - left_only_tables - right_only_tables - per_table_column_diffs: {table: {col: (left_def, right_def)}} where any of left_def/right_def can be None (missing), or both present but different. """ left_only = sorted(set(left.keys()) - set(right.keys())) right_only = sorted(set(right.keys()) - set(left.keys())) common = sorted(set(left.keys()) & set(right.keys())) per_table: Dict[str, Dict[str, Tuple[str | None, str | None]]] = {} for t in common: lcols = left[t].columns rcols = right[t].columns all_cols = sorted(set(lcols.keys()) | set(rcols.keys())) diffs: Dict[str, Tuple[str | None, str | None]] = {} for c in all_cols: ldef = lcols.get(c).raw_def if c in lcols else None rdef = rcols.get(c).raw_def if c in rcols else None if ldef != rdef: diffs[c] = (ldef, rdef) if diffs: per_table[t] = diffs return left_only, right_only, per_table def _h(s: str) -> str: return html.escape(s, quote=True) def render_html_report( left_name: str, right_name: str, left_only_tables: List[str], right_only_tables: List[str], per_table_column_diffs: Dict[str, Dict[str, Tuple[str | None, str | None]]], ) -> str: diff_table_count = len(per_table_column_diffs) diff_col_count = sum(len(v) for v in per_table_column_diffs.values()) parts: List[str] = [] parts.append("") parts.append('') parts.append("") parts.append('') parts.append('') parts.append("CRM 表结构差异报告") parts.append( """ """.strip() ) parts.append("") parts.append("") parts.append('

') parts.append('") parts.append('

') parts.append( f'

{len(left_only_tables)}

{_h(left_name)} 独有表

' ) parts.append( f'

{len(right_only_tables)}

{_h(right_name)} 独有表

' ) parts.append( f'

{diff_table_count}

同名但字段有差异的表

' ) parts.append( f'

{diff_col_count}

字段差异条目

' ) parts.append("

") parts.append('

') parts.append("

") parts.append(f"

{_h(left_name)} 比 {_h(right_name)} 多的表

") if left_only_tables: parts.append("

{_h(t)}

") else: parts.append(f"

无

") parts.append("

") parts.append("

") parts.append(f"

{_h(right_name)} 比 {_h(left_name)} 多的表

") if right_only_tables: parts.append("

{_h(t)}

") else: parts.append(f"

无

") parts.append("

") parts.append("

") if per_table_column_diffs: parts.append('

') parts.append("

同名表字段差异（红色为差异）

") parts.append('

') for t in sorted(per_table_column_diffs.keys()): parts.append(f"{_h(t)}") parts.append("

") parts.append("

") for t in sorted(per_table_column_diffs.keys()): diffs = per_table_column_diffs[t] parts.append(f"

") parts.append(f"

{_h(t)}

") parts.append("") parts.append("") parts.append( f"" ) parts.append("") parts.append("") for col in sorted(diffs.keys()): ldef, rdef = diffs[col] ltxt = ldef if ldef is not None else "(缺失)" rtxt = rdef if rdef is not None else "(缺失)" parts.append("") parts.append(f"") parts.append(f"") parts.append(f"") parts.append("") parts.append("") parts.append("

字段	{_h(left_name)}	{_h(right_name)}
{_h(col)}	{_h(ltxt)}	{_h(rtxt)}

") parts.append("

") parts.append("

") parts.append("") parts.append("") return "\n".join(parts) def main() -> int: p = argparse.ArgumentParser(description="Compare crm_* tables between two SQL dumps and output an HTML report.") p.add_argument("--left", required=True, help="Left SQL file path (e.g. shanxia2.0.sql)") p.add_argument("--right", required=True, help="Right SQL file path (e.g. sx.sql)") p.add_argument("--out", required=True, help="Output HTML path") p.add_argument("--left-name", default="shanxia2.0", help="Label for left side") p.add_argument("--right-name", default="crm(sx)", help="Label for right side") args = p.parse_args() left_text = Path(args.left).read_text(encoding="utf-8", errors="ignore") right_text = Path(args.right).read_text(encoding="utf-8", errors="ignore") left_tables = parse_crm_tables(left_text) right_tables = parse_crm_tables(right_text) left_only, right_only, per_table = diff_tables(left_tables, right_tables) report = render_html_report(args.left_name, args.right_name, left_only, right_only, per_table) out_path = Path(args.out) out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(report, encoding="utf-8") print(f"Parsed tables: left={len(left_tables)} right={len(right_tables)}") print(f"Left-only tables: {len(left_only)} Right-only tables: {len(right_only)}") print(f"Tables with column diffs: {len(per_table)}; diff items: {sum(len(v) for v in per_table.values())}") print(f"Wrote HTML: {out_path}") return 0 if __name__ == "__main__": raise SystemExit(main())