通过他人的订阅地址，自动洗可用节点并生成新的订阅地址

一套完整的、结合了 GitHub Actions（自动化测速清洗） 与 Cloudflare Workers + KV（稳定数据分发与分流） 的节点管理系统方案。

通过这套方案，你可以实现每日定时同步最新节点、自动添加日期前缀、历史记录按日期留存、获取全部历史去重节点，以及通过专属 API 一键无缝触发后台重测清洗。
※本方案为原理性方案，需要自行调试或让ai帮你调试；文中使用的节点订阅地址是：https://shadowmere.xyz/api/b64sub/，建议换成自己认为稳定的。※

总体架构与数据流向

• GitHub Actions：负责消耗性能的并发 TCP 探测、重命名、去重，以及调用 Cloudflare API 写入数据。
• Cloudflare Workers：负责路由解析，校验 UUID 安全访问，并秒级响应客户端的订阅请求。
• Cloudflare KV：作为数据中转站，存储明文和密文节点。

第一步：准备工作（获取各类密钥）

前提（自行准备）：Cloudflare账号及由CF解析的域名，GitHub账号。

在开始部署代码前，你需要收集并记录以下 4 个关键参数：

1. Cloudflare 账户 ID 与 KV ID：
   • 登录 Cloudflare，在首页右侧记录 账户 ID (Account ID)。
   • 进入 KV 页面，创建一个命名空间，命名为 SUB_STORE，复制其 ID。
2. Cloudflare API 令牌 (Token)：
   • 点击右上角头像 -> 我的个人资料 -> API 令牌 -> 创建令牌 -> 选择 编辑 Cloudflare Workers 模板，账户和区域选择你的对应资源，生成并保存该 Token。
3. GitHub 个人访问令牌 (PAT)：
   • 登录 GitHub，点击头像 -> Settings -> Developer settings -> Personal access tokens -> Tokens (classic) -> Generate new token (classic)。
   • Note 填写 CF_Worker_Trigger，必须勾选 repo 权限，生成并复制该 ghp_*** 字符串。

第二步：配置 GitHub 仓库与脚本

在 GitHub 上创建一个 私有仓库 (Private Repository)（确保节点隐私）。

1. 添加仓库 Secrets

进入仓库的 Settings -> Secrets and variables -> Actions -> New repository secret，依次添加：

• CF_ACCOUNT_ID：你的 Cloudflare 账户 ID
• CF_API_TOKEN：你的 Cloudflare API 令牌
• CF_KV_NAMESPACE_ID：你的 KV 空间 ID

2. 创建每日定时同步脚本 daily_check.py

在仓库根目录下新建文件 daily_check.py：

Python

import os
import base64
import requests
import asyncio
from datetime import datetime
from urllib.parse import urlparse, quote, unquote

SUB_URL = "https://shadowmere.xyz/api/b64sub/"

def decode_base64(data):
    missing_padding = len(data) % 4
    if missing_padding:
        data += '=' * (4 - missing_padding)
    return base64.b64decode(data).decode('utf-8', errors='ignore')

def encode_base64(data_str):
    return base64.b64encode(data_str.encode('utf-8')).decode('utf-8')

async def test_node(node_str):
    """简易高并发 TCP 端口存活测试"""
    if not node_str.strip(): return None
    try:
        parsed = urlparse(node_str.strip())
        host, port = None, None
        if parsed.scheme in ['ss', 'trojan', 'vless']:
            host_port = parsed.netloc.split('@')[-1] if '@' in parsed.netloc else parsed.netloc
            if ':' in host_port: host, port = host_port.split(':')[:2]
        if not host or not port: return node_str 
        
        port = int(port.split('?')[0])
        host = host.split('?')[0]
        conn = asyncio.open_connection(host, port)
        reader, writer = await asyncio.wait_for(conn, timeout=3.0)
        writer.close()
        await writer.wait_closed()
        return node_str
    except:
        return None

def rename_node(node_str, prefix):
    """在节点别名后面加上时间前缀"""
    try:
        if "#" in node_str:
            parts = node_str.split("#", 1)
            new_name = f"{prefix}{unquote(parts[1])}"
            return f"{parts[0]}#{quote(new_name)}"
        else:
            return f"{node_str}#{quote(prefix + 'Node')}"
    except:
        return node_str

