Update app.py

app.py

@@ -1,13 +1,12 @@
 from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel
-import requests
+import aiohttp
 import os
 from datetime import datetime, timedelta
 from groq import Groq
 from dotenv import load_dotenv
 import logging
 import asyncio
-import aiohttp
 from concurrent.futures import ThreadPoolExecutor
 from functools import partial
 
@@ -35,13 +34,17 @@ REPOSITORIES = [
     "SlimKQL/Hunting-Queries-Detection-Rules"
 ]
 DAYS_BACK = 1
-TIMEOUT = 10  # seconds for each request
+TIMEOUT = 10  # seconds
 
 # GitHub API base URL
 GITHUB_API_URL = "https://api.github.com"
 
 # Groq client setup
-groq_client = Groq(api_key=GROQ_API_KEY)
+try:
+    groq_client = Groq(api_key=GROQ_API_KEY)
+except Exception as e:
+    logger.error(f"Failed to initialize Groq client: {e}")
+    groq_client = None
 
 # FastAPI app
 app = FastAPI(docs_url=None, redoc_url=None)
@@ -53,65 +56,76 @@ class RepositoryDetails(BaseModel):
     description: str
     context: str
 
-async def fetch_with_aiohttp(url: str, headers: dict, params: dict, session: aiohttp.ClientSession) -> dict:
+async def fetch_with_aiohttp(url: str, headers: dict, params: dict, session: aiohttp.ClientSession) -> list:
     """Asynchronous HTTP GET request with timeout."""
     try:
         async with session.get(url, headers=headers, params=params, timeout=aiohttp.ClientTimeout(total=TIMEOUT)) as response:
             response.raise_for_status()
-            return await response.json()
-    except Exception as e:
+            data = await response.json()
+            return data if isinstance(data, list) else []
+    except aiohttp.ClientError as e:
         logger.error(f"Error fetching {url}: {e}")
         return []
+    except Exception as e:
+        logger.error(f"Unexpected error fetching {url}: {e}")
+        return []
 
 async def fetch_repository_changes(repo: str, days_back: int) -> list[str]:
     """Fetch recent commits and pull requests for a repository asynchronously."""
+    logger.debug(f"Fetching changes for repository: {repo}")
+    since_date = (datetime.now() - timedelta(days=days_back)).isoformat()
+    headers = {
+        "Authorization": f"token {GITHUB_TOKEN}",
+        "Accept": "application/vnd.github.v3+json"
+    }
+
+    async with aiohttp.ClientSession() as session:
+        # Fetch commits
+        commits_url = f"{GITHUB_API_URL}/repos/{repo}/commits"
+        commits_params = {"since": since_date, "per_page": 100}
+        logger.debug(f"Fetching commits from: {commits_url}")
+        commits = await fetch_with_aiohttp(commits_url, headers, commits_params, session)
+        logger.debug(f"Found {len(commits)} commits for {repo}")
+
+        # Fetch pull requests
+        prs_url = f"{GITHUB_API_URL}/repos/{repo}/pulls"
+        prs_params = {"state": "all", "sort": "updated", "direction": "desc", "per_page": 100}
+        logger.debug(f"Fetching pull requests from: {prs_url}")
+        prs = await fetch_with_aiohttp(prs_url, headers, prs_params, session)
+        logger.debug(f"Found {len(prs)} pull requests for {repo}")
+
+    # Extract changes
+    changes = []
     try:
-        logger.debug(f"Fetching changes for repository: {repo}")
-        since_date = (datetime.now() - timedelta(days=days_back)).isoformat()
-        headers = {
-            "Authorization": f"token {GITHUB_TOKEN}",
-            "Accept": "application/vnd.github.v3+json"
-        }
-
-        async with aiohttp.ClientSession() as session:
-            # Fetch commits
-            commits_url = f"{GITHUB_API_URL}/repos/{repo}/commits"
-            commits_params = {"since": since_date, "per_page": 100}  # Limit to 100 for efficiency
-            logger.debug(f"Fetching commits from: {commits_url}")
-            commits = await fetch_with_aiohttp(commits_url, headers, commits_params, session)
-            logger.debug(f"Found {len(commits)} commits for {repo}")
-
-            # Fetch pull requests
-            prs_url = f"{GITHUB_API_URL}/repos/{repo}/pulls"
-            prs_params = {"state": "all", "sort": "updated", "direction": "desc", "per_page": 100}
-            logger.debug(f"Fetching pull requests from: {prs_url}")
-            prs = await fetch_with_aiohttp(prs_url, headers, prs_params, session)
-            logger.debug(f"Found {len(prs)} pull requests for {repo}")
-
-        # Extract changes
-        changes = []
         for commit in commits:
-            changes.append(f"Commit: {commit['commit']['message']}")
-        cutoff_date = datetime.now() - timedelta(days=days_back)
-        for pr in prs:
-            updated_at = datetime.strptime(pr["updated_at"], "%Y-%m-%dT%H:%M:%SZ")
-            if updated_at >= cutoff_date:
-                changes.append(f"PR: {pr['title']} - {pr['body'] or 'No description'}")
+            if "commit" in commit and "message" in commit["commit"]:
+                changes.append(f"Commit: {commit['commit']['message']}")
+    except (TypeError, KeyError) as e:
+        logger.error(f"Error processing commits for {repo}: {e}")
 
