📁

目录结构与 mkdir

bash
mkdir -p /home/nginx /home/xray /home/python /home/cert/conf /home/cert/www /home/html

目录结构

    /home
     ├── nginx
     │    └── default.conf
     ├── xray
     │    └── config.json
     ├── python
     │    └── server.py
     ├── cert
     │    ├── conf
     │    └── www
     └── html
          ├── *.html
          └── test.bin
    
🐳

Docker 安装与常用命令

安装docker-compose-1.29.2,查看容器状态 & 日志
curl -L https://github.com/docker/compose/releases/download/1.29.2/docker-compose-linux-x86_64 \ > -o /usr/local/bin/docker-compose
chmod +x /usr/local/bin/docker-compose
hash -r
docker ps -a
docker logs -f nginx
强制杀死全部容器
docker kill $(docker ps -q)
拉取镜像 & 重建容器(yaml 变更后使用)
cd /home   # 进入 docker-compose.yaml 所在目录
docker-compose pull nginx xray python-api librespeed dashdot certbot
docker-compose up -d --force-recreate nginx xray python-api librespeed dashdot certbot
docker restart dashdot nginx xray python-api librespeed certbot
进入 / 退出容器
docker exec -it nginx sh
exit
📦

Docker Compose /home/docker-compose.yaml

docker-compose.yaml
 

version: "3.9"

services:
  dashdot:
    image: mauricenino/dashdot:latest
    container_name: dashdot
    restart: always
    privileged: true
    pid: host
    networks:
      - xray-net
    volumes:
      - /proc:/host/proc:ro
      - /sys:/host/sys:ro
      - /:/host/root:ro
      - /var/run/docker.sock:/var/run/docker.sock:ro
    environment:
      - DASHDOT_WIDGET_LIST=cpu,storage,ram,network
      - DASHDOT_NETWORK_INTERFACE=eth0

  xray:
    image: teddysun/xray
    container_name: xray
    restart: always
    volumes:
      - /home/xray/config.json:/etc/xray/config.json
    networks:
      - xray-net

  librespeed:
    image: linuxserver/librespeed
    container_name: librespeed
    restart: always
    environment:
      - MODE=standalone
    networks:
      - xray-net

  nginx:
    image: nginx:alpine
    container_name: nginx
    restart: always
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - /home/nginx/default.conf:/etc/nginx/conf.d/default.conf:ro
      - /home/nginx/.htpasswd:/etc/nginx/.htpasswd:ro
      - /home/cert/conf:/etc/letsencrypt
      - /home/cert/www:/var/www/certbot
      - /home/html:/usr/share/nginx/html
    networks:
      - xray-net

  certbot:
    image: certbot/certbot
    container_name: certbot
    restart: always
    volumes:
      - /home/cert/conf:/etc/letsencrypt
      - /home/cert/www:/var/www/certbot
    entrypoint: >
      sh -c "trap exit TERM;
      while :; do
        certbot renew --webroot -w /var/www/certbot --quiet;
        sleep 12h;
      done"

  python-api:
    image: docker:cli
    container_name: python-api
    restart: always
    working_dir: /app
    command: sh -c "apk add --no-cache python3 py3-pip openssl && python3 server.py"
    volumes:
      - /home/python:/app
      - /home/xray/config.json:/app/xray_config.json
      - /home/cert/conf:/etc/letsencrypt:ro
      - /var/run/docker.sock:/var/run/docker.sock
    expose:
      - "8080"
    networks:
      - xray-net

  warp:
    image: caomingjun/warp
    container_name: warp
    restart: always
    privileged: true
    cap_add:
      - NET_ADMIN
    devices:
      - /dev/net/tun
    environment:
      - WARP_SLEEP=2
      - WARP_LICENSE_KEY=
    networks:
      - xray-net

networks:
  xray-net:
    driver: bridge


    
🌐

Nginx 配置 /home/nginx/default.conf

default.conf
 

# =========================
# HTTP : 80 (Certbot + 首页)
# =========================
server {
    listen 80;
    server_name yourdomain.top www.yourdomain.top no-cdn.yourdomain.top www.no-cdn.yourdomain.top  seattle.yourdomain.top www.seattle.yourdomain.top  seattle-no-cdn.yourdomain.top www.seattle-no-cdn.yourdomain.top;

    root /usr/share/nginx/html;
    index index.html;

    location = /edit.html {
        auth_basic "Restricted";
        auth_basic_user_file /etc/nginx/.htpasswd;
    }
    
    location ~ ^/test.*\.bin$ {
        add_header Cache-Control "no-store, no-cache, must-revalidate, max-age=0";
        expires off;
    }

    location / {
        try_files $uri $uri/ =404;
    }

    # Certbot 验证
    location /.well-known/acme-challenge/ {
        alias /var/www/certbot/.well-known/acme-challenge/;
        default_type "text/plain";
    }

    access_log off;
    error_log /var/log/nginx/error.log warn;
}

