🛠 MVP 仓库

#!/usr/bin/env python3
"""Model Handoff: turn chats and repository state into a portable handoff package."""

from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import os
import re
import sqlite3
import subprocess
import sys
import tempfile
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any, Iterable


SCHEMA_VERSION = "1.0"
MAX_MODEL_CHARS = 60_000
ROLE_ALIASES = {
    "human": "user",
    "user": "user",
    "assistant": "assistant",
    "ai": "assistant",
    "bot": "assistant",
    "system": "system",
    "tool": "tool",
}
SECRET_PATTERNS = [
    (re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----", re.S), "[REDACTED_PRIVATE_KEY]"),
    (re.compile(r"(?i)\b(bearer)\s+[A-Za-z0-9._~+/=-]{12,}"), r"\1 [REDACTED_TOKEN]"),
    (re.compile(r"\b(?:sk|pk)-[A-Za-z0-9_-]{16,}\b"), "[REDACTED_API_KEY]"),
    (re.compile(r"\bgh[opusr]_[A-Za-z0-9]{20,}\b"), "[REDACTED_GITHUB_TOKEN]"),
    (re.compile(r"(?i)\b(api[_-]?key|token|password|secret)\s*([:=])\s*([^\s,;]+)"), r"\1\2[REDACTED]"),
    (re.compile(r"(?i)(https?://[^\s:/]+:)[^@\s]+(@)"), r"\1[REDACTED]\2"),
]
FILE_TOKEN = re.compile(
    r"(?<![\w.-])(?:`)?((?:\.?\.?/)?(?:[\w.@+-]+/)*[\w.@+-]+\.(?:py|js|ts|tsx|jsx|go|rs|java|rb|php|cs|cpp|c|h|md|json|ya?ml|toml|ini|sql|sh|html|css))(?:[:#]L?(\d+))?(?:`)?"
)
TODO_LINE = re.compile(r"(?i)\b(todo|fixme|remaining|next(?: step)?|待办|未完成|下一步|需要)\b")
DONE_LINE = re.compile(r"(?i)\b(done|completed|implemented|fixed|finished|已完成|实现了|修复了)\b")
DECISION_LINE = re.compile(r"(?i)\b(decided|decision|chosen|use|采用|决定|选择)\b")
COMMAND_LINE = re.compile(r"^\s*(?:\$\s*)?((?:git|python3?|pytest|npm|pnpm|yarn|cargo|go|make|docker)\s+[^\n]+)$", re.M)


def utc_now() -> str:
    return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()


# 导入常见 JSON 聊天格式或纯文本，并归一化为 role/content。
def load_chat(path: Path) -> tuple[list[dict[str, str]], str]:
    raw = path.read_text(encoding="utf-8", errors="replace")
    if path.suffix.lower() == ".json":
        try:
            data = json.loads(raw)
            messages = normalize_json_chat(data)
            if messages:
                return messages, "json"
        except json.JSONDecodeError as exc:
            raise ValueError(f"JSON 解析失败: {exc}") from exc
    return parse_plain_chat(raw), "text"


def normalize_json_chat(data: Any) -> list[dict[str, str]]:
    if isinstance(data, dict):
        for key in ("messages", "conversation", "chat", "items"):
            if isinstance(data.get(key), list):
                data = data[key]
                break
        else:
            # 支持 Claude/ChatGPT 导出中以 mapping 保存节点的形式。
            if isinstance(data.get("mapping"), dict):
                data = [v.get("message") for v in data["mapping"].values() if isinstance(v, dict)]
    if not isinstance(data, list):
        return []

    result: list[dict[str, str]] = []
    for item in data:
        if not isinstance(item, dict):
            continue
        role_value = item.get("role") or item.get("author") or item.get("sender") or "user"
        if isinstance(role_value, dict):
            role_value = role_value.get("role") or role_value.get("name") or "user"
        role = ROLE_ALIASES.get(str(role_value).lower(), str(role_value).lower())
        content: Any = item.get("content", item.get("text", item.get("message", "")))
        if isinstance(content, dict):
            content = content.get("text") or content.get("parts") or content.get("content") or ""
        if isinstance(content, list):
            parts = []
            for part in content:
                if isinstance(part, str):
                    parts.append(part)
                elif isinstance(part, dict):
                    parts.append(str(part.get("text") or part.get("content") or ""))
            content = "\n".join(p for p in parts if p)
        content = str(content).strip()
        if content:
            result.append({"role": role, "content": content})
    return result


def parse_plain_chat(raw: str) -> list[dict[str, str]]:
    marker = re.compile(r"(?im)^\s*(user|human|assistant|ai|system|tool|用户|助手|系统)\s*[:：]\s*")
    matches = list(marker.finditer(raw))
    if not matches:
        return [{"role": "user", "content": raw.strip()}] if raw.strip() else []
    localized = {"用户": "user", "助手": "assistant", "系统": "system"}
    result = []
    for index, match in enumerate(matches):
        end = matches[index + 1].start() if index + 1 < len(matches) else len(raw)
        content = raw[match.end():end].strip()
        role_key = match.group(1).lower()
        role = localized.get(role_key, ROLE_ALIASES.get(role_key, role_key))
        if content:
            result.append({"role": role, "content": content})
    return result


# 在任何外发或落盘前清理常见密钥；返回替换次数便于审计。
def redact_text(text: str) -> tuple[str, int]:
    count = 0
    for pattern, replacement in SECRET_PATTERNS:
        text, found = pattern.subn(replacement, text)
        count += found
    return text, count


def redact_messages(messages: Iterable[dict[str, str]]) -> tuple[list[dict[str, str]], int]:
    cleaned, total = [], 0
    for message in messages:
        content, count = redact_text(message["content"])
        cleaned.append({"role": message["role"], "content": content})
        total += count
    return cleaned, total


def run_git(repo: Path, *args: str, timeout: int = 8) -> str:
    try:
        proc = subprocess.run(
            ["git", "-C", str(repo), *args], capture_output=True, text=True,
            encoding="utf-8", errors="replace", timeout=timeout, check=False,
        )
        return proc.stdout.strip() if proc.returncode == 0 else ""
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return ""


# 只采集可复现交接所需的 Git 元数据，不读取提交中的敏感文件内容。
def collect_git(repo: Path) -> dict[str, Any]:
    inside = run_git(repo, "rev-parse", "--is-inside-work-tree") == "true"
    if not inside:
        return {"available": False, "repo": str(repo.resolve())}
    root_text = run_git(repo, "rev-parse", "--show-toplevel")
    root = Path(root_text).resolve() if root_text else repo.resolve()
    status = run_git(root, "status", "--short", "--untracked-files=all")
    return {
        "available": True,
        "repo": str(root),
        "branch": run_git(root, "branch", "--show-current") or "(detached)",
        "head": run_git(root, "rev-parse", "--short", "HEAD"),
        "last_commit": run_git(root, "log", "-1", "--pretty=%h %s"),
        "status": status.splitlines()[:200],
        "diff_stat": run_git(root, "diff", "--stat").splitlines()[:100],
    }


def compact_lines(messages: list[dict[str, str]], pattern: re.Pattern[str], limit: int = 12) -> list[str]:
    found: list[str] = []
    for message in reversed(messages):
        for line in reversed(message["content"].splitlines()):
            clean = re.sub(r"\s+", " ", line).strip(" -*#\t")
            if pattern.search(clean) and 4 <= len(clean) <= 300 and clean not in found:
                found.append(clean)
                if len(found) >= limit:
                    return list(reversed(found))
    return list(reversed(found))


def rule_summary(messages: list[dict[str, str]], git: dict[str, Any]) -> dict[str, Any]:
    user_messages = [m["content"] for m in messages if m["role"] == "user"]
    assistant_messages = [m["content"] for m in messages if m["role"] == "assistant"]
    objective = next((re.sub(r"\s+", " ", text)[:500] for text in user_messages if text.strip()), "未识别")
    last_exchange = messages[-4:]
    recent = []
    for message in last_exchange:
        compact = re.sub(r"\s+", " ", message["content"])[:280]
        recent.append(f'{message["role"]}: {compact}')
    completed = compact_lines(messages, DONE_LINE)
    remaining = compact_lines(messages, TODO_LINE)
    decisions = compact_lines(messages, DECISION_LINE, 8)
    if not completed and assistant_messages:
        completed = ["旧模型最后回复: " + re.sub(r"\s+", " ", assistant_messages[-1])[:400]]
    if not remaining:
        remaining = ["核对最近对话与 Git 未提交改动，确认下一步。"]
    return {
        "objective": objective,
        "summary": "；".join(recent) if recent else "没有可用对话内容。",
        "completed": completed,
        "remaining": remaining,
        "decisions": decisions,
        "constraints": [],
        "commands": list(dict.fromkeys(COMMAND_LINE.findall("\n".join(m["content"] for m in messages))))[:20],
        "errors": compact_lines(messages, re.compile(r"(?i)\b(error|exception|failed|失败|报错)\b"), 8),
    }


def extract_file_references(messages: list[dict[str, str]], repo: Path) -> list[dict[str, Any]]:
    candidates: dict[tuple[str, int | None], None] = {}
    for message in messages:
        for match in FILE_TOKEN.finditer(message["content"]):
            candidates[(match.group(1), int(match.group(2)) if match.group(2) else None)] = None

    root = repo.resolve()
    refs = []
    for raw, line in list(candidates)[:100]:
        candidate = (root / raw).resolve() if not Path(raw).is_absolute() else Path(raw).resolve()
        try:
            relative = candidate.relative_to(root)
            inside = True
        except ValueError:
            relative, inside = Path(raw), False
        exists = inside and candidate.is_file()
        line_valid: bool | None = None
        if exists and line is not None:
            try:
                with candidate.open("r", encoding="utf-8", errors="replace") as handle:
                    line_valid = sum(1 for _ in handle) >= line
            except OSError:
                line_valid = False
        refs.append({
            "path": relative.as_posix(), "line": line, "inside_repo": inside,
            "exists": exists, "line_valid": line_valid,
        })
    return refs


def model_prompt(messages: list[dict[str, str]], git: dict[str, Any]) -> str:
    transcript = "\n\n".join(f'[{m["role"]}]\n{m["content"]}' for m in messages)
    prompt = f"""你是软件项目交接助手。根据脱敏后的对话和 Git 元数据生成交接信息。
仅输出合法 JSON 对象，键必须是 objective, summary, completed, remaining, decisions, constraints, commands, errors。
除 objective 和 summary 为字符串外，其余值均为字符串数组。不要虚构已完成事项。

Git 元数据:
{json.dumps(git, ensure_ascii=False)}

对话:
{transcript}
"""
    return prompt[-MAX_MODEL_CHARS:]


# 调用 OpenAI Chat Completions 兼容端点；requests 缺失时退回标准库。
def call_compatible_model(url: str, api_key: str, model: str, prompt: str) -> dict[str, Any]:
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.1,
        "response_format": {"type": "json_object"},
    }
    headers = {"Content-Type": "application/json"}
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"
    body: Any
    try:
        import requests  # type: ignore
        response = requests.post(url, headers=headers, json=payload, timeout=60)
        response.raise_for_status()
        body = response.json()
    except ImportError:
        request = urllib.request.Request(url, data=json.dumps(payload).encode(), headers=headers, method="POST")
        with urllib.request.urlopen(request, timeout=60) as response:
            body = json.loads(response.read().decode())
    content = body["choices"][0]["message"]["content"]
    if isinstance(content, list):
        content = "".join(part.get("text", "") for part in content if isinstance(part, dict))
    match = re.search(r"\{.*\}", str(content), re.S)
    if not match:
        raise ValueError("模型响应中没有 JSON 对象")
    result = json.loads(match.group(0))
    if not isinstance(result, dict):
        raise ValueError("模型响应不是 JSON 对象")
    return normalize_handoff(result)


