chore(repo): 仓侧治理门禁 — ECR/migration 身份唯一
将编号与 schema 完整性留在 Repository Governance，不扩 ESS/Loop；TRACEABILITY 锚定 Next ECR=041、Max Migration=000050。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
@@ -0,0 +1,218 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Repository governance checks — outside ESS Loop.
|
||||
|
||||
Validates:
|
||||
1. Migration stem uniqueness under apps/api/migrations/
|
||||
2. ECR identity uniqueness (H1 / Canonical-ID) under docs/ECR/
|
||||
|
||||
Usage:
|
||||
python scripts/repo-governance-check.py
|
||||
python scripts/repo-governance-check.py --print-anchors
|
||||
|
||||
Exit 0 = PASS; non-zero = FAIL (merge/gate blocker).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
MIG_DIR = ROOT / "apps" / "api" / "migrations"
|
||||
ECR_DIR = ROOT / "docs" / "ECR"
|
||||
|
||||
# Full stem: 000018_admin_rbac (with or without .up/.down)
|
||||
MIG_STEM_RE = re.compile(r"^(\d{6}_.+?)(?:\.(?:up|down))?\.sql$", re.I)
|
||||
# Numeric prefix only (collision surface across branches)
|
||||
MIG_NUM_RE = re.compile(r"^(\d{6})_")
|
||||
ECR_ID_RE = re.compile(r"^ECR-\d+[A-Z]?(?:-[a-z0-9]+(?:-[a-z0-9]+)*)?$", re.I)
|
||||
H1_RE = re.compile(r"^#\s+(ECR-\S+)", re.M)
|
||||
CANON_RE = re.compile(r"(?im)^\*\*?Canonical-ID:\*\*?\s*`?(ECR-[A-Za-z0-9-]+)`?\s*$")
|
||||
|
||||
|
||||
def migration_stems(paths: list[Path]) -> dict[str, list[Path]]:
|
||||
by_stem: dict[str, list[Path]] = defaultdict(list)
|
||||
for p in paths:
|
||||
m = MIG_STEM_RE.match(p.name)
|
||||
if not m:
|
||||
continue
|
||||
by_stem[m.group(1)].append(p)
|
||||
return by_stem
|
||||
|
||||
|
||||
def migration_numbers(paths: list[Path]) -> dict[str, list[str]]:
|
||||
"""Map 6-digit version → distinct full stems (duplicate numbers = FAIL)."""
|
||||
by_num: dict[str, set[str]] = defaultdict(set)
|
||||
for p in paths:
|
||||
m = MIG_STEM_RE.match(p.name)
|
||||
if not m:
|
||||
continue
|
||||
stem = m.group(1)
|
||||
num_m = MIG_NUM_RE.match(stem)
|
||||
if not num_m:
|
||||
continue
|
||||
by_num[num_m.group(1)].add(stem)
|
||||
return {k: sorted(v) for k, v in by_num.items()}
|
||||
|
||||
|
||||
def ecr_identity(path: Path) -> str | None:
|
||||
text = path.read_text(encoding="utf-8")
|
||||
canon = CANON_RE.search(text)
|
||||
if canon:
|
||||
return canon.group(1)
|
||||
h1 = H1_RE.search(text)
|
||||
if h1:
|
||||
# Strip trailing punctuation from H1 token
|
||||
return h1.group(1).rstrip(".:)")
|
||||
return None
|
||||
|
||||
|
||||
def max_migration_num(paths: list[Path]) -> int:
|
||||
n = 0
|
||||
for p in paths:
|
||||
m = MIG_STEM_RE.match(p.name)
|
||||
if not m:
|
||||
continue
|
||||
num_m = MIG_NUM_RE.match(m.group(1))
|
||||
if num_m:
|
||||
n = max(n, int(num_m.group(1)))
|
||||
return n
|
||||
|
||||
|
||||
def next_ecr_suggestion(identities: list[str]) -> str:
|
||||
"""Next unused bare ECR-NNN (ignores letter/suffix variants for suggestion only)."""
|
||||
nums: list[int] = []
|
||||
for i in identities:
|
||||
m = re.match(r"ECR-(\d+)", i, re.I)
|
||||
if m:
|
||||
nums.append(int(m.group(1)))
|
||||
nxt = (max(nums) + 1) if nums else 1
|
||||
return f"ECR-{nxt:03d}"
|
||||
|
||||
|
||||
def check() -> int:
|
||||
errors: list[str] = []
|
||||
warnings: list[str] = []
|
||||
|
||||
if not MIG_DIR.is_dir():
|
||||
errors.append(f"missing migrations dir: {MIG_DIR}")
|
||||
mig_files: list[Path] = []
|
||||
else:
|
||||
mig_files = sorted(MIG_DIR.glob("*.sql"))
|
||||
|
||||
if not ECR_DIR.is_dir():
|
||||
errors.append(f"missing ECR dir: {ECR_DIR}")
|
||||
ecr_files: list[Path] = []
|
||||
else:
|
||||
ecr_files = sorted(ECR_DIR.glob("ECR-*.md"))
|
||||
|
||||
# --- migrations ---
|
||||
by_stem = migration_stems(mig_files)
|
||||
for stem, files in sorted(by_stem.items()):
|
||||
# Same stem with both .up/.down is OK if only those; identical duplicate names shouldn't happen
|
||||
names = [f.name for f in files]
|
||||
if len(files) > 2 or (len(files) == 2 and not _up_down_pair(names)):
|
||||
if len(set(names)) != len(names):
|
||||
errors.append(f"migration duplicate filename: {names}")
|
||||
elif len(files) > 2:
|
||||
errors.append(f"migration stem over-shared ({stem}): {names}")
|
||||
|
||||
by_num = migration_numbers(mig_files)
|
||||
for num, stems in sorted(by_num.items()):
|
||||
if len(stems) > 1:
|
||||
errors.append(
|
||||
f"migration number collision {num}: {', '.join(stems)} "
|
||||
f"(allocate next from TRACEABILITY Max Migration; never reuse)"
|
||||
)
|
||||
|
||||
# --- ECR identity ---
|
||||
by_id: dict[str, list[Path]] = defaultdict(list)
|
||||
missing: list[Path] = []
|
||||
invalid: list[tuple[Path, str]] = []
|
||||
for p in ecr_files:
|
||||
ident = ecr_identity(p)
|
||||
if not ident:
|
||||
missing.append(p)
|
||||
continue
|
||||
if not ECR_ID_RE.match(ident):
|
||||
invalid.append((p, ident))
|
||||
continue
|
||||
by_id[ident].append(p)
|
||||
|
||||
for p in missing:
|
||||
errors.append(f"ECR missing identity (H1 or Canonical-ID): {p.name}")
|
||||
for p, ident in invalid:
|
||||
errors.append(f"ECR invalid identity {ident!r} in {p.name}")
|
||||
|
||||
for ident, files in sorted(by_id.items()):
|
||||
if len(files) > 1:
|
||||
errors.append(
|
||||
f"ECR identity collision {ident}: "
|
||||
+ ", ".join(f.name for f in files)
|
||||
+ " (ECR-NNN is globally unique; use ECR-NNN-suffix or ECR-NNNA)"
|
||||
)
|
||||
|
||||
# Soft: bare number reserved — if ECR-012 and ECR-012-star both exist, OK;
|
||||
# if two different files both declare ECR-012, caught above.
|
||||
|
||||
identities = list(by_id.keys())
|
||||
max_mig = max_migration_num(mig_files)
|
||||
next_ecr = next_ecr_suggestion(identities)
|
||||
|
||||
if errors:
|
||||
print("REPO GOVERNANCE: FAIL")
|
||||
for e in errors:
|
||||
print(f" ERROR: {e}")
|
||||
for w in warnings:
|
||||
print(f" WARN: {w}")
|
||||
return 1
|
||||
|
||||
print("REPO GOVERNANCE: PASS")
|
||||
print(f" migrations: {len(by_num)} unique numbers (max {max_mig:06d})")
|
||||
print(f" ECR identities: {len(by_id)} unique")
|
||||
print(f" suggested Next ECR: {next_ecr}")
|
||||
print(f" suggested Max Migration: {max_mig:06d}")
|
||||
for w in warnings:
|
||||
print(f" WARN: {w}")
|
||||
return 0
|
||||
|
||||
|
||||
def _up_down_pair(names: list[str]) -> bool:
|
||||
if len(names) != 2:
|
||||
return False
|
||||
a, b = sorted(names)
|
||||
return a.endswith(".up.sql") and b.endswith(".down.sql") and a[: -len(".up.sql")] == b[: -len(".down.sql")]
|
||||
|
||||
|
||||
def print_anchors() -> int:
|
||||
mig_files = sorted(MIG_DIR.glob("*.sql")) if MIG_DIR.is_dir() else []
|
||||
ecr_files = sorted(ECR_DIR.glob("ECR-*.md")) if ECR_DIR.is_dir() else []
|
||||
identities = []
|
||||
for p in ecr_files:
|
||||
i = ecr_identity(p)
|
||||
if i:
|
||||
identities.append(i)
|
||||
max_mig = max_migration_num(mig_files)
|
||||
print(f"Next ECR: {next_ecr_suggestion(identities)}")
|
||||
print(f"Max Migration: {max_mig:06d}")
|
||||
return 0
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description=__doc__)
|
||||
ap.add_argument(
|
||||
"--print-anchors",
|
||||
action="store_true",
|
||||
help="Print Next ECR / Max Migration suggestions and exit 0",
|
||||
)
|
||||
args = ap.parse_args()
|
||||
if args.print_anchors:
|
||||
return print_anchors()
|
||||
return check()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user