# =========================
# HTTPS : 443 (TLS + gRPC + LibreSpeed)
# =========================
server {
    listen 443 ssl;
    http2 on;
    server_name yourdomain.top www.yourdomain.top no-cdn.yourdomain.top www.no-cdn.yourdomain.top  seattle.yourdomain.top www.seattle.yourdomain.top  seattle-no-cdn.yourdomain.top www.seattle-no-cdn.yourdomain.top;

    # 证书
    ssl_certificate /etc/letsencrypt/live/yourdomain.top/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/yourdomain.top/privkey.pem;

    # TLS 安全 & 兼容
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers HIGH:!aNULL:!MD5;
    ssl_prefer_server_ciphers off;
    ssl_session_cache shared:SSL:10m;
    ssl_session_timeout 1h;
    ssl_session_tickets off;
    ssl_buffer_size 8k;

    large_client_header_buffers 2 16k;

    root /usr/share/nginx/html;
    index index.html;
    
    keepalive_timeout 300s;
    keepalive_requests 1000;
    client_header_timeout 300s;
    client_body_timeout 300s;
    send_timeout 300s;

    location = /edit.html {
        auth_basic "Restricted";
        auth_basic_user_file /etc/nginx/.htpasswd;
    }
    location /api/ {
        proxy_pass http://python-api:8080;
        proxy_http_version 1.1;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_connect_timeout 600s;
        proxy_send_timeout 600s;
        proxy_read_timeout 600s;
    }
    # 首页
    location / {
        try_files $uri $uri/ =404;
    }

    error_page 404 = @go_home;
    location @go_home {
        return 301 /;
    }

    location /ws {
        proxy_pass http://xray:10002;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_read_timeout 300s;
        proxy_buffering off;
    }

    # LibreSpeed
    location /speed/ {
        proxy_pass http://librespeed:80/;
        proxy_http_version 1.1;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_buffering off;
    }

    # Certbot
    location /.well-known/acme-challenge/ {
        alias /var/www/certbot/.well-known/acme-challenge/;
        default_type "text/plain";
    }

    # 状态面板
    location /status/ {
        proxy_pass http://dashdot:3001/;
        proxy_http_version 1.1;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";    
    }

    location ~ ^/test.*\.bin$ {
        add_header Cache-Control "no-store, no-cache, must-revalidate, max-age=0";
        expires off;
    }
    location /ping {
        add_header Cache-Control "no-store, no-cache, must-revalidate, max-age=0";
        expires off;
        return 200;
    }

    access_log off;
    error_log /var/log/nginx/error.log warn;
}

    
🚀

clash 配置模板 /home/python/clash_template.yaml

clash_template.yaml
 mixed-port: 25555
allow-lan: true
mode: rule
log-level: info
dns:
  enable: true
  enhanced-mode: fake-ip
  fake-ip-range: 198.18.0.1/16
  nameserver:
    - 223.5.5.5
    - 119.29.29.29

# ---PROXIES---

rule-providers:
  cn:
    type: http
    behavior: classical
    url: https://raw.githubusercontent.com/Loyalsoldier/clash-rules/release/direct.txt
    path: ./rules/cn.yaml
    interval: 86400
  apple:
    type: http
    behavior: classical
    url: https://raw.githubusercontent.com/Loyalsoldier/clash-rules/release/apple.txt
    path: ./rules/apple.yaml
    interval: 86400
  proxy:
    type: http
    behavior: classical
    url: https://raw.githubusercontent.com/Loyalsoldier/clash-rules/release/proxy.txt
    path: ./rules/proxy.yaml
    interval: 86400
rules:
  - IP-CIDR,10.0.0.0/8,DIRECT,no-resolve
  - IP-CIDR,172.16.0.0/12,DIRECT,no-resolve
  - IP-CIDR,192.168.0.0/16,DIRECT,no-resolve
  - PROCESS-NAME,curl,DIRECT
  - PROCESS-NAME,python,DIRECT
  - PROCESS-NAME,python3,DIRECT
  - DOMAIN-SUFFIX,yourdomain.com,DIRECT
  - DOMAIN-SUFFIX,google.cn,AUTO
  - DOMAIN-SUFFIX,google.com,AUTO
  - DOMAIN-SUFFIX,googlevideo.com,AUTO
  - DOMAIN-SUFFIX,gstatic.com,AUTO
  - DOMAIN-SUFFIX,doubleclick.net,AUTO
  - DOMAIN-SUFFIX,gvt2.com,AUTO
  - DOMAIN-SUFFIX,gvt1.com,AUTO
  - DOMAIN-SUFFIX,googletagmanager.com,AUTO
  - DOMAIN-SUFFIX,googleapis.com,AUTO
  - RULE-SET,proxy,AUTO
  - RULE-SET,apple,DIRECT
  - RULE-SET,cn,DIRECT
  - GEOIP,CN,DIRECT
  - MATCH,AUTO
 
    
🚀

Xray 配置 /home/xray/config.json

config.json
 

{
  "log": {
    "loglevel": "warning",
    "access": "/var/log/xray/access.log",
    "error": "/var/log/xray/error.log"
  },
  "dns": {
    "servers": [
      "1.1.1.1","8.8.8.8"
    ]
  },
  "inbounds": [
    {
      "tag": "ws-in",
      "listen": "0.0.0.0",
      "port": 10002,
      "protocol": "vless",
      "settings": {
        "clients": [
          {
            "id": "yourUUID"
          }
        ],
        "decryption": "none"
      },
      "streamSettings": {
        "network": "ws",
        "security": "none",
        "wsSettings": {
          "path": "/ws",
          "headers": {}
        }
      }
    }
  ],
  "outbounds": [
    {
      "tag": "direct",
      "protocol": "freedom"
    },
    {
      "tag": "warp-out",
      "protocol": "socks",
      "settings": {
        "servers": [
          {
            "address": "warp",
            "port": 1080
          }
        ]
      }
    }
  ],
  "routing": {
    "domainStrategy": "AsIs",
    "rules": [
      {
        "type": "field",
        "domain": [
          "domain:openai.com",
          "domain:chatgpt.com",
          "domain:oaiusercontent.com",
          "domain:anthropic.com",
          "domain:claude.ai",
          "domain:poe.com",
          "domain:poecdn.net",
          "domain:perplexity.ai",
          "domain:googleapis.com",
          "domain:ai.google.dev",
          "domain:gemini.google.com",
          "domain:aistudio.google.com",
          "domain:generativelanguage.googleapis.com",
          "domain:huggingface.co",
          "domain:hf.co",
          "domain:cdn-lfs.huggingface.co",
          "domain:*.hf.space",
          "domain:replicate.com",
          "domain:stability.ai",
          "domain:cohere.ai",
          "domain:character.ai",
          "domain:runpod.io",
          "domain:together.ai",
          "domain:groq.com",
          "domain:workers.dev",
          "domain:disneyplus.com",
          "domain:disney-plus.net",
          "domain:bamgrid.com",
          "domain:reddit.com",
          "domain:redd.it",
          "domain:redditmedia.com",
          "domain:redditstatic.com",
          "domain:reuters.com"
        ],
        "outboundTag": "warp-out"
      },
      {
        "type": "field",
        "network": "tcp,udp",
        "outboundTag": "direct"
      }
    ]
  }
}

    
🔐

