run.py

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()