def normalize_handoff(value: dict[str, Any]) -> dict[str, Any]:
    result = {}
    for key in ("objective", "summary"):
        item = value.get(key, "")
        result[key] = item if isinstance(item, str) else json.dumps(item, ensure_ascii=False)
    for key in ("completed", "remaining", "decisions", "constraints", "commands", "errors"):
        item = value.get(key, [])
        if isinstance(item, str):
            item = [item]
        result[key] = [str(v) for v in item] if isinstance(item, list) else []
    return result


def build_package(args: argparse.Namespace) -> dict[str, Any]:
    input_path = Path(args.input).expanduser().resolve()
    repo = Path(args.repo or os.getcwd()).expanduser().resolve()
    messages, source_format = load_chat(input_path)
    if not messages:
        raise ValueError("输入中没有可识别的消息")
    clean_messages, redaction_count = redact_messages(messages)
    git = collect_git(repo)
    git_repo = Path(git["repo"]) if git.get("available") else repo
    handoff = rule_summary(clean_messages, git)
    generator = "rules"
    model_error = ""
    if args.model_url:
        try:
            api_key = os.environ.get(args.api_key_env, "")
            handoff = call_compatible_model(args.model_url, api_key, args.model, model_prompt(clean_messages, git))
            generator = f"compatible-model:{args.model}"
        except Exception as exc:  # 网络或供应商异常不阻断本地交接生成。
            model_error = f"{type(exc).__name__}: {exc}"

    refs = extract_file_references(clean_messages, git_repo)
    package = {
        "schema_version": SCHEMA_VERSION,
        "created_at": utc_now(),
        "source": {"path": str(input_path), "format": source_format, "sha256": hashlib.sha256(input_path.read_bytes()).hexdigest()},
        "conversation": {
            "message_count": len(clean_messages),
            "roles": {role: sum(m["role"] == role for m in clean_messages) for role in sorted({m["role"] for m in clean_messages})},
        },
        "privacy": {"redactions": redaction_count, "raw_chat_embedded": False},
        "generator": generator,
        "generator_warning": model_error,
        "git": git,
        "handoff": handoff,
        "file_references": refs,
    }
    validate_package(package)
    return package


