import os
import sys
import tempfile
import shutil
import requests
import time
import subprocess
import json
import traceback
import argparse
import re
import warnings
import numpy as np
from collections import defaultdict
from pathlib import Path
from urllib.parse import urlparse, quote
from github import Github
from git import Repo
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.exceptions import ConvergenceWarning
def run_semgrep(repo_path):
try:
result = subprocess.run(
["semgrep", "--config", "auto", "--json", repo_path],
capture_output=True,
text=True,
check=True
)
return json.loads(result.stdout)
except subprocess.CalledProcessError as e:
print(f"Semgrep error: {e}")
return None
except json.JSONDecodeError:
print("Failed to parse Semgrep output")
return None
def get_repo_info(input_str):
if input_str.startswith("http") or input_str.startswith("https"):
parsed_url = urlparse(input_str)
path_parts = parsed_url.path.strip("/").split("/")
return path_parts[0], path_parts[1]
else:
return input_str.split("/")
def clone_repo(owner, repo_name, temp_dir, repo_url):
repo_url = f"{repo_url}.git"
# Clone the repository
repo = Repo.clone_from(repo_url, temp_dir)
# Get the default branch
default_branch = repo.active_branch.name
# Get the full name of the repository (owner/repo_name)
full_name = f"{owner}/{repo_name}"
# Create the repo_info object
repo_info = {
"owner": owner,
"repo_name": repo_name,
"full_name": full_name,
"repo_url": repo_url,
"local_path": temp_dir,
"default_branch": default_branch,
"git_repo": repo
}
return repo_info
def analyze_code(repo_path):
file_types = defaultdict(int)
file_contents = {}
for root, _, files in os.walk(repo_path):
for file in files:
file_path = os.path.join(root, file)
_, ext = os.path.splitext(file)
file_types[ext] += 1
if ext in ['.py', '.js', '.java', '.cpp', '.cs', '.go', '.rb', '.php', 'ts', 'tsx', 'jsx']:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
file_contents[file_path] = f.read()
semgrep_results = run_semgrep(repo_path)
return {
"file_types": dict(file_types),
"file_contents": file_contents,
"semgrep_results": semgrep_results
}
def analyze_issues(github_repo, max_issues):
closed_issues = []
open_issues = []
for issue in github_repo.get_issues(state="all")[:max_issues]:
issue_data = {
"number": issue.number,
"title": issue.title,
"body": issue.body,
"state": issue.state,
"created_at": issue.created_at.isoformat(),
"closed_at": issue.closed_at.isoformat() if issue.closed_at else None,
"comments": []
}
for comment in issue.get_comments():
issue_data["comments"].append({
"body": comment.body,
"created_at": comment.created_at.isoformat()
})
if issue.state == "closed":
closed_issues.append(issue_data)
else:
open_issues.append(issue_data)
time.sleep(1) # Rate limiting
# Cluster and filter closed issues
if closed_issues:
filtered_closed_issues = cluster_and_filter_items(closed_issues, n_clusters=min(5, len(closed_issues)), n_items=min(10, len(closed_issues)))
else:
filtered_closed_issues = []
return {
'closed_issues': closed_issues,
'open_issues': open_issues,
'filtered_closed_issues': filtered_closed_issues
}
def analyze_pull_requests(github_repo, max_prs):
closed_prs = []
open_prs = []
for pr in github_repo.get_pulls(state="all")[:max_prs]:
pr_data = {
"number": pr.number,
"title": pr.title,
"body": pr.body,
"state": pr.state,
"created_at": pr.created_at.isoformat(),
"closed_at": pr.closed_at.isoformat() if pr.closed_at else None,
"comments": [],
"diff": pr.get_files()
}
for comment in pr.get_comments():
pr_data["comments"].append({
"body": comment.body,
"created_at": comment.created_at.isoformat()
})
if pr.state == "closed":
closed_prs.append(pr_data)
else:
open_prs.append(pr_data)
time.sleep(1) # Rate limiting
# Cluster and filter closed PRs
if closed_prs:
filtered_closed_prs = cluster_and_filter_items(closed_prs, n_clusters=min(5, len(closed_prs)), n_items=min(10, len(closed_prs)))
else:
filtered_closed_prs = []
return {
'closed_prs': closed_prs,
'open_prs': open_prs,
'filtered_closed_prs': filtered_closed_prs
}
def call_llm(client, prompt, model="nousresearch/hermes-3-llama-3.1-405b:free", max_tokens=4096):
response = client.chat.completions.create(
extra_headers={
"HTTP-Referer": "https://patched.codes",
"X-Title": "Patched GitHub Repo Analyzer"
},
model=model,
messages=[
{"role": "user", "content": prompt}
],
max_tokens=max_tokens,
)
return response.choices[0].message.content
def safe_call_llm(client, prompt, retries=3):
for attempt in range(retries):
try:
response = call_llm(client, prompt)
return parse_llm_response(response)
except Exception as e:
print(f"Error in LLM call (attempt {attempt + 1}/{retries}): {str(e)}")
if attempt == retries - 1:
print("All retries failed. Returning empty list.")
return []
return []
def parse_llm_response(response):
# Pattern to match JSON content within triple backticks, with or without 'json' specifier
matches = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', response)
if matches:
# If we found matches, use the first one (assuming there's only one JSON block)
response = matches.group(1)
try:
# First, try to parse the entire response as JSON
return json.loads(response)
except json.JSONDecodeError:
# If that fails, try to extract JSON from the response
try:
start = response.index('[')
end = response.rindex(']') + 1
json_str = response[start:end]
return json.loads(json_str)
except (ValueError, json.JSONDecodeError):
print(f"Warning: Failed to parse LLM response as JSON. Response: {response}...")
return []
def cluster_and_filter_items(items, n_clusters=5, n_items=10):
if len(items) == 0:
return []
if len(items) <= n_items:
return items
# Combine title and body for text analysis
texts = [f"{item['title']} {item['body']}" for item in items]
# Create TF-IDF vectors
vectorizer = TfidfVectorizer(stop_words='english', max_features=1000)
tfidf_matrix = vectorizer.fit_transform(texts)
# Determine the number of clusters
n_clusters = min(n_clusters, len(items))
# Perform clustering
with warnings.catch_warnings():
warnings.filterwarnings("ignore", category=ConvergenceWarning)
kmeans = KMeans(n_clusters=n_clusters)
kmeans.fit(tfidf_matrix)
# Get cluster centers
cluster_centers = kmeans.cluster_centers_
# Find items closest to cluster centers
filtered_items = []
for i in range(n_clusters):
cluster_items = [item for item, label in zip(items, kmeans.labels_) if label == i]
cluster_vectors = tfidf_matrix[kmeans.labels_ == i]
if cluster_vectors.shape[0] == 0:
continue
# Calculate similarities to cluster center
similarities = cosine_similarity(cluster_vectors, cluster_centers[i].reshape(1, -1)).flatten()
# Sort items by similarity and select top ones
sorted_items = [x for _, x in sorted(zip(similarities, cluster_items), key=lambda pair: pair[0], reverse=True)]
filtered_items.extend(sorted_items[:max(1, n_items // n_clusters)])
# If we didn't get enough items, add more from the original list
if len(filtered_items) < n_items:
remaining_items = [item for item in items if item not in filtered_items]
filtered_items.extend(remaining_items[:n_items - len(filtered_items)])
return filtered_items[:n_items]
def safe_filter_open_items(open_items, closed_patterns, n_items=10):
try:
# Combine title and body for text analysis
open_texts = [f"{item.get('title', '')} {item.get('body', '')}" for item in open_items]
pattern_texts = [f"{pattern.get('theme', '')} {pattern.get('description', '')}" for pattern in closed_patterns]
if not open_texts or not pattern_texts:
print("Warning: No open items or closed patterns to analyze.")
return []
# Create TF-IDF vectors
vectorizer = TfidfVectorizer(stop_words='english', max_features=1000)
tfidf_matrix = vectorizer.fit_transform(open_texts + pattern_texts)
# Split the matrix into open items and patterns
open_vectors = tfidf_matrix[:len(open_items)]
pattern_vectors = tfidf_matrix[len(open_items):]
# Calculate similarities between open items and patterns
similarities = cosine_similarity(open_vectors, pattern_vectors)
# Calculate the average similarity for each open item
avg_similarities = np.mean(similarities, axis=1)
# Sort open items by average similarity and select top ones
sorted_items = [x for _, x in sorted(zip(avg_similarities, open_items), key=lambda pair: pair[0], reverse=True)]
return sorted_items[:n_items]
except Exception as e:
print(f"Error in filtering open items: {str(e)}")
traceback.print_exc()
return open_items[:n_items] # Return first n_items if filtering fails
def filter_open_items(open_items, closed_patterns, n_items=10):
# Combine title and body for text analysis
open_texts = [f"{item['title']} {item['body']}" for item in open_items]
pattern_texts = [f"{pattern.get('theme', '')} {pattern.get('description', '')}" for pattern in closed_patterns]
# Create TF-IDF vectors
vectorizer = TfidfVectorizer(stop_words='english', max_features=1000)
tfidf_matrix = vectorizer.fit_transform(open_texts + pattern_texts)
# Split the matrix into open items and patterns
open_vectors = tfidf_matrix[:len(open_items)]
pattern_vectors = tfidf_matrix[len(open_items):]
# Calculate similarities between open items and patterns
similarities = cosine_similarity(open_vectors, pattern_vectors)
# Calculate the average similarity for each open item
avg_similarities = np.mean(similarities, axis=1)
# Sort open items by average similarity and select top ones
sorted_items = [x for _, x in sorted(zip(avg_similarities, open_items), key=lambda pair: pair[0], reverse=True)]
return sorted_items[:n_items]
def llm_analyze_closed_items(client, items, item_type):
prompt = f"""
Analyze the following closed GitHub {item_type}:
{items}
Based on these closed {item_type}, identify:
1. Common themes or recurring patterns
2. Areas where automation could streamline {item_type} management
3. Potential LLM-assisted workflows to improve the {item_type} process
4. Do not return anything other than the expected JSON object
For each identified pattern or theme, provide:
- A short title or theme name
- A brief description of the pattern
- Potential LLM-assisted solutions or workflows
Format your response as a list of JSON objects, like this:
[
{{
"theme": "Theme name",
"description": "Brief description of the pattern",
"llm_solution": "Potential LLM-assisted solution or workflow"
}},
...
]
"""
return safe_call_llm(client, prompt)
def llm_analyze_open_items(client, open_items, closed_patterns, item_type, repo_url):
prompt = f"""
Consider the following patterns identified in closed {item_type}:
{closed_patterns}
Now, analyze these open {item_type} in light of the above patterns:
{open_items}
For each open {item_type}:
1. Identify which pattern(s) it most closely matches
2. Suggest specific LLM-assisted workflows or automations that could be applied, based on the matched patterns
3. Explain how the suggested workflow would improve the handling of this {item_type}
4. Include the {item_type} number in your response
5. Do not return anything other than the expected JSON object
Format your response as a list of JSON objects, like this:
[
{{
"number": {item_type} number,
"matched_patterns": ["Pattern 1", "Pattern 2"],
"suggested_workflow": "Description of the suggested LLM-assisted workflow",
"expected_improvement": "Explanation of how this would improve the {item_type} handling"
}},
...
]
"""
return safe_call_llm(client, prompt)
def llm_analyze_issues(client, issues_data, repo_url):
filtered_closed_issues = issues_data['filtered_closed_issues']
all_closed_issues = issues_data['closed_issues']
open_issues = issues_data['open_issues']
closed_patterns = llm_analyze_closed_items(client, filtered_closed_issues, "issues")
relevant_open_issues = safe_filter_open_items(open_issues, closed_patterns, n_items=10)
open_issues_analysis = llm_analyze_open_items(client, relevant_open_issues, closed_patterns, "issues", repo_url)
summary_prompt = f"""
Summarize the analysis of closed and open issues:
Closed Issues Patterns:
{closed_patterns}
Open Issues Analysis:
{open_issues_analysis}
Provide a concise summary of:
1. Key patterns identified in closed issues
2. Most promising LLM-assisted workflows for handling open issues
3. Overall recommendations for improving issue management in this repository
4. For each suggested workflow, include the number of an open issue where it could be applied
5. Do not return anything other than the expected JSON object
Format your response as a JSON object with the following structure:
{{
"key_patterns": ["pattern1", "pattern2", ...],
"promising_workflows": [
{{
"workflow": "Description of the workflow",
"applicable_issue": issue_number
}},
...
],
"overall_recommendations": ["recommendation1", "recommendation2", ...]
}}
Total number of closed issues analyzed: {len(all_closed_issues)}
Total number of open issues: {len(open_issues)}
"""
summary = safe_call_llm(client, summary_prompt)
return {
'closed_patterns': closed_patterns,
'open_issues_analysis': open_issues_analysis,
'summary': summary
}
def llm_analyze_prs(client, prs_data, repo_url):
filtered_closed_prs = prs_data['filtered_closed_prs']
all_closed_prs = prs_data['closed_prs']
open_prs = prs_data['open_prs']
closed_patterns = llm_analyze_closed_items(client, filtered_closed_prs, "pull requests")
relevant_open_prs = safe_filter_open_items(open_prs, closed_patterns, n_items=10)
open_prs_analysis = llm_analyze_open_items(client, relevant_open_prs, closed_patterns, "pull requests", repo_url)
summary_prompt = f"""
Summarize the analysis of closed and open pull requests:
Closed PRs Patterns:
{closed_patterns}
Open PRs Analysis:
{open_prs_analysis}
Provide a concise summary of:
1. Key patterns identified in closed pull requests
2. Most promising LLM-assisted workflows for handling open pull requests
3. Overall recommendations for improving the PR process in this repository
4. For each suggested workflow, include the number of an open PR where it could be applied
5. Do not return anything other than the expected JSON object
Format your response as a JSON object with the following structure:
{{
"key_patterns": ["pattern1", "pattern2", ...],
"promising_workflows": [
{{
"workflow": "Description of the workflow",
"applicable_pr": pr_number
}},
...
],
"overall_recommendations": ["recommendation1", "recommendation2", ...]
}}
Total number of closed pull requests analyzed: {len(all_closed_prs)}
Total number of open pull requests: {len(open_prs)}
"""
summary = safe_call_llm(client, summary_prompt)
return {
'closed_patterns': closed_patterns,
'open_prs_analysis': open_prs_analysis,
'summary': summary
}
def llm_analyze_code(client, code_analysis):
semgrep_summary = "No Semgrep results available."
if code_analysis['semgrep_results']:
findings = code_analysis['semgrep_results'].get('results', [])
semgrep_summary = f"Semgrep found {len(findings)} potential issues:"
for finding in findings[:10]: # Limit to 10 findings to avoid token limits
semgrep_summary += f"\n- {finding['check_id']} in {finding['path']}: {finding['extra']['message']}"
file_contents_summary = ""
for file_path, content in code_analysis['file_contents'].items():
file_contents_summary += f"\n\nFile: {file_path}\nContent:\n{content[:1000]}..." # Limit content to avoid token limits
prompt = f"""
Analyze the following code structure, content, and Semgrep results:
File types: {code_analysis['file_types']}
Semgrep Analysis:
{semgrep_summary}
File Contents Summary:
{file_contents_summary}
Based on this information, provide an analysis covering:
1. Patterns in the codebase
2. Best practices being followed or missing
3. Areas for improvement
4. Potential security vulnerabilities or bugs (based on Semgrep results)
5. Opportunities for LLM-assisted automation in coding tasks
For LLM-assisted opportunities, consider tasks like code review, bug fixing, test generation, or documentation.
Respond ONLY with a JSON object in the following format:
{{
"patterns": ["pattern1", "pattern2", ...],
"best_practices": {{
"followed": ["practice1", "practice2", ...],
"missing": ["practice1", "practice2", ...]
}},
"areas_for_improvement": ["area1", "area2", ...],
"potential_vulnerabilities": [
{{
"description": "Description of the vulnerability",
"file_path": "Path to the affected file",
"severity": "High/Medium/Low"
}},
...
],
"llm_opportunities": [
{{
"task": "Description of the LLM-assisted task",
"file_path": "Path to the relevant file",
"improvement": "How LLM assistance would help"
}},
...
]
}}
Ensure your response is a valid JSON object and nothing else.
"""
return safe_call_llm(client, prompt)
def llm_synthesize_findings(client, code_analysis, issues_analysis, pr_analysis):
prompt = f"""
Synthesize the following analyses of a GitHub repository:
Code Analysis:
{code_analysis}
Issues Analysis:
{issues_analysis}
Pull Requests Analysis:
{pr_analysis}
Based on these analyses:
1. Summarize the key findings across all areas (code, issues, and PRs)
2. Identify the top 3-5 most promising opportunities for LLM-assisted workflows
3. For each opportunity, provide a specific example of how it could be implemented and the potential benefits
4. Suggest any additional areas of investigation or analysis that could provide further insights
"""
return call_llm(client, prompt, max_tokens=8192)
def generate_report(repo_info, code_analysis, issues_analysis, pr_analysis):
repo_url = f"https://github.com/{repo_info['owner']}/{repo_info['repo_name']}"
default_branch = repo_info.get('default_branch', 'master') # Assume 'master' if not provided
report = f"""# LLM-Assisted Workflow Analysis for {repo_info['owner']}/{repo_info['repo_name']}
## Repository Overview
- **Owner:** {repo_info['owner']}
- **Repository:** {repo_info['repo_name']}
- **URL:** [{repo_url}]({repo_url})
- **File types:** {', '.join(f"{ext} ({count})" for ext, count in code_analysis.get('file_types', {}).items())}
## Code Analysis
"""
if isinstance(code_analysis.get('llm_analysis'), dict):
code_llm_analysis = code_analysis['llm_analysis']
report += "### Patterns Identified\n"
for pattern in code_llm_analysis.get('patterns', []):
report += f"- {pattern}\n"
report += "\n### Best Practices\n"
report += "Followed
\n\n"
for practice in code_llm_analysis.get('best_practices', {}).get('followed', []):
report += f"- {practice}\n"
report += "Missing
\n\n"
for practice in code_llm_analysis.get('best_practices', {}).get('missing', []):
report += f"- {practice}\n"
report += "