awesome-kubernetes/src/pr_guardian.py

import os
import re
import sys
import httpx
import asyncio
import subprocess
from github import Github
from src.config import GH_TOKEN, TARGET_REPO
from src.gemini_utils import call_gemini_with_retry

async def analyze_pr_diff(diff_text: str) -> str:
    prompt = f"""
    You are the Nubenetes PR Guardian. Analyze this PR diff against the Nubenetes GEMINI.md mandates.
    Check for:
    1. Are new links added using the high-density description standard? (No generic clickbait)
    2. Are URLs normalized? (No tracking params)
    3. Are there emojis in H2/H3 titles? (Should be rejected)
    
    If the PR looks good, say '✅ PR Meets Nubenetes Standards.'
    If it violates mandates, list the violations clearly so the contributor can fix them.
    
    PR Diff:
    {diff_text}
    """
    try:
        # Mandate 50: Use Flash/Lite for rapid auditing and specify text format to avoid JSON parsing loops
        response = await call_gemini_with_retry(prompt, prefer_flash=True, response_format="text", role="PR-Guardian")
        await asyncio.sleep(1.0) # Safety delay to respect global RPM quota
        return str(response)
    except Exception as e:
        return f"⚠️ Guardian AI temporarily unavailable: {e}"

def auto_format_file(filepath: str):
    if not os.path.exists(filepath):
        return
    with open(filepath, "r", encoding="utf-8") as f:
        content = f.read()
    
    original_content = content
    
    # 1. Clean URLs
    from src.gemini_utils import normalize_url
    def repl_link(match):
        text = match.group(1)
        url = match.group(2)
        if url.startswith(("http://", "https://")):
            try:
                cleaned = normalize_url(url)
                return f"[{text}]({cleaned})"
            except Exception as e:
                print(f"[WARN] URL normalization failed for '{url[:60]}': {str(e)[:100]}")
                return match.group(0)
        return match.group(0)
    
    content = re.sub(r'\[([^\]]+)\]\((https?://[^\)]+)\)', repl_link, content)
    
    # 2. Clean headers (e.g., ampersands to "and", strip emojis)
    lines = content.splitlines()
    new_lines = []
    for line in lines:
        if line.strip().startswith("#"):
            line = line.replace("&", "and")
            line = re.sub(r'[\U00010000-\U0010ffff]', '', line)
            for char in "🧠☁️🌟🛡️🧪💎🕵️✨⚠️🔴🟡✅📍🎥":
                line = line.replace(char, "")
            line = re.sub(r'\s+', ' ', line).strip()
            match = re.match(r'^(#+)\s*(.*)', line)
            if match:
                level, title = match.groups()
                line = f"{level} {title.strip()}"
        new_lines.append(line)
    
    content = "\n".join(new_lines) + ("\n" if original_content.endswith("\n") else "")
    
    # 3. Clean <center> tags (Mandate 19)
    content = re.sub(r'<center(?! markdown="1")>', '<center markdown="1">', content)
    
    if content != original_content:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content)
        print(f"[*] Auto-formatted file: {filepath}")

def main():
    pr_number = os.getenv("PR_NUMBER")
    if not pr_number:
        print("No PR_NUMBER provided.")
        return

    g = Github(GH_TOKEN)
    repo = g.get_repo(TARGET_REPO)
    pr = repo.get_pull(int(pr_number))

    # Get changed files in PR and run auto-formatting
    try:
        files = pr.get_files()
        modified_files = [f.filename for f in files if f.status in ["modified", "added"]]
        for filepath in modified_files:
            # The v2-docs/ tree is generated by src/v2_optimizer.py, which is its
            # single source of truth for formatting. Auto-formatting it here
            # ("&"->"and" in headings, URL trailing-slash stripping) only fights
            # the generator: the rewrite lands on develop via the PR while master
            # gets the generator's raw form via release merges, producing a
            # perpetual cosmetic develop<->master drift. Skip the generated tree.
            if filepath.endswith(".md") and not filepath.startswith("v2-docs/"):
                auto_format_file(filepath)
    except Exception as e:
        print(f"Warning: Failed to fetch files for auto-formatting: {e}")

    # Fetch base branch and get the updated diff (with auto-fixes applied)
    diff_text = ""
    try:
        subprocess.run(["git", "fetch", "origin", pr.base.ref, "--depth=1"], capture_output=True, check=True)
        diff_process = subprocess.run(["git", "diff", f"origin/{pr.base.ref}"], capture_output=True, text=True, check=True)
        diff_text = diff_process.stdout
    except Exception as e:
        print(f"Warning: Failed to get git diff locally: {e}.")

    if not diff_text.strip():
        print("Diff is empty locally, falling back to GitHub PR diff.")
        try:
            # Mandate: URL Expansion & Normalization: Use follow_redirects=True for httpx
            diff_text = httpx.get(pr.diff_url, follow_redirects=True).text
        except Exception as e:
            print(f"Failed to fetch GitHub PR diff: {e}")

    # Skip if too huge or not markdown
    if len(diff_text) > 100000:
        print("Diff is too large for AI analysis (>100KB). Skipping detailed AI check, but auto-formatting was completed.")
        return
        
    analysis = asyncio.run(analyze_pr_diff(diff_text))
    
    # Check if comment already exists to avoid duplication
    bot_signature = "<!-- PR_GUARDIAN_SIGNATURE -->"
    existing_comment = None
    for comment in pr.get_issue_comments():
        if bot_signature in comment.body:
            existing_comment = comment
            break

    final_comment = f"{bot_signature}\n### 🛡️ Nubenetes PR Guardian\n\n{analysis}"
    
    if existing_comment:
        existing_comment.edit(final_comment)
        print("Existing Guardian comment updated.")
    else:
        pr.create_issue_comment(final_comment)
        print("Guardian comment posted.")

    # Determine if validation passed
    clean_analysis = analysis.strip().lower()
    if "pr meets nubenetes standards" in clean_analysis:
        print("✅ PR Meets Nubenetes Standards!")
        sys.exit(0)
    else:
        print("❌ PR violates Nubenetes mandates. Exiting with 1.")
        sys.exit(1)

if __name__ == "__main__":
    main()