Certbot 首次申请证书

bash
# 申请证书
docker run --rm -it \
  -v /home/cert/conf:/etc/letsencrypt \
  -v /home/cert/www:/var/www/certbot \
  certbot/certbot certonly \
  --webroot -w /var/www/certbot \
  -d yourdomain.com -d www.yourdomain.com

# 查看证书详情 & 过期时间
openssl x509 -in /home/cert/conf/live/yourdomain.com/fullchain.pem -noout -text
🐍

Python API /home/python/server.py

ℹ️
/edit.html 管理页面由 nginx HTTP basic auth 保护;当前订阅 API 未启用 token 验证,建议仅对可信用户开放。
server.py 完整代码
 

from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import os
import subprocess
from datetime import datetime

# =========================
# 全局配置区(唯一真源)
# =========================
CONFIG = {
    "":   {"file": "domains.json"},
    "CM": {"file": "domainsCM.json"},
    "CU": {"file": "domainsCU.json"},
    "CT": {"file": "domainsCT.json"},
}
XRAY_CONFIG = "/app/xray_config.json"
CLASH_TEMPLATE = "clash_template.yaml"
UUID = "yourUUID"
SERVER_IP = "yourdomain.top"
PORT = 443
PATH = "/ws"
CERT_RENEW_TIMEOUT_SECONDS = 600

# 综合评分权重(丢包率惩罚 300ms,与 GetBestDomains 一致)
WEIGHT_LATENCY  = 0.6
WEIGHT_LOSS     = 0.4
LOSS_PENALTY_MS = 300

# colo 代码 → 可读地名(不在表里的直接用 colo 代码)
COLO_NAME = {
    "HKG": "Hong Kong", "MFM": "Macau", "TPE": "Taipei", "TSA": "Taipei Songshan",
    "NRT": "Tokyo", "HND": "Tokyo Haneda", "KIX": "Osaka", "NGO": "Nagoya", "FUK": "Fukuoka", "CTS": "Sapporo",
    "ICN": "Seoul", "GMP": "Seoul Gimpo", "PUS": "Busan",
    "SIN": "Singapore",
    "LAX": "Los Angeles", "SFO": "San Francisco", "SJC": "San Jose", "OAK": "Oakland",
    "SEA": "Seattle", "PDX": "Portland", "SAN": "San Diego",
    "DFW": "Dallas", "DEN": "Denver", "PHX": "Phoenix", "LAS": "Las Vegas",
    "ORD": "Chicago", "MSP": "Minneapolis", "DTW": "Detroit",
    "ATL": "Atlanta", "MIA": "Miami", "IAD": "Washington DC",
    "EWR": "New York", "BOS": "Boston", "PHL": "Philadelphia", "CLT": "Charlotte", "MCO": "Orlando",
    "YVR": "Vancouver", "YYZ": "Toronto", "YUL": "Montreal",
    "HNL": "Honolulu",
    "LHR": "London", "CDG": "Paris", "FRA": "Frankfurt",
    "AMS": "Amsterdam", "MAD": "Madrid",
    "DUB": "Dublin", "BRU": "Brussels", "ZRH": "Zurich",
    "VIE": "Vienna", "CPH": "Copenhagen", "ARN": "Stockholm",
    "OSL": "Oslo", "HEL": "Helsinki",
    "WAW": "Warsaw", "PRG": "Prague", "BUD": "Budapest",
    "ATH": "Athens", "IST": "Istanbul",
    "MXP": "Milan", "FCO": "Rome",
    "BCN": "Barcelona", "LIS": "Lisbon",
    "BKK": "Bangkok", "KUL": "Kuala Lumpur", "MNL": "Manila", "SGN": "Ho Chi Minh City",
    "CGK": "Jakarta", "DPS": "Bali",
    "HAN": "Hanoi", "VTE": "Vientiane", "PNH": "Phnom Penh", "RGN": "Yangon",
    "BOM": "Mumbai", "DEL": "New Delhi", "MAA": "Chennai", "BLR": "Bangalore",
    "HYD": "Hyderabad", "CCU": "Kolkata",
    "DXB": "Dubai", "AUH": "Abu Dhabi", "DOH": "Doha",
    "RUH": "Riyadh", "JED": "Jeddah", "TLV": "Tel Aviv",
    "JNB": "Johannesburg", "CPT": "Cape Town",
    "CAI": "Cairo", "LOS": "Lagos",
    "NBO": "Nairobi", "ADD": "Addis Ababa",
    "CMN": "Casablanca",
    "GRU": "Sao Paulo", "GIG": "Rio de Janeiro",
    "EZE": "Buenos Aires", "SCL": "Santiago",
    "LIM": "Lima", "BOG": "Bogota",
    "SYD": "Sydney", "MEL": "Melbourne",
    "BNE": "Brisbane", "PER": "Perth",
    "AKL": "Auckland", "WLG": "Wellington",
}

