mirror of
https://github.com/gelusus/wxvl.git
synced 2026-10-27 19:10:56 +00:00
Your Name
ae2f047a3f
run.py: - 移除硬编码 doonsec session cookie (公开仓库凭据泄露, RSS 无需 cookie) - 新增 canon_url() 规范化去重: 长链按 __biz/mid/idx/sn 判重, 避免同一文章因 chksm/scene 跟踪参数不同被重复入库 - 新增 chainreactors 源: 从 archive 分支 dump frontmatter 提取短链 (长链对机房IP有 wappoc 验证码, CI 只收短链) - requests 全部加 30s 超时; 裸 except 改打印错误 - data.json 解析失败保留原文件并退出, 不再覆盖清空 - 修复 URL 提取正则字符类; 零抓取 exit 1 让 workflow 失败通知 - 删除 get_issue_url / issue 模式; 临时目录清理; 子进程 120s 超时 - rep_filename 同名冲突加序号防覆盖 README 更新数据源说明 新增: backfill.py (断档补录), crawl_todo.py (断点续爬)
257 lines
11 KiB
Python
257 lines
11 KiB
Python
import os
|
||
import re
|
||
import sys
|
||
import json
|
||
import xml.etree.ElementTree as ET
|
||
import platform
|
||
import tempfile
|
||
import requests
|
||
import shutil
|
||
import subprocess
|
||
import datetime
|
||
from urllib.parse import urlparse, parse_qsl, urlencode
|
||
|
||
|
||
def canon_url(u):
|
||
'''规范化微信文章URL: 短链保留token; 长链只留 __biz/mid/idx/sn,
|
||
去掉 chksm/scene 等跟踪参数和 #fragment —— 同一篇文章的多种写法归一,
|
||
否则不同源(裸参数版/带跟踪参数版)会被当成两篇导致重复入库'''
|
||
try:
|
||
p = urlparse(u)
|
||
if p.path.startswith('/s/') and len(p.path) > 3:
|
||
return 'https://mp.weixin.qq.com' + p.path
|
||
q = [(k, v) for k, v in parse_qsl(p.query) if k in ('__biz', 'mid', 'idx', 'sn')]
|
||
if not q:
|
||
return u
|
||
return 'https://mp.weixin.qq.com/s?' + urlencode(q)
|
||
except Exception:
|
||
return u
|
||
|
||
|
||
def write_json(path, data, encoding="utf8"):
|
||
"""写入json"""
|
||
with open(path, "w", encoding=encoding) as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=4)
|
||
|
||
|
||
def read_json(path, default_data={}, encoding="utf8"):
|
||
"""读取json, 解析失败时保留原文件并报错退出, 避免误清空历史记录"""
|
||
if not os.path.exists(path):
|
||
return default_data
|
||
try:
|
||
return json.loads(open(path, "r", encoding=encoding).read())
|
||
except Exception as e:
|
||
print("data.json 解析失败, 已停止以免覆盖历史记录: {}".format(e))
|
||
sys.exit(1)
|
||
|
||
def get_executable_path():
|
||
'''获取可执行文件路径'''
|
||
system = platform.system()
|
||
if system == 'Windows':
|
||
executable_path = './bin/wechatmp2markdown-v1.1.11_win64.exe'
|
||
else:
|
||
executable_path = './bin/wechatmp2markdown-v1.1.11_linux_amd64'
|
||
# 添加执行权限
|
||
os.chmod(executable_path, 0o755)
|
||
# 返回可执行文件的完整路径
|
||
return executable_path
|
||
|
||
def get_md_path(executable_path,url):
|
||
'''获取md文件路径'''
|
||
# finally 保证临时目录被清理, 子进程 120s 超时防止卡死整个 workflow
|
||
temp_directory = tempfile.mkdtemp()
|
||
command = [executable_path, url, temp_directory, '--image=url']
|
||
try:
|
||
subprocess.check_output(command, timeout=120)
|
||
for root, _, files in os.walk(temp_directory):
|
||
for file in files:
|
||
if file.endswith(".md"):
|
||
file_path = os.path.join(root, file)
|
||
yield file_path
|
||
finally:
|
||
shutil.rmtree(temp_directory, ignore_errors=True)
|
||
|
||
def get_chainreactors_url():
|
||
'''从 chainreactors picker (archive 分支) 当日 dump 的 frontmatter 提取微信文章链接。
|
||
|
||
dump 结构: archive/{年}/{月}/{日}/daily/{公众号_标题}.md, frontmatter 带 url 字段。
|
||
目录滞后一天生成(次日约 12:02 CST 提交), 故取昨天+今天两个目录。
|
||
只收 /s/ 短链: s?__biz= 长链自 2025-07 起对机房IP(含 GitHub runners)启用
|
||
wappoc 验证码, 家用IP实测约85%可爬但 CI 必失败, 为防烧坏 runner IP 直接过滤。
|
||
先按文件名(含标题)做关键词过滤, 只下载命中文件的 frontmatter, 减少请求量。
|
||
'''
|
||
keyword = re.compile(r'(?:复现|漏洞|CVE-\d+|CNVD-\d+|CNNVD-\d+|XVE-\d+|QVD-\d+|POC|EXP|0day|1day|nday|RCE|代码执行|命令执行)', re.I)
|
||
api_headers = {'Accept': 'application/vnd.github+json'}
|
||
token = os.environ.get('GH_TOKEN')
|
||
if token:
|
||
api_headers['Authorization'] = 'token ' + token
|
||
urls = []
|
||
today = datetime.date.today()
|
||
for delta in (1, 0):
|
||
d = today - datetime.timedelta(days=delta)
|
||
api = 'https://api.github.com/repos/chainreactors/picker/contents/archive/{}/{:02d}/{:02d}/daily?ref=archive'.format(d.year, d.month, d.day)
|
||
try:
|
||
resp = requests.get(api, headers=api_headers, timeout=30)
|
||
if resp.status_code != 200:
|
||
continue
|
||
for item in resp.json():
|
||
name = item.get('name', '')
|
||
if not name.endswith('.md') or not keyword.search(name):
|
||
continue
|
||
# Range 只取文件头, frontmatter (含 url 行) 在前 2KB 内
|
||
raw = requests.get(item['download_url'], timeout=30,
|
||
headers={'Range': 'bytes=0-2047'})
|
||
m = re.search(r'^url:\s*(https://mp\.weixin\.qq\.com/s/\S+)', raw.text, re.M)
|
||
if m:
|
||
urls.append(m.group(1).rstrip(').,，'))
|
||
except Exception as e:
|
||
print("chainreactors 源({})获取失败: {}".format(d, e))
|
||
return urls
|
||
|
||
def get_BruceFeIix_url():
|
||
'''获取今日url'''
|
||
current_date = datetime.datetime.now().strftime("%Y-%m-%d")
|
||
base_url = 'https://raw.githubusercontent.com/BruceFeIix/picker/refs/heads/master/archive/daily/{}/{}.md'.format(current_date[:4], current_date)
|
||
headers = {
|
||
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
|
||
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
|
||
'cache-control': 'no-cache',
|
||
'pragma': 'no-cache',
|
||
'priority': 'u=0, i',
|
||
'referer': 'https://github.com/BruceFeIix/picker',
|
||
'sec-ch-ua': '"Chromium";v="130", "Microsoft Edge";v="130", "Not?A_Brand";v="99"',
|
||
'sec-ch-ua-mobile': '?0',
|
||
'sec-ch-ua-platform': '"Windows"',
|
||
'sec-fetch-dest': 'document',
|
||
'sec-fetch-mode': 'navigate',
|
||
'sec-fetch-site': 'cross-site',
|
||
'sec-fetch-user': '?1',
|
||
'upgrade-insecure-requests': '1',
|
||
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36 Edg/130.0.0.0',
|
||
}
|
||
try:
|
||
response = requests.get(
|
||
base_url,
|
||
headers=headers,
|
||
timeout=30,
|
||
)
|
||
# 注意: 字符类必须覆盖 URL 全部合法字符 (/ ? = & 等), 否则链接被截断
|
||
urls = re.findall('(?:复现|漏洞|CVE-\d+|CNVD-\d+|CNNVD-\d+|XVE-\d+|QVD-\d+|POC|EXP|0day|1day|nday|RCE|代码执行|命令执行).*?(https://mp.weixin.qq.com/[A-Za-z0-9\-._~:/?#@!$&*+,;=%]+)',response.text,re.I)
|
||
urls = [url.rstrip(')') for url in urls]
|
||
return urls
|
||
except Exception as e:
|
||
print("BruceFeIix 源获取失败: {}".format(e))
|
||
return []
|
||
|
||
|
||
def get_doonsec_url():
|
||
'''从 Doonsec RSS 获取今日URL，使用 XML 解析'''
|
||
headers = {
|
||
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
|
||
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
|
||
'cache-control': 'no-cache',
|
||
'pragma': 'no-cache',
|
||
'priority': 'u=0, i',
|
||
'sec-ch-ua': '"Chromium";v="130", "Microsoft Edge";v="130", "Not?A_Brand";v="99"',
|
||
'sec-ch-ua-mobile': '?0',
|
||
'sec-ch-ua-platform': '"Windows"',
|
||
'sec-fetch-dest': 'document',
|
||
'sec-fetch-mode': 'navigate',
|
||
'sec-fetch-site': 'same-origin',
|
||
'sec-fetch-user': '?1',
|
||
'upgrade-insecure-requests': '1',
|
||
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36 Edg/130.0.0.0',
|
||
}
|
||
|
||
try:
|
||
response = requests.get('https://wechat.doonsec.com/rss.xml', headers=headers, timeout=30)
|
||
response.encoding = response.apparent_encoding
|
||
|
||
# XML 解析
|
||
root = ET.fromstring(response.text)
|
||
urls = []
|
||
for item in root.findall('./channel/item'):
|
||
title = item.findtext('title') or ''
|
||
link = item.findtext('link') or ''
|
||
if re.search(r'(复现|漏洞|CVE-\d+|CNVD-\d+|CNNVD-\d+|XVE-\d+|QVD-\d+|POC|EXP|0day|1day|nday|RCE|代码执行|命令执行)', title, re.I) and link.startswith('https://mp.weixin.qq.com/'):
|
||
urls.append(link.rstrip(')'))
|
||
|
||
return urls
|
||
except Exception as e:
|
||
print("Error parsing Doonsec RSS:", e)
|
||
return []
|
||
|
||
|
||
|
||
def rep_filename(result_path):
|
||
'''
|
||
替换不能用于文件名的字符; 同名冲突时追加序号避免覆盖已有文章
|
||
'''
|
||
for root, _, files in os.walk(result_path):
|
||
for file in files:
|
||
if file.endswith(".md"):
|
||
file_path = os.path.join(root, file)
|
||
new_file = re.sub(r'[\/\\\:\*\?\"\<\>\|]', '', file)
|
||
if new_file == file:
|
||
continue
|
||
dest = os.path.join(root, new_file)
|
||
i = 2
|
||
while os.path.exists(dest):
|
||
stem, ext = os.path.splitext(new_file)
|
||
dest = os.path.join(root, '{} ({}).{}'.format(stem, i, ext))
|
||
i += 1
|
||
shutil.move(file_path, dest)
|
||
|
||
def main():
|
||
'''主函数'''
|
||
data_file = 'data.json'
|
||
data = {}
|
||
executable_path = get_executable_path()
|
||
base_result_path = 'doc'
|
||
# 创建基于当前年月的子目录 (格式: YYYY-MM)
|
||
current_month = datetime.datetime.now().strftime("%Y-%m")
|
||
result_path = os.path.join(base_result_path, current_month)
|
||
os.makedirs(result_path, exist_ok=True)
|
||
# 读取历史记录
|
||
data = read_json(data_file, default_data=data)
|
||
urls = []
|
||
if len(sys.argv) == 2 and sys.argv[1] == 'today':
|
||
print("正在获取URL...")
|
||
chainreactors_urls = get_chainreactors_url()
|
||
print(f"chainreactors: {len(chainreactors_urls)} 个URL")
|
||
|
||
brucefelix_urls = get_BruceFeIix_url()
|
||
print(f"BruceFeIix: {len(brucefelix_urls)} 个URL")
|
||
|
||
doonsec_urls = get_doonsec_url()
|
||
print(f"Doonsec: {len(doonsec_urls)} 个URL")
|
||
|
||
urls = list(set(chainreactors_urls + brucefelix_urls + doonsec_urls))
|
||
print(f"总共去重后: {len(urls)} 个URL")
|
||
# 三个源同时归零说明数据源已失效, 退出非零让 workflow 失败并通知,
|
||
# 防止再次出现断档空转无人察觉 (2025-08~12 的教训)
|
||
if not urls:
|
||
sys.exit(1)
|
||
# 规范化去重: 同一篇文章在不同源里可能带不同跟踪参数, 按规范化键判重
|
||
known = {canon_url(u) for u in data}
|
||
for url in urls:
|
||
if url in data or canon_url(url) in known:
|
||
continue
|
||
try:
|
||
for file_path in get_md_path(executable_path, url):
|
||
name = os.path.splitext(os.path.basename(file_path))[0]
|
||
if name == '.md':
|
||
continue
|
||
shutil.copy2(file_path,result_path)
|
||
data[url] = name
|
||
known.add(canon_url(url))
|
||
write_json(data_file,data)
|
||
print(name,end='、')
|
||
except Exception as e:
|
||
print(f"\n处理URL失败 {url}: {e}")
|
||
continue
|
||
rep_filename(result_path)
|
||
|
||
if __name__ == '__main__':
|
||
main()
|