#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 公开青龙脚本采集器（多来源首轮版） 目标： - 从公开代码平台/公开网页入口搜索青龙/QL 脚本； - 按 APP版/抓包、APP版/账密、web版/抓包、web版/账密、微信小程序/协议版、微信小程序/抓包版分类； - 以代码内容 SHA256 去重； - 同项目不同代码放入同一项目目录； - 记录源仓库、源路径、源 URL、上传/提交时间。 说明： - GitHub 搜索 API 未配置 token 时有限流；可设置 GITHUB_TOKEN 提升额度。 - “上传时间”优先取该文件最近一次 commit 的 commit.committer.date；失败时退化为仓库 pushed_at。 """ from __future__ import annotations import base64 import csv import hashlib import json import os import re import time import unicodedata import urllib.parse from dataclasses import dataclass, asdict from pathlib import Path from typing import Dict, Iterable, List, Optional, Tuple try: import requests except ModuleNotFoundError: # 便于在新环境直接运行 import subprocess import sys as _sys subprocess.check_call([_sys.executable, "-m", "pip", "install", "requests", "-q", "--break-system-packages"]) import requests ROOT = Path(__file__).resolve().parent OUT_DIR = ROOT / "脚本库" META_DIR = ROOT / "metadata" INDEX_CSV = ROOT / "INDEX.csv" INDEX_JSON = ROOT / "INDEX.json" README = ROOT / "README.md" CACHE = META_DIR / "collector_cache.json" MAX_REPOS_PER_QUERY = int(os.getenv("MAX_REPOS_PER_QUERY", "12")) 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() SESSION = requests.Session() SESSION.headers.update({ "User-Agent": "ql-script-classifier/1.0", "Accept": "application/vnd.github+json", }) if GITHUB_TOKEN: SESSION.headers["Authorization"] = f"Bearer {GITHUB_TOKEN}" SEARCH_QUERIES = [ # 青龙/QL 常用关键词 "青龙 脚本 language:JavaScript", "青龙 脚本 language:Python", "qinglong script language:JavaScript", "qinglong script language:Python", "ql repo script language:JavaScript", '"new Env" "cron" language:JavaScript', '"new Env" "青龙" language:JavaScript', '"ql repo" "new Env" language:JavaScript', '"青龙面板" "new Env" language:JavaScript', '"微信小程序" "青龙" language:Python', '"wx_token" "wx_cloud" language:Python', '"getMiniProgramCode" language:Python', '"getMiniProgramCode" language:JavaScript', ] GITEE_SEARCH_QUERIES = [ "青龙 脚本", "new Env 青龙", "wx_token wx_cloud", "getMiniProgramCode", ] # 其它公开网页入口：只处理公开可访问的代码页/raw 链接；不处理需要登录、私密分享或绕权的网盘。 WEB_SEARCH_QUERIES = [ '青龙 脚本 "new Env"', '青龙脚本 "ql repo"', '青龙脚本 "wx_token" "wx_cloud"', '青龙脚本 "getMiniProgramCode"', '青龙脚本 site:gitee.com', '青龙脚本 site:gitcode.com', '青龙脚本 site:agit.ai', '青龙脚本 site:raw.githubusercontent.com', ] VALID_SUFFIXES = {".js", ".py", ".ts", ".sh"} SKIP_PATH_KEYWORDS = { "node_modules", ".git", "dist/", "build/", "package-lock", "yarn.lock", "readme", "license", "docker", "workflow", ".github", "icon", "image", "png", "jpg", } CATEGORY_ORDER = [ "APP版/抓包", "APP版/账密", "web版/抓包", "web版/账密", "微信小程序/协议版", "微信小程序/抓包版", "未分类", ] @dataclass class Record: category: str project: str filename: str sha256: str source_platform: str source_repo: str source_path: str source_url: str raw_url: str uploaded_at: str saved_path: str auth_type: str platform_type: str evidence: str def safe_segment(text: str, fallback: str = "unknown", max_len: int = 80) -> str: text = unicodedata.normalize("NFKC", text or "") text = re.sub(r"[\\/:*?\"<>|\s]+", "_", text).strip("._-") text = re.sub(r"_+", "_", text) if not text: text = fallback return text[:max_len] def normalized_hash(content: str) -> str: # 去 BOM、统一换行、去末尾空白；不删除注释，避免误合并不同版本 norm = content.replace("\r\n", "\n").replace("\r", "\n").lstrip("\ufeff") norm = "\n".join(line.rstrip() for line in norm.split("\n")).strip() + "\n" return hashlib.sha256(norm.encode("utf-8", "ignore")).hexdigest() def github_get(url: str, **kwargs): time.sleep(REQUEST_SLEEP) r = SESSION.get(url, timeout=30, **kwargs) if r.status_code == 403 and "rate limit" in r.text.lower(): reset = r.headers.get("X-RateLimit-Reset") print(f"[WARN] GitHub rate limited: {url} reset={reset}") raise RuntimeError("github_rate_limited") if r.status_code >= 400: print(f"[WARN] HTTP {r.status_code}: {url} {r.text[:160]}") return None return r def search_github_repos() -> List[dict]: repos: Dict[str, dict] = {} for q in SEARCH_QUERIES: url = "https://api.github.com/search/repositories" params = {"q": q, "sort": "updated", "order": "desc", "per_page": MAX_REPOS_PER_QUERY} print(f"[SEARCH] {q}", flush=True) r = github_get(url, params=params) if not r: continue data = r.json() for item in data.get("items", []): full = item.get("full_name") if full and full not in repos: repos[full] = item return list(repos.values()) def generic_get(url: str, **kwargs): time.sleep(REQUEST_SLEEP) headers = kwargs.pop("headers", {}) or {} merged = {"User-Agent": "Mozilla/5.0 ql-script-classifier/1.0", **headers} try: r = SESSION.get(url, timeout=30, headers=merged, **kwargs) if r.status_code >= 400: print(f"[WARN] HTTP {r.status_code}: {url} {r.text[:120]}") return None return r except Exception as e: print(f"[WARN] GET failed: {url} {e}") return None def search_gitee_repos() -> List[dict]: """Gitee 公开搜索。API 搜索经常返回空，因此同时解析公开搜索页。""" repos: Dict[str, dict] = {} 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}) if r: try: data = r.json() if isinstance(data, list): for item in data: full = item.get("full_name") or item.get("path_with_namespace") if full: repos[full] = item except Exception: pass # HTML 搜索页尝试 html_url = "https://gitee.com/search" # HTML 搜索容易被前端页面拖慢/反爬，默认仅使用 API；如需扩展可后续加入定向种子仓库。 return list(repos.values()) def list_gitee_tree(repo: dict) -> List[dict]: full = repo.get("full_name") or repo.get("path_with_namespace") if not full: return [] 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}) if not r: continue try: data = r.json() tree = data.get("tree") if isinstance(data, dict) else None if isinstance(tree, list): repo["default_branch"] = b return tree except Exception: continue return [] def get_gitee_file_content(repo_full: str, path: str, branch: str) -> Tuple[Optional[str], str, str]: # 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) 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 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}) if r: try: data = r.json() if isinstance(data, list) and data: c = data[0] return ((c.get("commit") or {}).get("committer") or {}).get("date") or c.get("created_at") or fallback or "" except Exception: pass return fallback or "" def discover_public_code_links() -> List[dict]: """通过公开搜索页发现 raw/js/py 链接。使用 DuckDuckGo HTML，无登录。""" links: Dict[str, dict] = {} if os.getenv("ENABLE_WEB_SEARCH", "false").strip().lower() != "true": print("[WEB_SEARCH] disabled; set ENABLE_WEB_SEARCH=true to enable public web search") return [] for q in WEB_SEARCH_QUERIES: print(f"[WEB_SEARCH] {q}") url = "https://duckduckgo.com/html/" r = generic_get(url, params={"q": q}) if not r: continue html = r.text for m in re.finditer(r'href="([^"]+)"', html): href = m.group(1) href = href.replace("&", "&") if "uddg=" in href: qs = urllib.parse.parse_qs(urllib.parse.urlparse(href).query) href = qs.get("uddg", [href])[0] if not re.search(r"\.(js|py|ts|sh)(\?|$)", href, flags=re.I): continue if not any(host in href for host in ["githubusercontent.com", "gitee.com", "gitcode", "agit.ai", "raw"]): continue links[href] = {"url": href} return list(links.values()) def get_public_link_content(url: str) -> Tuple[Optional[str], str, str, str]: raw_url = url html_url = url # GitHub blob -> raw m = re.match(r"https://github\.com/([^/]+/[^/]+)/blob/([^/]+)/(.+)", url) if m: raw_url = f"https://raw.githubusercontent.com/{m.group(1)}/{m.group(2)}/{m.group(3)}" # Gitee blob -> raw m = re.match(r"https://gitee\.com/([^/]+/[^/]+)/blob/([^/]+)/(.+)", url) if m: raw_url = f"https://gitee.com/{m.group(1)}/raw/{m.group(2)}/{m.group(3)}" r = generic_get(raw_url) 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, "" def list_github_tree(repo: dict) -> List[dict]: full = repo["full_name"] branch = repo.get("default_branch") or "main" url = f"https://api.github.com/repos/{full}/git/trees/{urllib.parse.quote(branch)}?recursive=1" r = github_get(url) if not r: return [] data = r.json() return data.get("tree", []) def is_candidate_path(path: str) -> bool: lower = path.lower() if any(k in lower for k in SKIP_PATH_KEYWORDS): return False if Path(path).suffix.lower() not in VALID_SUFFIXES: return False # 常见脚本路径或文件内容关键词由后续判断；这里不过窄过滤 return True def get_file_content(repo_full: str, path: str, branch: str) -> Tuple[Optional[str], str, str]: encoded = urllib.parse.quote(path, safe="") url = f"https://api.github.com/repos/{repo_full}/contents/{encoded}?ref={urllib.parse.quote(branch)}" r = github_get(url) if not r: return None, "", "" data = r.json() raw_url = data.get("download_url", "") html_url = data.get("html_url", "") content = data.get("content") if data.get("encoding") == "base64" and content: try: b = base64.b64decode(content) if len(b) > 1024 * 1024: return None, raw_url, html_url text = b.decode("utf-8", "ignore") return text, raw_url, html_url except Exception: return None, raw_url, html_url if raw_url: rr = github_get(raw_url) if rr: return rr.text, raw_url, html_url return None, raw_url, html_url def get_file_uploaded_at(repo_full: str, path: str, fallback: str) -> str: encoded = urllib.parse.quote(path, safe="") url = f"https://api.github.com/repos/{repo_full}/commits?path={encoded}&per_page=1" r = github_get(url) if not r: return fallback or "" arr = r.json() if isinstance(arr, list) and arr: commit = arr[0].get("commit", {}) return (commit.get("committer") or {}).get("date") or (commit.get("author") or {}).get("date") or fallback or "" return fallback or "" def looks_like_ql_script(content: str, path: str) -> bool: lower = content.lower() score = 0 keywords = [ "new env", "cron:", "ql repo", "青龙", "notify", "sendnotify", "process.env", "requests.", "axios", "got(", "wx_token", "jd_cookie", "cookie", "authorization", "getminiprogramcode", "wx_cloud", "wx.login", "小程序", ] for k in keywords: if k in lower: score += 1 if Path(path).suffix.lower() in {".js", ".py", ".sh"}: score += 1 return score >= 2 def extract_project(content: str, path: str) -> str: patterns = [ r"new\s+Env\([\"']([^\"']{2,80})[\"']\)", r"new\s+Env\s*\([`\"]([^`\"]{2,80})[`\"]\)", r"#\s*new\s+Env\([\"']([^\"']{2,80})[\"']\)", r"项目[:：]\s*([^\n\r]{2,60})", r"name[:：]\s*([^\n\r]{2,60})", ] for pat in patterns: m = re.search(pat, content, flags=re.I) if m: name = m.group(1).strip() name = re.sub(r"[【】\[\]()（）]", "", name).strip() if name: return safe_segment(name) stem = Path(path).stem stem = re.sub(r"^(ql_|jd_|jx_|ks_|wx_|applet_)", "", stem, flags=re.I) return safe_segment(stem) def classify(content: str, path: str) -> Tuple[str, str, str, str]: text = content lower = text.lower() evidence = [] is_wx_protocol = any(k in lower for k in [ "getminiprogramcode", "wx_cloud", "wx_token", "miniprogramcode", "养鸡场", "wx.login", "code2session" ]) is_wx = is_wx_protocol or any(k in lower for k in [ "appid", "openid", "session_key", "encrypteddata", "小程序", "wechat/login", "wxappid" ]) has_pwd = any(k in lower for k in [ "password", "passwd", "pwd", "username", "user_name", "account", "账密", "手机号登录", "mobile", "smscode", "验证码" ]) and any(k in lower for k in ["login", "登录", "/auth", "/user/login"]) has_capture = any(k in lower for k in [ "cookie", "authorization", "bearer", "token", "x-token", "session", "抓包", "header", "headers", "ck" ]) web_hint = any(k in lower for k in [ "web", "h5", "pc", "browser", "chrome", "网页登录", "www.", "passport", "csrf" ]) app_hint = any(k in lower for k in [ "app", "android", "ios", "okhttp", "x-requested-with", "user-agent", "deviceid", "imei", "oaid" ]) if is_wx_protocol: evidence.append("getMiniProgramCode/wx_cloud/wx_token/code协议链路") return "微信小程序/协议版", "微信小程序", "协议版", "; ".join(evidence) if is_wx: evidence.append("appid/openid/session_key/encryptedData等小程序字段") return "微信小程序/抓包版", "微信小程序", "抓包版", "; ".join(evidence) if web_hint and has_pwd: evidence.append("web/H5关键词 + username/password/login") return "web版/账密", "web版", "账密", "; ".join(evidence) if web_hint and has_capture: evidence.append("web/H5关键词 + cookie/token/header") return "web版/抓包", "web版", "抓包", "; ".join(evidence) if has_pwd: evidence.append("username/password/mobile/login") return "APP版/账密", "APP版", "账密", "; ".join(evidence) if has_capture: evidence.append("cookie/token/authorization/header") return "APP版/抓包", "APP版", "抓包", "; ".join(evidence) if app_hint: evidence.append("APP/Android/iOS/device关键词") return "APP版/抓包", "APP版", "抓包", "; ".join(evidence) return "未分类", "未分类", "未分类", "规则未命中" def unique_filename(project_dir: Path, uploaded_at: str, original: str, sha: str) -> str: date = (uploaded_at or "unknown")[:10] or "unknown" stem = safe_segment(Path(original).stem, "script", 60) suffix = Path(original).suffix or ".txt" name = f"{date}_{stem}_{sha[:8]}{suffix}" i = 2 while (project_dir / name).exists(): name = f"{date}_{stem}_{sha[:8]}_{i}{suffix}" i += 1 return name def write_record_file(record: Record, content: str): out_path = ROOT / record.saved_path out_path.parent.mkdir(parents=True, exist_ok=True) header = ( f"# Source: {record.source_url}\n" f"# Raw: {record.raw_url}\n" f"# Repo: {record.source_repo}\n" f"# Path: {record.source_path}\n" f"# UploadedAt: {record.uploaded_at}\n" f"# SHA256: {record.sha256}\n" f"# Category: {record.category}\n" f"# Evidence: {record.evidence}\n\n" ) if out_path.suffix.lower() == ".py": header = header elif out_path.suffix.lower() in {".js", ".ts"}: header = "// " + header.replace("\n", "\n// ").rstrip("// ") + "\n" elif out_path.suffix.lower() == ".sh": header = header out_path.write_text(header + content, encoding="utf-8", errors="ignore") def load_seen_hashes() -> set: if INDEX_JSON.exists(): try: arr = json.loads(INDEX_JSON.read_text(encoding="utf-8")) return {x.get("sha256") for x in arr if x.get("sha256")} except Exception: return set() return set() def save_indexes(records: List[Record]): existing = [] if INDEX_JSON.exists(): try: existing = json.loads(INDEX_JSON.read_text(encoding="utf-8")) except Exception: existing = [] by_sha = {x.get("sha256"): x for x in existing if x.get("sha256")} for r in records: by_sha[r.sha256] = asdict(r) arr = sorted(by_sha.values(), key=lambda x: (x.get("category", ""), x.get("project", ""), x.get("uploaded_at", ""))) INDEX_JSON.write_text(json.dumps(arr, ensure_ascii=False, indent=2), encoding="utf-8") with INDEX_CSV.open("w", newline="", encoding="utf-8-sig") as f: fields = list(asdict(records[0]).keys()) if records else [ "category", "project", "filename", "sha256", "source_platform", "source_repo", "source_path", "source_url", "raw_url", "uploaded_at", "saved_path", "auth_type", "platform_type", "evidence" ] w = csv.DictWriter(f, fieldnames=fields) w.writeheader() for item in arr: w.writerow(item) def update_readme(): arr = [] if INDEX_JSON.exists(): arr = json.loads(INDEX_JSON.read_text(encoding="utf-8")) counts = {c: 0 for c in CATEGORY_ORDER} projects = {} for x in arr: counts[x.get("category", "未分类")] = counts.get(x.get("category", "未分类"), 0) + 1 projects.setdefault(x.get("category", "未分类"), set()).add(x.get("project", "unknown")) lines = [ "# 青龙脚本分类库\n", "\n", "本仓库用于归档公开青龙脚本，并按运行端与凭证方式分类。\n", "\n", "## 分类\n", "\n", ] for c in CATEGORY_ORDER: lines.append(f"- `{c}`：{counts.get(c,0)} 个脚本，{len(projects.get(c,set()))} 个项目\n") lines += [ "\n", "## 目录规则\n", "\n", "```text\n", "脚本库/APP版/抓包/<项目>/YYYY-MM-DD_原文件名_.ext\n", "脚本库/APP版/账密/<项目>/YYYY-MM-DD_原文件名_.ext\n", "脚本库/web版/抓包/<项目>/YYYY-MM-DD_原文件名_.ext\n", "脚本库/web版/账密/<项目>/YYYY-MM-DD_原文件名_.ext\n", "脚本库/微信小程序/协议版/<项目>/YYYY-MM-DD_原文件名_.ext\n", "脚本库/微信小程序/抓包版/<项目>/YYYY-MM-DD_原文件名_.ext\n", "```\n", "\n", "## 元数据\n", "\n", "- `INDEX.csv`：脚本索引，含分类、项目、源仓库、源路径、源 URL、上传时间、SHA256。\n", "- `INDEX.json`：同上，JSON 格式。\n", "- 文件名中的日期优先使用源文件最近一次 commit 时间。\n", "- 相同代码以归一化 SHA256 去重。\n", "\n", "## 采集器\n", "\n", "运行：\n", "\n", "```bash\n", "python3 collector.py\n", "```\n", "\n", "可选环境变量：\n", "\n", "```bash\n", "export GITHUB_TOKEN=你的GitHubToken # 可提升 GitHub API 限额\n", "export MAX_REPOS_PER_QUERY=20\n", "export MAX_FILES_TOTAL=500\n", "```\n", ] README.write_text("".join(lines), encoding="utf-8") def add_content_record( records: List[Record], seen: set, content: str, source_platform: str, source_repo: str, source_path: str, source_url: str, raw_url: str, uploaded_at: str, ) -> bool: if not content or not looks_like_ql_script(content, source_path): return False sha = normalized_hash(content) if sha in seen: return False category, platform_type, auth_type, evidence = classify(content, source_path) project = extract_project(content, source_path) project_dir = OUT_DIR / category / project fname = unique_filename(project_dir, uploaded_at, Path(source_path).name, sha) saved_path = str((project_dir / fname).relative_to(ROOT)) rec = Record( category=category, project=project, filename=fname, sha256=sha, source_platform=source_platform, source_repo=source_repo, source_path=source_path, source_url=source_url, raw_url=raw_url, uploaded_at=uploaded_at, saved_path=saved_path, auth_type=auth_type, platform_type=platform_type, evidence=evidence, ) write_record_file(rec, content) records.append(rec) seen.add(sha) print(f" [ADD] {category} / {project} / {fname}") return True def collect_github(records: List[Record], seen: set, files_done: int) -> int: repos = search_github_repos() print(f"[INFO] github_repos={len(repos)}") for repo in repos: if files_done >= MAX_FILES_TOTAL: break full = repo.get("full_name") if not full: continue full = str(full) branch = str(repo.get("default_branch") or "main") print(f"[GITHUB_REPO] {full}") tree = list_github_tree(repo) candidates = [x for x in tree if x.get("type") == "blob" and is_candidate_path(x.get("path", ""))] candidates.sort(key=lambda x: ( 0 if re.search(r"(?i)(script|scripts|ql|青龙|jd_|wx_|applet|env)", x.get("path", "")) else 1, len(x.get("path", "")) )) for item in candidates[:80]: if files_done >= MAX_FILES_TOTAL: break path = item.get("path", "") content, raw_url, html_url = get_file_content(full, path, branch) uploaded_at = get_file_uploaded_at(full, path, repo.get("pushed_at", "")) if content else "" if add_content_record(records, seen, content or "", "GitHub", full, path, html_url, raw_url, uploaded_at): files_done += 1 return files_done def collect_gitee(records: List[Record], seen: set, files_done: int) -> int: repos = search_gitee_repos() print(f"[INFO] gitee_repos={len(repos)}") for repo in repos: if files_done >= MAX_FILES_TOTAL: break full = repo.get("full_name") or repo.get("path_with_namespace") if not full: continue full = str(full) branch = str(repo.get("default_branch") or "master") print(f"[GITEE_REPO] {full}") tree = list_gitee_tree(repo) branch = str(repo.get("default_branch") or branch) candidates = [x for x in tree if x.get("type") in ("blob", "file") and is_candidate_path(x.get("path", ""))] candidates.sort(key=lambda x: ( 0 if re.search(r"(?i)(script|scripts|ql|青龙|jd_|wx_|applet|env)", x.get("path", "")) else 1, len(x.get("path", "")) )) for item in candidates[:80]: if files_done >= MAX_FILES_TOTAL: break path = item.get("path", "") content, raw_url, html_url = get_gitee_file_content(full, path, branch) uploaded_at = get_gitee_file_uploaded_at(full, path, repo.get("pushed_at", "")) if content else "" if add_content_record(records, seen, content or "", "Gitee", full, path, html_url, raw_url, uploaded_at): files_done += 1 return files_done def collect_public_links(records: List[Record], seen: set, files_done: int) -> int: links = discover_public_code_links() print(f"[INFO] public_links={len(links)}") for item in links: if files_done >= MAX_FILES_TOTAL: break url = item.get("url", "") content, raw_url, html_url, uploaded_at = get_public_link_content(url) source_repo = urllib.parse.urlparse(url).netloc source_path = urllib.parse.urlparse(url).path.strip("/") or Path(url).name if add_content_record(records, seen, content or "", "PublicWeb", source_repo, source_path, html_url, raw_url, uploaded_at): files_done += 1 return files_done def rebuild_index_from_files(): """当采集被限流/中断后，可从已落盘脚本头部重建 INDEX。""" records = [] for path in OUT_DIR.rglob("*"): if not path.is_file() or path.suffix.lower() not in VALID_SUFFIXES: continue text = path.read_text(encoding="utf-8", errors="ignore") head = "\n".join(text.splitlines()[:12]) def pick(name): m = re.search(rf"{name}:\s*(.+)", head) return m.group(1).strip().lstrip("// ").strip() if m else "" rel = str(path.relative_to(ROOT)) parts = path.relative_to(OUT_DIR).parts if len(parts) < 4: continue category = f"{parts[0]}/{parts[1]}" if parts[0] in {"APP版", "web版", "微信小程序"} else parts[0] project = parts[2] if len(parts) >= 4 else "unknown" rec = Record( category=category, project=project, filename=path.name, sha256=pick("SHA256") or normalized_hash(text), source_platform="GitHub", source_repo=pick("Repo"), source_path=pick("Path"), source_url=pick("Source"), raw_url=pick("Raw"), uploaded_at=pick("UploadedAt"), saved_path=rel, auth_type=category.split("/")[-1] if "/" in category else "未分类", platform_type=category.split("/")[0] if "/" in category else "未分类", evidence=pick("Evidence"), ) records.append(rec) save_indexes(records) update_readme() print(f"[REBUILD] records={len(records)}") def main(): OUT_DIR.mkdir(parents=True, exist_ok=True) META_DIR.mkdir(parents=True, exist_ok=True) seen = load_seen_hashes() records: List[Record] = [] files_done = 0 try: files_done = collect_github(records, seen, files_done) except RuntimeError as e: if str(e) == "github_rate_limited": print("[WARN] GitHub 限流，跳过 GitHub 剩余采集，继续其它来源") else: raise if files_done < MAX_FILES_TOTAL: files_done = collect_gitee(records, seen, files_done) if files_done < MAX_FILES_TOTAL: files_done = collect_public_links(records, seen, files_done) if records: save_indexes(records) update_readme() elif any(OUT_DIR.rglob("*")): rebuild_index_from_files() else: save_indexes(records) update_readme() print(f"[DONE] added={len(records)} total_files_done={files_done}") if __name__ == "__main__": main()