# 校验稳定 schema 和文件引用字段，便于另一个模型或工具可靠消费。
def validate_package(package: dict[str, Any]) -> list[str]:
    errors = []
    required = {"schema_version", "created_at", "source", "conversation", "privacy", "git", "handoff", "file_references"}
    missing = sorted(required - package.keys())
    if missing:
        errors.append("缺少字段: " + ", ".join(missing))
    handoff = package.get("handoff", {})
    if not isinstance(handoff, dict):
        errors.append("handoff 必须是对象")
    else:
        for key in ("objective", "summary"):
            if not isinstance(handoff.get(key), str):
                errors.append(f"handoff.{key} 必须是字符串")
        for key in ("completed", "remaining", "decisions", "constraints", "commands", "errors"):
            if not isinstance(handoff.get(key), list):
                errors.append(f"handoff.{key} 必须是数组")
    if not isinstance(package.get("file_references"), list):
        errors.append("file_references 必须是数组")
    return errors


def markdown_list(items: list[Any], empty: str = "- 暂无") -> str:
    return "\n".join(f"- {str(item).strip()}" for item in items if str(item).strip()) or empty


def render_markdown(package: dict[str, Any]) -> str:
    handoff, git = package["handoff"], package["git"]
    refs = package["file_references"]
    ref_lines = []
    for ref in refs:
        suffix = f':{ref["line"]}' if ref["line"] else ""
        state = "有效" if ref["exists"] and ref["line_valid"] is not False else "缺失或越界"
        if not ref["inside_repo"]:
            state = "仓库外引用"
        ref_lines.append(f'{ref["path"]}{suffix} ({state})')
    git_status = git.get("status", []) if git.get("available") else ["非 Git 仓库或 Git 不可用"]
    metadata = "\n".join([
        "# Model Handoff",
        "",
        f'- Schema: {package["schema_version"]}',
        f'- 生成时间: {package["created_at"]}',
        f'- 生成方式: {package["generator"]}',
        f'- 消息数: {package["conversation"]["message_count"]}',
        f'- 隐私替换: {package["privacy"]["redactions"]}',
    ])
    sections = [metadata]
    if package.get("generator_warning"):
        sections.append(f'> 模型调用失败，已回退到规则摘要：{package["generator_warning"]}')
    sections.extend([
        f'## 目标\n\n{handoff["objective"]}',
        f'## 当前上下文\n\n{handoff["summary"]}',
        f'## 已完成\n\n{markdown_list(handoff["completed"])}',
        f'## 剩余事项\n\n{markdown_list(handoff["remaining"])}',
        f'## 决策\n\n{markdown_list(handoff["decisions"])}',
        f'## 约束\n\n{markdown_list(handoff["constraints"])}',
        f'## 相关文件\n\n{markdown_list(ref_lines)}',
        f'## 已运行命令\n\n{markdown_list(handoff["commands"])}',
        f'## 错误与阻塞\n\n{markdown_list(handoff["errors"])}',
        "\n".join([
            "## Git 快照", "",
            f'- 仓库: {git.get("repo", "")}',
            f'- 分支: {git.get("branch", "")}',
            f'- HEAD: {git.get("head", "")}', "", "```text",
            os.linesep.join(git_status) or "工作区干净", "```",
        ]),
    ])
    return "\n\n".join(sections).strip() + "\n"