# =========================
# colo → 地区映射
# =========================
REGION_MAP = {
    # 港澳
    "HKG": "HONGKONG", "MFM": "HONGKONG",

    # 东亚(台 / 日 / 韩)
    "TPE": "EastAsia", "TSA": "EastAsia", "SIN": "EastAsia",
    "NRT": "EastAsia", "HND": "EastAsia", "KIX": "EastAsia",
    "NGO": "EastAsia", "FUK": "EastAsia", "CTS": "EastAsia",
    "ICN": "EastAsia", "GMP": "EastAsia", "PUS": "EastAsia",

    # 北美(美 / 加 / 墨)
    "LAX": "NorthAmerica", "SFO": "NorthAmerica", "SJC": "NorthAmerica", "OAK": "NorthAmerica",
    "SEA": "NorthAmerica", "PDX": "NorthAmerica", "SAN": "NorthAmerica",
    "DFW": "NorthAmerica", "DEN": "NorthAmerica", "PHX": "NorthAmerica", "LAS": "NorthAmerica",
    "ORD": "NorthAmerica", "MSP": "NorthAmerica", "DTW": "NorthAmerica",
    "ATL": "NorthAmerica", "MIA": "NorthAmerica", "IAD": "NorthAmerica",
    "EWR": "NorthAmerica", "BOS": "NorthAmerica", "PHL": "NorthAmerica",
    "CLT": "NorthAmerica", "MCO": "NorthAmerica",
    "YVR": "NorthAmerica", "YYZ": "NorthAmerica", "YUL": "NorthAmerica",
    "HNL": "NorthAmerica",

    # 欧洲
    "LHR": "Europe", "CDG": "Europe", "FRA": "Europe",
    "AMS": "Europe", "MAD": "Europe",
    "DUB": "Europe", "BRU": "Europe", "ZRH": "Europe",
    "VIE": "Europe", "CPH": "Europe", "ARN": "Europe",
    "OSL": "Europe", "HEL": "Europe",
    "WAW": "Europe", "PRG": "Europe", "BUD": "Europe",
    "ATH": "Europe", "IST": "Europe",
    "MXP": "Europe", "FCO": "Europe",
    "BCN": "Europe", "LIS": "Europe",

    # 东南亚 / 南亚
    "BKK": "SoutheastAsia", "KUL": "SoutheastAsia", "MNL": "SoutheastAsia", "SGN": "SoutheastAsia",
    "CGK": "SoutheastAsia", "DPS": "SoutheastAsia",
    "HAN": "SoutheastAsia", "VTE": "SoutheastAsia", "PNH": "SoutheastAsia", "RGN": "SoutheastAsia",
    "BOM": "SoutheastAsia", "DEL": "SoutheastAsia", "MAA": "SoutheastAsia", "BLR": "SoutheastAsia",
    "HYD": "SoutheastAsia", "CCU": "SoutheastAsia",

    # 中东
    "DXB": "MiddleEast", "AUH": "MiddleEast", "DOH": "MiddleEast",
    "RUH": "MiddleEast", "JED": "MiddleEast", "TLV": "MiddleEast",

    # 非洲
    "JNB": "Africa", "CPT": "Africa",
    "CAI": "Africa", "LOS": "Africa",
    "NBO": "Africa", "ADD": "Africa",
    "CMN": "Africa",

    # 南美
    "GRU": "SouthAmerica", "GIG": "SouthAmerica",
    "EZE": "SouthAmerica", "SCL": "SouthAmerica",
    "LIM": "SouthAmerica", "BOG": "SouthAmerica",

    # 大洋洲
    "SYD": "Oceania", "MEL": "Oceania",
    "BNE": "Oceania", "PER": "Oceania",
    "AKL": "Oceania", "WLG": "Oceania",
}

# 地区显示顺序(控制 proxy-groups 里地区组的排列)
REGION_ORDER = [
    "HONGKONG",
    "EastAsia",
    "SoutheastAsia",
    "NorthAmerica",
    "Europe",
    "MiddleEast",
    "Africa",
    "SouthAmerica",
    "Oceania",
]

def get_region(colo):
    """返回 colo 所属地区标签,未知 colo 返回 None(不建分组)"""
    return REGION_MAP.get(colo.upper())

# =========================
# 初始化 JSON 文件
# =========================
def ensure_files():
    for v in CONFIG.values():
        if not os.path.exists(v["file"]):
            with open(v["file"], "w") as f:
                f.write("[]")
    if not os.path.exists(CLASH_TEMPLATE):
        with open(CLASH_TEMPLATE, "w") as f:
            f.write("# ---PROXIES---\n")

ensure_files()

# =========================
# 工具函数
# =========================
def read_file(path, default=""):
    try:
        with open(path) as f:
            return f.read()
    except:
        return default

def write_file(path, data):
    with open(path, "w") as f:
        f.write(data)

def read_json(path):
    try:
        with open(path) as f:
            return json.load(f)
    except:
        return []

def write_json(path, data):
    with open(path, "w") as f:
        json.dump(data, f)

# =========================
# ⭐ 动态文件解析
# =========================
def get_dynamic_file(prefix, path):
    if not path.startswith(prefix):
        return None
    name = path[len(prefix):].strip("/")
    if not name:
        return None
    if ".." in name or "/" in name:
        return None
    return f"{name}.json"

# =========================
# 节点数据结构适配
# =========================
def normalize_nodes(raw_list):
    """
    兼容新旧两种数据格式:
      旧格式:["1.2.3.4", "example.com", ...]  → 仅有 ip/domain,无探测数据
      新格式:[{"ip":"1.2.3.4","colo":"HKG","lat":88.2,"loss":0.0,"source":"xx.com"}, ...]
    统一返回 node dict 列表,保证每个节点都有:
      ip, colo, lat, loss, score, source, name(节点显示名)
    """
    nodes = []
    for item in raw_list:
        if isinstance(item, str):
            # 旧格式纯字符串
            nodes.append({
                "ip":     item,
                "colo":   "",
                "lat":    9999.0,
                "loss":   0.0,
                "score":  9999.0,
                "source": "",
                "name":   item,
            })
        elif isinstance(item, dict):
            ip     = item.get("ip", "").strip()
            colo   = item.get("colo", "").strip().upper()
            lat    = float(item.get("lat", 9999))
            loss   = float(item.get("loss", 0))
            source = item.get("source", "").strip()
            score  = lat * WEIGHT_LATENCY + loss * 100 * LOSS_PENALTY_MS * WEIGHT_LOSS
            colo_label = COLO_NAME.get(colo, colo) if colo else ""
            # 节点名:优先 colo 地名,没有就用 ip
            name = f"{colo_label}-{ip}" if colo_label else ip
            nodes.append({
                "ip":     ip,
                "colo":   colo,
                "lat":    round(lat, 1),
                "loss":   round(loss, 2),
                "score":  round(score, 2),
                "source": source,
                "name":   name,
            })
    return [n for n in nodes if n["ip"]]

