Pr Review Expert

#!/usr/bin/env python3
"""Calculate PR blast radius by analyzing import chains and dependency trees.

Given a set of changed files and a source root, walks the file system to build
an import/dependency graph, then computes how many files are directly and
transitively affected by the changes. Supports Python, JavaScript/TypeScript,
Go, and generic import patterns.

Usage:
    python blast_radius_calculator.py --changed src/lib/auth.ts src/utils/hash.py --root ./src
    git diff --name-only main...HEAD | python blast_radius_calculator.py --root .
    python blast_radius_calculator.py --changed api/models.py --root . --json --depth 5
"""

import argparse
import json
import os
import re
import sys
from collections import defaultdict, deque
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple


# --- Import pattern extractors ---

# Python: import foo, from foo import bar, from . import bar
PY_IMPORT_RE = [
    re.compile(r"^\s*import\s+([\w.]+)", re.MULTILINE),
    re.compile(r"^\s*from\s+([\w.]+)\s+import", re.MULTILINE),
]

# JavaScript/TypeScript: import ... from '...', require('...')
JS_IMPORT_RE = [
    re.compile(r"""(?:import|export)\s+.*?from\s+['"](\.[\w./\\-]+)['"]""", re.MULTILINE),
    re.compile(r"""require\(\s*['"](\.[\w./\\-]+)['"]\s*\)""", re.MULTILINE),
]

# Go: import "path" or import ( "path" )
GO_IMPORT_RE = [
    re.compile(r"""(?:import\s+)?"([\w./\\-]+)"$""", re.MULTILINE),
]

SUPPORTED_EXTENSIONS = {
    ".py": "python",
    ".js": "javascript",
    ".ts": "typescript",
    ".tsx": "typescript",
    ".jsx": "javascript",
    ".mjs": "javascript",
    ".cjs": "javascript",
    ".go": "go",
}

SKIP_DIRS = {
    "node_modules", ".git", "__pycache__", ".venv", "venv",
    "dist", "build", ".next", ".nuxt", "vendor", ".tox",
    "coverage", ".pytest_cache", ".mypy_cache",
}


def discover_files(root: str) -> List[str]:
    """Walk the source tree and collect supported source files."""
    files = []
    root_path = Path(root).resolve()
    for dirpath, dirnames, filenames in os.walk(root_path):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        for fname in filenames:
            ext = os.path.splitext(fname)[1]
            if ext in SUPPORTED_EXTENSIONS:
                full = os.path.join(dirpath, fname)
                files.append(full)
    return files


def normalize_path(path: str, root: str) -> str:
    """Convert absolute path to root-relative with forward slashes."""
    try:
        return str(Path(path).resolve().relative_to(Path(root).resolve()))
    except ValueError:
        return path


def extract_imports_python(content: str, file_path: str, root: str) -> List[str]:
    """Extract import targets from Python source."""
    imports = []
    for pattern in PY_IMPORT_RE:
        for match in pattern.finditer(content):
            module = match.group(1)
            # Convert dotted module to path
            module_path = module.replace(".", "/")
            imports.append(module_path)
    return imports


def extract_imports_js(content: str, file_path: str, root: str) -> List[str]:
    """Extract import targets from JS/TS source."""
    imports = []
    for pattern in JS_IMPORT_RE:
        for match in pattern.finditer(content):
            rel_path = match.group(1)
            # Resolve relative to file's directory
            file_dir = os.path.dirname(file_path)
            resolved = os.path.normpath(os.path.join(file_dir, rel_path))
            imports.append(normalize_path(resolved, root))
    return imports


def extract_imports_go(content: str, file_path: str, root: str) -> List[str]:
    """Extract import targets from Go source."""
    imports = []
    for pattern in GO_IMPORT_RE:
        for match in pattern.finditer(content):
            imports.append(match.group(1))
    return imports


EXTRACTORS = {
    "python": extract_imports_python,
    "javascript": extract_imports_js,
    "typescript": extract_imports_js,
    "go": extract_imports_go,
}


def resolve_import_to_file(imp: str, known_files: Set[str]) -> Optional[str]:
    """Try to resolve an import string to a known file path."""
    # Direct match
    if imp in known_files:
        return imp

    # Try common extensions and index files
    candidates = [
        imp,
        imp + ".py",
        imp + ".ts",
        imp + ".tsx",
        imp + ".js",
        imp + ".jsx",
        imp + ".mjs",
        imp + ".go",
        os.path.join(imp, "index.ts"),
        os.path.join(imp, "index.tsx"),
        os.path.join(imp, "index.js"),
        os.path.join(imp, "__init__.py"),
        os.path.join(imp, "mod.go"),
    ]
    for c in candidates:
        normalized = os.path.normpath(c)
        if normalized in known_files:
            return normalized
    return None