def write_output(package: dict[str, Any], output: Path) -> None:
    output.parent.mkdir(parents=True, exist_ok=True)
    if output.suffix.lower() == ".json":
        content = json.dumps(package, ensure_ascii=False, indent=2) + "\n"
    else:
        content = render_markdown(package)
    output.write_text(content, encoding="utf-8")


def index_package(db_path: Path, package: dict[str, Any], output: Path) -> None:
    db_path.parent.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(db_path) as conn:
        conn.execute("""CREATE TABLE IF NOT EXISTS handoffs (
            id INTEGER PRIMARY KEY, created_at TEXT NOT NULL, output_path TEXT NOT NULL UNIQUE,
            source_path TEXT NOT NULL, objective TEXT NOT NULL, head TEXT, package_json TEXT NOT NULL
        )""")
        conn.execute(
            "INSERT OR REPLACE INTO handoffs(created_at, output_path, source_path, objective, head, package_json) VALUES(?,?,?,?,?,?)",
            (package["created_at"], str(output.resolve()), package["source"]["path"], package["handoff"]["objective"], package["git"].get("head", ""), json.dumps(package, ensure_ascii=False)),
        )


def self_test() -> None:
    with tempfile.TemporaryDirectory() as temp:
        root = Path(temp)
        (root / "main.py").write_text("print('ok')\n", encoding="utf-8")
        chat = root / "chat.json"
        chat.write_text(json.dumps({"messages": [
            {"role": "user", "content": "实现导入器，token=super-secret-value，检查 main.py:1"},
            {"role": "assistant", "content": "已完成 JSON 导入。TODO: 增加引用校验。\npython main.py"},
        ]}, ensure_ascii=False), encoding="utf-8")
        args = argparse.Namespace(input=str(chat), repo=str(root), model_url="", api_key_env="MODEL_API_KEY", model="test")
        package = build_package(args)
        assert not validate_package(package)
        assert package["privacy"]["redactions"] == 1
        assert package["file_references"][0]["exists"] is True
        assert "super-secret-value" not in json.dumps(package)
        md = render_markdown(package)
        assert "## 剩余事项" in md and "main.py:1 (有效)" in md
        assert md.startswith("# Model Handoff")
        md_output = root / "handoff.md"
        json_output = root / "handoff.json"
        db_output = root / "handoffs.db"
        write_output(package, md_output)
        write_output(package, json_output)
        index_package(db_output, package, md_output)
        assert not validate_package(json.loads(json_output.read_text(encoding="utf-8")))
        with sqlite3.connect(db_output) as conn:
            assert conn.execute("SELECT COUNT(*) FROM handoffs").fetchone()[0] == 1
    print("SELF_TEST_OK")