def node_score(n):
    """综合评分,越低越好"""
    return n["score"]

def group_by_colo(nodes):
    """按 colo 分组,返回 {colo: [node, ...]} 按 score 排序"""
    groups = {}
    for n in nodes:
        colo = n["colo"] or "UNKNOWN"
        groups.setdefault(colo, []).append(n)
    for colo in groups:
        groups[colo].sort(key=node_score)
    return groups

def top_nodes_global(nodes, n):
    """全局去重(同IP只取最优)后取前 n 个"""
    seen = {}
    for node in nodes:
        ip = node["ip"]
        if ip not in seen or node["score"] < seen[ip]["score"]:
            seen[ip] = node
    deduped = sorted(seen.values(), key=node_score)
    return deduped[:n]

# =========================
# VLESS 订阅生成
# 按地区配额选优:
#   港澳       top 5
#   北美       top 5
#   东亚(日韩台新) top 5
#   欧洲       top 2
# 旧格式(无 colo):全部输出
# =========================

# colo → 订阅地区分类(与 REGION_MAP 独立,颗粒度更细)
VLESS_REGION_MAP = {
    # 港澳
    "HKG": "hkmo", "MFM": "hkmo",

    # 东亚:日本 / 韩国 / 台湾 / 新加坡
    "TPE": "eastasia", "TSA": "eastasia",
    "NRT": "eastasia", "HND": "eastasia", "KIX": "eastasia",
    "NGO": "eastasia", "FUK": "eastasia", "CTS": "eastasia",
    "ICN": "eastasia", "GMP": "eastasia", "PUS": "eastasia",
    "SIN": "eastasia",

    # 北美
    "LAX": "nam", "SFO": "nam", "SJC": "nam", "OAK": "nam",
    "SEA": "nam", "PDX": "nam", "SAN": "nam",
    "DFW": "nam", "DEN": "nam", "PHX": "nam", "LAS": "nam",
    "ORD": "nam", "MSP": "nam", "DTW": "nam",
    "ATL": "nam", "MIA": "nam", "IAD": "nam",
    "EWR": "nam", "BOS": "nam", "PHL": "nam",
    "CLT": "nam", "MCO": "nam",
    "YVR": "nam", "YYZ": "nam", "YUL": "nam",
    "HNL": "nam",

    # 欧洲
    "LHR": "eu", "CDG": "eu", "FRA": "eu",
    "AMS": "eu", "MAD": "eu",
    "DUB": "eu", "BRU": "eu", "ZRH": "eu",
    "VIE": "eu", "CPH": "eu", "ARN": "eu",
    "OSL": "eu", "HEL": "eu",
    "WAW": "eu", "PRG": "eu", "BUD": "eu",
    "ATH": "eu", "IST": "eu",
    "MXP": "eu", "FCO": "eu",
    "BCN": "eu", "LIS": "eu",
}

# 各地区配额 & 显示标签(顺序即输出顺序)
VLESS_REGION_QUOTA = [
    ("hkmo",    4, "HONGKONG"),
    ("eastasia", 1, "EastAsia"),
    ("nam",     4, "NorthAmerica"),
    ("eu",      1, "Europe"),
]

def pick_top_by_region(nodes, region_key, quota):
    """
    从 nodes(已全局去重、按 score 升序)中,
    按 VLESS_REGION_MAP 过滤出指定地区的节点,去重后取 top quota 个。
    """
    seen_ip = set()
    result  = []
    for n in nodes:
        vr = VLESS_REGION_MAP.get(n["colo"])
        if vr != region_key:
            continue
        if n["ip"] in seen_ip:
            continue
        seen_ip.add(n["ip"])
        result.append(n)
        if len(result) >= quota:
            break
    return result

def generate_vless(raw_list):
    nodes = normalize_nodes(raw_list)
    if not nodes:
        return ""

    has_colo = any(n["colo"] for n in nodes)

    if not has_colo:
        # 旧格式:无 colo 信息,全部输出
        selected = nodes
    else:
        # 全局去重并按综合评分排序
        seen = {}
        for n in nodes:
            ip = n["ip"]
            if ip not in seen or n["score"] < seen[ip]["score"]:
                seen[ip] = n
        sorted_nodes = sorted(seen.values(), key=node_score)

        selected = []
        for region_key, quota, label in VLESS_REGION_QUOTA:
            top = pick_top_by_region(sorted_nodes, region_key, quota)
            selected.extend(top)

    lines = []
    used_names = {}
    for n in selected:
        base = n["name"]
        cnt  = used_names.get(base, 0) + 1
        used_names[base] = cnt
        display = base if cnt == 1 else f"{base}-{cnt}"

        line = (
            f"vless://{UUID}@{n['ip']}:{PORT}"
            f"?security=tls&type=ws"
            f"&host={SERVER_IP}"
            f"&sni={SERVER_IP}"
            f"&path={PATH}"
            f"#{display}"
        )
        lines.append(line)

    return "\n".join(lines)


