python脚本合并订阅

拾荒开拓者2024-08-012024-09-28

Python 脚本合并订阅

在当今的网络环境中，使用代理服务器来保护隐私、突破地理限制已经成为许多用户的常态。随着越来越多的协议如 Trojan、Vmess、VLESS、SS、Hysteria2、TUIC 等的出现，如何有效管理和合并这些协议的订阅链接成为了一个重要的问题。在这篇博客中，我将介绍如何编写一个 Python 脚本，自动获取并合并多个订阅链接中的节点信息，同时确保它们的协议内容保持一致。

背景介绍

随着不同代理协议的兴起，各类订阅链接纷至沓来。然而，由于这些协议有着不同的格式和加密方式，手动合并它们的内容不仅耗时，而且容易出错。为了简化这一过程，我编写了一个 Python 脚本，能够自动化地获取多个订阅链接，将它们的节点信息进行合并，并处理协议中需要解码和替换的部分。

功能概述

这个脚本的核心功能包括：

支持多种协议：包括 Trojan、Vmess、VLESS、SS、Hysteria2 和 TUIC。
自动解码与编码：对于使用 Base64 编码的节点信息，脚本能够解码为明文状态，进行字符串替换后再重新编码。
自动化操作：通过 GitHub Actions 自动每45分钟运行一次，保持订阅内容的最新和一致性。

脚本实现

以下是 Python 脚本的核心代码片段：

import os
import base64
import requests
import json

NAME = "免费节点"

# 定义需要替换的关键字列表
replace_keywords = [
    "xxxxx",
    "xxxxx"
]

# 定义你的订阅链接列表
subscription_urls = [
    "https://abcd.com/sub1",
    "https://abcd.com/sub2"
]

# 获取和处理每个订阅链接的内容
def fetch_and_process_subscriptions(urls):
    all_contents = []

    for url in urls:
        response = requests.get(url)
        if response.status_code == 200:
            content = response.text
            if is_base64_encoded(content):
                decoded_content = base64.b64decode(content).decode('utf-8')
                processed_content = process_content(decoded_content)
                all_contents.append(processed_content)
            else:
                processed_content = process_content(content)
                all_contents.append(processed_content)
        else:
            print(f"Failed to fetch content from {url}")

    combined_content = "\n".join(all_contents)
    final_encoded_content = base64.b64encode(combined_content.encode('utf-8')).decode('utf-8')
    return final_encoded_content

# 检查内容是否为base64编码
def is_base64_encoded(data):
    try:
        if isinstance(data, str):
            data_bytes = data.encode('utf-8')
        elif isinstance(data, bytes):
            data_bytes = data
        else:
            raise ValueError("Input must be a string or bytes")

        return base64.b64encode(base64.b64decode(data_bytes)) == data_bytes
    except Exception:
        return False

# 处理解码后的内容，只修改指定信息
def process_content(content):
    lines = content.splitlines()
    processed_lines = []

    for line in lines:
        if line.startswith("vmess://"):
            vmess_encoded_part = line[len("vmess://"):]

            # 调整Base64字符串长度为4的倍数
            vmess_encoded_part = fix_base64_padding(vmess_encoded_part)

            vmess_decoded = base64.b64decode(vmess_encoded_part).decode('utf-8')
            vmess_dict = json.loads(vmess_decoded)

            # 只修改'ps'字段，保留其余内容
            vmess_dict['ps'] = replace_specified_info(vmess_dict.get('ps', ''))

            vmess_encoded_modified = base64.b64encode(json.dumps(vmess_dict).encode('utf-8')).decode('utf-8')
            processed_lines.append(f"vmess://{vmess_encoded_modified}")
        else:
            # 对非vmess协议的内容，替换指定的关键字
            processed_line = replace_specified_info(line)
            processed_lines.append(processed_line)

    return "\n".join(processed_lines)

# 替换指定信息的函数
def replace_specified_info(text):
    for keyword in replace_keywords:
        text = text.replace(keyword, NAME)
    return text

# 修正Base64字符串长度
def fix_base64_padding(data):
    missing_padding = len(data) % 4
    if missing_padding != 0:
        data += '=' * (4 - missing_padding)
    return data

final_content = fetch_and_process_subscriptions(subscription_urls)

# 将结果写入sub.txt文件
with open('sub.txt', 'w') as file:
    file.write(final_content)

print("总订阅节点已写入sub.txt文件")

代码讲解

协议支持：脚本通过 base64_protocols 列表定义了需要 Base64 解码的协议。对于这些协议的节点信息，脚本会进行解码、替换字符串、再重新编码的处理。
内容替换：replace_content 函数用于将节点信息中不合适的内容替换为“由盒子在互联网上收集”，以保持内容的一致性和安全性。
节点处理：process_node 函数是脚本的核心，负责处理每一个节点，包括判断协议类型、解码与编码、以及内容替换。

GitHub Actions 集成

为了实现自动化，每 45 分钟从订阅链接获取最新的内容并更新合并后的文件，我将脚本与 GitHub Actions 集成。以下是相关的配置：

yaml复制代码name: Merge Subscriptions

on:
  schedule:
    - cron: '*/45 * * * *'
  push:
    branches:
      - main

jobs:
  merge:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout repository
        uses: actions/checkout@v2

      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: 3.8

      - name: Install dependencies
        run: pip install requests

      - name: Run merge script
        run: python script.py

      - name: Display combined_subscription.txt
        run: cat combined_subscription.txt

      - name: Commit and push changes
        run: |
          git config --global user.name 'github-actions[bot]'
          git config --global user.email 'github-actions[bot]@users.noreply.github.com'
          git add combined_subscription.txt
          git commit -m 'Update combined subscription' || echo "No changes to commit"
          git pull --rebase origin main
          git push

GitHub Actions 的权限

1. 进入仓库设置

打开你的 GitHub 仓库。
点击仓库页面右上角的 Settings（设置）选项。

2. 配置 GitHub Actions 权限

在左侧导航栏中，向下滚动找到 Actions，并点击它。
在 Actions permissions 部分，你将看到以下选项：
- Allow all actions and reusable workflows：允许所有 GitHub Actions 和可重用工作流。
- Allow local actions and reusable workflows only：只允许本地 Actions 和可重用工作流。
- Allow select actions and reusable workflows：只允许选择的 Actions 和可重用工作流。
你可以根据需要选择合适的权限范围。
如果你选择了 Allow select actions and reusable workflows，你可以在下面进一步配置你想要允许的特定 Actions 或工作流。

3. 设置默认的工作流权限

向下滚动到 Workflow permissions 部分。这里有两个选项：
- Read repository contents permission：默认只允许工作流读取仓库内容。
- Read and write permissions：允许工作流读取和写入仓库内容。
根据你的需求选择其中一个选项。如果你想要更多的控制，也可以启用 Allow GitHub Actions to create and approve pull requests 选项。
最后，点击页面底部的 Save 按钮保存更改。