diff --git a/collector.py b/collector.py index e1b3744..8c7efa6 100644 --- a/collector.py +++ b/collector.py @@ -51,6 +51,7 @@ MAX_FILES_TOTAL = int(os.getenv("MAX_FILES_TOTAL", "260")) REQUEST_SLEEP = float(os.getenv("REQUEST_SLEEP", "0.4")) GITHUB_TOKEN = os.getenv("GITHUB_TOKEN", "").strip() +GITEE_TOKEN = os.getenv("GITEE_TOKEN", "").strip() GITHUB_RATE_LIMIT_RESET = 0 SESSION = requests.Session() SESSION.headers.update({ @@ -204,13 +205,19 @@ def generic_get(url: str, **kwargs): def search_gitee_repos() -> List[dict]: - """Gitee 公开搜索。API 搜索经常返回空,因此同时解析公开搜索页。""" + """Gitee 公开搜索。支持 GITEE_TOKEN 提升额度/访问能力。""" repos: Dict[str, dict] = {} + gitee_headers = {} + if GITEE_TOKEN: + gitee_headers["Authorization"] = f"token {GITEE_TOKEN}" for q in GITEE_SEARCH_QUERIES: print(f"[GITEE_SEARCH] {q}") # API 尝试 api = "https://gitee.com/api/v5/search/repositories" - r = generic_get(api, params={"q": q, "page": 1, "per_page": MAX_REPOS_PER_QUERY}) + params = {"q": q, "page": 1, "per_page": MAX_REPOS_PER_QUERY} + if GITEE_TOKEN: + params["access_token"] = GITEE_TOKEN + r = generic_get(api, params=params, headers=gitee_headers) if r: try: data = r.json() @@ -234,7 +241,10 @@ def list_gitee_tree(repo: dict) -> List[dict]: branch = repo.get("default_branch") or "master" for b in [branch, "master", "main"]: api = f"https://gitee.com/api/v5/repos/{urllib.parse.quote(str(full), safe='')}/git/trees/{urllib.parse.quote(str(b))}" - r = generic_get(api, params={"recursive": 1}) + params = {"recursive": 1} + if GITEE_TOKEN: + params["access_token"] = GITEE_TOKEN + r = generic_get(api, params=params, headers={"Authorization": f"token {GITEE_TOKEN}"} if GITEE_TOKEN else {}) if not r: continue try: @@ -252,7 +262,7 @@ def get_gitee_file_content(repo_full: str, path: str, branch: str) -> Tuple[Opti # raw URL 不需要 token 的公开仓库通常可直接访问 raw_url = f"https://gitee.com/{repo_full}/raw/{urllib.parse.quote(branch)}/{urllib.parse.quote(path)}" html_url = f"https://gitee.com/{repo_full}/blob/{urllib.parse.quote(branch)}/{urllib.parse.quote(path)}" - r = generic_get(raw_url) + r = generic_get(raw_url, headers={"Authorization": f"token {GITEE_TOKEN}"} if GITEE_TOKEN else {}) if r and len(r.content) <= 1024 * 1024: return r.content.decode("utf-8", "ignore"), raw_url, html_url return None, raw_url, html_url @@ -262,7 +272,10 @@ def get_gitee_file_uploaded_at(repo_full: str, path: str, fallback: str) -> str: # Gitee commit API 兼容性不稳定;失败时退化到 repo pushed_at 或空。 encoded_repo = urllib.parse.quote(repo_full, safe="") api = f"https://gitee.com/api/v5/repos/{encoded_repo}/commits" - r = generic_get(api, params={"path": path, "page": 1, "per_page": 1}) + params = {"path": path, "page": 1, "per_page": 1} + if GITEE_TOKEN: + params["access_token"] = GITEE_TOKEN + r = generic_get(api, params=params, headers={"Authorization": f"token {GITEE_TOKEN}"} if GITEE_TOKEN else {}) if r: try: data = r.json()