Update app.py

app.py

@@ -6,6 +6,10 @@ 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
 
 # Configure logging
 logging.basicConfig(level=logging.DEBUG)
@@ -14,9 +18,9 @@ logger = logging.getLogger(__name__)
 # Load environment variables
 load_dotenv()
 
-# Configuration (Use environment variables instead of hardcoding)
-GITHUB_TOKEN = "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"
-GROQ_API_KEY = "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"
+# Configuration
+GITHUB_TOKEN = os.getenv("GITHUB_TOKEN", "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")
+GROQ_API_KEY = os.getenv("GROQ_API_KEY", "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")
 REPOSITORIES = [
     "falcosecurity/rules",
     "SigmaHQ/sigma",
@@ -31,6 +35,7 @@ REPOSITORIES = [
     "SlimKQL/Hunting-Queries-Detection-Rules"
 ]
 DAYS_BACK = 1
+TIMEOUT = 10  # seconds for each request
 
 # GitHub API base URL
 GITHUB_API_URL = "https://api.github.com"
@@ -48,10 +53,18 @@ class RepositoryDetails(BaseModel):
     description: str
     context: str
 
-def fetch_repository_changes(repo: str, days_back: int) -> list[str]:
-    """
-    Fetch recent commits and pull requests for a repository.
-    """
+async def fetch_with_aiohttp(url: str, headers: dict, params: dict, session: aiohttp.ClientSession) -> dict:
+    """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:
+        logger.error(f"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."""
     try:
         logger.debug(f"Fetching changes for repository: {repo}")
         since_date = (datetime.now() - timedelta(days=days_back)).isoformat()
@@ -60,55 +73,51 @@ def fetch_repository_changes(repo: str, days_back: int) -> list[str]:
             "Accept": "application/vnd.github.v3+json"
         }
 
-        # Fetch commits
-        commits_url = f"{GITHUB_API_URL}/repos/{repo}/commits"
-        commits_params = {"since": since_date}
-        logger.debug(f"Fetching commits from: {commits_url} with params: {commits_params}")
-        commits_response = requests.get(commits_url, headers=headers, params=commits_params)
-        commits_response.raise_for_status()
-        commits = commits_response.json()
-        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"}
-        logger.debug(f"Fetching pull requests from: {prs_url} with params: {prs_params}")
-        prs_response = requests.get(prs_url, headers=headers, params=prs_params)
-        prs_response.raise_for_status()
-        prs = prs_response.json()
-        logger.debug(f"Found {len(prs)} pull requests for {repo}")
+        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 >= datetime.now() - timedelta(days=days_back):
+            if updated_at >= cutoff_date:
                 changes.append(f"PR: {pr['title']} - {pr['body'] or 'No description'}")
 
         logger.debug(f"Total changes for {repo}: {len(changes)}")
         return changes
 
-    except requests.exceptions.RequestException as e:
+    except Exception as e:
         logger.error(f"Error fetching changes for {repo}: {e}")
-        raise HTTPException(status_code=500, detail=f"Error fetching changes for {repo}: {e}")
+        return []
 
 def summarize_changes_with_deepseek(repo: str, changes: list[str]) -> dict:
-    """
-    Use Groq's DeepSeek model to summarize changes and provide insights.
-    """
+    """Use Groq's DeepSeek model to summarize changes (synchronous due to Groq SDK limitations)."""
     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)}
+        {', '.join(changes[:50])}  # Limit to 50 changes to avoid token overflow
         
         Provide a detailed response with two sections:
         - Description: Summarize what changes were made.
         - Context: Explain why these changes might be required.
         """
-        logger.debug(f"Sending prompt to DeepSeek: {prompt[:100]}...")  # Truncate for brevity in logs
+        logger.debug(f"Sending prompt to DeepSeek: {prompt[:100]}...")
         response = groq_client.chat.completions.create(
             model="deepseek-chat",
             messages=[{"role": "user", "content": prompt}],
@@ -125,48 +134,55 @@ def summarize_changes_with_deepseek(repo: str, changes: list[str]) -> dict:
             description = summary.split("Description:")[1].split("Context:")[0].strip()
             context = summary.split("Context:")[1].strip()
         else:
-            description = summary  # Fallback to full summary if sections aren't clear
+            description = summary
 
         return {"description": description, "context": context}
 
     except Exception as e:
         logger.error(f"Error summarizing changes for {repo}: {e}")
-        raise HTTPException(status_code=500, detail=f"Error summarizing changes for {repo}: {e}")
+        return {"description": "Error occurred.", "context": "Error occurred."}
+
+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:
+        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."
+        )
 
 @app.get("/monitor", response_model=list[RepositoryDetails])
 async def monitor_repositories():
-    """
-    Single API endpoint to fetch and summarize changes for all repositories.
-    """
+    """Single API endpoint to fetch and summarize changes for all repositories."""
     try:
         logger.debug("Starting to monitor repositories")
-        results = []
-        for repo in REPOSITORIES:
-            logger.debug(f"Processing repository: {repo}")
-            changes = fetch_repository_changes(repo, DAYS_BACK)
-
-            if changes:
-                logger.debug(f"Summarizing changes for {repo}")
-                summary = summarize_changes_with_deepseek(repo, changes)
-                results.append(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:
-                logger.debug(f"No changes detected for {repo}")
-                results.append(RepositoryDetails(
-                    repo_name=f"{repo} (No changes)",
-                    repo_url=f"https://github.com/{repo}",
-                    changes="No changes detected in the last 7 days.",
-                    description="No changes detected.",
-                    context="No context available."
-                ))
+        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 results
+        return final_results
 
     except Exception as e:
         logger.error(f"Error in monitor_repositories: {e}")