def build_dependency_graph(root: str) -> Tuple[Dict[str, Set[str]], Dict[str, Set[str]], Set[str]]:
    """Build forward (imports) and reverse (imported-by) dependency graphs.

    Returns:
        imports_graph: file -> set of files it imports
        reverse_graph: file -> set of files that import it
        all_files: set of all discovered file paths (root-relative)
    """
    all_abs_files = discover_files(root)
    all_files = set()
    file_abs_map = {}

    for abs_path in all_abs_files:
        rel = normalize_path(abs_path, root)
        all_files.add(rel)
        file_abs_map[rel] = abs_path

    imports_graph: Dict[str, Set[str]] = defaultdict(set)
    reverse_graph: Dict[str, Set[str]] = defaultdict(set)

    for rel_path, abs_path in file_abs_map.items():
        ext = os.path.splitext(abs_path)[1]
        lang = SUPPORTED_EXTENSIONS.get(ext)
        if not lang or lang not in EXTRACTORS:
            continue

        try:
            with open(abs_path, "r", encoding="utf-8", errors="replace") as f:
                content = f.read()
        except (OSError, IOError):
            continue

        extractor = EXTRACTORS[lang]
        raw_imports = extractor(content, rel_path, root)

        for imp in raw_imports:
            resolved = resolve_import_to_file(imp, all_files)
            if resolved and resolved != rel_path:
                imports_graph[rel_path].add(resolved)
                reverse_graph[resolved].add(rel_path)

    return dict(imports_graph), dict(reverse_graph), all_files


def compute_blast_radius(
    changed_files: List[str],
    reverse_graph: Dict[str, Set[str]],
    max_depth: int,
) -> Dict[str, Dict]:
    """BFS from changed files through reverse dependency graph.

    Returns per-changed-file impact analysis with depth tracking.
    """
    results = {}

    for changed in changed_files:
        visited: Dict[str, int] = {}
        queue: deque = deque()

        # Seed with direct dependents
        if changed in reverse_graph:
            for dep in reverse_graph[changed]:
                if dep not in changed_files:
                    queue.append((dep, 1))
                    visited[dep] = 1

        # BFS
        while queue:
            current, depth = queue.popleft()
            if depth >= max_depth:
                continue
            if current in reverse_graph:
                for dep in reverse_graph[current]:
                    if dep not in visited and dep not in changed_files:
                        visited[dep] = depth + 1
                        queue.append((dep, depth + 1))

        direct = [f for f, d in visited.items() if d == 1]
        transitive = [f for f, d in visited.items() if d > 1]

        results[changed] = {
            "direct_dependents": sorted(direct),
            "transitive_dependents": sorted(transitive),
            "direct_count": len(direct),
            "transitive_count": len(transitive),
            "total_affected": len(visited),
            "max_chain_depth": max(visited.values()) if visited else 0,
        }

    return results


def classify_severity(total_affected: int) -> str:
    """Classify blast radius severity based on total affected files."""
    if total_affected >= 20:
        return "CRITICAL"
    elif total_affected >= 10:
        return "HIGH"
    elif total_affected >= 3:
        return "MEDIUM"
    else:
        return "LOW"


def generate_report(
    changed_files: List[str],
    root: str,
    max_depth: int,
) -> Dict:
    """Full blast radius analysis report."""
    imports_graph, reverse_graph, all_files = build_dependency_graph(root)

    # Normalize changed file paths
    normalized_changed = []
    for cf in changed_files:
        norm = normalize_path(cf, root)
        if norm in all_files:
            normalized_changed.append(norm)
        else:
            # Try without leading path components
            basename = os.path.basename(cf)
            matches = [f for f in all_files if f.endswith(cf) or os.path.basename(f) == basename]
            if matches:
                normalized_changed.extend(matches)
            else:
                normalized_changed.append(norm)

    per_file = compute_blast_radius(normalized_changed, reverse_graph, max_depth)

    # Aggregate
    all_affected: Set[str] = set()
    for info in per_file.values():
        all_affected.update(info["direct_dependents"])
        all_affected.update(info["transitive_dependents"])

    total_affected = len(all_affected)
    overall_severity = classify_severity(total_affected)

    # Hotspots: most-imported files among the changed set
    hotspots = []
    for cf in normalized_changed:
        dep_count = len(reverse_graph.get(cf, set()))
        if dep_count > 0:
            hotspots.append({"file": cf, "dependent_count": dep_count})
    hotspots.sort(key=lambda x: x["dependent_count"], reverse=True)

    return {
        "overall_severity": overall_severity,
        "total_files_in_project": len(all_files),
        "changed_files": normalized_changed,
        "changed_file_count": len(normalized_changed),
        "total_affected_files": total_affected,
        "impact_percentage": round(total_affected / max(len(all_files), 1) * 100, 1),
        "max_depth_searched": max_depth,
        "hotspots": hotspots[:10],
        "per_file_analysis": per_file,
        "all_affected_files": sorted(all_affected),
    }


