wehub-resource-sync
497 lines
18 KiB
Python
497 lines
18 KiB
Python
import os
|
||
import re
|
||
import json
|
||
import requests
|
||
import datetime
|
||
import time
|
||
from pathlib import Path
|
||
|
||
from typing import Dict, List, Optional, Any
|
||
from models import GitHubProfile
|
||
from pdf import logger
|
||
from prompts.template_manager import TemplateManager
|
||
from prompt import DEFAULT_MODEL, MODEL_PARAMETERS
|
||
from llm_utils import initialize_llm_provider, extract_json_from_response
|
||
from config import DEVELOPMENT_MODE
|
||
|
||
|
||
def _create_cache_filename(api_url: str, params: dict = None) -> str:
|
||
url_parts = api_url.replace("https://api.github.com/", "").replace("/", "_")
|
||
|
||
if params:
|
||
param_str = "_".join([f"{k}_{v}" for k, v in sorted(params.items())])
|
||
filename = f"cache/gh_githubcache_{url_parts}_{param_str}.json"
|
||
else:
|
||
filename = f"cache/gh_githubcache_{url_parts}.json"
|
||
return filename
|
||
|
||
|
||
def _fetch_github_api(api_url, params=None):
|
||
headers = {}
|
||
github_token = os.environ.get("GITHUB_TOKEN")
|
||
if github_token:
|
||
headers["Authorization"] = f"token {github_token}"
|
||
|
||
cache_filename = _create_cache_filename(api_url, params)
|
||
if DEVELOPMENT_MODE and os.path.exists(cache_filename):
|
||
print(f"Loading cached GitHub data from {cache_filename}")
|
||
try:
|
||
cached_data = json.loads(Path(cache_filename).read_text(encoding="utf-8"))
|
||
if not cached_data:
|
||
raise ValueError("Cached data is empty")
|
||
return 200, cached_data
|
||
except Exception as e:
|
||
print(f"⚠️ Warning: Error reading cache file {cache_filename}: {e}")
|
||
try:
|
||
os.remove(cache_filename)
|
||
except Exception as delete_err:
|
||
print(
|
||
f"Failed to delete invalid cache file {cache_filename}: {delete_err}"
|
||
)
|
||
|
||
response = requests.get(api_url, params, timeout=10, headers=headers)
|
||
status_code = response.status_code
|
||
|
||
# Check GitHub rate limit headers
|
||
rate_limit_remaining = response.headers.get("X-RateLimit-Remaining")
|
||
rate_limit_limit = response.headers.get("X-RateLimit-Limit")
|
||
rate_limit_reset = response.headers.get("X-RateLimit-Reset")
|
||
logger.info(
|
||
f"{rate_limit_remaining}/{rate_limit_limit}. Reset at {rate_limit_reset}"
|
||
)
|
||
|
||
if rate_limit_remaining is not None and rate_limit_limit is not None:
|
||
remaining = int(rate_limit_remaining)
|
||
limit = int(rate_limit_limit)
|
||
|
||
# Log rate limit information and handle proactively
|
||
if remaining < 10 and rate_limit_reset:
|
||
reset_timestamp = int(rate_limit_reset)
|
||
current_timestamp = int(time.time())
|
||
wait_seconds = (
|
||
max(0, reset_timestamp - current_timestamp) + 5
|
||
) # Add 5 second buffer
|
||
reset_time = datetime.datetime.fromtimestamp(reset_timestamp)
|
||
|
||
# Cap maximum wait time at 1 hour
|
||
max_wait = 3600
|
||
if wait_seconds > max_wait:
|
||
print(
|
||
f"⚠️ Rate limit reset time is too far in the future ({wait_seconds}s). Capping wait to {max_wait}s"
|
||
)
|
||
wait_seconds = max_wait
|
||
|
||
logger.error(
|
||
f"⚠️ GitHub API rate limit low: {remaining}/{limit} requests remaining. Resets at {reset_time}"
|
||
)
|
||
print(
|
||
f"💡 Tip: Set GITHUB_TOKEN environment variable to increase rate limits (60/hour → 5000/hour)"
|
||
)
|
||
|
||
if wait_seconds > 0:
|
||
logger.info(
|
||
f"⏳ Proactively sleeping for {wait_seconds} seconds until rate limit resets..."
|
||
)
|
||
time.sleep(wait_seconds)
|
||
print(f"✅ Rate limit should be reset now. Continuing...")
|
||
elif remaining < 100:
|
||
logger.info(
|
||
f"ℹ️ GitHub API rate limit: {remaining}/{limit} requests remaining"
|
||
)
|
||
|
||
data = response.json() if response.status_code == 200 else {}
|
||
|
||
if DEVELOPMENT_MODE and status_code == 200:
|
||
try:
|
||
os.makedirs("cache", exist_ok=True)
|
||
Path(cache_filename).write_text(
|
||
json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"Error caching GitHub data to {cache_filename}: {e}")
|
||
|
||
return status_code, data
|
||
|
||
|
||
def extract_github_username(github_url: str) -> Optional[str]:
|
||
if not github_url:
|
||
return None
|
||
|
||
github_url = github_url.replace(" ", "")
|
||
github_url = github_url.strip()
|
||
|
||
patterns = [
|
||
r"https?://github\.com/([^/]+)",
|
||
r"github\.com/([^/]+)",
|
||
r"@([^/]+)",
|
||
r"^([a-zA-Z0-9-]+)$",
|
||
]
|
||
|
||
for pattern in patterns:
|
||
match = re.search(pattern, github_url)
|
||
if match:
|
||
username = match.group(1)
|
||
# Remove query parameters if present (e.g., "?tab=repositories")
|
||
if "?" in username:
|
||
username = username.split("?", 1)[0]
|
||
return username
|
||
return None
|
||
|
||
|
||
def fetch_github_profile(github_url: str) -> Optional[GitHubProfile]:
|
||
try:
|
||
username = extract_github_username(github_url)
|
||
logger.info(f"{username}")
|
||
if not username:
|
||
print(f"Could not extract username from: {github_url}")
|
||
return None
|
||
|
||
api_url = f"https://api.github.com/users/{username}"
|
||
|
||
status_code, data = _fetch_github_api(api_url)
|
||
|
||
if status_code == 200:
|
||
profile = GitHubProfile(
|
||
username=username,
|
||
name=data.get("name"),
|
||
bio=data.get("bio"),
|
||
location=data.get("location"),
|
||
company=data.get("company"),
|
||
public_repos=data.get("public_repos"),
|
||
followers=data.get("followers"),
|
||
following=data.get("following"),
|
||
created_at=data.get("created_at"),
|
||
updated_at=data.get("updated_at"),
|
||
avatar_url=data.get("avatar_url"),
|
||
blog=data.get("blog"),
|
||
twitter_username=data.get("twitter_username"),
|
||
hireable=data.get("hireable"),
|
||
)
|
||
|
||
return profile
|
||
elif status_code == 404:
|
||
print(f"GitHub user not found: {username}")
|
||
return None
|
||
else:
|
||
print(f"GitHub API error: {status_code} - {data}")
|
||
return None
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"Error fetching GitHub profile: {e}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"Unexpected error fetching GitHub profile: {e}")
|
||
return None
|
||
|
||
|
||
def fetch_contributions_count(owner: str, contributors_data):
|
||
user_contributions = 0
|
||
total_contributions = 0
|
||
|
||
for contributor in contributors_data:
|
||
if isinstance(contributor, dict):
|
||
contributions = contributor.get("contributions", 0)
|
||
total_contributions += contributions
|
||
|
||
if contributor.get("login", "").lower() == owner.lower():
|
||
user_contributions = contributions
|
||
|
||
return user_contributions, total_contributions
|
||
|
||
|
||
def fetch_repo_contributors(owner: str, repo_name: str) -> list[dict]:
|
||
try:
|
||
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/contributors"
|
||
|
||
status_code, contributors_data = _fetch_github_api(api_url)
|
||
|
||
if status_code == 200:
|
||
return contributors_data
|
||
else:
|
||
return []
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error fetching contributors for {owner}/{repo_name}: {e}")
|
||
return []
|
||
|
||
|
||
def fetch_all_github_repos(github_url: str, max_repos: int = 100) -> List[Dict]:
|
||
try:
|
||
username = extract_github_username(github_url)
|
||
if not username:
|
||
print(f"Could not extract username from: {github_url}")
|
||
return []
|
||
|
||
api_url = f"https://api.github.com/users/{username}/repos"
|
||
|
||
params = {"sort": "updated", "per_page": min(max_repos, 100), "type": "all"}
|
||
|
||
status_code, repos_data = _fetch_github_api(api_url, params=params)
|
||
|
||
if status_code == 200:
|
||
projects = []
|
||
for repo in repos_data:
|
||
if repo.get("fork") and repo.get("forks_count", 0) < 5:
|
||
continue
|
||
|
||
repo_name = repo.get("name")
|
||
|
||
contributors_data = fetch_repo_contributors(username, repo_name)
|
||
contributor_count = len(contributors_data)
|
||
|
||
user_contributions, total_contributions = fetch_contributions_count(
|
||
username, contributors_data
|
||
)
|
||
|
||
project_type = (
|
||
"open_source" if contributor_count > 1 else "self_project"
|
||
)
|
||
|
||
project = {
|
||
"name": repo.get("name"),
|
||
"description": repo.get("description"),
|
||
"github_url": repo.get("html_url"),
|
||
"live_url": repo.get("homepage") if repo.get("homepage") else None,
|
||
"technologies": (
|
||
[repo.get("language")] if repo.get("language") else []
|
||
),
|
||
"project_type": project_type,
|
||
"contributor_count": contributor_count,
|
||
"author_commit_count": user_contributions,
|
||
"total_commit_count": total_contributions,
|
||
"github_details": {
|
||
"stars": repo.get("stargazers_count", 0),
|
||
"forks": repo.get("forks_count", 0),
|
||
"language": repo.get("language"),
|
||
"description": repo.get("description"),
|
||
"created_at": repo.get("created_at"),
|
||
"updated_at": repo.get("updated_at"),
|
||
"topics": repo.get("topics", []),
|
||
"open_issues": repo.get("open_issues_count", 0),
|
||
"size": repo.get("size", 0),
|
||
"fork": repo.get("fork", False),
|
||
"archived": repo.get("archived", False),
|
||
"default_branch": repo.get("default_branch"),
|
||
"contributors": contributor_count,
|
||
},
|
||
}
|
||
projects.append(project)
|
||
|
||
projects.sort(key=lambda x: x["github_details"]["stars"], reverse=True)
|
||
|
||
open_source_count = sum(
|
||
1 for p in projects if p["project_type"] == "open_source"
|
||
)
|
||
self_project_count = sum(
|
||
1 for p in projects if p["project_type"] == "self_project"
|
||
)
|
||
|
||
print(f"✅ Found {len(projects)} repositories")
|
||
print(
|
||
f"📊 Project classification: {open_source_count} open source, {self_project_count} self projects"
|
||
)
|
||
return projects
|
||
|
||
elif status_code == 404:
|
||
print(f"GitHub user not found: {username}")
|
||
return []
|
||
else:
|
||
print(f"GitHub API error: {status_code} - {repos_data}")
|
||
return []
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"Error fetching GitHub repositories: {e}")
|
||
return []
|
||
except Exception as e:
|
||
print(f"Unexpected error fetching GitHub repositories: {e}")
|
||
return []
|
||
|
||
|
||
def generate_profile_json(profile: GitHubProfile) -> Dict:
|
||
if not profile:
|
||
return {}
|
||
|
||
profile_data = {
|
||
"username": profile.username,
|
||
"name": profile.name,
|
||
"bio": profile.bio,
|
||
"location": profile.location,
|
||
"company": profile.company,
|
||
"public_repos": profile.public_repos,
|
||
"followers": profile.followers,
|
||
"following": profile.following,
|
||
"created_at": profile.created_at,
|
||
"updated_at": profile.updated_at,
|
||
"avatar_url": profile.avatar_url,
|
||
"blog": profile.blog,
|
||
"twitter_username": profile.twitter_username,
|
||
"hireable": profile.hireable,
|
||
}
|
||
|
||
return profile_data
|
||
|
||
|
||
def generate_projects_json(projects: List[Dict]) -> List[Dict]:
|
||
if not projects:
|
||
return []
|
||
|
||
try:
|
||
projects_data = []
|
||
for project in projects:
|
||
if project.get("author_commit_count") == 0:
|
||
continue
|
||
|
||
project_data = {
|
||
"name": project.get("name"),
|
||
"description": project.get("description"),
|
||
"github_url": project.get("github_url"),
|
||
"live_url": project.get("live_url"),
|
||
"technologies": project.get("technologies", []),
|
||
"project_type": project.get("project_type", "self_project"),
|
||
"contributor_count": project.get("contributor_count", 1),
|
||
"author_commit_count": project.get("author_commit_count", 0),
|
||
"total_commit_count": project.get("total_commit_count", 0),
|
||
"github_details": project.get("github_details", {}),
|
||
}
|
||
projects_data.append(project_data)
|
||
|
||
projects_json = json.dumps(projects_data, indent=2)
|
||
|
||
template_manager = TemplateManager()
|
||
prompt = template_manager.render_template(
|
||
"github_project_selection", projects_data=projects_json
|
||
)
|
||
|
||
print(
|
||
f"🤖 Using LLM to select top 5 projects from {len(projects)} repositories..."
|
||
)
|
||
|
||
# Initialize the LLM provider
|
||
provider = initialize_llm_provider(DEFAULT_MODEL)
|
||
|
||
# Get model parameters
|
||
model_params = MODEL_PARAMETERS.get(
|
||
DEFAULT_MODEL, {"temperature": 0.1, "top_p": 0.9}
|
||
)
|
||
|
||
# Prepare chat parameters
|
||
chat_params = {
|
||
"model": DEFAULT_MODEL,
|
||
"messages": [
|
||
{
|
||
"role": "system",
|
||
"content": "You are an expert technical recruiter analyzing GitHub repositories to identify the most impressive projects. CRITICAL: You must select exactly 7 UNIQUE projects - no duplicates allowed. Each project must be different from the others.",
|
||
},
|
||
{"role": "user", "content": prompt},
|
||
],
|
||
"options": model_params,
|
||
}
|
||
|
||
# Call the LLM provider
|
||
response = provider.chat(**chat_params)
|
||
|
||
response_text = response["message"]["content"]
|
||
|
||
try:
|
||
response_text = response_text.strip()
|
||
response_text = extract_json_from_response(response_text)
|
||
|
||
selected_projects = json.loads(response_text)
|
||
|
||
unique_projects = []
|
||
seen_names = set()
|
||
|
||
for project in selected_projects:
|
||
project_name = project.get("name", "")
|
||
if project_name and project_name not in seen_names:
|
||
unique_projects.append(project)
|
||
seen_names.add(project_name)
|
||
|
||
if len(unique_projects) < 7:
|
||
print(
|
||
f"⚠️ LLM selected {len(selected_projects)} projects but {len(unique_projects)} are unique"
|
||
)
|
||
|
||
for project in projects_data:
|
||
if len(unique_projects) >= 7:
|
||
break
|
||
project_name = project.get("name", "")
|
||
if project_name and project_name not in seen_names:
|
||
unique_projects.append(project)
|
||
seen_names.add(project_name)
|
||
|
||
project_names = ", ".join(
|
||
[proj.get("name", "N/A") for proj in unique_projects]
|
||
)
|
||
print(
|
||
f"✅ LLM selected {len(unique_projects)} unique top projects: {project_names}"
|
||
)
|
||
return unique_projects
|
||
|
||
except json.JSONDecodeError as e:
|
||
print(f"ERROR: Error parsing LLM response: {e}")
|
||
print(f"ERROR: Raw response: {response_text}")
|
||
|
||
print("🔄 Falling back to first 7 projects")
|
||
return projects_data[:7]
|
||
|
||
except Exception as e:
|
||
print(f"Error using LLM for project selection: {e}")
|
||
print("🔄 Falling back to first 7 projects")
|
||
|
||
projects_data = []
|
||
for project in projects[:7]:
|
||
project_data = {
|
||
"name": project.get("name"),
|
||
"description": project.get("description"),
|
||
"github_url": project.get("github_url"),
|
||
"live_url": project.get("live_url"),
|
||
"technologies": project.get("technologies", []),
|
||
"project_type": project.get("project_type", "self_project"),
|
||
"contributor_count": project.get("contributor_count", 1),
|
||
"github_details": project.get("github_details", {}),
|
||
}
|
||
projects_data.append(project_data)
|
||
|
||
return projects_data
|
||
|
||
|
||
def fetch_and_display_github_info(github_url: str) -> Dict:
|
||
logger.info(f"{github_url}")
|
||
github_profile = fetch_github_profile(github_url)
|
||
if not github_profile:
|
||
print("\n❌ Failed to fetch GitHub profile details.")
|
||
return {}
|
||
|
||
print("🔍 Fetching all repository details...")
|
||
projects = fetch_all_github_repos(github_url)
|
||
|
||
if not projects:
|
||
print("\n❌ No repositories found or failed to fetch repository details.")
|
||
|
||
profile_json = generate_profile_json(github_profile)
|
||
projects_json = generate_projects_json(projects)
|
||
|
||
result = {
|
||
"profile": profile_json,
|
||
"projects": projects_json,
|
||
"total_projects": len(projects_json),
|
||
}
|
||
|
||
return result
|
||
|
||
|
||
def main(github_url):
|
||
result = fetch_and_display_github_info(github_url)
|
||
print("\n" + "=" * 60)
|
||
print("JSON DATA OUTPUT")
|
||
print("=" * 60)
|
||
print(json.dumps(result, indent=2, ensure_ascii=False))
|
||
print("=" * 60)
|
||
|
||
return result
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main("https://github.com/PavitKaur05")
|