# =========================
# Clash 订阅生成
# - 每个 colo 建独立分组,取前5
# - PROXIES 默认分组:全局综合最优8个
# - 保留 URLTEST / FALLBACK / LOADBALANCE / AUTO
# =========================
def generate_clash(raw_list):
    template = read_file(CLASH_TEMPLATE)
    nodes = normalize_nodes(raw_list)
    manual_raw = read_json(CONFIG[""]["file"])
    manual_nodes = normalize_nodes(manual_raw)
    if not nodes:
        return template.replace("# ---PROXIES---", "proxies: []\nproxy-groups: []")

    has_colo = any(n["colo"] for n in nodes)

    # ===== 1. 全量去重排序 =====
    seen = {}
    for n in nodes:
        ip = n["ip"]
        if ip not in seen or n["score"] < seen[ip]["score"]:
            seen[ip] = n
    all_nodes = sorted(seen.values(), key=node_score)

    # ===== 2. 确定实际需要写入 proxies 块的节点集合 =====
    # PROXIES/FALLBACK/URLTEST/LOADBALANCE 共用 top8
    proxies_top = all_nodes[:8]
    proxies_top_ips = {n["ip"] for n in proxies_top}

    # 每个地区的节点集合(先按 score 排序,后面取 top N)
    region_map_nodes = {}  # region_label → [node, ...]
    if has_colo:
        for n in all_nodes:
            region = get_region(n["colo"])
            if region:
                region_map_nodes.setdefault(region, []).append(n)
    # 每个地区去重取 top8
    region_top_map = {}  # region_label → [node, ...]
    for region, ns in region_map_nodes.items():
        seen_ip2 = set()
        top_list = []
        for n in ns:
            if n["ip"] not in seen_ip2:
                seen_ip2.add(n["ip"])
                top_list.append(n)
                if len(top_list) >= 8:
                    break
        if top_list:
            region_top_map[region] = top_list

    # 所有被引用的 IP = top8(全局) ∪ 各地区 top8
    used_ips = set(proxies_top_ips)
    for ns in region_top_map.values():
        for n in ns:
            used_ips.add(n["ip"])

    # 按全局 score 排序,只保留被引用的节点
    used_nodes = [n for n in all_nodes if n["ip"] in used_ips]
    used_ip_set = {x["ip"] for x in used_nodes}
    for n in manual_nodes:
        if n["ip"] not in used_ip_set:
            used_nodes.append(n)
            used_ip_set.add(n["ip"])

    # ===== 3. 生成 proxy 块(仅 used_nodes)=====
    used_names   = {}
    node_to_name = {}

    def get_proxy_name(n):
        base = n["name"]
        cnt  = used_names.get(base, 0) + 1
        used_names[base] = cnt
        return base if cnt == 1 else f"{base}-{cnt}"

    proxy_lines = []
    for n in used_nodes:
        pname = get_proxy_name(n)
        node_to_name[n["ip"]] = pname
        proxy_lines += [
            f"  - name: \"{pname}\"",
            "    type: vless",
            f"    server: {n['ip']}",
            f"    servername: {SERVER_IP}",
            f"    port: {PORT}",
            f"    uuid: {UUID}",
            "    network: ws",
            "    tls: true",
            "    udp: true",
            "    ws-opts:",
            f"      path: {PATH}",
            "      headers:",
            f"        Host: {SERVER_IP}",
        ]

    def names_block(ns, indent=8):
        pad = " " * indent
        return "\n".join(f"{pad}- \"{node_to_name[n['ip']]}\"" for n in ns if n["ip"] in node_to_name)

    # ===== 4. 按地区分组 yaml =====
    region_groups_yaml = []
    # 按 REGION_ORDER 定义的顺序排列,有节点的地区才输出
    ordered_regions = [r for r in REGION_ORDER if r in region_top_map]
    # REGION_ORDER 里没有的地区(理论上不会有,但防御一下)
    extra_regions = [r for r in sorted(region_top_map) if r not in REGION_ORDER]
    for region in ordered_regions + extra_regions:
        top_nodes = region_top_map[region]
        region_groups_yaml += [
            f"  - name: \"{region}\"",
            "    type: url-test",
            "    url: https://la-no-cdn.yourdomain.top/test.bin",
            "    interval: 180",
            "    tolerance: 1",
            "    proxies:",
            names_block(top_nodes),
        ]

    # ===== 5. 策略分组(PROXIES/FALLBACK/URLTEST/LOADBALANCE 共用 top8)=====
    proxies_names_block = names_block(proxies_top)

    # 收集地区分组名,用于 AUTO
    region_group_names = ordered_regions + extra_regions

    auto_proxies = (
        ["      - URLTEST", "      - FALLBACK", "      - LOADBALANCE", "      - PROXIES", "      - MANUAL"]
        + [f"      - \"{g}\"" for g in region_group_names]
        + ["      - DIRECT"]
    )

    manual_names = []
    for n in manual_nodes:
        if n["ip"] in node_to_name:
            manual_names.append(f'      - "{node_to_name[n["ip"]]}"')
    
    manual_group = [
        "  - name: MANUAL",
        "    type: select",
        "    proxies:",
    ] + manual_names if manual_names else []

    strategy_groups = manual_group +[
        "  - name: PROXIES",
        "    type: select",
        "    proxies:",
        proxies_names_block,
        "",
        "  - name: FALLBACK",
        "    type: fallback",
        "    url: https://la-no-cdn.yourdomain.top/test.bin",
        "    interval: 90",
        "    proxies:",
        proxies_names_block,
        "",
        "  - name: URLTEST",
        "    type: url-test",
        "    url: https://la-no-cdn.yourdomain.top/test.bin",
        "    interval: 150",
        "    tolerance: 1",
        "    proxies:",
        proxies_names_block,
        "",
        "  - name: LOADBALANCE",
        "    type: load-balance",
        "    strategy: round-robin",
        "    proxies:",
        proxies_names_block,
        "",
        "  - name: AUTO",
        "    type: select",
        "    proxies:",
        "\n".join(auto_proxies),
    ]

    # ===== 6. 拼装 =====
    proxy_block  = "proxies:\n" + "\n".join(proxy_lines)
    groups_block = "proxy-groups:\n" + "\n".join(region_groups_yaml) + (
        "\n" if region_groups_yaml else ""
    ) + "\n".join(strategy_groups)

    final_yaml = template.replace(
        "# ---PROXIES---",
        proxy_block + "\n\n" + groups_block
    )
    return final_yaml