def format_human(result: Dict) -> str:
    """Format blast radius report for human reading."""
    lines = []
    lines.append("=" * 60)
    lines.append("  BLAST RADIUS ANALYSIS")
    lines.append("=" * 60)
    lines.append("")
    lines.append(f"Overall Severity:    {result['overall_severity']}")
    lines.append(f"Project Files:       {result['total_files_in_project']}")
    lines.append(f"Changed Files:       {result['changed_file_count']}")
    lines.append(f"Affected Files:      {result['total_affected_files']}")
    lines.append(f"Impact:              {result['impact_percentage']}% of project")
    lines.append(f"Max Depth Searched:  {result['max_depth_searched']}")
    lines.append("")

    # Hotspots
    if result["hotspots"]:
        lines.append("Dependency Hotspots (most imported changed files):")
        for hs in result["hotspots"]:
            lines.append(f"  {hs['file']} <- {hs['dependent_count']} dependents")
        lines.append("")

    # Per-file breakdown
    lines.append("-" * 60)
    lines.append("PER-FILE IMPACT")
    lines.append("-" * 60)

    for changed_file, info in result["per_file_analysis"].items():
        sev = classify_severity(info["total_affected"])
        lines.append("")
        lines.append(f"[{sev}] {changed_file}")
        lines.append(f"  Direct dependents:     {info['direct_count']}")
        lines.append(f"  Transitive dependents: {info['transitive_count']}")
        lines.append(f"  Total affected:        {info['total_affected']}")
        if info["max_chain_depth"] > 0:
            lines.append(f"  Longest chain depth:   {info['max_chain_depth']}")

        if info["direct_dependents"]:
            lines.append("  Direct:")
            for dep in info["direct_dependents"][:10]:
                lines.append(f"    -> {dep}")
            if len(info["direct_dependents"]) > 10:
                lines.append(f"    ... and {len(info['direct_dependents']) - 10} more")

        if info["transitive_dependents"]:
            lines.append("  Transitive:")
            for dep in info["transitive_dependents"][:10]:
                lines.append(f"    ~> {dep}")
            if len(info["transitive_dependents"]) > 10:
                lines.append(f"    ... and {len(info['transitive_dependents']) - 10} more")

    # Summary
    lines.append("")
    lines.append("-" * 60)
    lines.append("AFFECTED FILE LIST")
    lines.append("-" * 60)
    if result["all_affected_files"]:
        for af in result["all_affected_files"]:
            lines.append(f"  {af}")
    else:
        lines.append("  No downstream dependencies found.")

    lines.append("")
    lines.append("=" * 60)
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Calculate PR blast radius by analyzing import chains and "
                    "dependency trees from changed files.",
        epilog="Example: git diff --name-only main...HEAD | python blast_radius_calculator.py --root .",
    )
    parser.add_argument(
        "--changed", "-c",
        nargs="+",
        help="List of changed file paths. If omitted, reads from stdin.",
    )
    parser.add_argument(
        "--root", "-r",
        default=".",
        help="Source root directory to scan for dependencies (default: current directory).",
    )
    parser.add_argument(
        "--depth", "-d",
        type=int,
        default=5,
        help="Maximum transitive dependency depth to traverse (default: 5).",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results as JSON.",
    )
    args = parser.parse_args()

    root = os.path.abspath(args.root)
    if not os.path.isdir(root):
        print(f"Error: Root directory not found: {root}", file=sys.stderr)
        sys.exit(1)

    if args.changed:
        changed_files = args.changed
    else:
        if sys.stdin.isatty():
            print("Reading changed file paths from stdin (one per line)...", file=sys.stderr)
        changed_files = [line.strip() for line in sys.stdin if line.strip()]

    if not changed_files:
        print("Error: No changed files provided.", file=sys.stderr)
        sys.exit(1)

    result = generate_report(changed_files, root, args.depth)

    if args.json_output:
        print(json.dumps(result, indent=2))
    else:
        print(format_human(result))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Analyze git diff output for risk indicators.

Scans unified diff input for large file changes, sensitive path patterns,
configuration modifications, breaking change signals, and security red flags.
Produces a prioritized risk report in human-readable or JSON format.

Usage:
    git diff main...HEAD | python diff_analyzer.py
    python diff_analyzer.py --file /tmp/pr-123.diff
    python diff_analyzer.py --file /tmp/pr-123.diff --json