def make_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="生成跨模型项目交接包")
    sub = parser.add_subparsers(dest="command")
    build = sub.add_parser("build", help="从聊天生成交接包")
    build.add_argument("input", help="JSON 或纯文本聊天文件")
    build.add_argument("-r", "--repo", help="Git 仓库路径，默认当前目录")
    build.add_argument("-o", "--output", default="handoff.md", help="输出 .md 或 .json")
    build.add_argument("--db", help="可选 SQLite 索引路径")
    build.add_argument("--model-url", help="OpenAI Chat Completions 兼容端点")
    build.add_argument("--model", default="gpt-4o-mini", help="兼容端点的模型名")
    build.add_argument("--api-key-env", default="MODEL_API_KEY", help="读取 API key 的环境变量名")
    validate = sub.add_parser("validate", help="校验 JSON 交接包")
    validate.add_argument("package")
    sub.add_parser("self-test", help="运行内置样例测试")
    return parser


def print_usage() -> None:
    print("Model Handoff - 跨模型项目交接包")
    print("生成: python 40_mvp.py build chat.json -r ./repo -o handoff.md")
    print("接模型: 添加 --model-url URL --model MODEL（密钥默认读 MODEL_API_KEY）")
    print("校验: python 40_mvp.py validate handoff.json")
    print("测试: python 40_mvp.py self-test")


def main(argv: list[str] | None = None) -> int:
    parser = make_parser()
    args = parser.parse_args(argv)
    if not args.command:
        print_usage()
        return 0
    try:
        if args.command == "self-test":
            self_test()
            return 0
        if args.command == "validate":
            package = json.loads(Path(args.package).read_text(encoding="utf-8"))
            errors = validate_package(package)
            print("VALID" if not errors else "INVALID: " + "; ".join(errors))
            return 0 if not errors else 1
        package = build_package(args)
        output = Path(args.output).expanduser()
        write_output(package, output)
        if args.db:
            index_package(Path(args.db).expanduser(), package, output)
        print(f"已生成: {output.resolve()}")
        if package.get("generator_warning"):
            print(f'提示: {package["generator_warning"]}', file=sys.stderr)
        return 0
    except (OSError, ValueError, KeyError, json.JSONDecodeError, urllib.error.URLError) as exc:
        print(f"错误: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())