claude-skills/markdown-html/skills/md-review/scripts/diff_parser.py

#!/usr/bin/env python3
"""diff_parser.py - Extract unified-diff hunks from markdown code review notes.

Stdlib-only. Scans markdown for ```diff fenced code blocks (the convention used
in PR writeups), parses each one as a unified diff, and returns structured
hunk data the renderer can lay out as a 2-column annotated diff.

Pattern (the standard unified-diff shape):

    ```diff
    --- a/path/to/file.py
    +++ b/path/to/file.py
    @@ -10,7 +10,8 @@ def existing_context
     context line (unchanged)
    -removed line
    +added line
    +another added line
     context line
    @@ -50,3 +51,4 @@
     ...
    ```

Also accepts:
  - Multiple ```diff blocks (each treated as a separate "section")
  - Inline-fenced diffs (no language tag) IF --infer-diff is passed
  - Per-hunk @@ header context strings (preserved in output)

NO LLM CALLS. Pure regex + state machine.

Usage:
    python diff_parser.py --input review.md --output hunks.json
    python diff_parser.py --sample
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any

FENCE_OPEN_RE = re.compile(r"^```(diff)?\s*$")
FENCE_CLOSE_RE = re.compile(r"^```\s*$")
FILE_OLD_RE = re.compile(r"^---\s+(?:a/)?(.+?)\s*$")
FILE_NEW_RE = re.compile(r"^\+\+\+\s+(?:b/)?(.+?)\s*$")
HUNK_RE = re.compile(
    r"^@@\s+-(\d+)(?:,(\d+))?\s+\+(\d+)(?:,(\d+))?\s+@@\s*(.*)$"
)


def _parse_hunk_body(lines: list[str], old_start: int, new_start: int) -> list[dict[str, Any]]:
    """Walk hunk body lines and assign source-side and target-side line numbers."""
    body: list[dict[str, Any]] = []
    old_num = old_start
    new_num = new_start
    for ln in lines:
        if not ln:
            # treat as context (truly empty line in diff)
            body.append({"kind": "context", "text": "", "old": old_num, "new": new_num})
            old_num += 1
            new_num += 1
            continue
        prefix = ln[0]
        text = ln[1:] if len(ln) > 1 else ""
        if prefix == "+":
            body.append({"kind": "addition", "text": text, "old": None, "new": new_num})
            new_num += 1
        elif prefix == "-":
            body.append({"kind": "deletion", "text": text, "old": old_num, "new": None})
            old_num += 1
        elif prefix == " ":
            body.append({"kind": "context", "text": text, "old": old_num, "new": new_num})
            old_num += 1
            new_num += 1
        elif prefix == "\\":
            body.append({"kind": "meta", "text": text.strip(), "old": None, "new": None})
        else:
            # Unknown line — preserve as context with the raw prefix
            body.append({"kind": "context", "text": ln, "old": old_num, "new": new_num})
            old_num += 1
            new_num += 1
    return body


def _parse_single_diff(block_lines: list[str]) -> list[dict[str, Any]]:
    """Parse one fenced diff block. Returns a list of file entries.

    File entry shape:
        {
            "path_old": str | None,
            "path_new": str | None,
            "hunks": [
                {"old_start": int, "new_start": int, "header_context": str,
                 "lines": [...], "hunk_index_in_block": int}
            ]
        }
    """
    files: list[dict[str, Any]] = []
    current_file: dict[str, Any] | None = None
    current_hunk: dict[str, Any] | None = None
    hunk_body: list[str] = []
    hunk_index_in_block = 0

    def flush_hunk() -> None:
        nonlocal current_hunk, hunk_body, hunk_index_in_block
        if current_hunk is None or current_file is None:
            current_hunk = None
            hunk_body = []
            return
        current_hunk["lines"] = _parse_hunk_body(
            hunk_body, current_hunk["old_start"], current_hunk["new_start"]
        )
        current_hunk["hunk_index_in_block"] = hunk_index_in_block
        hunk_index_in_block += 1
        current_file["hunks"].append(current_hunk)
        current_hunk = None
        hunk_body = []

    def flush_file() -> None:
        nonlocal current_file
        flush_hunk()
        if current_file:
            files.append(current_file)
            current_file = None

    for raw in block_lines:
        m_old = FILE_OLD_RE.match(raw)
        m_new = FILE_NEW_RE.match(raw)
        m_hunk = HUNK_RE.match(raw)

        if m_old:
            # New file boundary; flush previous
            flush_file()
            current_file = {"path_old": m_old.group(1), "path_new": None, "hunks": []}
            continue
        if m_new:
            if current_file is None:
                current_file = {"path_old": None, "path_new": m_new.group(1), "hunks": []}
            else:
                current_file["path_new"] = m_new.group(1)
            continue
        if m_hunk:
            flush_hunk()
            current_hunk = {
                "old_start": int(m_hunk.group(1)),
                "old_count": int(m_hunk.group(2) or 1),
                "new_start": int(m_hunk.group(3)),
                "new_count": int(m_hunk.group(4) or 1),
                "header_context": m_hunk.group(5).strip(),
                "lines": [],
            }
            continue
        if current_hunk is not None:
            hunk_body.append(raw)

    flush_file()
    return files


def parse_markdown_for_diffs(text: str, infer_unfenced: bool = False) -> dict[str, Any]:
    """Top-level: find every ```diff block and parse it.

    Returns:
        {
            "blocks": [
                {"block_index": int, "files": [...]},
                ...
            ],
            "summary": {"total_files": int, "total_hunks": int, "total_blocks": int}
        }
    """
    lines = text.splitlines()
    in_block = False
    is_diff_block = False
    block_buf: list[str] = []
    blocks: list[dict[str, Any]] = []
    block_index = 0
    block_line_starts: list[int] = []

    i = 0
    while i < len(lines):
        ln = lines[i]
        if not in_block:
            m = FENCE_OPEN_RE.match(ln)
            if m:
                in_block = True
                lang = m.group(1)
                is_diff_block = (lang == "diff")
                block_buf = []
                block_line_starts.append(i)
        else:
            if FENCE_CLOSE_RE.match(ln):
                if is_diff_block:
                    files = _parse_single_diff(block_buf)
                    blocks.append({
                        "block_index": block_index,
                        "source_line": block_line_starts[block_index],
                        "files": files,
                    })
                    block_index += 1
                elif infer_unfenced:
                    # Try to detect if this looks like a diff (starts with --- / +++ / @@)
                    head = "\n".join(block_buf[:3])
                    if "@@" in head or head.startswith("---") or head.startswith("+++"):
                        files = _parse_single_diff(block_buf)
                        blocks.append({
                            "block_index": block_index,
                            "source_line": block_line_starts[block_index],
                            "files": files,
                            "inferred": True,
                        })
                        block_index += 1
                in_block = False
                is_diff_block = False
                block_buf = []
            else:
                block_buf.append(ln)
        i += 1

    total_files = sum(len(b["files"]) for b in blocks)
    total_hunks = sum(
        sum(len(f["hunks"]) for f in b["files"]) for b in blocks
    )

    return {
        "blocks": blocks,
        "summary": {
            "total_files": total_files,
            "total_hunks": total_hunks,
            "total_blocks": len(blocks),
        },
    }


SAMPLE_MARKDOWN = """# PR Review: Add payment retry logic

Two changes worth flagging.

```diff
--- a/payments/retry.py
+++ b/payments/retry.py
@@ -10,7 +10,8 @@ def schedule_retry(payment_id, attempt):
     if attempt > MAX_ATTEMPTS:
         return None
     delay = 2 ** attempt