"""

import argparse
import json
import re
import sys
from collections import defaultdict
from typing import Dict, List, Tuple


# --- Risk pattern definitions ---

SENSITIVE_PATHS = [
    (r"(^|/)\.env", "Environment config file", "HIGH"),
    (r"(^|/)auth/", "Authentication module", "HIGH"),
    (r"(^|/)security/", "Security module", "HIGH"),
    (r"(^|/)middleware/", "Middleware layer", "MEDIUM"),
    (r"(^|/)migrations?/", "Database migration", "HIGH"),
    (r"(^|/)config/", "Configuration directory", "MEDIUM"),
    (r"(^|/)secrets?/", "Secrets directory", "CRITICAL"),
    (r"(^|/)payments?/", "Payment processing", "CRITICAL"),
    (r"(^|/)crypto/", "Cryptography module", "HIGH"),
    (r"(^|/)docker-compose", "Docker orchestration", "MEDIUM"),
    (r"(^|/)Dockerfile", "Container definition", "MEDIUM"),
    (r"(^|/)k8s/|kubernetes/", "Kubernetes config", "HIGH"),
    (r"(^|/)terraform/|\.tf$", "Infrastructure as code", "HIGH"),
    (r"(^|/)ci/|\.github/workflows/", "CI/CD pipeline", "MEDIUM"),
    (r"(^|/)nginx|apache|caddy", "Web server config", "MEDIUM"),
]

CONFIG_FILE_PATTERNS = [
    r"\.ya?ml$", r"\.json$", r"\.toml$", r"\.ini$", r"\.cfg$",
    r"\.conf$", r"\.env", r"Makefile$", r"\.lock$",
    r"requirements.*\.txt$", r"package\.json$", r"go\.mod$",
    r"Cargo\.toml$", r"pom\.xml$", r"build\.gradle$",
]

BREAKING_PATTERNS = [
    (r"^\+.*\bDROP\s+(TABLE|COLUMN|INDEX)", "SQL destructive operation", "CRITICAL"),
    (r"^\+.*\bTRUNCATE\b", "SQL truncate", "CRITICAL"),
    (r"^\+.*\bALTER.*NOT\s+NULL", "Non-nullable column addition", "HIGH"),
    (r"^-.*\brouter\.(get|post|put|delete|patch)\(", "API endpoint removal", "HIGH"),
    (r"^-.*\b@(app|api)\.(get|post|put|delete)\b", "API route removal", "HIGH"),
    (r"^-.*\bexport\s+(interface|type)\s+", "Exported type removal", "HIGH"),
    (r"^-.*\bexport\s+(function|const|class)\s+", "Exported symbol removal", "HIGH"),
    (r"^\+.*\bDEPRECAT", "Deprecation notice added", "MEDIUM"),
    (r"^-.*\benv\.[A-Z_]+", "Environment variable removal", "MEDIUM"),
]

SECURITY_PATTERNS = [
    (r"^\+.*\b(password|secret|api_key|token|private_key)\s*=\s*['\"][^'\"]{8,}", "Hardcoded secret", "CRITICAL"),
    (r"^\+.*AKIA[0-9A-Z]{16}", "AWS access key", "CRITICAL"),
    (r"^\+.*\beval\s*\(", "Dynamic code execution (eval)", "HIGH"),
    (r"^\+.*\bexec\s*\(", "Dynamic code execution (exec)", "HIGH"),
    (r"^\+.*dangerouslySetInnerHTML", "XSS vector (React)", "HIGH"),
    (r"^\+.*innerHTML\s*=", "XSS vector (innerHTML)", "HIGH"),
    (r"^\+.*__proto__", "Prototype pollution", "HIGH"),
    (r"^\+.*\bcreateHash\(['\"]md5", "Weak hash (MD5)", "MEDIUM"),
    (r"^\+.*\bcreateHash\(['\"]sha1", "Weak hash (SHA1)", "MEDIUM"),
    (r"^\+.*subprocess\.call\(.*shell\s*=\s*True", "Shell injection risk", "HIGH"),
    (r"^\+.*path\.join\(.*req\.", "Potential path traversal", "MEDIUM"),
]

# Thresholds
LARGE_FILE_THRESHOLD = 200  # lines changed
LARGE_PR_THRESHOLD = 500    # total lines changed


def parse_diff(diff_text: str) -> List[Dict]:
    """Parse unified diff into per-file records."""
    files = []
    current_file = None
    current_lines = []

    for line in diff_text.splitlines():
        if line.startswith("diff --git"):
            if current_file:
                files.append({"path": current_file, "lines": current_lines})
            match = re.search(r"b/(.+)$", line)
            current_file = match.group(1) if match else "unknown"
            current_lines = []
        elif current_file is not None:
            current_lines.append(line)

    if current_file:
        files.append({"path": current_file, "lines": current_lines})

    return files


def count_changes(lines: List[str]) -> Tuple[int, int]:
    """Count additions and deletions in diff lines."""
    additions = sum(1 for l in lines if l.startswith("+") and not l.startswith("+++"))
    deletions = sum(1 for l in lines if l.startswith("-") and not l.startswith("---"))
    return additions, deletions


def check_sensitive_paths(file_path: str) -> List[Dict]:
    """Check if file path matches sensitive patterns."""
    findings = []
    for pattern, desc, severity in SENSITIVE_PATHS:
        if re.search(pattern, file_path, re.IGNORECASE):
            findings.append({
                "type": "sensitive_path",
                "severity": severity,
                "description": desc,
                "file": file_path,
            })
    return findings


def check_config_file(file_path: str) -> bool:
    """Check if a file is a configuration file."""
    return any(re.search(p, file_path, re.IGNORECASE) for p in CONFIG_FILE_PATTERNS)


def scan_patterns(lines: List[str], file_path: str, patterns, category: str) -> List[Dict]:
    """Scan diff lines against a list of regex patterns."""
    findings = []
    for i, line in enumerate(lines, 1):
        for pattern, desc, severity in patterns:
            if re.search(pattern, line, re.IGNORECASE):
                findings.append({
                    "type": category,
                    "severity": severity,
                    "description": desc,
                    "file": file_path,
                    "line_number": i,
                    "content": line.strip()[:120],
                })
    return findings


def analyze_diff(diff_text: str, large_file_thresh: int = LARGE_FILE_THRESHOLD, large_pr_thresh: int = LARGE_PR_THRESHOLD) -> Dict:
    """Run full analysis on diff text and return structured results."""
    files = parse_diff(diff_text)
    all_findings = []
    file_stats = []
    total_additions = 0
    total_deletions = 0
    large_files = []
    config_files = []

    for f in files:
        path = f["path"]
        lines = f["lines"]
        additions, deletions = count_changes(lines)
        total_additions += additions
        total_deletions += deletions
        total_changed = additions + deletions

        stat = {
            "path": path,
            "additions": additions,
            "deletions": deletions,
            "total": total_changed,
        }
        file_stats.append(stat)

        # Large file check
        if total_changed > large_file_thresh:
            large_files.append(stat)
            all_findings.append({
                "type": "large_file",
                "severity": "MEDIUM",
                "description": f"Large change: {total_changed} lines modified",
                "file": path,
            })

        # Sensitive path check
        all_findings.extend(check_sensitive_paths(path))

        # Config file check
        if check_config_file(path):
            config_files.append(path)
            all_findings.append({
                "type": "config_change",
                "severity": "MEDIUM",
                "description": "Configuration file modified",
                "file": path,
            })

        # Breaking pattern scan
        all_findings.extend(scan_patterns(lines, path, BREAKING_PATTERNS, "breaking_change"))

        # Security pattern scan
        all_findings.extend(scan_patterns(lines, path, SECURITY_PATTERNS, "security"))

    # PR-level size check
    total_changed = total_additions + total_deletions
    if total_changed > large_pr_thresh:
        all_findings.append({
            "type": "large_pr",
            "severity": "HIGH",
            "description": f"Large PR: {total_changed} total lines changed across {len(files)} files. Consider splitting.",
            "file": "PR-level",
        })

    # Deduplicate findings
    seen = set()
    unique_findings = []
    for f in all_findings:
        key = (f["type"], f["severity"], f["file"], f.get("line_number", 0))
        if key not in seen:
            seen.add(key)
            unique_findings.append(f)

    # Compute overall risk
    severity_order = {"CRITICAL": 4, "HIGH": 3, "MEDIUM": 2, "LOW": 1}
    max_sev = max((severity_order.get(f["severity"], 0) for f in unique_findings), default=0)
    risk_map = {4: "CRITICAL", 3: "HIGH", 2: "MEDIUM", 1: "LOW", 0: "LOW"}
    overall_risk = risk_map[max_sev]

    severity_counts = defaultdict(int)
    for f in unique_findings:
        severity_counts[f["severity"]] += 1

    return {
        "overall_risk": overall_risk,
        "total_files": len(files),
        "total_additions": total_additions,
        "total_deletions": total_deletions,
        "total_changed": total_changed,
        "severity_counts": dict(severity_counts),
        "large_files": large_files,
        "config_files_changed": config_files,
        "findings": sorted(unique_findings, key=lambda x: severity_order.get(x["severity"], 0), reverse=True),
    }


def format_human(result: Dict) -> str:
    """Format analysis results for human consumption."""
    lines = []
    lines.append("=" * 60)
    lines.append("  DIFF RISK ANALYSIS REPORT")
    lines.append("=" * 60)
    lines.append("")
    lines.append(f"Overall Risk:   {result['overall_risk']}")
    lines.append(f"Files Changed:  {result['total_files']}")
    lines.append(f"Lines Added:    +{result['total_additions']}")
    lines.append(f"Lines Removed:  -{result['total_deletions']}")
    lines.append(f"Total Changed:  {result['total_changed']}")
    lines.append("")

    sc = result["severity_counts"]
    if sc:
        lines.append("Findings by Severity:")
        for sev in ["CRITICAL", "HIGH", "MEDIUM", "LOW"]:
            if sev in sc:
                lines.append(f"  {sev}: {sc[sev]}")
        lines.append("")

    if result["config_files_changed"]:
        lines.append("Config Files Modified:")
        for cf in result["config_files_changed"]:
            lines.append(f"  - {cf}")
        lines.append("")

    if result["large_files"]:
        lines.append("Large File Changes (>{} lines):".format(LARGE_FILE_THRESHOLD))
        for lf in result["large_files"]:
            lines.append(f"  - {lf['path']} (+{lf['additions']}/-{lf['deletions']})")
        lines.append("")

    findings = result["findings"]
    if findings:
        lines.append("-" * 60)
        lines.append("FINDINGS")
        lines.append("-" * 60)
        for i, f in enumerate(findings, 1):
            lines.append("")
            lines.append(f"[{f['severity']}] #{i}: {f['description']}")
            lines.append(f"  Type: {f['type']}")
            lines.append(f"  File: {f['file']}")
            if "line_number" in f:
                lines.append(f"  Line: {f['line_number']}")
            if "content" in f:
                lines.append(f"  Code: {f['content']}")
    else:
        lines.append("No risk indicators found.")

    lines.append("")
    lines.append("=" * 60)
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Analyze git diff output for risk indicators (large files, "
                    "sensitive paths, config changes, breaking patterns, security red flags).",
        epilog="Example: git diff main...HEAD | python diff_analyzer.py",
    )
    parser.add_argument(
        "--file", "-f",
        help="Path to a diff file. If omitted, reads from stdin.",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results as JSON.",
    )
    parser.add_argument(
        "--large-file-threshold",
        type=int,
        default=LARGE_FILE_THRESHOLD,
        help=f"Lines changed to flag a file as large (default: {LARGE_FILE_THRESHOLD}).",
    )
    parser.add_argument(
        "--large-pr-threshold",
        type=int,
        default=LARGE_PR_THRESHOLD,
        help=f"Total lines changed to flag PR as large (default: {LARGE_PR_THRESHOLD}).",
    )
    args = parser.parse_args()

    file_threshold = args.large_file_threshold
    pr_threshold = args.large_pr_threshold

    if args.file:
        try:
            with open(args.file, "r", encoding="utf-8", errors="replace") as fh:
                diff_text = fh.read()
        except FileNotFoundError:
            print(f"Error: File not found: {args.file}", file=sys.stderr)
            sys.exit(1)
    else:
        if sys.stdin.isatty():
            print("Reading diff from stdin (pipe a diff or use --file)...", file=sys.stderr)
        diff_text = sys.stdin.read()

    if not diff_text.strip():
        print("Error: Empty diff input.", file=sys.stderr)
        sys.exit(1)

    result = analyze_diff(diff_text, file_threshold, pr_threshold)

    if args.json_output:
        print(json.dumps(result, indent=2))
    else:
        print(format_human(result))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Generate PR review checklists based on changed file types and patterns.

Reads a list of changed file paths (from git diff --name-only or stdin) and
produces a tailored review checklist. Checklist items are selected based on
which file categories appear in the change set: API routes, database files,
frontend components, config, tests, infrastructure, etc.

Usage:
    git diff --name-only main...HEAD | python review_checklist_generator.py
    python review_checklist_generator.py --files src/auth.ts src/db/migrate.sql
    python review_checklist_generator.py --diff-file /tmp/pr-123.diff --json
"""

