Spaces:
Sleeping
Sleeping
| import httpx | |
| import os | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| GITHUB_TOKEN = os.getenv("GITHUB_TOKEN") | |
| HEADERS = { | |
| "Authorization": f"Bearer {GITHUB_TOKEN}", | |
| "Accept": "application/vnd.github.v3+json" | |
| } | |
| async def fetch_pr_files(owner: str, repo: str, pr_number: int): | |
| url = f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}/files" | |
| async with httpx.AsyncClient() as client: | |
| response = await client.get(url, headers=HEADERS) | |
| response.raise_for_status() | |
| return response.json() | |
| async def post_pr_comment(owner: str, repo: str, pr_number: int, body: str): | |
| url = f"https://api.github.com/repos/{owner}/{repo}/issues/{pr_number}/comments" | |
| async with httpx.AsyncClient() as client: | |
| response = await client.post(url, headers=HEADERS, json={"body": body}) | |
| response.raise_for_status() | |
| return response.json() | |
| async def post_file_review(owner: str, repo: str, pr_number: int, filename: str, ai_feedback: list): | |
| """ | |
| Posts a review comment for a single file immediately after analysis. | |
| """ | |
| logger.info(f"📝 Formatting review for {filename} ({len(ai_feedback)} comments)...") | |
| review_body = f"### 🤖 AI Review: `{filename}`\n\n" | |
| if not ai_feedback: | |
| review_body += "✅ No critical issues found in this file.\n\n" | |
| else: | |
| for c in ai_feedback: | |
| line = c.get("line", 0) | |
| severity = c.get("severity", "info").capitalize() | |
| comment = c.get("comment", "").strip() | |
| emoji_map = {"low": "🟢", "medium": "🟠", "high": "🔴", "info": "💡"} | |
| emoji = emoji_map.get(c.get("severity", "").lower(), "💡") | |
| review_body += f"{emoji} **{severity}** (Line {line}): {comment}\n\n" | |
| review_body += "_Posted by PRism AI Reviewer_ 🤖" | |
| # Post the comment | |
| result = await post_pr_comment(owner, repo, pr_number, review_body) | |
| logger.info(f"✅ Posted review for {filename} (ID: {result.get('id', 'unknown')})") | |
| return result | |
| async def post_review_summary(owner: str, repo: str, pr_number: int, review_comments: list): | |
| """ | |
| Formats all AI feedback into a single markdown summary | |
| and posts it as a PR comment. | |
| """ | |
| logger.info(f"📝 Formatting review summary ({len(review_comments)} comments)...") | |
| if not review_comments: | |
| summary_body = "✅ **No issue found by AI reviewer!**" | |
| logger.info("✨ No issues found - posting clean review") | |
| else: | |
| summary_body = "### 🤖 AI Code Review Summary\n\n" | |
| summary_body += "The following findings were automatically generated by the AI reviewer:\n\n" | |
| for c in review_comments: | |
| file = c.get("file", "unknown") | |
| line = c.get("line", 0) | |
| severity = c.get("severity", "info").capitalize() | |
| comment = c.get("comment", "").strip() | |
| emoji = {"Low": "🟢", "Medium": "🟠", "High": "🔴"}.get(c.get("severity", "").lower(), "💡") | |
| summary_body += f"{emoji} **{severity}** — `{file}` (L{line}): {comment}\n" | |
| summary_body += "\n---\n_This review was auto-generated by **PRism AI Reviewer**._ 🤖" | |
| logger.info(f"📋 Summary length: {len(summary_body)} characters") | |
| # Post the formatted comment | |
| logger.info(f"🚀 Posting comment to PR #{pr_number}...") | |
| result = await post_pr_comment(owner, repo, pr_number, summary_body) | |
| logger.info(f"✅ Comment posted successfully (ID: {result.get('id', 'unknown')})") | |
| return result | |