82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
||
"""Stop hook: 检查是否有逻辑改动未验证（未跑测试）、DDL 变更未建迁移"""
|
||
import json, re, subprocess, sys, os
|
||
|
||
project_dir = os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd())
|
||
|
||
try:
|
||
r = subprocess.run(
|
||
["git", "diff", "--name-only"],
|
||
capture_output=True, text=True, timeout=10, cwd=project_dir,
|
||
)
|
||
staged = subprocess.run(
|
||
["git", "diff", "--name-only", "--cached"],
|
||
capture_output=True, text=True, timeout=10, cwd=project_dir,
|
||
)
|
||
changed = set((r.stdout + "\n" + staged.stdout).strip().splitlines())
|
||
except Exception:
|
||
sys.exit(0)
|
||
|
||
if not changed:
|
||
sys.exit(0)
|
||
|
||
warnings = []
|
||
|
||
# --- 1. 逻辑改动未验证检查 ---
|
||
LOGIC_PATTERNS = [
|
||
r"^apps/etl/connectors/feiqiu/(tasks|loaders|scd|orchestration|config|database|models|quality)/",
|
||
r"^apps/backend/app/(routers|services|auth|schemas)/",
|
||
r"^packages/shared/",
|
||
]
|
||
|
||
logic_files = [
|
||
f for f in changed
|
||
if any(re.search(p, f) for p in LOGIC_PATTERNS)
|
||
]
|
||
|
||
if logic_files:
|
||
# 检查是否有测试结果文件被修改（间接判断是否跑了测试）
|
||
# 更可靠的方式：检查变更中是否包含测试文件
|
||
test_files = [f for f in changed if re.search(r"tests?/", f)]
|
||
if not test_files:
|
||
count = len(logic_files)
|
||
warnings.append(
|
||
f"本次会话修改了 {count} 个逻辑文件但未发现测试文件变更，"
|
||
"建议运行相关测试验证（单元/集成/lint）"
|
||
)
|
||
|
||
# --- 2. DDL 变更未建迁移检查 ---
|
||
schema_files = [f for f in changed if re.search(r"^db/[^/]+/schemas/.*\.sql$", f)]
|
||
migration_files = [f for f in changed if re.search(r"^db/[^/]+/migrations/.*\.sql$", f)]
|
||
|
||
if schema_files and not migration_files:
|
||
# 也检查 untracked 的迁移文件
|
||
try:
|
||
untracked = subprocess.run(
|
||
["git", "ls-files", "--others", "--exclude-standard", "db/"],
|
||
capture_output=True, text=True, timeout=10, cwd=project_dir,
|
||
)
|
||
new_migrations = [
|
||
f for f in untracked.stdout.strip().splitlines()
|
||
if re.search(r"^db/[^/]+/migrations/.*\.sql$", f)
|
||
]
|
||
except Exception:
|
||
new_migrations = []
|
||
|
||
if not new_migrations:
|
||
dbs = set()
|
||
for f in schema_files:
|
||
m = re.match(r"^db/([^/]+)/", f)
|
||
if m:
|
||
dbs.add(m.group(1))
|
||
db_list = ", ".join(sorted(dbs))
|
||
warnings.append(
|
||
f"检测到 {db_list} 的 schemas/ DDL 已变更但无对应迁移脚本，"
|
||
"建议在 db/*/migrations/ 创建增量迁移文件"
|
||
)
|
||
|
||
# --- 输出 ---
|
||
if warnings:
|
||
msg = "[verify-check] " + " | ".join(warnings)
|
||
print(json.dumps({"systemMessage": msg}))
|