import argparse
import json
import re
import sys
from collections import OrderedDict
from typing import Dict, List, Set, Tuple


# --- File category classifiers ---

FILE_CATEGORIES: List[Tuple[str, str, List[str]]] = [
    # (category_name, description, [regex patterns])
    ("api", "API Routes & Endpoints", [
        r"(routes?|controllers?|handlers?|endpoints?|api)/",
        r"\.(controller|handler|route)\.(ts|js|py|go|rb)$",
    ]),
    ("database", "Database & Migrations", [
        r"migrations?/", r"schemas?/", r"models?/",
        r"\.sql$", r"(prisma|drizzle|knex|sequelize|typeorm)",
        r"(alembic|flyway|liquibase)",
    ]),
    ("auth", "Authentication & Authorization", [
        r"auth/", r"(login|signup|session|oauth|jwt|rbac|permission)",
        r"middleware.*(auth|token|session)",
    ]),
    ("frontend", "Frontend Components & UI", [
        r"\.(tsx|jsx|vue|svelte)$",
        r"components?/", r"pages?/", r"views?/",
        r"styles?/", r"\.(css|scss|less|styled)$",
    ]),
    ("test", "Tests & Test Infrastructure", [
        r"\.(test|spec)\.", r"__tests__/", r"tests?/",
        r"(jest|vitest|pytest|mocha|cypress|playwright)",
        r"fixtures?/", r"factories/", r"mocks?/",
    ]),
    ("config", "Configuration Files", [
        r"\.(ya?ml|json|toml|ini|cfg|conf|env)$",
        r"(Makefile|Rakefile|Taskfile)",
        r"\.(eslint|prettier|babel|webpack|vite|tsconfig)",
        r"requirements.*\.txt$", r"package\.json$",
        r"(go\.mod|Cargo\.toml|pom\.xml|build\.gradle)",
    ]),
    ("infra", "Infrastructure & DevOps", [
        r"(docker|Docker)", r"k8s/", r"kubernetes/",
        r"terraform/", r"\.tf$", r"ansible/",
        r"helm/", r"pulumi/", r"cloudformation",
        r"\.github/(workflows|actions)/", r"(ci|cd)/",
        r"(Jenkinsfile|\.gitlab-ci)",
    ]),
    ("security", "Security-Sensitive Code", [
        r"(crypto|encrypt|decrypt|hash|sign|verify)/",
        r"(payment|billing|checkout|stripe|paypal)",
        r"(secrets?|credentials?|certs?)/",
        r"(cors|csp|helmet|sanitiz)",
    ]),
    ("shared_lib", "Shared Libraries & Utilities", [
        r"(lib|libs|shared|common|utils?|helpers?|packages?)/",
        r"(types|interfaces|contracts)/",
    ]),
    ("docs", "Documentation", [
        r"\.(md|mdx|rst|adoc|txt)$",
        r"(docs?|documentation)/",
        r"(README|CHANGELOG|CONTRIBUTING|LICENSE)",
    ]),
    ("deps", "Dependencies", [
        r"(package-lock|yarn\.lock|pnpm-lock|Gemfile\.lock|Pipfile\.lock|poetry\.lock)",
        r"(go\.sum|Cargo\.lock|composer\.lock)",
        r"(requirements.*\.txt|package\.json|Gemfile|Pipfile)$",
    ]),
]

