846 lines
32 KiB
Python
846 lines
32 KiB
Python
#!/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 html
|
||
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()
|
||
GITEE_TOKEN = os.getenv("GITEE_TOKEN", "").strip()
|
||
GITHUB_RATE_LIMIT_RESET = 0
|
||
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",
|
||
}
|
||
|
||
JD_EXCLUDE_KEYWORDS = [
|
||
"京东", "东东", "jd_", "JD_", "jdCookie", "JD_COOKIE", "pt_key", "pt_pin",
|
||
"bean", "京豆", "joy", "宠汪汪", "东东农场", "东东萌宠", "京喜", "京东到家",
|
||
"api.m.jd.com", "m.jd.com", "jingdong", "jdc", "jd.com"
|
||
]
|
||
|
||
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():
|
||
global GITHUB_RATE_LIMIT_RESET
|
||
reset = r.headers.get("X-RateLimit-Reset")
|
||
try:
|
||
GITHUB_RATE_LIMIT_RESET = max(GITHUB_RATE_LIMIT_RESET, int(reset or 0))
|
||
except Exception:
|
||
pass
|
||
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 公开搜索。支持 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"
|
||
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()
|
||
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))}"
|
||
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:
|
||
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, 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
|
||
|
||
|
||
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"
|
||
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()
|
||
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 链接。优先 Bing；DuckDuckGo 可选。"""
|
||
links: Dict[str, dict] = {}
|
||
search_urls = []
|
||
for q in WEB_SEARCH_QUERIES:
|
||
search_urls.append(("Bing", "https://www.bing.com/search", {"q": q}))
|
||
if os.getenv("ENABLE_DDG_SEARCH", "false").strip().lower() == "true":
|
||
search_urls.append(("DuckDuckGo", "https://duckduckgo.com/html/", {"q": q}))
|
||
|
||
for engine, url, params in search_urls:
|
||
print(f"[WEB_SEARCH:{engine}] {params.get('q')}", flush=True)
|
||
r = generic_get(url, params=params)
|
||
if not r:
|
||
continue
|
||
page = html.unescape(r.text)
|
||
candidates = set()
|
||
# Bing 结果常在 href="https://..." 或 h="..." 内
|
||
for m in re.finditer(r'https?://[^"\'<>\s]+', page):
|
||
href = m.group(0)
|
||
href = href.replace("&", "&")
|
||
if "uddg=" in href:
|
||
qs = urllib.parse.parse_qs(urllib.parse.urlparse(href).query)
|
||
href = qs.get("uddg", [href])[0]
|
||
# 清理 Bing 跳转尾巴
|
||
href = href.split('"')[0].split('"')[0].split("'")[0]
|
||
candidates.add(href)
|
||
for href in candidates:
|
||
# 代码文件直链
|
||
if re.search(r"\.(js|py|ts|sh)(\?|$)", href, flags=re.I) and any(host in href for host in [
|
||
"githubusercontent.com", "github.com", "gitee.com", "gitcode", "agit.ai", "raw"
|
||
]):
|
||
links[href] = {"url": href}
|
||
# 仓库页面也先记录，后续可定向扩展
|
||
elif any(host in href for host in ["gitee.com", "gitcode.com", "agit.ai"]):
|
||
if re.match(r"https?://[^/]+/[^/?#]+/[^/?#]+/?$", href):
|
||
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 is_jd_related(content: str, path: str = "", project: str = "") -> bool:
|
||
hay = f"{path}\n{project}\n{content[:20000]}"
|
||
lower = hay.lower()
|
||
for kw in JD_EXCLUDE_KEYWORDS:
|
||
if kw.lower() in lower:
|
||
return True
|
||
return False
|
||
|
||
|
||
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_原文件名_<sha8>.ext\n",
|
||
"脚本库/APP版/账密/<项目>/YYYY-MM-DD_原文件名_<sha8>.ext\n",
|
||
"脚本库/web版/抓包/<项目>/YYYY-MM-DD_原文件名_<sha8>.ext\n",
|
||
"脚本库/web版/账密/<项目>/YYYY-MM-DD_原文件名_<sha8>.ext\n",
|
||
"脚本库/微信小程序/协议版/<项目>/YYYY-MM-DD_原文件名_<sha8>.ext\n",
|
||
"脚本库/微信小程序/抓包版/<项目>/YYYY-MM-DD_原文件名_<sha8>.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
|
||
project = extract_project(content, source_path)
|
||
if is_jd_related(content, source_path, project):
|
||
print(f" [SKIP_JD] {source_repo}/{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_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"
|
||
if is_jd_related(text, rel, project):
|
||
try:
|
||
path.unlink()
|
||
print(f"[REMOVE_JD] {rel}")
|
||
except Exception:
|
||
pass
|
||
continue
|
||
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
|
||
github_limited = False
|
||
|
||
# 先跑 GitHub；若限流，马上切换其它来源，不阻塞。
|
||
try:
|
||
files_done = collect_github(records, seen, files_done)
|
||
except RuntimeError as e:
|
||
if str(e) == "github_rate_limited":
|
||
github_limited = True
|
||
print("[WARN] GitHub 限流，立即切到 Gitee/公开网页等其它来源", flush=True)
|
||
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)
|
||
|
||
# 其它来源处理完后，如果 GitHub reset 时间已到或等待时间较短，则回头再试一次。
|
||
if github_limited and files_done < MAX_FILES_TOTAL:
|
||
now = int(time.time())
|
||
wait = max(0, GITHUB_RATE_LIMIT_RESET - now + 2)
|
||
max_wait = int(os.getenv("GITHUB_RETRY_MAX_WAIT", "180"))
|
||
if wait <= max_wait:
|
||
print(f"[GITHUB_RETRY] 等待 {wait}s 后回头继续 GitHub", flush=True)
|
||
time.sleep(wait)
|
||
try:
|
||
files_done = collect_github(records, seen, files_done)
|
||
except RuntimeError as e:
|
||
if str(e) == "github_rate_limited":
|
||
print("[WARN] GitHub 仍限流，本轮结束；下轮继续", flush=True)
|
||
else:
|
||
raise
|
||
else:
|
||
print(f"[GITHUB_RETRY] reset 等待时间 {wait}s 超过阈值 {max_wait}s，本轮先不等", flush=True)
|
||
|
||
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()
|