# =========================
# HTTP 处理器
# =========================
class Handler(BaseHTTPRequestHandler):
    def send_text(self, text, content_type="text/plain"):
        self.send_response(200)
        self.send_header("Content-Type", content_type)
        self.end_headers()
        self.wfile.write(text.encode())

    def do_GET(self):
        # ===== CONFIG 固定路由 =====
        for key, cfg in CONFIG.items():
            if self.path == f"/api/domains{key}":
                data = read_json(cfg["file"])
                return self.send_text(json.dumps(data), "application/json")
            if self.path == f"/api/sub{key}":
                nodes = read_json(cfg["file"])
                return self.send_text(generate_vless(nodes))
            if self.path == f"/api/clash{key}":
                nodes = read_json(cfg["file"])
                return self.send_text(generate_clash(nodes))

        # ===== 动态路由 =====
        if self.path.startswith("/api/domains/"):
            # clash/template 优先匹配,不要被 /api/domains/ 前缀误吃
            pass
        if self.path.startswith("/api/domains/") and not self.path == "/api/clash/template":
            file = get_dynamic_file("/api/domains/", self.path)
            if not file:
                return self._404()
            if not os.path.exists(file):
                write_json(file, [])
            data = read_json(file)
            return self.send_text(json.dumps(data), "application/json")

        if self.path.startswith("/api/sub/"):
            file = get_dynamic_file("/api/sub/", self.path)
            if not file:
                return self._404()
            nodes = read_json(file) if os.path.exists(file) else []
            return self.send_text(generate_vless(nodes))

        if self.path == "/api/clash/template":
            return self.send_text(read_file(CLASH_TEMPLATE))

        if self.path.startswith("/api/clash/"):
            file = get_dynamic_file("/api/clash/", self.path)
            if not file:
                return self._404()
            nodes = read_json(file) if os.path.exists(file) else []
            return self.send_text(generate_clash(nodes))

        if self.path == "/api/xray/config":
            return self.send_text(read_file(XRAY_CONFIG), "application/json")

        if self.path == "/api/cert/info":
            data = get_cert_info()
            return self.send_text(json.dumps(data), "application/json")

        self.send_response(404)
        self.end_headers()

    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        body   = self.rfile.read(length)

        # ===== CONFIG 固定路由 =====
        for key, cfg in CONFIG.items():
            if self.path == f"/api/domains{key}":
                write_json(cfg["file"], json.loads(body))
                self.send_response(200)
                self.end_headers()
                return

        # ===== 动态写入 =====
        if self.path.startswith("/api/domains/"):
            file = get_dynamic_file("/api/domains/", self.path)
            if not file:
                return self._404()
            write_json(file, json.loads(body))
            self.send_response(200)
            self.end_headers()
            return

        # ===== Xray config =====
        if self.path == "/api/xray/config":
            write_file(XRAY_CONFIG, body.decode())
            self.send_response(200)
            self.end_headers()
            return

        # ===== Clash 模板 =====
        if self.path == "/api/clash/template":
            write_file(CLASH_TEMPLATE, body.decode())
            self.send_response(200)
            self.end_headers()
            return

        # ===== Docker restart =====
        if self.path == "/api/docker/restart":
            subprocess.call(["docker", "restart", "xray", "dashdot", "librespeed", "certbot"])
            self.send_response(200)
            self.end_headers()
            return

        # ===== Nginx restart =====
        if self.path == "/api/nginx/restart":
            result = restart_nginx_async()
            self.send_text(json.dumps(result), "application/json")
            return

        # ===== Certificate renew =====
        if self.path == "/api/cert/renew":
            result = renew_certificates()
            self.send_text(json.dumps(result), "application/json")
            return

        self.send_response(404)
        self.end_headers()

    def _404(self):
        self.send_response(404)
        self.end_headers()

