compounding-intelligence/scripts/review_comment_generator.py

#!/usr/bin/env python3
"""
Review Comment Generator — Issue #126
Reads JSONL findings, deduplicates, posts as Gitea PR comments.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import urllib.request
import urllib.error
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Optional

SCRIPT_DIR = Path(__file__).resolve().parent
REPO_ROOT = SCRIPT_DIR.parent

DEFAULT_API_BASE = os.environ.get(
    "GITEA_API_BASE",
    "https://forge.alexanderwhitestone.com"
)
TOKEN_PATHS = [
    os.path.expanduser("~/.config/gitea/token"),
    os.path.expanduser("~/.hermes/gitea.token"),
    os.environ.get("GITEA_TOKEN", ""),
]

def load_token() -> Optional[str]:
    token = os.environ.get("GITEA_TOKEN", "")
    if token:
        return token
    for path in TOKEN_PATHS:
        if path and os.path.exists(path):
            with open(path) as f:
                t = f.read().strip()
                if t:
                    return t
    return None

class GiteaClient:
    def __init__(self, base_url: str, token: str, org: str, repo: str):
        self.base_url = base_url.rstrip("/")
        self.token = token
        self.org = org
        self.repo = repo

    def _post(self, path: str, data: Dict) -> Optional[Dict]:
        url = f"{self.base_url}/api/v1{path}"
        body = json.dumps(data).encode("utf-8")
        req = urllib.request.Request(url, data=body, method="POST")
        req.add_header("Authorization", f"token {self.token}")
        req.add_header("Content-Type", "application/json")
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                return json.loads(resp.read().decode())
        except urllib.error.HTTPError as e:
            err = e.read().decode() if e.read() else str(e)
            print(f"[ERROR] HTTP {e.code}: {err}", file=sys.stderr)
            return None
        except Exception as e:
            print(f"[ERROR] {e}", file=sys.stderr)
            return None

    def post_issue_comment(self, issue_num: int, body: str) -> Optional[Dict]:
        return self._post(
            f"/repos/{self.org}/{self.repo}/issues/{issue_num}/comments",
            {"body": body}
        )

def content_hash(finding: Dict) -> str:
    key = f"{finding['file']}:{finding['line']}:{finding['text']}"
    return hashlib.sha256(key.encode("utf-8")).hexdigest()

def format_comment(finding: Dict) -> str:
    emoji = {
        "error": "🛑",
        "warning": "⚠️",
        "info": "ℹ️",
    }.get(finding.get("severity", ""), "📝")
    f = finding["file"]
    ln = finding["line"]
    txt = finding["text"]
    return f"{emoji} **Review Comment**\n\nFile: `{f}`\nLine: {ln}\n\n> {txt}\n"

def load_findings(path: Optional[Path], from_stdin: bool) -> List[Dict]:
    import fileinput
    findings = []
    sources = ["-"] if from_stdin else [str(path)]
    for line in fileinput.input(files=sources):
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        try:
            f = json.loads(line)
            for key in ("file", "line", "text"):
                if key not in f:
                    raise ValueError(f"Missing key: {key}")
            findings.append(f)
        except json.JSONDecodeError as e:
            print(f"WARNING: Skipping invalid JSON: {e}", file=sys.stderr)
    return findings

def main() -> int:
    parser = argparse.ArgumentParser(
        description="Post review findings as comments to a Gitea PR/issue"
    )
    parser.add_argument("--pr", type=int, required=True, help="PR/issue number")
    parser.add_argument("--org", default="Timmy_Foundation", help="Gitea org")
    parser.add_argument("--repo", default="compounding-intelligence", help="Repo name")
    parser.add_argument("--api-base", default=DEFAULT_API_BASE, help="Gitea API base")
    parser.add_argument("--token", default=None, help="API token (or env/file)")
    parser.add_argument("--input", type=Path, default=None, help="JSONL input file")
    parser.add_argument("--stdin", action="store_true", help="Read from stdin")
    parser.add_argument("--dry-run", action="store_true", help="Show without posting")
    parser.add_argument("--json", action="store_true", help="Emit JSON report")

    args = parser.parse_args()

    if not args.stdin and args.input is None:
        print("ERROR: --input or --stdin required", file=sys.stderr)
        return 1
    if args.stdin and args.input:
        print("ERROR: --stdin and --input exclusive", file=sys.stderr)
        return 1

    token = args.token or load_token()
    if not token:
        print("ERROR: Token not found. Set GITEA_TOKEN or ~/.config/gitea/token", file=sys.stderr)
        return 1

    findings = load_findings(args.input, args.stdin)
    if not findings:
        print("ERROR: No findings loaded", file=sys.stderr)
        return 1

    if not args.json: print(f"Loaded {len(findings)} finding(s)")

    seen: Dict[str, Dict] = {}
    for f in findings:
        h = content_hash(f)
        if h not in seen:
            seen[h] = f

    unique = list(seen.values())
    if not args.json: print(f"After dedup: {len(unique)} unique")

    if args.json:
        report = {
            "total": len(findings),
            "unique": len(unique),
            "findings": unique,
            "generated_at": datetime.now(timezone.utc).isoformat(),
        }
        print(json.dumps(report, indent=2))
        return 0

    if args.dry_run:
        print("\n=== DRY RUN — would post ===")
        for i, f in enumerate(unique, 1):
            print(f"\n--- Comment {i}/{len(unique)} ---")
            print(format_comment(f))
        return 0

    client = GiteaClient(args.api_base, token, args.org, args.repo)
    posted = 0
    for f in unique:
        body = format_comment(f)
        result = client.post_issue_comment(args.pr, body)
        if result:
            print(f"✅ Posted: {f['file']}:{f['line']} (id={result.get('id')})")
            posted += 1
        else:
            print(f"❌ Failed: {f['file']}:{f['line']}")

    print(f"\nPosted {posted}/{len(unique)} to PR #{args.pr}")
    return 0 if posted == len(unique) else 1

if __name__ == "__main__":
    sys.exit(main())