def get_kv(key):
    url = f"https://api.cloudflare.com/client/v4/accounts/{os.environ['CF_ACCOUNT_ID']}/storage/kv/namespaces/{os.environ['CF_KV_NAMESPACE_ID']}/values/{key}"
    headers = {"Authorization": f"Bearer {os.environ['CF_API_TOKEN']}"}
    res = requests.get(url, headers=headers)
    return res.text if res.status_code == 200 else ""

def put_kv(key, value):
    url = f"https://api.cloudflare.com/client/v4/accounts/{os.environ['CF_ACCOUNT_ID']}/storage/kv/namespaces/{os.environ['CF_KV_NAMESPACE_ID']}/values/{key}"
    headers = {"Authorization": f"Bearer {os.environ['CF_API_TOKEN']}", "Content-Type": "text/plain"}
    requests.put(url, headers=headers, data=value.encode('utf-8'))

async def main():
    now = datetime.utcnow()
    prefix_date = now.strftime("%y/%m/%d").replace("/0", "/") # 格式例：26/6/15
    history_key_date = now.strftime("%y%m%d") # 格式：260615

    res = requests.get(SUB_URL, timeout=10)
    decoded_nodes = decode_base64(res.text.strip()).splitlines()
    
    tasks = [test_node(n) for n in decoded_nodes if n]
    live_nodes = [n for n in await asyncio.gather(*tasks) if n]
    
    renamed_nodes = [rename_node(n, prefix_date) for n in live_nodes]
    if not renamed_nodes:
        print("今日无可用节点。")
        return

    today_nodes_str = "\n".join(renamed_nodes)
    put_kv("latest_sub", encode_base64(today_nodes_str))
    put_kv(f"history_{history_key_date}", today_nodes_str)
    
    old_all_str = get_kv("all_history_nodes")
    old_all_list = old_all_str.splitlines() if old_all_str else []
    total_set = set(old_all_list + renamed_nodes)
    new_all_str = "\n".join(list(total_set))
    put_kv("all_history_nodes", new_all_str)
    
    print(f"同步成功。今日有效: {len(renamed_nodes)} 个，历史总库: {len(total_set)} 个。")

if __name__ == "__main__":
    asyncio.run(main())

3. 创建历史重测脚本 retest_all.py

在仓库根目录下新建文件 retest_all.py：

Python

import os
import asyncio
from daily_check import test_node, get_kv, put_kv 

async def main():
    print("开始异步重测全部历史节点...")
    all_nodes_str = get_kv("all_history_nodes")
    if not all_nodes_str.strip():
        print("历史库为空。")
        return
        
    all_nodes = all_nodes_str.splitlines()
    tasks = [test_node(n) for n in all_nodes if n]
    valid_nodes = [n for n in await asyncio.gather(*tasks) if n]
    
    new_all_str = "\n".join(valid_nodes)
    put_kv("all_history_nodes", new_all_str)
    print(f"清洗完成！保留可用历史节点: {len(valid_nodes)} 个。")

if __name__ == "__main__":
    asyncio.run(main())

4. 配置 GitHub Workflow 工作流

在仓库内创建目录 .github/workflows/，并放入以下两个 .yml 文件：

文件一：.github/workflows/daily.yml (每日定时任务)

YAML

name: Daily Sync and Test
on:
  schedule:
    - cron: '0 0 * * *'
  workflow_dispatch:

jobs:
  run-daily:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v4
    - uses: actions/setup-python@v5
      with:
        python-version: '3.10'
    - name: Install dependencies
      run: pip install requests
    - name: Run
      env:
        CF_ACCOUNT_ID: ${{ secrets.CF_ACCOUNT_ID }}
        CF_API_TOKEN: ${{ secrets.CF_API_TOKEN }}
        CF_KV_NAMESPACE_ID: ${{ secrets.CF_KV_NAMESPACE_ID }}
      run: python daily_check.py

文件二：.github/workflows/retest.yml (外部触发重测任务)

YAML

name: Retest All History Nodes
on:
  workflow_dispatch: 
  repository_dispatch:
    types: [retest_event]

jobs:
  run-retest:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v4
    - uses: actions/setup-python@v5
      with:
        python-version: '3.10'
    - name: Install dependencies
      run: pip install requests
    - name: Run Retest
      env:
        CF_ACCOUNT_ID: ${{ secrets.CF_ACCOUNT_ID }}
        CF_API_TOKEN: ${{ secrets.CF_API_TOKEN }}
        CF_KV_NAMESPACE_ID: ${{ secrets.CF_KV_NAMESPACE_ID }}
      run: python retest_all.py