# --- Checklist items per category ---

CHECKLIST_ITEMS: Dict[str, List[Tuple[str, str]]] = {
    "always": [
        ("scope", "PR title accurately describes the change"),
        ("scope", "PR description explains WHY, not just WHAT"),
        ("scope", "No unrelated changes included (scope creep)"),
        ("scope", "Linked ticket/issue exists and matches scope"),
    ],
    "api": [
        ("breaking", "No API endpoints removed without deprecation period"),
        ("breaking", "No required fields added to existing request/response schemas"),
        ("security", "Auth/authorization middleware applied to all new endpoints"),
        ("security", "Input validation present on all new parameters"),
        ("security", "Rate limiting considered for public endpoints"),
        ("security", "CORS configured correctly for new endpoints"),
        ("testing", "Integration tests cover new/modified endpoints"),
        ("testing", "Error response codes tested (400, 401, 403, 404, 500)"),
        ("docs", "API documentation updated (OpenAPI/Swagger if applicable)"),
        ("perf", "Pagination added for list endpoints returning unbounded data"),
    ],
    "database": [
        ("breaking", "Migration is reversible (has rollback/down method)"),
        ("breaking", "No destructive operations (DROP TABLE/COLUMN) without migration plan"),
        ("breaking", "NOT NULL columns include a default value or two-phase migration"),
        ("perf", "Indexes added for new query patterns"),
        ("perf", "No unbounded queries without LIMIT"),
        ("security", "Parameterized queries used (no string interpolation in SQL)"),
        ("testing", "Migration tested against production-like data volume"),
        ("data", "Data backfill strategy documented if needed"),
    ],
    "auth": [
        ("security", "No auth bypass patterns (skip, noauth, TODO)"),
        ("security", "Session/token expiration configured correctly"),
        ("security", "Failed login attempts are rate-limited"),
        ("security", "Sensitive data not exposed in error messages or logs"),
        ("security", "Password hashing uses bcrypt/argon2 (not MD5/SHA1)"),
        ("testing", "Auth edge cases tested (expired token, revoked session, role escalation)"),
        ("testing", "Both positive and negative auth paths have test coverage"),
    ],
    "frontend": [
        ("security", "No XSS vectors (innerHTML, dangerouslySetInnerHTML)"),
        ("security", "User input sanitized before rendering"),
        ("a11y", "Accessibility attributes present (aria-*, alt text, semantic HTML)"),
        ("perf", "No unnecessary re-renders or missing memoization"),
        ("perf", "Images and assets optimized"),
        ("ux", "Loading and error states handled"),
        ("ux", "Responsive design verified"),
        ("testing", "Component tests cover user interactions"),
    ],
    "test": [
        ("quality", "Test names clearly describe what they verify"),
        ("quality", "Edge cases covered (empty, null, boundary values)"),
        ("quality", "Error paths tested (not just happy path)"),
        ("quality", "No tests deleted without clear justification"),
        ("quality", "Test fixtures do not contain real secrets or PII"),
        ("quality", "Flaky test patterns avoided (timeouts, sleep, order dependency)"),
    ],
    "config": [
        ("breaking", "New environment variables documented in .env.example"),
        ("breaking", "Removed config values verified as unused in all environments"),
        ("security", "No secrets or credentials in config files"),
        ("ops", "Config changes are backward-compatible with rolling deploys"),
        ("ops", "Feature flags used for risky config changes"),
    ],
    "infra": [
        ("security", "No secrets in Dockerfiles or CI configs"),
        ("security", "Container images use specific version tags (not :latest)"),
        ("security", "Least privilege applied to IAM/RBAC changes"),
        ("ops", "Resource limits defined (CPU, memory)"),
        ("ops", "Health checks configured"),
        ("ops", "Rollback procedure documented for infrastructure changes"),
        ("testing", "Infrastructure changes tested in staging first"),
    ],
    "security": [
        ("security", "No hardcoded secrets, API keys, or credentials"),
        ("security", "Encryption algorithms are current (AES-256, SHA-256+)"),
        ("security", "Sensitive data encrypted at rest and in transit"),
        ("security", "Audit logging added for security-relevant operations"),
        ("testing", "Security-critical paths have near-100% test coverage"),
        ("compliance", "Changes reviewed against relevant compliance requirements"),
    ],
    "shared_lib": [
        ("breaking", "Backward-compatible for all consumers"),
        ("breaking", "Version bump follows semver"),
        ("blast_radius", "All downstream consumers identified and checked"),
        ("blast_radius", "Cross-service dependencies verified"),
        ("testing", "Shared code has comprehensive unit tests"),
        ("docs", "Public API documented with usage examples"),
    ],
    "docs": [
        ("quality", "Documentation matches actual code behavior"),
        ("quality", "Code examples are tested or verified"),
        ("quality", "Links are valid and not broken"),
    ],
    "deps": [
        ("security", "New dependencies checked for known CVEs"),
        ("security", "Dependencies are actively maintained (not abandoned)"),
        ("perf", "Bundle size impact assessed for frontend dependencies"),
        ("ops", "Lock file committed alongside dependency changes"),
        ("license", "New dependency licenses compatible with project license"),
    ],
}


