chore: import upstream snapshot with attribution
This commit is contained in:
wehub-resource-sync
@@ -0,0 +1,496 @@
|
||||
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")
|
||||
Reference in New Issue
Block a user