-        logger.debug(f"Total changes for {repo}: {len(changes)}")
-        return changes
+    cutoff_date = datetime.now() - timedelta(days=days_back)
+    try:
+        for pr in prs:
+            if "updated_at" in pr and pr["updated_at"]:
+                updated_at = datetime.strptime(pr["updated_at"], "%Y-%m-%dT%H:%M:%SZ")
+                if updated_at >= cutoff_date:
+                    changes.append(f"PR: {pr.get('title', 'No title')} - {pr.get('body', 'No description')}")
+    except (TypeError, KeyError, ValueError) as e:
+        logger.error(f"Error processing PRs for {repo}: {e}")
 
-    except Exception as e:
-        logger.error(f"Error fetching changes for {repo}: {e}")
-        return []
+    logger.debug(f"Total changes for {repo}: {len(changes)}")
+    return changes
 
 def summarize_changes_with_deepseek(repo: str, changes: list[str]) -> dict:
-    """Use Groq's DeepSeek model to summarize changes (synchronous due to Groq SDK limitations)."""
+    """Use Groq's DeepSeek model to summarize changes."""
+    if not groq_client:
+        return {"description": "Groq client not initialized.", "context": "API unavailable."}
+    
     try:
         logger.debug(f"Summarizing changes for repository: {repo}")
         prompt = f"""
         Analyze the following changes made to detection rules in the GitHub repository {repo}:
-        {', '.join(changes[:50])}  # Limit to 50 changes to avoid token overflow
+        {', '.join(changes[:50])}
         
         Provide a detailed response with two sections:
         - Description: Summarize what changes were made.
@@ -140,53 +154,58 @@ def summarize_changes_with_deepseek(repo: str, changes: list[str]) -> dict:
 
     except Exception as e:
         logger.error(f"Error summarizing changes for {repo}: {e}")
-        return {"description": "Error occurred.", "context": "Error occurred."}
+        return {"description": "Error occurred during summarization.", "context": str(e)}
 
 async def process_repository(repo: str, days_back: int) -> RepositoryDetails:
     """Process a single repository and return its details."""
-    changes = await fetch_repository_changes(repo, days_back)
-    if changes:
-        summary = await asyncio.get_event_loop().run_in_executor(
-            None, partial(summarize_changes_with_deepseek, repo, changes)
-        )
-        return RepositoryDetails(
-            repo_name=f"{repo} (+{len(changes)})",
-            repo_url=f"https://github.com/{repo}",
-            changes="\n".join(changes),
-            description=summary["description"],
-            context=summary["context"]
-        )
-    else:
+    try:
+        changes = await fetch_repository_changes(repo, days_back)
+        if changes:
+            summary = await asyncio.get_event_loop().run_in_executor(
+                None, partial(summarize_changes_with_deepseek, repo, changes)
+            )
+            return RepositoryDetails(
+                repo_name=f"{repo} (+{len(changes)})",
+                repo_url=f"https://github.com/{repo}",
+                changes="\n".join(changes),
+                description=summary["description"],
+                context=summary["context"]
+            )
+        else:
+            return RepositoryDetails(
+                repo_name=f"{repo} (No changes)",
+                repo_url=f"https://github.com/{repo}",
+                changes=f"No changes detected in the last {days_back} day(s).",
+                description="No changes detected.",
+                context="No context available."
+            )
+    except Exception as e:
+        logger.error(f"Error processing repository {repo}: {e}")
         return RepositoryDetails(
-            repo_name=f"{repo} (No changes)",
+            repo_name=f"{repo} (Error)",
             repo_url=f"https://github.com/{repo}",
-            changes=f"No changes detected in the last {days_back} day(s).",
-            description="No changes detected.",
-            context="No context available."
+            changes="Error occurred.",
+            description=str(e),
+            context="Processing failed."
         )
 
 @app.get("/monitor", response_model=list[RepositoryDetails])
 async def monitor_repositories():
     """Single API endpoint to fetch and summarize changes for all repositories."""
-    try:
-        logger.debug("Starting to monitor repositories")
-        tasks = [process_repository(repo, DAYS_BACK) for repo in REPOSITORIES]
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-        
-        # Handle any exceptions in results
-        final_results = []
-        for result in results:
-            if isinstance(result, Exception):
-                logger.error(f"Error processing a repository: {result}")
-                continue
-            final_results.append(result)
-
-        logger.debug("Finished monitoring repositories")
-        return final_results
-
-    except Exception as e:
-        logger.error(f"Error in monitor_repositories: {e}")
-        raise HTTPException(status_code=500, detail=str(e))
+    logger.debug("Starting to monitor repositories")
+    tasks = [process_repository(repo, DAYS_BACK) for repo in REPOSITORIES]
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+    
+    # Filter out exceptions and log them
+    final_results = []
+    for result in results:
+        if isinstance(result, Exception):
+            logger.error(f"Task failed with exception: {result}")
+            continue
+        final_results.append(result)
+
+    logger.debug(f"Finished monitoring repositories. Processed {len(final_results)} results.")
+    return final_results
 
 if __name__ == "__main__":
     import uvicorn