def classify_files(file_paths: List[str]) -> Dict[str, List[str]]:
    """Classify each file path into zero or more categories."""
    categories: Dict[str, List[str]] = {}
    for path in file_paths:
        matched = False
        for cat_name, _, patterns in FILE_CATEGORIES:
            for pattern in patterns:
                if re.search(pattern, path, re.IGNORECASE):
                    categories.setdefault(cat_name, []).append(path)
                    matched = True
                    break
        if not matched:
            categories.setdefault("other", []).append(path)
    return categories


def get_category_label(cat_name: str) -> str:
    """Get human-readable label for a category."""
    for name, label, _ in FILE_CATEGORIES:
        if name == cat_name:
            return label
    return cat_name.replace("_", " ").title()


def build_checklist(categories: Dict[str, List[str]]) -> OrderedDict:
    """Build a tailored checklist based on detected categories."""
    checklist = OrderedDict()

    # Always include base items
    checklist["Scope & Context"] = [
        {"check": item[1], "tag": item[0]} for item in CHECKLIST_ITEMS["always"]
    ]

    # Add category-specific items
    for cat_name in categories:
        if cat_name in CHECKLIST_ITEMS:
            section_label = get_category_label(cat_name)
            items = [
                {"check": item[1], "tag": item[0]}
                for item in CHECKLIST_ITEMS[cat_name]
            ]
            checklist[section_label] = items

    return checklist