第三步：配置 Cloudflare Worker

1. 绑定变量与域名

进入你创建的 Worker 控制台 -> Settings -> Variables：

• KV Namespace Bindings：绑定你创建的 KV 空间，变量名必须叫 SUB_STORE。
• Environment Variables：
  • 添加明文变量 SUB_UUID，填写你自定义的 UUID。
  • 添加加密变量 (Secret) GH_TOKEN，填写你从 GitHub 申请到的个人访问令牌 (ghp_***)。
• 并在 Triggers 中绑定你自己的自定义域名。

2. Worker 完整分发代码

JavaScript

export default {
  async fetch(request, env, ctx) {
    const url = new URL(request.url);
    const paths = url.pathname.split('/').filter(Boolean);
    const validUuid = env.SUB_UUID;

    // 安全校验：第一层路径必须精准匹配 UUID
    if (paths.length < 2 || paths[0] !== validUuid) {
      return new Response("Not Found", { status: 404 });
    }

    const action = paths[1]; 

    // 功能一：一键重测所有历史节点 API (https://我的域名/UUID/retestall)
    if (action === 'retestall') {
      // ⚠️请在此处修改为你真实的 GitHub 账号名和仓库名称
      const githubOwner = "你的GitHub用户名"; 
      const githubRepo = "你的仓库名称"; 
      
      const ghUrl = `https://api.github.com/repos/${githubOwner}/${githubRepo}/dispatches`;
      
      const init = {
        method: 'POST',
        headers: {
          'Authorization': `token ${env.GH_TOKEN}`,
          'Accept': 'application/vnd.github.v3+json',
          'User-Agent': 'Cloudflare-Worker-Trigger'
        },
        body: JSON.stringify({ event_type: 'retest_event' })
      };

      try {
        const ghResponse = await fetch(ghUrl, init);
        if (ghResponse.status === 204) {
          return new Response(JSON.stringify({ status: "success", message: "已成功触发 GitHub 后台历史节点重测任务，清洗通常需要 1-2 分钟，请稍后下载 history/all 订阅。" }), {
            headers: { "Content-Type": "application/json; charset=utf-8" }
          });
        } else {
          const errText = await ghResponse.text();
          return new Response(JSON.stringify({ status: "failed", error: errText }), {
            status: 500, headers: { "Content-Type": "application/json; charset=utf-8" }
          });
        }
      } catch (err) {
        return new Response(JSON.stringify({ status: "error", message: err.message }), {
          status: 500, headers: { "Content-Type": "application/json; charset=utf-8" }
        });
      }
    }

    // 功能二：订阅分发模块 (https://我的域名/UUID/sub)
    if (action === 'sub') {
      // 1. 默认最新订阅： /UUID/sub
      if (paths.length === 2) {
        const subData = await env.SUB_STORE.get("latest_sub");
        return subData 
          ? new Response(subData, { headers: { "Content-Type": "text/plain; charset=utf-8" } })
          : new Response("今日订阅尚未生成，请等待工作流运行或手动触发。", { status: 404 });
      }

      // 2. 获取全部去重历史总订阅： /UUID/sub/all
      if (paths.length === 3 && paths[2] === 'all') {
        const allNodes = await env.SUB_STORE.get("all_history_nodes");
        if (!allNodes) return new Response("历史库为空。", { status: 404 });
        const encodedAll = btoa(unescape(encodeURIComponent(allNodes)));
        return new Response(encodedAll, { headers: { "Content-Type": "text/plain; charset=utf-8" } });
      }

      // 3. 获取指定日期的历史快照： /UUID/sub/260615
      if (paths.length === 3) {
        const dateStr = paths[2]; 
        const historyNodes = await env.SUB_STORE.get(`history_${dateStr}`);
        if (!historyNodes) return new Response(`未找到日期为 ${dateStr} 的历史订阅。`, { status: 404 });
        const encodedHistory = btoa(unescape(encodeURIComponent(historyNodes)));
        return new Response(encodedHistory, { headers: { "Content-Type": "text/plain; charset=utf-8" } });
      }
    }

    return new Response("Not Found", { status: 404 });
  }
};

最终效果与 API 调用清单

配置完成后，请先前往 GitHub 点击一次手动运行（Workflow_dispatch）以生成初始数据。生成后，你可通过以下标准链接在各类 2Ray、Clash、Shadowrocket 客户端中使用：