# =========================
# 证书管理函数
# =========================
def get_cert_info():
    """从 /etc/letsencrypt/live/ 目录读取证书信息"""
    try:
        import re

        def extract_cert_domains(cert_text, fallback_domain):
            """Extract all DNS names covered by the certificate SAN extension."""
            san_lines = []
            lines = cert_text.splitlines()
            for idx, line in enumerate(lines):
                if "Subject Alternative Name" not in line:
                    continue

                inline_san = line.split("Subject Alternative Name:", 1)[1].strip()
                if inline_san and inline_san.lower() != "critical":
                    san_lines.append(inline_san)

                for san_line in lines[idx + 1:]:
                    stripped = san_line.strip()
                    if not stripped:
                        continue
                    if stripped.startswith("X509v3 ") or stripped.startswith("Signature Algorithm"):
                        break
                    san_lines.append(stripped)
                break

            names = re.findall(r"DNS:([^,\s]+)", " ".join(san_lines))

            if not names:
                subject_match = re.search(r"Subject:\s*(.+?)(?=\n|$)", cert_text)
                if subject_match:
                    cn_match = re.search(r"CN\s*=\s*([^,]+)", subject_match.group(1))
                    if cn_match:
                        names.append(cn_match.group(1).strip())

            if not names and fallback_domain:
                names.append(fallback_domain)

            seen = set()
            return [name for name in names if not (name in seen or seen.add(name))]
        
        # 支持多个可能的路径
        possible_paths = [
            "/etc/letsencrypt/live",
            "/home/cert/conf/live"
        ]
        
        cert_dir = None
        for path in possible_paths:
            if os.path.exists(path):
                cert_dir = path
                break
        
        if not cert_dir:
            return {"domains": [], "error": f"证书目录不存在 (尝试过: {', '.join(possible_paths)})"}
        
        domains = []
        for domain_dir in os.listdir(cert_dir):
            domain_path = os.path.join(cert_dir, domain_dir)
            if not os.path.isdir(domain_path):
                continue
            
            # 优先使用 fullchain.pem,其次 cert.pem
            cert_file = None
            for filename in ["fullchain.pem", "cert.pem"]:
                candidate = os.path.join(domain_path, filename)
                if os.path.exists(candidate):
                    cert_file = candidate
                    break
            
            if not cert_file:
                continue
            
            try:
                # 使用 openssl 提取证书信息
                result = subprocess.run(
                    ["openssl", "x509", "-in", cert_file, "-text", "-noout"],
                    capture_output=True, text=True, timeout=5
                )
                
                if result.returncode != 0:
                    continue
                
                cert_text = result.stdout
                
                # 提取信息 - 使用正则表达式以处理不同的格式
                issuer = ""
                valid_from = ""
                valid_to = ""
                
                # Issuer: C = US, O = Let's Encrypt, CN = E8
                issuer_match = re.search(r'Issuer:\s*(.+?)(?=\n|$)', cert_text)
                if issuer_match:
                    issuer = issuer_match.group(1).strip()
                
                # Not Before: Apr  6 09:12:16 2026 GMT
                not_before_match = re.search(r'Not Before:\s*(.+?)(?=\n|$)', cert_text)
                if not_before_match:
                    valid_from = not_before_match.group(1).strip()
                
                # Not After : Jul  5 09:12:15 2026 GMT
                not_after_match = re.search(r'Not After\s*:\s*(.+?)(?=\n|$)', cert_text)
                if not_after_match:
                    valid_to = not_after_match.group(1).strip()
                
                # 计算剩余天数
                try:
                    # 从日期字符串中提取标准格式 "Mon DD HH:MM:SS YYYY"
                    # 原始格式:Apr  6 09:12:16 2026 GMT(可能有多个空格)
                    date_pattern = r'(\w+)\s+(\d+)\s+(\d+):(\d+):(\d+)\s+(\d+)'
                    
                    match = re.search(date_pattern, valid_to)
                    if match:
                        month_str, day, hour, minute, second, year = match.groups()
                        # 构建标准日期字符串
                        normalized_date = f"{month_str} {int(day):02d} {hour}:{minute}:{second} {year}"
                        expires_at = datetime.strptime(normalized_date, "%b %d %H:%M:%S %Y")
                        now = datetime.utcnow()
                        days_left = max(0, (expires_at - now).days)
                    else:
                        print(f"Date pattern not matched for: {valid_to}")
                        days_left = -1
                        expires_at = None
                except Exception as e:
                    print(f"Error parsing date {valid_to}: {e}")
                    days_left = -1
                    expires_at = None
                
                domains.append({
                    "domain": domain_dir,
                    "covered_domains": extract_cert_domains(cert_text, domain_dir),
                    "issuer": issuer,
                    "valid_from": valid_from,
                    "valid_to": valid_to,
                    "expires_at": expires_at.isoformat() if expires_at else valid_to,
                    "days_left": days_left
                })
            except Exception as e:
                print(f"Error parsing certificate for {domain_dir}: {e}")
                continue
        
        return {"domains": sorted(domains, key=lambda x: x["days_left"])}
    
    except Exception as e:
        return {"error": f"获取证书信息失败: {str(e)}"}

def renew_certificates():
    """通过 docker exec 执行 certbot renew"""
    def clean_output(value):
        if value is None:
            return ""
        if isinstance(value, bytes):
            return value.decode(errors="replace")
        return str(value)

    try:
        # 使用 docker exec 在 certbot 容器中执行 renew 命令
        try:
            result = subprocess.run(
                [
                    "docker", "exec", "certbot",
                    "certbot", "renew",
                    "--webroot", "-w", "/var/www/certbot",
                    "--non-interactive", "--agree-tos",
                    "--no-random-sleep-on-renew"
                ],
                capture_output=True,
                text=True,
                timeout=CERT_RENEW_TIMEOUT_SECONDS
            )
        except subprocess.TimeoutExpired as e:
            output = clean_output(e.stdout) + clean_output(e.stderr)
            message = f"certbot renew 超过 {CERT_RENEW_TIMEOUT_SECONDS} 秒,已停止等待。"
            if output.strip():
                message += "\n\n超时前输出:\n" + output
            else:
                message += "\n\n没有捕获到 certbot 输出。"
            return {
                "success": False,
                "timeout": True,
                "stage": "certbot_renew",
                "message": message,
                "timeout_seconds": CERT_RENEW_TIMEOUT_SECONDS
            }

        output = result.stdout + result.stderr
        success = result.returncode == 0
        response = {
            "success": success,
            "message": output or ("证书续期成功" if success else "证书续期过程中发生错误"),
            "returncode": result.returncode,
            "timeout_seconds": CERT_RENEW_TIMEOUT_SECONDS
        }

        return response

    except Exception as e:
        return {"success": False, "message": f"执行失败: {str(e)}"}

def restart_nginx_async():
    """后台重启 nginx,避免当前 nginx 代理请求被重启过程打断。"""
    try:
        subprocess.Popen(
            ["sh", "-c", "sleep 1; docker restart nginx"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True
        )
        return {
            "success": True,
            "message": "nginx 重启任务已提交,页面可能会短暂断开,请稍后刷新证书信息。"
        }
    except Exception as e:
        return {"success": False, "message": f"提交 nginx 重启失败: {str(e)}"}

# =========================
# 启动服务器
# =========================
if __name__ == "__main__":
    print("Python API server running on 0.0.0.0:8080")
    HTTPServer(("0.0.0.0", 8080), Handler).serve_forever()


    
🖥

订阅地址

订阅地址格式