def extract_files_from_diff(diff_text: str) -> List[str]:
    """Extract file paths from unified diff format."""
    files = []
    for line in diff_text.splitlines():
        if line.startswith("diff --git"):
            match = re.search(r"b/(.+)$", line)
            if match:
                files.append(match.group(1))
    return files


def generate_report(file_paths: List[str]) -> Dict:
    """Generate the full checklist report."""
    categories = classify_files(file_paths)
    checklist = build_checklist(categories)

    total_items = sum(len(items) for items in checklist.values())

    category_summary = {}
    for cat_name, files in sorted(categories.items()):
        category_summary[cat_name] = {
            "label": get_category_label(cat_name),
            "file_count": len(files),
            "files": files,
        }

    return {
        "total_files": len(file_paths),
        "categories_detected": list(categories.keys()),
        "total_checklist_items": total_items,
        "category_summary": category_summary,
        "checklist": {k: v for k, v in checklist.items()},
    }


def format_human(result: Dict) -> str:
    """Format the checklist report for human reading."""
    lines = []
    lines.append("=" * 60)
    lines.append("  PR REVIEW CHECKLIST")
    lines.append("=" * 60)
    lines.append("")
    lines.append(f"Files Analyzed:      {result['total_files']}")
    lines.append(f"Categories Detected: {', '.join(result['categories_detected'])}")
    lines.append(f"Checklist Items:     {result['total_checklist_items']}")
    lines.append("")

    # Category summary
    lines.append("File Categories:")
    for cat_name, info in result["category_summary"].items():
        lines.append(f"  [{info['label']}] ({info['file_count']} files)")
        for f in info["files"][:5]:
            lines.append(f"    - {f}")
        if info["file_count"] > 5:
            lines.append(f"    ... and {info['file_count'] - 5} more")
    lines.append("")

    # Checklist
    lines.append("-" * 60)
    lines.append("CHECKLIST")
    lines.append("-" * 60)

    for section, items in result["checklist"].items():
        lines.append("")
        lines.append(f"### {section}")
        for item in items:
            lines.append(f"  [ ] [{item['tag']}] {item['check']}")

    lines.append("")
    lines.append("=" * 60)
    lines.append(f"Total: {result['total_checklist_items']} items to verify")
    lines.append("=" * 60)
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Generate PR review checklists based on changed file types and patterns.",
        epilog="Example: git diff --name-only main...HEAD | python review_checklist_generator.py",
    )
    group = parser.add_mutually_exclusive_group()
    group.add_argument(
        "--files", "-f",
        nargs="+",
        help="List of changed file paths to analyze.",
    )
    group.add_argument(
        "--diff-file", "-d",
        help="Path to a unified diff file (extracts file paths from diff headers).",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results as JSON.",
    )
    args = parser.parse_args()

    if args.files:
        file_paths = args.files
    elif args.diff_file:
        try:
            with open(args.diff_file, "r", encoding="utf-8", errors="replace") as fh:
                diff_text = fh.read()
        except FileNotFoundError:
            print(f"Error: File not found: {args.diff_file}", file=sys.stderr)
            sys.exit(1)
        file_paths = extract_files_from_diff(diff_text)
    else:
        if sys.stdin.isatty():
            print("Reading file paths from stdin (one per line)...", file=sys.stderr)
        file_paths = [line.strip() for line in sys.stdin if line.strip()]

    if not file_paths:
        print("Error: No file paths provided.", file=sys.stderr)
        sys.exit(1)

    result = generate_report(file_paths)

    if args.json_output:
        print(json.dumps(result, indent=2))
    else:
        print(format_human(result))


if __name__ == "__main__":
    main()