MoFin/scripts/check-prompt-deps.py

#!/usr/bin/env python3
"""
检查 MoFin prompt-manager 的跨提示词依赖一致性。

功能：
1. 对每个有 depends_on 的 prompt，验证上游 prompt 的 current_version 匹配
2. 对每个有 impacts 的 prompt，验证下游 prompt 的 depends_on.version 是否已正确更新
3. 输出状态：✅ 全部一致 / ⚠️ 有差异（列出差异详情）

用法：
  python3 check-prompt-deps.py
  python3 check-prompt-deps.py --registry /path/to/registry.json

集成方式：
  每次做策略分析前，或每次更新提示词后，运行一次确认一致性。
"""

import json
import sys
import os
from pathlib import Path

DEFAULT_REGISTRY = "/home/hmo/projects/MoFin/data/prompts/registry.json"

def load_registry(path: str) -> dict:
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)

def prompt_map(registry: dict) -> dict:
    """将 prompts 数组转为 id -> prompt 的字典"""
    return {p["id"]: p for p in registry.get("prompts", [])}

def check_depends_on(pm: dict) -> list:
    """检查所有 depends_on 声明是否匹配上游的 current_version"""
    issues = []
    for p in pm.values():
        deps = p.get("depends_on")
        if not deps:
            continue
        target_id = deps.get("prompt")
        expected_ver = deps.get("version")
        target = pm.get(target_id)
        if not target:
            issues.append({
                "type": "MISSING_UPSTREAM",
                "prompt": p["id"],
                "detail": f"依赖的上游提示词 '{target_id}' 在 registry 中不存在"
            })
            continue
        actual_ver = target.get("current_version")
        if actual_ver != expected_ver:
            issues.append({
                "type": "VERSION_MISMATCH",
                "prompt": p["id"],
                "detail": (
                    f"{p['id']}.depends_on → {target_id}@{expected_ver}, "
                    f"但 {target_id}.current_version = {actual_ver}"
                )
            })
    return issues

def check_impacts_fulfilled(pm: dict) -> list:
    """检查有 impacts 声明的 prompt，其下游是否已更新 depends_on 到当前版本"""
    issues = []
    for p in pm.values():
        impacted = p.get("impacts")
        if not impacted:
            continue
        my_id = p["id"]
        my_ver = p.get("current_version")
        for downstream_id in impacted:
            downstream = pm.get(downstream_id)
            if not downstream:
                issues.append({
                    "type": "MISSING_DOWNSTREAM",
                    "prompt": my_id,
                    "detail": f"impacts 中列出的 '{downstream_id}' 在 registry 中不存在"
                })
                continue
            dd = downstream.get("depends_on", {})
            expected_ver = dd.get("version") if dd.get("prompt") == my_id else None
            if expected_ver is None:
                issues.append({
                    "type": "MISSING_DEPENDS_ON",
                    "prompt": my_id,
                    "detail": (
                        f"{my_id}.impacts 声明了 {downstream_id}, "
                        f"但 {downstream_id} 没有 depends_on → {my_id}"
                    )
                })
            elif expected_ver != my_ver:
                issues.append({
                    "type": "STALE_DEPENDENCY",
                    "prompt": my_id,
                    "detail": (
                        f"{my_id}.current_version = {my_ver}, "
                        f"但 {downstream_id}.depends_on.version = {expected_ver}"
                    )
                })
    return issues

def check_integrity(pm: dict) -> list:
    """基本格式校验"""
    issues = []
    for p in pm.values():
        deps = p.get("depends_on")
        if deps:
            if not deps.get("prompt") or not deps.get("version"):
                issues.append({
                    "type": "INVALID_DEPENDS_ON",
                    "prompt": p["id"],
                    "detail": "depends_on 结构不完整，需包含 prompt 和 version"
                })
            if not deps.get("reason"):
                issues.append({
                    "type": "MISSING_REASON",
                    "prompt": p["id"],
                    "detail": "depends_on 缺少 reason 字段（说明为什么依赖）"
                })
        impacts = p.get("impacts")
        if impacts and not isinstance(impacts, list):
            issues.append({
                "type": "INVALID_IMPACTS",
                "prompt": p["id"],
                "detail": "impacts 必须是数组"
            })
    return issues

def main():
    reg_path = DEFAULT_REGISTRY
    if len(sys.argv) > 2 and sys.argv[1] == "--registry":
        reg_path = sys.argv[2]
    if not os.path.exists(reg_path):
        print(f"❌ 找不到 registry 文件: {reg_path}")
        sys.exit(1)

    registry = load_registry(reg_path)
    pm = prompt_map(registry)

    all_issues = []
    all_issues.extend(check_integrity(pm))
    all_issues.extend(check_depends_on(pm))
    all_issues.extend(check_impacts_fulfilled(pm))

    print(f"📋 MoFin 提示词版本依赖检查 ({registry.get('updated_at', '?')})")
    print(f"   共 {len(pm)} 个提示词")
    print()

    if not all_issues:
        print("✅ 全部一致，无依赖问题")
        # 打印当前依赖状态一览
        print()
        for p in pm.values():
            deps = p.get("depends_on")
            impacts = p.get("impacts")
            status = p["id"]
            if deps:
                status += f" ← 依赖 {deps['prompt']}@{deps['version']}"
            if impacts:
                status += f" → 影响 {', '.join(impacts)}"
            print(f"  {status}")
        sys.exit(0)

    # 按类型分组输出
    categories = {
        "MISSING_UPSTREAM": "上游提示词缺失",
        "VERSION_MISMATCH": "依赖版本不匹配",
        "MISSING_DOWNSTREAM": "下游提示词缺失",
        "MISSING_DEPENDS_ON": "下游缺少 depends_on 声明",
        "STALE_DEPENDENCY": "下游依赖版本未同步更新",
        "INVALID_DEPENDS_ON": "depends_on 结构不完整",
        "MISSING_REASON": "缺少依赖原因说明",
        "INVALID_IMPACTS": "impacts 格式错误",
    }

    has_error = False
    for issue in all_issues:
        cat = categories.get(issue["type"], issue["type"])
        marker = "❌" if issue["type"] in ("VERSION_MISMATCH", "STALE_DEPENDENCY", "MISSING_UPSTREAM") else "⚠️"
        if marker == "❌":
            has_error = True
        print(f"{marker} [{issue['prompt']}] {cat}")
        print(f"   {issue['detail']}")

    print()
    if has_error:
        print("❗ 存在必须修复的依赖问题，请在更新提示词前先处理。")
    else:
        print("💡 无版本冲突，但上面有信息缺失/规范问题，建议补充。")

    sys.exit(1 if has_error else 0)

if __name__ == "__main__":
    main()