Code Review

{
  "version": "1.0",
  "generated_at": "",
  "review_target": {
    "type": "pr",
    "id": ""
  },
  "status": [
    {
      "comment_id": 0,
      "path": "",
      "line": 0,
      "side": "RIGHT",
      "body": "",
      "snippet": "",
      "snippet_fingerprint": "",
      "status_flow": "open",
      "status_tech": "not-fixed",
      "mapped_finding_id": "",
      "notes": ""
    }
  ]
}

{
  "version": "1.0",
  "repo": "",
  "review_target": {
    "type": "pr",
    "id": "",
    "base_ref": "",
    "head_ref": "",
    "base_sha": "",
    "head_sha": ""
  },
  "strategy": {
    "commit_map_threshold": 0.9,
    "hunk_match_threshold": 0.8,
    "fallback_on_low_confidence": true
  },
  "review_round": 1,
  "generated_at": ""
}

{
  "version": "1.0",
  "review_rounds": [],
  "commits": [
    {
      "commit_sha": "",
      "patch_id": "",
      "review_round": 1,
      "reviewed_at": "",
      "hunk_fingerprints": [],
      "files": [],
      "mapping": {
        "method": "direct|patch-id|hunk-similarity|none",
        "mapped_from_commit": "",
        "confidence": 0.0
      }
    }
  ]
}

#!/usr/bin/env python3
import argparse
import hashlib
import json
import re
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Set, Tuple


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


def sanitize_branch_name(branch_name: str) -> str:
    return branch_name.replace("/", "-")


def resolve_target(args: argparse.Namespace) -> Tuple[str, str]:
    if args.target_type and args.target_id:
        target_type = args.target_type
        target_id = args.target_id
    elif args.pr_number is not None:
        target_type = "pr"
        target_id = str(args.pr_number)
    elif args.branch_name:
        target_type = "branch"
        target_id = args.branch_name
    else:
        raise SystemExit("must provide either --target-type/--target-id or --pr-number or --branch-name")
    if target_type not in {"pr", "branch"}:
        raise SystemExit("target type must be pr or branch")
    return target_type, target_id


def run_git(repo_root: Path, args: List[str]) -> str:
    proc = subprocess.run(["git", *args], cwd=repo_root, text=True, capture_output=True, check=True)
    return proc.stdout


def run_git_no_check(repo_root: Path, args: List[str]) -> subprocess.CompletedProcess:
    return subprocess.run(["git", *args], cwd=repo_root, text=True, capture_output=True, check=False)


def normalize_file_content(text: str) -> str:
    lines = [re.sub(r"\s+", " ", line.strip()) for line in text.splitlines()]
    return "\n".join(lines)


def stable_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def get_changed_files(repo_root: Path, base_sha: str, head_sha: str) -> List[str]:
    out = run_git(repo_root, ["diff", "--name-only", f"{base_sha}..{head_sha}"])
    return sorted({line.strip() for line in out.splitlines() if line.strip()})


def get_file_content_at_commit(repo_root: Path, commit_sha: str, path: str) -> str:
    proc = run_git_no_check(repo_root, ["show", f"{commit_sha}:{path}"])
    if proc.returncode != 0:
        return ""
    return proc.stdout


