361 lines
11 KiB
Python
361 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Postmortem 匹配检查脚本
|
||
检查当前变更是否触发历史 postmortem
|
||
|
||
使用方法:
|
||
python tools/postmortem_check.py [--base main] [--mode warn|block]
|
||
|
||
返回码:
|
||
0 - 无匹配或仅警告
|
||
1 - 发现高置信度匹配（blocking mode）
|
||
2 - 错误
|
||
"""
|
||
import argparse
|
||
import fnmatch
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
import yaml
|
||
|
||
POSTMORTEM_DIR = Path(__file__).parent.parent / "postmortem"
|
||
|
||
|
||
@dataclass
|
||
class MatchResult:
|
||
"""单个匹配结果"""
|
||
|
||
pm_id: str
|
||
reason: str
|
||
confidence: float
|
||
|
||
|
||
@dataclass
|
||
class AggregatedMatch:
|
||
"""聚合后的匹配结果"""
|
||
|
||
pm_id: str
|
||
reasons: List[str] = field(default_factory=list)
|
||
max_confidence: float = 0.0
|
||
match_count: int = 0
|
||
final_confidence: float = 0.0
|
||
has_file_match: bool = False
|
||
has_content_match: bool = False
|
||
|
||
|
||
class PostmortemMatcher:
|
||
"""Postmortem 匹配器"""
|
||
|
||
# 匹配权重
|
||
WEIGHT_FILE_EXACT = 0.8
|
||
WEIGHT_FILE_PATTERN = 0.6
|
||
WEIGHT_FUNCTION = 0.7
|
||
WEIGHT_PATTERN = 0.5
|
||
WEIGHT_KEYWORD = 0.4
|
||
|
||
def __init__(self):
|
||
self.postmortems = self._load_all_postmortems()
|
||
|
||
def _load_all_postmortems(self) -> List[Dict]:
|
||
"""加载所有 postmortem 文件"""
|
||
pms = []
|
||
if not POSTMORTEM_DIR.exists():
|
||
return pms
|
||
|
||
for f in POSTMORTEM_DIR.glob("PM-*.yaml"):
|
||
try:
|
||
with open(f, encoding="utf-8") as fp:
|
||
pm = yaml.safe_load(fp)
|
||
if pm:
|
||
pms.append(pm)
|
||
except Exception as e:
|
||
print(f"Warning: Failed to load {f}: {e}", file=sys.stderr)
|
||
|
||
return pms
|
||
|
||
def match_files(self, changed_files: List[str]) -> List[MatchResult]:
|
||
"""匹配修改的文件"""
|
||
matches = []
|
||
|
||
for pm in self.postmortems:
|
||
triggers = pm.get("triggers", {})
|
||
pm_files = triggers.get("files", [])
|
||
pm_id = pm.get("id", "unknown")
|
||
|
||
for changed in changed_files:
|
||
for pattern in pm_files:
|
||
if self._file_matches(changed, pattern):
|
||
# 精确匹配 vs 模式匹配
|
||
is_exact = "*" not in pattern and "?" not in pattern
|
||
confidence = (
|
||
self.WEIGHT_FILE_EXACT if is_exact else self.WEIGHT_FILE_PATTERN
|
||
)
|
||
matches.append(
|
||
MatchResult(
|
||
pm_id=pm_id,
|
||
reason=f"File: {changed} ~ {pattern}",
|
||
confidence=confidence,
|
||
)
|
||
)
|
||
|
||
return matches
|
||
|
||
def match_diff_content(self, diff: str) -> List[MatchResult]:
|
||
"""匹配 diff 内容中的函数名和模式"""
|
||
matches = []
|
||
|
||
for pm in self.postmortems:
|
||
triggers = pm.get("triggers", {})
|
||
pm_id = pm.get("id", "unknown")
|
||
|
||
# 函数名匹配
|
||
for func in triggers.get("functions", []):
|
||
if not func:
|
||
continue
|
||
try:
|
||
if re.search(rf"\b{re.escape(func)}\b", diff):
|
||
matches.append(
|
||
MatchResult(
|
||
pm_id=pm_id,
|
||
reason=f"Function: {func}",
|
||
confidence=self.WEIGHT_FUNCTION,
|
||
)
|
||
)
|
||
except re.error:
|
||
continue
|
||
|
||
# 正则模式匹配
|
||
for pattern in triggers.get("patterns", []):
|
||
if not pattern:
|
||
continue
|
||
try:
|
||
if re.search(pattern, diff, re.IGNORECASE):
|
||
matches.append(
|
||
MatchResult(
|
||
pm_id=pm_id,
|
||
reason=f"Pattern: {pattern}",
|
||
confidence=self.WEIGHT_PATTERN,
|
||
)
|
||
)
|
||
except re.error:
|
||
continue
|
||
|
||
# 关键词匹配
|
||
for keyword in triggers.get("keywords", []):
|
||
if not keyword:
|
||
continue
|
||
if keyword.lower() in diff.lower():
|
||
matches.append(
|
||
MatchResult(
|
||
pm_id=pm_id,
|
||
reason=f"Keyword: {keyword}",
|
||
confidence=self.WEIGHT_KEYWORD,
|
||
)
|
||
)
|
||
|
||
return matches
|
||
|
||
def _file_matches(self, filepath: str, pattern: str) -> bool:
|
||
"""检查文件路径是否匹配模式"""
|
||
# 支持 glob 模式
|
||
if fnmatch.fnmatch(filepath, pattern):
|
||
return True
|
||
# 也尝试匹配文件名部分
|
||
if fnmatch.fnmatch(Path(filepath).name, pattern):
|
||
return True
|
||
# 检查是否包含（用于简单的路径匹配）
|
||
if "*" not in pattern and "?" not in pattern:
|
||
return pattern in filepath
|
||
return False
|
||
|
||
def aggregate_matches(
|
||
self, file_matches: List[MatchResult], content_matches: List[MatchResult]
|
||
) -> Dict[str, AggregatedMatch]:
|
||
"""聚合匹配结果，计算综合置信度"""
|
||
result: Dict[str, AggregatedMatch] = {}
|
||
|
||
all_matches = file_matches + content_matches
|
||
for match in all_matches:
|
||
if match.pm_id not in result:
|
||
result[match.pm_id] = AggregatedMatch(pm_id=match.pm_id)
|
||
|
||
agg = result[match.pm_id]
|
||
agg.reasons.append(match.reason)
|
||
agg.max_confidence = max(agg.max_confidence, match.confidence)
|
||
agg.match_count += 1
|
||
|
||
if "File" in match.reason:
|
||
agg.has_file_match = True
|
||
if "Function" in match.reason or "Pattern" in match.reason:
|
||
agg.has_content_match = True
|
||
|
||
# 计算综合置信度
|
||
for pm_id, agg in result.items():
|
||
base = agg.max_confidence
|
||
# 每增加一个匹配项，加 0.1 bonus（最多 0.3）
|
||
count_bonus = min(0.3, 0.1 * (agg.match_count - 1))
|
||
# 如果同时有文件匹配和内容匹配，额外加 0.1
|
||
cross_bonus = 0.1 if (agg.has_file_match and agg.has_content_match) else 0
|
||
agg.final_confidence = min(1.0, base + count_bonus + cross_bonus)
|
||
|
||
return result
|
||
|
||
def get_postmortem_details(self, pm_id: str) -> Optional[Dict]:
|
||
"""获取 postmortem 详情"""
|
||
for pm in self.postmortems:
|
||
if pm.get("id") == pm_id:
|
||
return pm
|
||
return None
|
||
|
||
|
||
def get_changed_files(base_ref: str) -> List[str]:
|
||
"""获取相对于 base 的变更文件"""
|
||
cwd = POSTMORTEM_DIR.parent
|
||
|
||
# 尝试不同的 diff 方式
|
||
# 1. 三点 diff（用于 PR）
|
||
cmd = ["git", "diff", "--name-only", f"{base_ref}...HEAD"]
|
||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd)
|
||
|
||
if result.returncode != 0:
|
||
# 2. 两点 diff
|
||
cmd = ["git", "diff", "--name-only", f"{base_ref}..HEAD"]
|
||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd)
|
||
|
||
if result.returncode != 0:
|
||
# 3. 直接对比
|
||
cmd = ["git", "diff", "--name-only", base_ref]
|
||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd)
|
||
|
||
return [f for f in result.stdout.strip().split("\n") if f]
|
||
|
||
|
||
def get_diff_content(base_ref: str) -> str:
|
||
"""获取相对于 base 的 diff 内容"""
|
||
cwd = POSTMORTEM_DIR.parent
|
||
|
||
# 只看代码文件的 diff
|
||
extensions = ["*.py", "*.js", "*.ts", "*.jsx", "*.tsx"]
|
||
cmd = ["git", "diff", f"{base_ref}...HEAD", "--"] + extensions
|
||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd)
|
||
|
||
if result.returncode != 0:
|
||
cmd = ["git", "diff", f"{base_ref}..HEAD", "--"] + extensions
|
||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd)
|
||
|
||
if result.returncode != 0:
|
||
cmd = ["git", "diff", base_ref, "--"] + extensions
|
||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd)
|
||
|
||
return result.stdout
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Postmortem Check")
|
||
parser.add_argument("--base", default="main", help="Base branch/commit")
|
||
parser.add_argument(
|
||
"--mode",
|
||
choices=["warn", "block"],
|
||
default="warn",
|
||
help="warn: only print, block: exit 1 on high confidence",
|
||
)
|
||
parser.add_argument(
|
||
"--threshold",
|
||
type=float,
|
||
default=0.7,
|
||
help="Confidence threshold for blocking (default: 0.7)",
|
||
)
|
||
parser.add_argument(
|
||
"--output",
|
||
choices=["text", "json"],
|
||
default="text",
|
||
help="Output format",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
if not POSTMORTEM_DIR.exists():
|
||
print("No postmortem directory found. Run postmortem_init.py first.")
|
||
sys.exit(0)
|
||
|
||
matcher = PostmortemMatcher()
|
||
|
||
if not matcher.postmortems:
|
||
print("No postmortems found.")
|
||
sys.exit(0)
|
||
|
||
# 获取变更
|
||
changed_files = get_changed_files(args.base)
|
||
diff_content = get_diff_content(args.base)
|
||
|
||
if not changed_files:
|
||
print("No changes detected.")
|
||
sys.exit(0)
|
||
|
||
print(f"Checking {len(changed_files)} changed files against {len(matcher.postmortems)} postmortems...")
|
||
|
||
# 执行匹配
|
||
file_matches = matcher.match_files(changed_files)
|
||
content_matches = matcher.match_diff_content(diff_content)
|
||
|
||
# 聚合结果
|
||
results = matcher.aggregate_matches(file_matches, content_matches)
|
||
|
||
if not results:
|
||
print("No postmortem matches found.")
|
||
sys.exit(0)
|
||
|
||
# 输出结果
|
||
blocking = []
|
||
warnings = []
|
||
|
||
sorted_results = sorted(
|
||
results.items(), key=lambda x: x[1].final_confidence, reverse=True
|
||
)
|
||
|
||
for pm_id, agg in sorted_results:
|
||
confidence = agg.final_confidence
|
||
is_blocking = confidence >= args.threshold
|
||
level = "BLOCK" if is_blocking else "WARN"
|
||
|
||
# 获取 postmortem 详情
|
||
pm = matcher.get_postmortem_details(pm_id)
|
||
if not pm:
|
||
pm = {"title": "Unknown", "severity": "unknown"}
|
||
|
||
print(f"\n[{level}] {pm_id} ({confidence:.0%} confidence)")
|
||
print(f" Title: {pm.get('title', 'N/A')}")
|
||
print(f" Severity: {pm.get('severity', 'N/A')}")
|
||
print(f" Reasons:")
|
||
for reason in agg.reasons[:5]: # 最多显示5个原因
|
||
print(f" - {reason}")
|
||
|
||
if pm.get("verification"):
|
||
print(f" Verification checklist:")
|
||
for check in pm["verification"]:
|
||
print(f" [ ] {check}")
|
||
|
||
if is_blocking:
|
||
blocking.append(pm_id)
|
||
else:
|
||
warnings.append(pm_id)
|
||
|
||
# 总结
|
||
print(f"\n{'=' * 50}")
|
||
print(f"Summary: {len(blocking)} blocking, {len(warnings)} warnings")
|
||
|
||
# 决定退出码
|
||
if blocking and args.mode == "block":
|
||
print(f"\n{len(blocking)} postmortem(s) triggered with high confidence.")
|
||
print("Please review the changes and verify they don't reintroduce past issues.")
|
||
sys.exit(1)
|
||
|
||
sys.exit(0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|