303 lines
8.6 KiB
Python
303 lines
8.6 KiB
Python
import requests
|
|
import os
|
|
import time
|
|
import datetime
|
|
import base64
|
|
import json
|
|
import xml.etree.ElementTree as ET
|
|
import subprocess
|
|
import re
|
|
import yaml
|
|
|
|
# 路径配置
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
PROJECT_ROOT = os.path.dirname(SCRIPT_DIR)
|
|
|
|
DATA_DIR = os.path.join(PROJECT_ROOT, "data")
|
|
REPO_DIR = os.path.join(DATA_DIR, "repos1")
|
|
LOG_DIR = os.path.join(PROJECT_ROOT, "logs")
|
|
|
|
OUTPUT_JSON = os.path.join(DATA_DIR, "collected_repos.json")
|
|
CLONE_FAIL_LOG = os.path.join(LOG_DIR, "clone_fail.txt")
|
|
RATE_LIMIT_LOG = os.path.join(LOG_DIR, "rate_limit.log")
|
|
MISSING_REPO_LOG = os.path.join(LOG_DIR, "missing_repo.log")
|
|
|
|
for d in [DATA_DIR, REPO_DIR, LOG_DIR]:
|
|
os.makedirs(d, exist_ok=True)
|
|
|
|
# 读取配置文件
|
|
def load_config():
|
|
config_path = os.path.join(SCRIPT_DIR, ".env.yml")
|
|
if not os.path.exists(config_path):
|
|
print(f"Config file not found: {config_path}")
|
|
print(f"Please create it based on .env.example.yml")
|
|
exit(1)
|
|
|
|
with open(config_path, "r") as f:
|
|
return yaml.safe_load(f)
|
|
|
|
# 加载配置
|
|
config = load_config()
|
|
|
|
# GitHub 配置
|
|
GITHUB_TOKEN = config.get("github_token")
|
|
HEADERS = {"Accept": "application/vnd.github+json"}
|
|
if GITHUB_TOKEN:
|
|
HEADERS["Authorization"] = f"token {GITHUB_TOKEN}"
|
|
|
|
BASE_URL = config.get("base_url", "https://api.github.com")
|
|
MIN_STARS = config.get("min_stars", 20)
|
|
REQUEST_INTERVAL = config.get("request_interval", 0.8)
|
|
MAX_RETRY_5XX = config.get("max_retry_5xx", 3)
|
|
|
|
# clone 控制参数
|
|
CLONE_RETRY = config.get("clone_retry", 2)
|
|
CLONE_TIMEOUT = config.get("clone_timeout", 180)
|
|
|
|
# 日志工具
|
|
def log_to_file(path, msg):
|
|
with open(path, "a", encoding="utf-8") as f:
|
|
f.write(msg + "\n")
|
|
|
|
# GitHub GET
|
|
def github_get(url, params=None):
|
|
retry_5xx = 0
|
|
while True:
|
|
r = requests.get(url, headers=HEADERS, params=params)
|
|
if r.status_code == 200:
|
|
time.sleep(REQUEST_INTERVAL)
|
|
return r
|
|
elif r.status_code == 403:
|
|
reset = r.headers.get("X-RateLimit-Reset")
|
|
if reset:
|
|
sleep_time = int(reset) - int(time.time()) + 1
|
|
log_to_file(RATE_LIMIT_LOG, f"403 rate limit → sleep {sleep_time}s, url={url}")
|
|
time.sleep(max(sleep_time, 1))
|
|
continue
|
|
return None
|
|
elif r.status_code == 404:
|
|
log_to_file(MISSING_REPO_LOG, f"404 missing {url}")
|
|
return None
|
|
elif 500 <= r.status_code < 600:
|
|
retry_5xx += 1
|
|
if retry_5xx <= MAX_RETRY_5XX:
|
|
time.sleep(2)
|
|
continue
|
|
return None
|
|
else:
|
|
return None
|
|
|
|
# 搜索仓库（按日期）
|
|
def search_repositories_by_stars(start_date, end_date):
|
|
query = f"language:Java stars:>={MIN_STARS} created:{start_date}..{end_date}"
|
|
repos = []
|
|
|
|
for page in range(1, 11):
|
|
r = github_get(
|
|
f"{BASE_URL}/search/repositories",
|
|
params={
|
|
"q": query,
|
|
"sort": "stars",
|
|
"order": "desc",
|
|
"per_page": 100,
|
|
"page": page
|
|
},
|
|
)
|
|
if not r:
|
|
break
|
|
items = r.json().get("items", [])
|
|
if not items:
|
|
break
|
|
repos.extend(items)
|
|
return repos
|
|
|
|
# 文件
|
|
def get_file_content(repo, path):
|
|
r = github_get(f"{BASE_URL}/repos/{repo}/contents/{path}")
|
|
if not r:
|
|
return None
|
|
c = r.json().get("content")
|
|
if not c:
|
|
return None
|
|
return base64.b64decode(c).decode("utf-8", errors="ignore")
|
|
|
|
# Maven 解析
|
|
def analyze_pom(pom_text):
|
|
uses_test = False
|
|
uses_spring = False
|
|
|
|
try:
|
|
root = ET.fromstring(pom_text)
|
|
except ET.ParseError:
|
|
return False, False
|
|
|
|
for dep in root.findall(".//dependencies/dependency"):
|
|
gid = dep.findtext("groupId")
|
|
if not gid:
|
|
continue
|
|
gid = gid.strip()
|
|
|
|
# 测试框架
|
|
if gid in {"junit", "org.junit.jupiter", "org.testng"}:
|
|
uses_test = True
|
|
|
|
# Spring Framework
|
|
if gid == "org.springframework":
|
|
uses_spring = True
|
|
|
|
return uses_test, uses_spring
|
|
|
|
# Gradle 解析（正则）
|
|
SPRING_RE = re.compile(r'["\']org\.springframework:([^"\']+)["\']')
|
|
JUNIT_RE = re.compile(r'["\'](junit:junit|org\.junit\.jupiter:[^"\']+)["\']')
|
|
|
|
def analyze_gradle(gradle_text):
|
|
uses_spring = bool(SPRING_RE.search(gradle_text))
|
|
uses_test = bool(JUNIT_RE.search(gradle_text))
|
|
return uses_test, uses_spring
|
|
|
|
# 克隆仓库
|
|
def clone_repo(repo_full_name):
|
|
local_name = repo_full_name.replace("/", "_")
|
|
target = os.path.join(REPO_DIR, local_name)
|
|
|
|
if os.path.exists(target):
|
|
print(f" [SKIP] {repo_full_name} already cloned")
|
|
return target
|
|
|
|
for attempt in range(1, CLONE_RETRY + 1):
|
|
print(f" [CLONE] {repo_full_name} (attempt {attempt}/{CLONE_RETRY})")
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "clone", "--depth", "1",
|
|
f"https://github.com/{repo_full_name}.git", target],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
timeout=CLONE_TIMEOUT
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
msg = f"timeout after {CLONE_TIMEOUT}s"
|
|
print(f" ↳ {msg}")
|
|
if attempt == CLONE_RETRY:
|
|
log_to_file(CLONE_FAIL_LOG, f"{repo_full_name} → {msg}")
|
|
continue
|
|
|
|
if result.returncode == 0:
|
|
return target
|
|
|
|
error_msg = result.stderr.strip() or "unknown error"
|
|
print(f" ↳ git error: {error_msg}")
|
|
|
|
if attempt == CLONE_RETRY:
|
|
log_to_file(CLONE_FAIL_LOG, f"{repo_full_name} → {error_msg}")
|
|
|
|
time.sleep(2)
|
|
|
|
return None
|
|
|
|
# 遍历多模块 Maven 项目
|
|
def find_all_poms(local_repo_path):
|
|
pom_paths = []
|
|
for root, dirs, files in os.walk(local_repo_path):
|
|
if "pom.xml" in files:
|
|
pom_paths.append(os.path.join(root, "pom.xml"))
|
|
return pom_paths
|
|
|
|
def analyze_maven_multimodule(local_repo_path):
|
|
uses_test = False
|
|
uses_spring = False
|
|
poms = find_all_poms(local_repo_path)
|
|
for pom_file in poms:
|
|
try:
|
|
with open(pom_file, "r", encoding="utf-8") as f:
|
|
text = f.read()
|
|
except Exception:
|
|
continue
|
|
t, s = analyze_pom(text)
|
|
uses_test |= t
|
|
uses_spring |= s
|
|
if uses_test and uses_spring:
|
|
break
|
|
return uses_test, uses_spring
|
|
|
|
# 分析仓库
|
|
def analyze_repository(repo):
|
|
name = repo["full_name"]
|
|
stars = repo.get("stargazers_count", 0)
|
|
|
|
gradle = get_file_content(name, "build.gradle")
|
|
gradle_kts = get_file_content(name, "build.gradle.kts")
|
|
|
|
# 克隆仓库
|
|
local_path = clone_repo(name)
|
|
if not local_path:
|
|
return None
|
|
|
|
# Maven 分析（多模块）
|
|
pom_paths = find_all_poms(local_path)
|
|
uses_test, uses_spring = False, False
|
|
if pom_paths:
|
|
uses_test, uses_spring = analyze_maven_multimodule(local_path)
|
|
build_tool = "maven"
|
|
else:
|
|
# Gradle 分析
|
|
gradle_files = [f for f in [gradle, gradle_kts] if f]
|
|
for g in gradle_files:
|
|
t, s = analyze_gradle(g)
|
|
uses_test |= t
|
|
uses_spring |= s
|
|
build_tool = "gradle"
|
|
|
|
if not uses_test:
|
|
print(f" [SKIP] {name} no junit/testng detected")
|
|
return None
|
|
if not uses_spring:
|
|
print(f" [SKIP] {name} no Spring Framework dependency")
|
|
return None
|
|
|
|
print(f" [OK] {name} passed all filters")
|
|
return {
|
|
"repo": name,
|
|
"stars": stars,
|
|
"build_tool": build_tool,
|
|
"local_path": local_path
|
|
}
|
|
|
|
def main():
|
|
results = []
|
|
|
|
start_date = datetime.date(2025, 2, 8)
|
|
end_date = datetime.date(2025, 6, 30)
|
|
|
|
delta = datetime.timedelta(days=1)
|
|
current_date = start_date
|
|
|
|
while current_date <= end_date:
|
|
next_date = current_date + delta
|
|
print(f"\n Searching date: {current_date}")
|
|
|
|
repos = search_repositories_by_stars(
|
|
current_date.isoformat(),
|
|
(next_date - datetime.timedelta(seconds=1)).isoformat()
|
|
)
|
|
print(f" found {len(repos)} repos")
|
|
|
|
for repo in repos:
|
|
info = analyze_repository(repo)
|
|
if info:
|
|
results.append(info)
|
|
|
|
current_date = next_date
|
|
|
|
with open(OUTPUT_JSON, "w", encoding="utf-8") as f:
|
|
json.dump(results, f, indent=2)
|
|
|
|
print(f"\n Total collected: {len(results)} repositories")
|
|
print(f" Saved to {OUTPUT_JSON}")
|
|
print(f" Clone failed repos saved to {CLONE_FAIL_LOG}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|