def main() -> None:
    parser = argparse.ArgumentParser(description="Build snapshot baseline for incremental review.")
    parser.add_argument("--repo-root", required=True)
    parser.add_argument("--target-type", choices=["pr", "branch"])
    parser.add_argument("--target-id")
    parser.add_argument("--pr-number", type=int, help="PR number (legacy compatible)")
    parser.add_argument("--branch-name", help="Branch name (legacy compatible)")
    parser.add_argument("--base", required=True, help="Base commit SHA")
    parser.add_argument("--head", required=True, help="Head commit SHA")
    parser.add_argument("--review-round", required=True, type=int)
    args = parser.parse_args()

    repo_root = Path(args.repo_root).resolve()
    target_type, target_id_raw = resolve_target(args)
    target_id_dir = sanitize_branch_name(target_id_raw) if target_type == "branch" else target_id_raw
    review_dir = repo_root / "code-review" / f"{target_type}-{target_id_dir}"

    snapshot_root = review_dir / "snapshot" / f"round-{args.review_round}"
    files_root = snapshot_root / "files"
    files_root.mkdir(parents=True, exist_ok=True)

    changed_files = get_changed_files(repo_root, args.base, args.head)
    manifest_files: List[Dict[str, object]] = []

    for rel_path in changed_files:
        content = get_file_content_at_commit(repo_root, args.head, rel_path)
        if content == "":
            # Deleted file at head; keep entry for audit but no content snapshot.
            manifest_files.append(
                {"path": rel_path, "exists_in_head": False, "raw_hash": "", "normalized_hash": "", "size": 0}
            )
            continue

        out_path = files_root / rel_path
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(content, encoding="utf-8")
        manifest_files.append(
            {
                "path": rel_path,
                "exists_in_head": True,
                "raw_hash": stable_hash(content),
                "normalized_hash": stable_hash(normalize_file_content(content)),
                "size": len(content.encode("utf-8")),
            }
        )

    manifest = {
        "version": "1.0",
        "review_target": {"type": target_type, "id": target_id_raw},
        "review_round": args.review_round,
        "base_sha": args.base,
        "head_sha": args.head,
        "generated_at": utc_now(),
        "files": manifest_files,
    }
    manifest_path = snapshot_root / "manifest.json"
    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")

    latest = {
        "latest_round": args.review_round,
        "manifest": str(manifest_path.relative_to(review_dir)),
        "updated_at": utc_now(),
    }
    latest_path = review_dir / "snapshot" / "latest.json"
    latest_path.parent.mkdir(parents=True, exist_ok=True)
    latest_path.write_text(json.dumps(latest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")

    print(
        json.dumps(
            {
                "review_target": {"type": target_type, "id": target_id_raw},
                "review_round": args.review_round,
                "files": len(manifest_files),
                "manifest": str(manifest_path),
            },
            ensure_ascii=False,
        )
    )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
import argparse
import json
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Tuple


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


def run_cmd(args: List[str], cwd: Path) -> str:
    proc = subprocess.run(args, cwd=cwd, text=True, capture_output=True, check=False)
    if proc.returncode != 0:
        raise SystemExit(f"command failed: {' '.join(args)}\n{proc.stderr.strip()}")
    return proc.stdout


def sanitize_branch_name(branch_name: str) -> str:
    return branch_name.replace("/", "-")


def resolve_target(args: argparse.Namespace) -> Tuple[str, str]:
    if args.target_type and args.target_id:
        target_type = args.target_type
        target_id = args.target_id
    elif args.pr_number is not None:
        target_type = "pr"
        target_id = str(args.pr_number)
    elif args.branch_name:
        target_type = "branch"
        target_id = args.branch_name
    else:
        raise SystemExit("must provide either --target-type/--target-id or --pr-number or --branch-name")
    if target_type not in {"pr", "branch"}:
        raise SystemExit("target type must be pr or branch")
    return target_type, target_id


def parse_name_with_owner(repo_root: Path) -> Tuple[str, str]:
    out = run_cmd(["gh", "repo", "view", "--json", "nameWithOwner", "--jq", ".nameWithOwner"], repo_root).strip()
    if "/" not in out:
        raise SystemExit(f"invalid repository nameWithOwner: {out}")
    owner, name = out.split("/", 1)
    return owner, name


def get_viewer_login(repo_root: Path) -> str:
    out = run_cmd(["gh", "api", "user", "--jq", ".login"], repo_root).strip()
    return out


def run_graphql(repo_root: Path, owner: str, name: str, pr_number: int, cursor: str) -> Dict[str, Any]:
    # Query review threads instead of plain review comments so we can
    # persist thread-level resolution state deterministically.
    query = """
query($owner:String!, $name:String!, $number:Int!, $endCursor:String) {
  repository(owner:$owner, name:$name) {
    pullRequest(number:$number) {
      reviewThreads(first:100, after:$endCursor) {
        pageInfo { hasNextPage endCursor }
        nodes {
          isResolved
          comments(first:100) {
            nodes {
              databaseId
              body
              path
              line
              originalLine
              createdAt
              updatedAt
              author { login }
              originalCommit { oid }
              replyTo { databaseId }
            }
          }
        }
      }
    }
  }
}
"""
    cmd = [
        "gh",
        "api",
        "graphql",
        "-f",
        f"query={query}",
        "-F",
        f"owner={owner}",
        "-F",
        f"name={name}",
        "-F",
        f"number={pr_number}",
    ]
    if cursor:
        cmd.extend(["-F", f"endCursor={cursor}"])
    out = run_cmd(cmd, repo_root)
    return json.loads(out)


def main() -> None:
    parser = argparse.ArgumentParser(description="Fetch PR review comments to stable schema file.")
    parser.add_argument("--repo-root", required=True)
    parser.add_argument("--target-type", choices=["pr", "branch"])
    parser.add_argument("--target-id")
    parser.add_argument("--pr-number", type=int, help="PR number (legacy compatible)")
    parser.add_argument("--branch-name", help="Branch name (legacy compatible)")
    args = parser.parse_args()

    repo_root = Path(args.repo_root).resolve()
    target_type, target_id_raw = resolve_target(args)
    target_id_dir = sanitize_branch_name(target_id_raw) if target_type == "branch" else target_id_raw
    review_dir = repo_root / "code-review" / f"{target_type}-{target_id_dir}"
    comments_path = review_dir / "comments" / "review-comments.json"
    comments_path.parent.mkdir(parents=True, exist_ok=True)

    if target_type != "pr":
        payload = {
            "version": "1.0",
            "source": "branch_review_comments",
            "fetched_at": utc_now(),
            "review_target": {"type": target_type, "id": target_id_raw},
            "comments": [],
        }
        comments_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        print(json.dumps({"target": f"{target_type}-{target_id_raw}", "threads": 0, "comments": 0, "resolved_threads": 0}))
        return

    owner, name = parse_name_with_owner(repo_root)
    viewer_login = get_viewer_login(repo_root)
    pr_number = int(target_id_raw)

    cursor = ""
    has_next = True
    comments: List[Dict[str, Any]] = []
    total_threads = 0
    resolved_threads = 0

    while has_next:
        # Paginate until all review threads are collected.
        data = run_graphql(repo_root, owner, name, pr_number, cursor)
        threads_obj = data["data"]["repository"]["pullRequest"]["reviewThreads"]
        page_info = threads_obj["pageInfo"]
        threads = threads_obj["nodes"] or []
        total_threads += len(threads)
        for thread in threads:
            is_resolved = bool(thread.get("isResolved", False))
            if is_resolved:
                resolved_threads += 1
            thread_comments = thread.get("comments", {}).get("nodes", []) or []
            for c in thread_comments:
                author = (c.get("author") or {}).get("login", "")
                original_commit = (c.get("originalCommit") or {}).get("oid", "")
                reply_to = (c.get("replyTo") or {}).get("databaseId")
                # Use originalLine as a stable anchor because line can be null
                # after code evolves on newer commits.
                original_line = c.get("originalLine")
                line = original_line if isinstance(original_line, int) else 0
                comments.append(
                    {
                        "comment_id": c.get("databaseId"),
                        "author": author,
                        "created_at": c.get("createdAt", ""),
                        "updated_at": c.get("updatedAt", ""),
                        "path": c.get("path", ""),
                        "line": line,
                        "side": "RIGHT",
                        "commit_id": original_commit,
                        "in_reply_to_id": reply_to,
                        "body": c.get("body", ""),
                        "thread_resolved": is_resolved,
                    }
                )
        has_next = bool(page_info.get("hasNextPage"))
        cursor = page_info.get("endCursor") if has_next else ""

    payload = {
        "version": "1.0",
        "source": "github_pr_review_comments",
        "fetched_at": utc_now(),
        "review_target": {"type": "pr", "id": target_id_raw},
        "viewer_login": viewer_login,
        "comments": comments,
    }
    comments_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(
        json.dumps(
            {
                "target": f"pr-{target_id_raw}",
                "threads": total_threads,
                "comments": len(comments),
                "resolved_threads": resolved_threads,
            },
            ensure_ascii=False,
        )
    )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
import argparse
import json
from pathlib import Path
from typing import Dict, List, Tuple


def sanitize_branch_name(branch_name: str) -> str:
    return branch_name.replace("/", "-")


def resolve_target(args: argparse.Namespace) -> Tuple[str, str]:
    if args.target_type and args.target_id:
        target_type = args.target_type
        target_id = args.target_id
    elif args.pr_number is not None:
        target_type = "pr"
        target_id = str(args.pr_number)
    elif args.branch_name:
        target_type = "branch"
        target_id = args.branch_name
    else:
        raise SystemExit("must provide either --target-type/--target-id or --pr-number or --branch-name")
    if target_type not in {"pr", "branch"}:
        raise SystemExit("target type must be pr or branch")
    return target_type, target_id


def read_json(path: Path) -> Dict:
    if not path.exists():
        raise SystemExit(f"missing file: {path}")
    return json.loads(path.read_text(encoding="utf-8"))


def esc_cell(text: str) -> str:
    return (text or "").replace("\n", " ").replace("|", "\\|").replace("`", "").strip()


def build_comment_meta_map(review_comments_payload: Dict) -> Dict[int, Dict]:
    comments = review_comments_payload.get("comments", [])
    meta_map: Dict[int, Dict] = {}
    if not isinstance(comments, list):
        return meta_map
    for c in comments:
        cid = c.get("comment_id")
        if isinstance(cid, int):
            meta_map[cid] = c
    return meta_map


def build_markdown(target_type: str, target_id: str, items: List[Dict], comment_meta: Dict[int, Dict]) -> str:
    lines = []
    lines.append(f"# Comment Status Report ({target_type}-{target_id})")
    lines.append("")
    lines.append(f"- Total: {len(items)}")
    lines.append("")
    lines.append("| 评论时间 | File | Line | 作者 | Comment | Flow | Tech |")
    lines.append("|---|---|---:|---|---|---|---|")
    for item in items:
        cid = item.get("comment_id", "")
        meta = comment_meta.get(cid, {})
        created_at = esc_cell(str(meta.get("created_at", "")))
        author = esc_cell(str(meta.get("author", "")))
        path = esc_cell(str(item.get("path", "")))
        line = item.get("line", 0)
        body = esc_cell(str(item.get("body", "")))
        if len(body) > 160:
            body = body[:157] + "..."
        status_flow = esc_cell(str(item.get("status_flow", "")))
        status_tech = esc_cell(str(item.get("status_tech", "")))
        lines.append(
            f"| {created_at} | `{path}` | {line} | {author} | {body} | {status_flow} | {status_tech} |"
        )
    lines.append("")
    return "\n".join(lines)


def main() -> None:
    parser = argparse.ArgumentParser(description="Generate markdown report from comment-status.json.")
    parser.add_argument("--repo-root", required=True)
    parser.add_argument("--target-type", choices=["pr", "branch"])
    parser.add_argument("--target-id")
    parser.add_argument("--pr-number", type=int, help="PR number (legacy compatible)")
    parser.add_argument("--branch-name", help="Branch name (legacy compatible)")
    args = parser.parse_args()

    repo_root = Path(args.repo_root).resolve()
    target_type, target_id_raw = resolve_target(args)
    target_id_dir = sanitize_branch_name(target_id_raw) if target_type == "branch" else target_id_raw
    review_dir = repo_root / "code-review" / f"{target_type}-{target_id_dir}"

    status_path = review_dir / "comments" / "comment-status.json"
    review_comments_path = review_dir / "comments" / "review-comments.json"
    report_path = review_dir / "comments" / "comment-status.md"

    payload = read_json(status_path)
    if not isinstance(payload, dict) or not isinstance(payload.get("status"), list):
        raise SystemExit("invalid comment-status.json: root must be object and `status` must be list")
    review_comments_payload = read_json(review_comments_path)
    if not isinstance(review_comments_payload, dict):
        raise SystemExit("invalid review-comments.json: root must be object")

    comment_meta = build_comment_meta_map(review_comments_payload)
    markdown = build_markdown(target_type, target_id_raw, payload["status"], comment_meta)
    report_path.write_text(markdown + "\n", encoding="utf-8")
    print(str(report_path))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
import argparse
import hashlib
import json
import re
import subprocess
from pathlib import Path
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Dict, List, Optional, Set, Tuple


def run_git(repo_root: Path, args: List[str]) -> str:
    result = subprocess.run(
        ["git", *args],
        cwd=repo_root,
        text=True,
        capture_output=True,
        check=True,
    )
    return result.stdout


def run_git_no_check(repo_root: Path, args: List[str]) -> subprocess.CompletedProcess:
    return subprocess.run(
        ["git", *args],
        cwd=repo_root,
        text=True,
        capture_output=True,
        check=False,
    )


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


def normalize_code_line(line: str) -> str:
    return re.sub(r"\s+", " ", line.strip())


def stable_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def normalize_file_content(text: str) -> str:
    # Keep line boundaries but normalize whitespace noise for robust matching.
    lines = [re.sub(r"\s+", " ", line.strip()) for line in text.splitlines()]
    return "\n".join(lines)


def compute_patch_id(repo_root: Path, commit_sha: str) -> str:
    patch_text = run_git(repo_root, ["show", "--pretty=format:", "--no-color", commit_sha])
    proc = subprocess.run(
        ["git", "patch-id", "--stable"],
        cwd=repo_root,
        text=True,
        input=patch_text,
        capture_output=True,
        check=True,
    )
    output = proc.stdout.strip()
    return output.split()[0] if output else ""


def get_commit_files(repo_root: Path, commit_sha: str) -> List[str]:
    out = run_git(repo_root, ["show", "--pretty=format:", "--name-only", "--no-color", commit_sha])
    return sorted({line.strip() for line in out.splitlines() if line.strip()})


def get_file_content_at_commit(repo_root: Path, commit_sha: str, path: str) -> Optional[str]:
    proc = run_git_no_check(repo_root, ["show", f"{commit_sha}:{path}"])
    if proc.returncode != 0:
        return None
    return proc.stdout


def load_latest_snapshot_map(review_dir: Path) -> Dict[str, str]:
    latest_path = review_dir / "snapshot" / "latest.json"
    if not latest_path.exists():
        return {}
    try:
        latest = json.loads(latest_path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    manifest_rel = latest.get("manifest")
    if not isinstance(manifest_rel, str) or not manifest_rel:
        return {}
    manifest_path = review_dir / manifest_rel
    if not manifest_path.exists():
        return {}
    try:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    mapping: Dict[str, str] = {}
    for item in manifest.get("files", []):
        path = item.get("path")
        n_hash = item.get("normalized_hash")
        if isinstance(path, str) and isinstance(n_hash, str):
            mapping[path] = n_hash
    return mapping


def compute_snapshot_match_rate(
    repo_root: Path, head_sha: str, changed_files: Set[str], snapshot_map: Dict[str, str]
) -> Optional[float]:
    if not snapshot_map or not changed_files:
        return None
    overlap = [p for p in changed_files if p in snapshot_map]
    if not overlap:
        return None
    matched = 0
    for path in overlap:
        content = get_file_content_at_commit(repo_root, head_sha, path)
        if content is None:
            continue
        current_hash = stable_hash(normalize_file_content(content))
        if current_hash == snapshot_map[path]:
            matched += 1
    return matched / len(overlap)


def parse_hunk_fingerprints(repo_root: Path, commit_sha: str) -> List[str]:
    patch = run_git(repo_root, ["show", "--pretty=format:", "--no-color", "-U3", commit_sha])
    lines = patch.splitlines()
    file_path = ""
    hunk_header = ""
    hunk_lines: List[str] = []
    fps: List[str] = []

    def flush() -> None:
        nonlocal hunk_lines, hunk_header
        if not hunk_lines:
            return
        key = file_path + "\n" + hunk_header + "\n" + "\n".join(hunk_lines)
        fps.append(stable_hash(key))
        hunk_lines = []
        hunk_header = ""

    for line in lines:
        if line.startswith("diff --git "):
            flush()
            m = re.search(r" b/(.+)$", line)
            file_path = m.group(1) if m else ""
            continue
        if line.startswith("@@"):
            flush()
            hunk_header = line
            continue
        if line.startswith("+") or line.startswith("-"):
            if line.startswith("+++") or line.startswith("---"):
                continue
            hunk_lines.append(normalize_code_line(line[1:]))

    flush()
    return sorted(set(fps))


def jaccard(a: Set[str], b: Set[str]) -> float:
    if not a and not b:
        return 1.0
    if not a or not b:
        return 0.0
    return len(a & b) / len(a | b)


@dataclass
class CommitRecord:
    commit_sha: str
    patch_id: str
    hunk_fingerprints: List[str]
    review_round: int
    reviewed_at: str
    mapping: Dict[str, object]


def load_json(path: Path) -> Dict:
    if not path.exists():
        return {}
    return json.loads(path.read_text(encoding="utf-8"))


def save_json(path: Path, payload: Dict) -> None:
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def sanitize_branch_name(branch_name: str) -> str:
    return branch_name.replace("/", "-")


def resolve_target(args: argparse.Namespace) -> Tuple[str, str]:
    if args.target_type and args.target_id:
        target_type = args.target_type
        target_id = args.target_id
    elif args.pr_number is not None:
        target_type = "pr"
        target_id = str(args.pr_number)
    elif args.branch_name:
        target_type = "branch"
        target_id = args.branch_name
    else:
        raise SystemExit("must provide either --target-type/--target-id or --pr-number or --branch-name")
    if target_type not in {"pr", "branch"}:
        raise SystemExit("target type must be pr or branch")
    return target_type, target_id


def ensure_commit_exists(repo_root: Path, sha: str, target_type: str, target_id: str) -> bool:
    exists = run_git_no_check(repo_root, ["cat-file", "-e", f"{sha}^{{commit}}"])
    if exists.returncode == 0:
        return True

    # First generic fetch to cover normal branch updates.
    run_git_no_check(repo_root, ["fetch", "--all", "--prune", "--tags"])
    exists = run_git_no_check(repo_root, ["cat-file", "-e", f"{sha}^{{commit}}"])
    if exists.returncode == 0:
        return True

    # Then PR-specific fetch for detached PR heads.
    if target_type == "pr":
        run_git_no_check(repo_root, ["fetch", "origin", f"pull/{target_id}/head"])
        exists = run_git_no_check(repo_root, ["cat-file", "-e", f"{sha}^{{commit}}"])
        if exists.returncode == 0:
            return True
    return False


def main() -> None:
    parser = argparse.ArgumentParser(description="Map reviewed commits for incremental PR/branch review.")
    parser.add_argument("--repo-root", required=True)
    parser.add_argument("--target-type", choices=["pr", "branch"])
    parser.add_argument("--target-id")
    parser.add_argument("--pr-number", type=int, help="PR number (legacy compatible)")
    parser.add_argument("--branch-name", help="Branch name (legacy compatible)")
    parser.add_argument("--base", required=True, help="Base commit SHA for comparison")
    parser.add_argument("--head", required=True, help="Head commit SHA for comparison")
    parser.add_argument("--review-round", required=True, type=int)
    parser.add_argument("--commit-map-threshold", type=float, default=0.9)
    parser.add_argument("--hunk-match-threshold", type=float, default=0.8)
    parser.add_argument("--snapshot-match-threshold", type=float, default=0.9)
    args = parser.parse_args()

    repo_root = Path(args.repo_root).resolve()
    target_type, target_id_raw = resolve_target(args)
    target_id_dir = sanitize_branch_name(target_id_raw) if target_type == "branch" else target_id_raw
    review_dir = repo_root / "code-review" / f"{target_type}-{target_id_dir}"
    reviewed_path = review_dir / "reviewed_commits.json"
    reviewed = load_json(reviewed_path) or {"version": "1.0", "review_rounds": [], "commits": []}
    old_commits = reviewed.get("commits", [])
    snapshot_map = load_latest_snapshot_map(review_dir)

    if not ensure_commit_exists(repo_root, args.base, target_type, target_id_raw):
        raise SystemExit(
            f"missing base commit object: {args.base}. "
            "Please fetch the base branch history, then retry."
        )
    if not ensure_commit_exists(repo_root, args.head, target_type, target_id_raw):
        raise SystemExit(
            f"missing head commit object: {args.head}. "
            "For PR review, try: git fetch origin pull/<pr-number>/head"
        )

    try:
        rev_list_output = run_git(repo_root, ["rev-list", "--reverse", f"{args.base}..{args.head}"])
    except subprocess.CalledProcessError as e:
        stderr = (e.stderr or "").strip()
        raise SystemExit(
            f"failed to build commit range {args.base}..{args.head}: {stderr or 'unknown git error'}"
        )

    current_commits = [sha for sha in rev_list_output.splitlines() if sha]
    current_set = set(current_commits)
    commit_files_map: Dict[str, List[str]] = {}
    current_changed_files: Set[str] = set()
    for sha in current_commits:
        files = get_commit_files(repo_root, sha)
        commit_files_map[sha] = files
        current_changed_files.update(files)

    old_by_sha = {c.get("commit_sha"): c for c in old_commits if c.get("commit_sha")}
    old_by_patch_id: Dict[str, Dict] = {}
    for c in old_commits:
        pid = c.get("patch_id")
        if pid and pid not in old_by_patch_id:
            old_by_patch_id[pid] = c

    mapped: Dict[str, CommitRecord] = {}
    unchanged_by_sha = 0
    for sha in current_commits:
        if sha in old_by_sha:
            oc = old_by_sha[sha]
            mapped[sha] = CommitRecord(
                commit_sha=sha,
                patch_id=oc.get("patch_id", ""),
                hunk_fingerprints=oc.get("hunk_fingerprints", []),
                review_round=oc.get("review_round", args.review_round),
                reviewed_at=oc.get("reviewed_at", utc_now()),
                mapping={"method": "direct", "mapped_from_commit": sha, "confidence": 1.0},
            )
            unchanged_by_sha += 1

    for sha in current_commits:
        if sha in mapped:
            continue
        pid = compute_patch_id(repo_root, sha)
        if pid and pid in old_by_patch_id:
            oc = old_by_patch_id[pid]
            mapped[sha] = CommitRecord(
                commit_sha=sha,
                patch_id=pid,
                hunk_fingerprints=oc.get("hunk_fingerprints", []),
                review_round=oc.get("review_round", args.review_round),
                reviewed_at=oc.get("reviewed_at", utc_now()),
                mapping={"method": "patch-id", "mapped_from_commit": oc.get("commit_sha", ""), "confidence": 0.98},
            )

    old_unmapped = [c for c in old_commits if c.get("commit_sha") not in current_set]
    old_hunk_sets = {
        c.get("commit_sha", ""): set(c.get("hunk_fingerprints", [])) for c in old_unmapped if c.get("commit_sha")
    }

    for sha in current_commits:
        if sha in mapped:
            continue
        new_hunks = set(parse_hunk_fingerprints(repo_root, sha))
        best_score = 0.0
        best_old = ""
        for old_sha, old_hunks in old_hunk_sets.items():
            score = jaccard(new_hunks, old_hunks)
            if score > best_score:
                best_score = score
                best_old = old_sha
        if best_old and best_score >= args.hunk_match_threshold:
            mapped[sha] = CommitRecord(
                commit_sha=sha,
                patch_id=compute_patch_id(repo_root, sha),
                hunk_fingerprints=sorted(new_hunks),
                review_round=args.review_round,
                reviewed_at=utc_now(),
                mapping={"method": "hunk-similarity", "mapped_from_commit": best_old, "confidence": round(best_score, 4)},
            )

    need_review: List[str] = [sha for sha in current_commits if sha not in mapped]

    # commit_map_rate measures "how many OLD commits are accounted for in the
    # new commit set", NOT "what fraction of current commits are mapped".
    # Denominator = old commit count (the baseline we reviewed before).
    # This way appending new commits doesn't penalise the rate, while rebase
    # that loses old commits correctly lowers it.
    old_commits_covered: Set[str] = set()
    for rec in mapped.values():
        from_sha = rec.mapping.get("mapped_from_commit", "")
        if from_sha:
            old_commits_covered.add(from_sha)
    old_commit_count = len(old_commits)
    commit_map_rate = (len(old_commits_covered) / old_commit_count) if old_commit_count > 0 else 1.0

    if need_review:
        hunk_scores: List[float] = []
        for sha in need_review:
            new_hunks = set(parse_hunk_fingerprints(repo_root, sha))
            best = 0.0
            for old_hunks in old_hunk_sets.values():
                best = max(best, jaccard(new_hunks, old_hunks))
            hunk_scores.append(best)
        hunk_match_rate = (sum(hunk_scores) / len(hunk_scores)) if hunk_scores else 1.0
    else:
        hunk_match_rate = 1.0

    if commit_map_rate >= args.commit_map_threshold:
        recommendation = "incremental"
    elif hunk_match_rate >= args.hunk_match_threshold:
        recommendation = "partial"
    else:
        recommendation = "full"

    snapshot_match_rate = compute_snapshot_match_rate(repo_root, args.head, current_changed_files, snapshot_map)
    if recommendation == "full" and snapshot_match_rate is not None and snapshot_match_rate >= args.snapshot_match_threshold:
        # For squash/rebase-conflict scenarios, snapshot evidence can safely
        # downgrade from full to partial.
        recommendation = "partial"

    round_record = {
        "review_round": args.review_round,
        "generated_at": utc_now(),
        "base": args.base,
        "head": args.head,
        "stats": {
            "total_commits": len(current_commits),
            "mapped_commits": len(mapped),
            "direct_sha_hits": unchanged_by_sha,
            "commit_map_rate": round(commit_map_rate, 4),
            "hunk_match_rate": round(hunk_match_rate, 4),
            "snapshot_match_rate": round(snapshot_match_rate, 4) if snapshot_match_rate is not None else None,
            "recommendation": recommendation,
        },
        "need_review_commits": need_review,
    }

    merged_commits = [c for c in old_commits if c.get("commit_sha") not in current_set]
    for sha in current_commits:
        if sha in mapped:
            c = mapped[sha]
            merged_commits.append(
                {
                    "commit_sha": c.commit_sha,
                    "patch_id": c.patch_id,
                    "review_round": c.review_round,
                    "reviewed_at": c.reviewed_at,
                    "hunk_fingerprints": c.hunk_fingerprints,
                    "files": commit_files_map.get(sha, []),
                    "mapping": c.mapping,
                }
            )
        else:
            merged_commits.append(
                {
                    "commit_sha": sha,
                    "patch_id": compute_patch_id(repo_root, sha),
                    "review_round": args.review_round,
                    "reviewed_at": "",
                    "hunk_fingerprints": parse_hunk_fingerprints(repo_root, sha),
                    "files": commit_files_map.get(sha, []),
                    "mapping": {"method": "none", "mapped_from_commit": "", "confidence": 0.0},
                }
            )

    reviewed["commits"] = merged_commits
    reviewed.setdefault("review_rounds", []).append(round_record)
    save_json(reviewed_path, reviewed)

    print(
        json.dumps(
            {
                "review_target": {"type": target_type, "id": target_id_raw},
                "total_commits": len(current_commits),
                "need_review_commits": need_review,
                "commit_map_rate": round(commit_map_rate, 4),
                "hunk_match_rate": round(hunk_match_rate, 4),
                "snapshot_match_rate": round(snapshot_match_rate, 4) if snapshot_match_rate is not None else None,
                "recommendation": recommendation,
            },
            ensure_ascii=False,
        )
    )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
import argparse
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Tuple


SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_DIR = SCRIPT_DIR.parent
REF_DIR = SKILL_DIR / "references"


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


def read_json(path: Path) -> Dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def write_json(path: Path, data: Dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def ensure_file_from_template(target: Path, template: Path, mutate=None) -> None:
    if target.exists():
        return
    payload = read_json(template)
    if mutate:
        mutate(payload)
    write_json(target, payload)


def sanitize_branch_name(branch_name: str) -> str:
    return branch_name.replace("/", "-")


def resolve_target(args: argparse.Namespace) -> Tuple[str, str]:
    if args.target_type and args.target_id:
        target_type = args.target_type
        target_id = args.target_id
    elif args.pr_number is not None:
        target_type = "pr"
        target_id = str(args.pr_number)
    elif args.branch_name:
        target_type = "branch"
        target_id = args.branch_name
    else:
        raise SystemExit("must provide either --target-type/--target-id or --pr-number or --branch-name")
    if target_type not in {"pr", "branch"}:
        raise SystemExit("target type must be pr or branch")
    return target_type, target_id


def main() -> None:
    parser = argparse.ArgumentParser(description="Initialize code-review workspace for PR or branch.")
    parser.add_argument("--repo-root", required=True, help="Repository root path")
    parser.add_argument("--target-type", choices=["pr", "branch"], help="Review target type")
    parser.add_argument("--target-id", help="Review target id (PR number or branch name)")
    parser.add_argument("--pr-number", type=int, help="PR number (legacy compatible)")
    parser.add_argument("--branch-name", help="Branch name (legacy compatible)")
    parser.add_argument("--base-ref", default="", help="PR base ref")
    parser.add_argument("--head-ref", default="", help="PR head ref")
    parser.add_argument("--base-sha", default="", help="PR base sha")
    parser.add_argument("--head-sha", default="", help="PR head sha")
    args = parser.parse_args()

    repo_root = Path(args.repo_root).resolve()
    target_type, target_id_raw = resolve_target(args)
    target_id_dir = sanitize_branch_name(target_id_raw) if target_type == "branch" else target_id_raw
    review_dir = repo_root / "code-review" / f"{target_type}-{target_id_dir}"
    comments_dir = review_dir / "comments"

    comments_dir.mkdir(parents=True, exist_ok=True)

    meta_path = review_dir / "meta.json"
    reviewed_commits_path = review_dir / "reviewed_commits.json"
    review_comments_path = comments_dir / "review-comments.json"
    comment_status_path = comments_dir / "comment-status.json"

    def mutate_meta(payload: Dict[str, Any]) -> None:
        payload["repo"] = str(repo_root)
        payload["review_target"]["type"] = target_type
        payload["review_target"]["id"] = target_id_raw
        payload["review_target"]["base_ref"] = args.base_ref
        payload["review_target"]["head_ref"] = args.head_ref
        payload["review_target"]["base_sha"] = args.base_sha
        payload["review_target"]["head_sha"] = args.head_sha
        payload["generated_at"] = utc_now()

    def create_review_comments_payload() -> Dict[str, Any]:
        return {
            "version": "1.0",
            "source": "github_pr_review_comments" if target_type == "pr" else "branch_review_comments",
            "fetched_at": utc_now(),
            "review_target": {"type": target_type, "id": target_id_raw},
            "viewer_login": "",
            "comments": [],
        }

    def mutate_comment_status(payload: Dict[str, Any]) -> None:
        payload["review_target"]["type"] = target_type
        payload["review_target"]["id"] = target_id_raw
        payload["generated_at"] = utc_now()
        payload["status"] = []

    def mutate_reviewed_commits(payload: Dict[str, Any]) -> None:
        payload["review_rounds"] = []
        payload["commits"] = []

    ensure_file_from_template(meta_path, REF_DIR / "meta.template.json", mutate_meta)
    ensure_file_from_template(
        reviewed_commits_path, REF_DIR / "reviewed_commits.template.json", mutate_reviewed_commits
    )
    if not review_comments_path.exists():
        write_json(review_comments_path, create_review_comments_payload())
    ensure_file_from_template(
        comment_status_path, REF_DIR / "comment-status.template.json", mutate_comment_status
    )

    print(str(review_dir))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
import argparse
import hashlib
import json
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Tuple


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


def stable_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def normalize_text(text: str) -> str:
    text = re.sub(r"\s+", " ", text.strip())
    return text


def infer_flow_status_from_comment(comment: Dict) -> str:
    # Single deterministic rule from upstream schema.
    return "resolved" if comment.get("thread_resolved") is True else "open"


def read_json(path: Path) -> Dict:
    if not path.exists():
        return {}
    return json.loads(path.read_text(encoding="utf-8"))


def write_json(path: Path, payload: Dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def default_status(comment: Dict) -> Dict:
    body = comment.get("body", "")
    snippet = normalize_text(body)[:300]
    fingerprint_seed = "|".join(
        [
            str(comment.get("path", "")),
            str(comment.get("line", 0)),
            str(comment.get("side", "RIGHT")),
            snippet,
        ]
    )
    return {
        "comment_id": comment.get("comment_id"),
        "path": comment.get("path", ""),
        "line": comment.get("line", 0),
        "side": comment.get("side", "RIGHT"),
        "body": body,
        "snippet": snippet,
        "snippet_fingerprint": stable_hash(fingerprint_seed),
        "status_flow": infer_flow_status_from_comment(comment),
        # status_tech is owned by model review in later phase.
        "status_tech": "not-fixed",
        "mapped_finding_id": "",
        "notes": "",
    }


def sanitize_branch_name(branch_name: str) -> str:
    return branch_name.replace("/", "-")


def resolve_target(args: argparse.Namespace) -> Tuple[str, str]:
    if args.target_type and args.target_id:
        target_type = args.target_type
        target_id = args.target_id
    elif args.pr_number is not None:
        target_type = "pr"
        target_id = str(args.pr_number)
    elif args.branch_name:
        target_type = "branch"
        target_id = args.branch_name
    else:
        raise SystemExit("must provide either --target-type/--target-id or --pr-number or --branch-name")
    if target_type not in {"pr", "branch"}:
        raise SystemExit("target type must be pr or branch")
    return target_type, target_id


def validate_payload(raw: Dict) -> List[Dict]:
    if not isinstance(raw, dict):
        raise SystemExit("invalid review-comments.json: root must be object")
    comments = raw.get("comments")
    if not isinstance(comments, list):
        raise SystemExit("invalid review-comments.json: `comments` must be list")
    return comments


def validate_comment(comment: Dict) -> None:
    required = ["comment_id", "path", "line", "side", "body", "thread_resolved"]
    missing = [k for k in required if k not in comment]
    if missing:
        raise SystemExit(
            "invalid review comment record: missing required fields "
            + ",".join(missing)
        )


def infer_manual_tech_override(replies: List[Dict], viewer_login: str) -> str:
    if not viewer_login:
        return ""
    # Prefer the latest explicit override from current reviewer account.
    for reply in reversed(replies):
        if str(reply.get("author", "")).lower() != viewer_login.lower():
            continue
        text = normalize_text(str(reply.get("body", ""))).lower()
        if "false-positive" in text or "false positive" in text or "假阳性" in text or "误判" in text:
            return "false-positive"
        if re.search(r"\bfixed\b", text) or "已修复" in text:
            return "fixed"
    return ""


def main() -> None:
    parser = argparse.ArgumentParser(description="Build comment-status.json from review comments.")
    parser.add_argument("--repo-root", required=True)
    parser.add_argument("--target-type", choices=["pr", "branch"])
    parser.add_argument("--target-id")
    parser.add_argument("--pr-number", type=int, help="PR number (legacy compatible)")
    parser.add_argument("--branch-name", help="Branch name (legacy compatible)")
    args = parser.parse_args()

    repo_root = Path(args.repo_root).resolve()
    target_type, target_id_raw = resolve_target(args)
    target_id_dir = sanitize_branch_name(target_id_raw) if target_type == "branch" else target_id_raw
    review_dir = repo_root / "code-review" / f"{target_type}-{target_id_dir}"
    comments_path = review_dir / "comments" / "review-comments.json"
    status_path = review_dir / "comments" / "comment-status.json"

    comments_payload = read_json(comments_path)
    if not comments_payload:
        raise SystemExit(f"missing comments file: {comments_path}")
    comments = validate_payload(comments_payload)
    viewer_login = str(comments_payload.get("viewer_login", "")).strip()

    replies_by_parent: Dict[int, List[Dict]] = {}
    root_comments: List[Dict] = []
    for c in comments:
        parent = c.get("in_reply_to_id")
        if parent is None:
            root_comments.append(c)
        else:
            replies_by_parent.setdefault(parent, []).append(c)

    previous = read_json(status_path)
    previous_map = {item.get("comment_id"): item for item in previous.get("status", [])}

    status: List[Dict] = []
    seen_fp = set()
    for comment in root_comments:
        validate_comment(comment)
        cid = comment.get("comment_id")
        if cid in previous_map:
            item = previous_map[cid]
            # Preserve manual/model edits on status_tech/notes, always sync flow status from source.
            item["status_flow"] = infer_flow_status_from_comment(comment)
        else:
            item = default_status(comment)

        fp = item.get("snippet_fingerprint", "")
        if fp and fp in seen_fp:
            item["notes"] = (item.get("notes", "") + " duplicate-fingerprint").strip()
        manual_override = infer_manual_tech_override(replies_by_parent.get(cid, []), viewer_login)
        if manual_override:
            item["status_tech"] = manual_override
            item["notes"] = (item.get("notes", "") + f" manual-tech-override:{manual_override}").strip()
        seen_fp.add(fp)
        status.append(item)

    payload = {
        "version": "1.0",
        "generated_at": utc_now(),
        "review_target": {"type": target_type, "id": target_id_raw},
        "status": status,
    }
    write_json(status_path, payload)

    print(
        json.dumps(
            {"review_target": {"type": target_type, "id": target_id_raw}, "status_count": len(status)},
            ensure_ascii=False,
        )
    )


if __name__ == "__main__":
    main()