-    return _enqueue(payment_id, delay)
+    jitter = random.uniform(0, delay * 0.1)
+    return _enqueue(payment_id, delay + jitter)
```

> [!MAJOR]
> This `random.uniform()` call is not seeded. In test runs it'll produce
> non-deterministic retry delays — make tests flaky.

```diff
--- a/payments/queue.py
+++ b/payments/queue.py
@@ -40,3 +40,7 @@ def _enqueue(payment_id, delay):
     redis.zadd("retries", {payment_id: time.time() + delay})
     log.info("scheduled", payment_id=payment_id, delay=delay)
+
+def cancel_retries(payment_id):
+    redis.zrem("retries", payment_id)
+    log.info("cancelled retries", payment_id=payment_id)
```

> [!NIT]
> Minor — would prefer `log.info("retries.cancelled", ...)` (dotted event name).
"""


def main(argv: list[str]) -> int:
    p = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    p.add_argument("--input", help="Path to markdown file, or '-' for stdin")
    p.add_argument("--output", help="Path to write JSON output (else stdout)")
    p.add_argument("--infer-diff", action="store_true",
                   help="Also parse unfenced or language-less blocks that look like diffs")
    p.add_argument("--sample", action="store_true",
                   help="Parse a built-in sample PR-review markdown")
    args = p.parse_args(argv)

    if args.sample:
        text = SAMPLE_MARKDOWN
    elif args.input:
        text = sys.stdin.read() if args.input == "-" else Path(args.input).read_text(encoding="utf-8")
    else:
        p.print_help()
        return 0

    result = parse_markdown_for_diffs(text, infer_unfenced=args.infer_diff)
    payload = json.dumps(result, indent=2)
    if args.output:
        Path(args.output).write_text(payload, encoding="utf-8")
        print(f"wrote {args.output}: {result['summary']['total_files']} files, "
              f"{result['summary']['total_hunks']} hunks across "
              f"{result['summary']['total_blocks']} diff blocks")
    else:
        print(payload)
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))