main.py

#!/usr/bin/env python3
"""
Cloudflare IP 优选工具 (TCP筛选 + IP可用性二次筛选 + HTTP检测 + curl带宽测速 + WxPusher通知)
依赖：requests, curl, aiohttp
配置文件：同目录下的 config.json
结果保存到 ProxyIP_A.txt，并自动推送到 GitHub，同时批量更新到 Cloudflare DNS
支持 Windows / Linux
"""

import requests
import socket
import time
import sys
import re
import os
import subprocess
import shutil
import json
import asyncio
import aiohttp
import ipaddress
import warnings
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse
from urllib3.exceptions import InsecureRequestWarning
from requests.adapters import HTTPAdapter
from urllib3.connection import HTTPSConnection

if sys.platform == 'win32':
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
warnings.filterwarnings("ignore", message="Unverified HTTPS request")

_original_session_request = requests.Session.request

def _new_session_request(self, method, url, *args, **kwargs):
    kwargs.setdefault("verify", False)
    return _original_session_request(self, method, url, *args, **kwargs)

requests.Session.request = _new_session_request

_IS_TTY = hasattr(sys.stdout, 'isatty') and sys.stdout.isatty()
_last_progress_len = 0

def log(level, msg, end="\n"):
    global _last_progress_len
    line = f"[{time.strftime('%H:%M:%S')}] [{level}] {msg}"
    if end == "\n" and level != "进度":
        print(line)
    elif end == "\n" and level == "进度":
        if _IS_TTY:
            pad = max(0, _last_progress_len - len(line))
            print(f"\r{line}{' ' * pad}")
        else:
            print(line)
        _last_progress_len = 0
    elif _IS_TTY:
        pad = max(0, _last_progress_len - len(line))
        print(f"\r{line}{' ' * pad}", end="", flush=True)
        _last_progress_len = len(line)

NODE_PATTERN = re.compile(r"^(\d+\.\d+\.\d+\.\d+):(\d+)#(.+)$")
IP_PORT_PATTERN = re.compile(r"^(\d+\.\d+\.\d+\.\d+):(\d+)#")

CN_TO_CODE = {
    "阿富汗": "AF", "奥兰群岛": "AX", "阿尔巴尼亚": "AL", "阿尔及利亚": "DZ",
    "美属萨摩亚": "AS", "安道尔": "AD", "安哥拉": "AO", "安圭拉": "AI",
    "南极洲": "AQ", "安提瓜和巴布达": "AG", "阿根廷": "AR", "亚美尼亚": "AM",
    "阿鲁巴": "AW", "澳大利亚": "AU", "奥地利": "AT", "阿塞拜疆": "AZ",
    "巴哈马": "BS", "巴林": "BH", "孟加拉国": "BD", "孟加拉": "BD",
    "巴巴多斯": "BB", "白俄罗斯": "BY", "比利时": "BE", "伯利兹": "BZ",
    "贝宁": "BJ", "百慕大": "BM", "不丹": "BT", "玻利维亚": "BO",
    "波黑": "BA", "波斯尼亚和黑塞哥维那": "BA", "博茨瓦纳": "BW",
    "布维岛": "BV", "巴西": "BR", "英属印度洋领地": "IO",
    "文莱": "BN", "保加利亚": "BG", "布基纳法索": "BF", "布隆迪": "BI",
    "柬埔寨": "KH", "喀麦隆": "CM", "加拿大": "CA", "佛得角": "CV",
    "开曼群岛": "KY", "中非": "CF", "乍得": "TD", "智利": "CL",
    "中国": "CN", "圣诞岛": "CX", "科科斯(基林)群岛": "CC",
    "哥伦比亚": "CO", "科摩罗": "KM", "刚果(布)": "CG", "刚果（布）": "CG",
    "刚果(金)": "CD", "刚果（金）": "CD", "库克群岛": "CK",
    "哥斯达黎加": "CR", "科特迪瓦": "CI", "克罗地亚": "HR", "古巴": "CU",
    "塞浦路斯": "CY", "捷克": "CZ", "丹麦": "DK", "吉布提": "DJ",
    "多米尼克": "DM", "多米尼加": "DO", "厄瓜多尔": "EC", "埃及": "EG",
    "萨尔瓦多": "SV", "赤道几内亚": "GQ", "厄立特里亚": "ER",
    "爱沙尼亚": "EE", "埃塞俄比亚": "ET", "福克兰群岛(马尔维纳斯)": "FK",
    "法罗群岛": "FO", "斐济": "FJ", "芬兰": "FI", "法国": "FR",
    "法属圭亚那": "GF", "法属波利尼西亚": "PF", "法属南部领地": "TF",
    "加蓬": "GA", "冈比亚": "GM", "格鲁吉亚": "GE", "德国": "DE",
    "加纳": "GH", "直布罗陀": "GI", "希腊": "GR", "格陵兰": "GL",
    "格林纳达": "GD", "瓜德罗普": "GP", "关岛": "GU", "危地马拉": "GT",
    "根西岛": "GG", "几内亚": "GN", "几内亚比绍": "GW", "圭亚那": "GY",
    "海地": "HT", "赫德岛和麦克唐纳群岛": "HM", "梵蒂冈": "VA",
    "洪都拉斯": "HN", "香港": "HK", "匈牙利": "HU", "冰岛": "IS",
    "印度": "IN", "印度尼西亚": "ID", "伊朗": "IR", "伊拉克": "IQ",
    "爱尔兰": "IE", "马恩岛": "IM", "以色列": "IL", "意大利": "IT",
    "牙买加": "JM", "日本": "JP", "泽西岛": "JE", "约旦": "JO",
    "哈萨克斯坦": "KZ", "肯尼亚": "KE", "基里巴斯": "KI", "朝鲜": "KP",
    "韩国": "KR", "科威特": "KW", "吉尔吉斯斯坦": "KG", "老挝": "LA",
    "拉脱维亚": "LV", "黎巴嫩": "LB", "莱索托": "LS", "利比里亚": "LR",
    "利比亚": "LY", "列支敦士登": "LI", "立陶宛": "LT", "卢森堡": "LU",
    "澳门": "MO", "北马其顿": "MK", "马其顿": "MK", "马达加斯加": "MG",
    "马拉维": "MW", "马来西亚": "MY", "马尔代夫": "MV", "马里": "ML",
    "马耳他": "MT", "马绍尔群岛": "MH", "马提尼克": "MQ",
    "毛里塔尼亚": "MR", "毛里求斯": "MU", "马约特": "YT", "墨西哥": "MX",
    "密克罗尼西亚": "FM", "摩尔多瓦": "MD", "摩纳哥": "MC", "蒙古": "MN",
    "黑山": "ME", "蒙特塞拉特": "MS", "摩洛哥": "MA", "莫桑比克": "MZ",
    "缅甸": "MM", "纳米比亚": "NA", "瑙鲁": "NR", "尼泊尔": "NP",
    "荷兰": "NL", "新喀里多尼亚": "NC", "新西兰": "NZ", "尼加拉瓜": "NI",
    "尼日尔": "NE", "尼日利亚": "NG", "纽埃": "NU", "诺福克岛": "NF",
    "北马里亚纳群岛": "MP", "挪威": "NO", "阿曼": "OM", "巴基斯坦": "PK",
    "帕劳": "PW", "巴勒斯坦": "PS", "巴拿马": "PA", "巴布亚新几内亚": "PG",
    "巴拉圭": "PY", "秘鲁": "PE", "菲律宾": "PH", "皮特凯恩": "PN",
    "波兰": "PL", "葡萄牙": "PT", "波多黎各": "PR", "卡塔尔": "QA",
    "留尼汪": "RE", "罗马尼亚": "RO", "俄罗斯": "RU", "卢旺达": "RW",
    "圣巴泰勒米": "BL", "圣赫勒拿": "SH", "圣基茨和尼维斯": "KN",
    "圣卢西亚": "LC", "圣马丁": "MF", "圣皮埃尔和密克隆": "PM",
    "圣文森特和格林纳丁斯": "VC", "萨摩亚": "WS", "圣马力诺": "SM",
    "圣多美和普林西比": "ST", "沙特阿拉伯": "SA", "沙特": "SA",
    "塞内加尔": "SN", "塞尔维亚": "RS", "塞舌尔": "SC", "塞拉利昂": "SL",
    "新加坡": "SG", "圣马丁(荷兰)": "SX", "斯洛伐克": "SK",
    "斯洛文尼亚": "SI", "所罗门群岛": "SB", "索马里": "SO", "南非": "ZA",
    "南乔治亚和南桑威奇群岛": "GS", "南苏丹": "SS", "西班牙": "ES",
    "斯里兰卡": "LK", "苏丹": "SD", "苏里南": "SR", "斯瓦尔巴和扬马延": "SJ",
    "斯威士兰": "SZ", "瑞典": "SE", "瑞士": "CH", "叙利亚": "SY",
    "台湾": "TW", "塔吉克斯坦": "TJ", "坦桑尼亚": "TZ", "泰国": "TH",
    "东帝汶": "TL", "多哥": "TG", "托克劳": "TK", "汤加": "TO",
    "特立尼达和多巴哥": "TT", "突尼斯": "TN", "土耳其": "TR",
    "土库曼斯坦": "TM", "特克斯和凯科斯群岛": "TC", "图瓦卢": "TV",
    "乌干达": "UG", "乌克兰": "UA", "阿联酋": "AE", "英国": "GB",
    "美国": "US", "美国本土外小岛屿": "UM", "乌拉圭": "UY",
    "乌兹别克斯坦": "UZ", "瓦努阿图": "VU", "委内瑞拉": "VE",
    "越南": "VN", "英属维尔京群岛": "VG", "美属维尔京群岛": "VI",
    "瓦利斯和富图纳": "WF", "西撒哈拉": "EH", "也门": "YE",
    "赞比亚": "ZM", "津巴布韦": "ZW",
}

ALPHA3_TO_ALPHA2 = {
    "AFG": "AF", "ALA": "AX", "ALB": "AL", "DZA": "DZ", "ASM": "AS",
    "AND": "AD", "AGO": "AO", "AIA": "AI", "ATA": "AQ", "ATG": "AG",
    "ARG": "AR", "ARM": "AM", "ABW": "AW", "AUS": "AU", "AUT": "AT",
    "AZE": "AZ", "BHS": "BS", "BHR": "BH", "BGD": "BD", "BRB": "BB",
    "BLR": "BY", "BEL": "BE", "BLZ": "BZ", "BEN": "BJ", "BMU": "BM",
    "BTN": "BT", "BOL": "BO", "BIH": "BA", "BWA": "BW", "BVT": "BV",
    "BRA": "BR", "IOT": "IO", "BRN": "BN", "BGR": "BG", "BFA": "BF",
    "BDI": "BI", "KHM": "KH", "CMR": "CM", "CAN": "CA", "CPV": "CV",
    "CYM": "KY", "CAF": "CF", "TCD": "TD", "CHL": "CL", "CHN": "CN",
    "CXR": "CX", "CCK": "CC", "COL": "CO", "COM": "KM", "COG": "CG",
    "COD": "CD", "COK": "CK", "CRI": "CR", "CIV": "CI", "HRV": "HR",
    "CUB": "CU", "CYP": "CY", "CZE": "CZ", "DNK": "DK", "DJI": "DJ",
    "DMA": "DM", "DOM": "DO", "ECU": "EC", "EGY": "EG", "SLV": "SV",
    "GNQ": "GQ", "ERI": "ER", "EST": "EE", "ETH": "ET", "FLK": "FK",
    "FRO": "FO", "FJI": "FJ", "FIN": "FI", "FRA": "FR", "GUF": "GF",
    "PYF": "PF", "ATF": "TF", "GAB": "GA", "GMB": "GM", "GEO": "GE",
    "DEU": "DE", "GHA": "GH", "GIB": "GI", "GRC": "GR", "GRL": "GL",
    "GRD": "GD", "GLP": "GP", "GUM": "GU", "GTM": "GT", "GGY": "GG",
    "GIN": "GN", "GNB": "GW", "GUY": "GY", "HTI": "HT", "HMD": "HM",
    "VAT": "VA", "HND": "HN", "HKG": "HK", "HUN": "HU", "ISL": "IS",
    "IND": "IN", "IDN": "ID", "IRN": "IR", "IRQ": "IQ", "IRL": "IE",
    "IMN": "IM", "ISR": "IL", "ITA": "IT", "JAM": "JM", "JPN": "JP",
    "JEY": "JE", "JOR": "JO", "KAZ": "KZ", "KEN": "KE", "KIR": "KI",
    "PRK": "KP", "KOR": "KR", "KWT": "KW", "KGZ": "KG", "LAO": "LA",
    "LVA": "LV", "LBN": "LB", "LSO": "LS", "LBR": "LR", "LBY": "LY",
    "LIE": "LI", "LTU": "LT", "LUX": "LU", "MAC": "MO", "MKD": "MK",
    "MDG": "MG", "MWI": "MW", "MYS": "MY", "MDV": "MV", "MLI": "ML",
    "MLT": "MT", "MHL": "MH", "MTQ": "MQ", "MRT": "MR", "MUS": "MU",
    "MYT": "YT", "MEX": "MX", "FSM": "FM", "MDA": "MD", "MCO": "MC",
    "MNG": "MN", "MNE": "ME", "MSR": "MS", "MAR": "MA", "MOZ": "MZ",
    "MMR": "MM", "NAM": "NA", "NRU": "NR", "NPL": "NP", "NLD": "NL",
    "NCL": "NC", "NZL": "NZ", "NIC": "NI", "NER": "NE", "NGA": "NG",
    "NIU": "NU", "NFK": "NF", "MNP": "MP", "NOR": "NO", "OMN": "OM",
    "PAK": "PK", "PLW": "PW", "PSE": "PS", "PAN": "PA", "PNG": "PG",
    "PRY": "PY", "PER": "PE", "PHL": "PH", "PCN": "PN", "POL": "PL",
    "PRT": "PT", "PRI": "PR", "QAT": "QA", "REU": "RE", "ROU": "RO",
    "RUS": "RU", "RWA": "RW", "BLM": "BL", "SHN": "SH", "KNA": "KN",
    "LCA": "LC", "MAF": "MF", "SPM": "PM", "VCT": "VC", "WSM": "WS",
    "SMR": "SM", "STP": "ST", "SAU": "SA", "SEN": "SN", "SRB": "RS",
    "SYC": "SC", "SLE": "SL", "SGP": "SG", "SXM": "SX", "SVK": "SK",
    "SVN": "SI", "SLB": "SB", "SOM": "SO", "ZAF": "ZA", "SGS": "GS",
    "SSD": "SS", "ESP": "ES", "LKA": "LK", "SDN": "SD", "SUR": "SR",
    "SJM": "SJ", "SWZ": "SZ", "SWE": "SE", "CHE": "CH", "SYR": "SY",
    "TWN": "TW", "TJK": "TJ", "TZA": "TZ", "THA": "TH", "TLS": "TL",
    "TGO": "TG", "TKL": "TK", "TON": "TO", "TTO": "TT", "TUN": "TN",
    "TUR": "TR", "TKM": "TM", "TCA": "TC", "TUV": "TV", "UGA": "UG",
    "UKR": "UA", "ARE": "AE", "GBR": "GB", "USA": "US", "UMI": "UM",
    "URY": "UY", "UZB": "UZ", "VUT": "VU", "VEN": "VE", "VNM": "VN",
    "VGB": "VG", "VIR": "VI", "WLF": "WF", "ESH": "EH", "YEM": "YE",
    "ZMB": "ZM", "ZWE": "ZW",
}

CODE_SET = set(CN_TO_CODE.values())

CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")

DEFAULT_UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/154.0.0.0 Safari/537.36"

def load_config():
    defaults = {
        "USE_GLOBAL_MODE": True,
        "GLOBAL_TOP_N": 15,
        "PER_COUNTRY_TOP_N": 1,
        "BANDWIDTH_CANDIDATES": 300,
        "TCP_PROBES": 1,
        "MIN_SUCCESS_RATE": 1.0,
        "TCP_LATENCY_WEIGHT": 0.0,
        "TIMEOUT": 3.0,
        "SOCKET_DEFAULT_TIMEOUT": 3,
        "PROGRESS_PRINT_INTERVAL": 1,
        "FILTER_COUNTRIES_ENABLED": False,
        "ALLOWED_COUNTRIES": ["HK"],
        "PRE_FILTER_BLOCKED_ENABLED": True,
        "PRE_FILTER_BLOCKED_COUNTRIES": ["CN"],
        "PRE_FILTER_PORT_ENABLED": True,
        "PRE_FILTER_PORTS": [443],
        "ENABLE_TG": True,
        "TG_BOT_TOKEN": "",
        "TG_CHAT_ID": "",
        "TG_API_URL": "https://api.telegram.org",
        "NOTIFY_TIMEOUT": 3,
        "NOTIFY_CONNECT_TIMEOUT": 3,
        "CF_ENABLED": True,
        "CF_API_TOKEN": "your_CF_API_TOKEN",
        "CF_ZONE_ID": "your_CF_ZONE_ID",
        "CF_DNS_RECORD_NAME": "your_CF_DNS_RECORD_NAME",
        "CF_TTL": 60,
        "CF_PROXIED": False,
        "CF_DNS_CONNECT_TIMEOUT": 3,
        "CF_DNS_READ_TIMEOUT": 3,
        "DNS_RECORD_TYPE": "TXT",
        "ADDITIONAL_SOURCES": [
            {"url": "https://zip.cm.edu.kg/all.txt", "enabled": True},
            {"url": "https://countrymerge.pages.dev/all.txt", "enabled": True}
        ],
        "FETCH_MAX_RETRIES": 3,
        "FETCH_RETRY_DELAY": 3,
        "FETCH_TIMEOUT": 3,
        "FETCH_CONNECT_TIMEOUT": 3,
        "IP_CALIBRATION_ENABLED": False,
        "TOKEN_FAILURE_THRESHOLD": 3,
        "IP_CALIBRATION_MIN_INTERVAL": 0.1,
        "IP_CALIBRATION_REQUEST_TIMEOUT": 3,
        "IP_CALIBRATION_TOKEN_FILE": "valid_tokens.txt",
        "IP_CALIBRATION_CACHE_FILE": "ipinfo_cache.txt",
        "OUTPUT_FILE": "ProxyIP_A.txt",
        "ENABLE_LOGGING": False,
        "LOG_FILE": "cfnb.log",
        "FORCE_DIRECT": False,
        "USER_AGENT": DEFAULT_UA,
        "TEST_AVAILABILITY": True,
        "PROBE_HOST": "cloudflare-ech.com",
        "PROBE_PATH": "/cdn-cgi/trace",
        "AVAILABILITY_TIMEOUT": 3,
        "AVAILABILITY_CONNECT_TIMEOUT": 3,
        "AVAILABILITY_RETRY_MAX": 3,
        "AVAILABILITY_RETRY_DELAY": 3,
        "AVAILABILITY_INNER_RETRY_ENABLED": True,
        "AVAILABILITY_INNER_RETRY_MAX": 3,
        "AVAILABILITY_INNER_RETRY_DELAY": 3,
        "AVAILABILITY_TEST_ROUNDS": 3,
        "HTTP_LATENCY_ENABLED": False,
        "HTTP_LATENCY_WEIGHT": 3.0,
        "JITTER_WEIGHT": 3.0,
        "OUTPUT_FILTER_COUNTRIES_ENABLED": False,
        "OUTPUT_ALLOWED_COUNTRIES": ["HK"],
        "FILTER_IPV6_AVAILABILITY": True,
        "FILTER_BLOCKED_COUNTRIES_ENABLED": True,
        "BLOCKED_COUNTRIES": [
            "BD", "BI", "BY", "CD", "CF", "CN", "CU", "DE", "ET", "HK",
            "IR", "KP", "LY", "MO", "NG", "NL", "PK", "RU", "SD", "SO",
            "SY", "TH", "TW", "UA", "VE", "VN", "YE", "ZW"
        ],
        "DNS_IP_RISK_FILTER_ENABLED": False,
        "DNS_IP_RISK_MAX_LEVEL": "高风险",
        "IP_RISK_TIMEOUT": 3,
        "DNS_UPDATE_TARGET_COUNT": 15,
        "BANDWIDTH_SIZE_MB": 1.0,
        "BANDWIDTH_TIMEOUT": 3,
        "BANDWIDTH_RETRY_MAX": 3,
        "BANDWIDTH_RETRY_DELAY": 3,
        "BANDWIDTH_URL_TEMPLATE": "{scheme}://speed.cloudflare.com:{port}/__down?bytes={bytes}",
        "BANDWIDTH_PROCESS_BUFFER": 3,
        "BANDWIDTH_CONNECT_TIMEOUT": 3,
        "SPEED_WEIGHT": 3.0,
        "IP_CALIBRATION_CONCURRENCY": 300,
        "MAX_WORKERS": 300,
        "AVAILABILITY_WORKERS": 300,
        "FALLBACK_WORKERS": 300,
        "BANDWIDTH_WORKERS": 3,
        "IP_RISK_WORKERS": 300,
        "DNS_UPDATE_MAX_RETRIES": 3,
        "DNS_UPDATE_RETRY_DELAY": 3,
        "GITHUB_SYNC_MAX_RETRIES": 3,
        "GITHUB_SYNC_RETRY_DELAY": 3,
        "GIT_SYNC_PROCESS_TIMEOUT": 180,
        "AD_HEADER_ENABLED": False,
        "AD_HEADER_LINES": [],
        "AD_FOOTER_ENABLED": False,
        "AD_FOOTER_LINES": [],
        "AD_PERLINE_ENABLED": False,
        "AD_PERLINE_TEXT": "",
        "IP_TXT_SHOW_BANDWIDTH": False,
        "IP_TXT_SHOW_HTTP_LATENCY": False,
        "IP_TXT_SHOW_HTTP_JITTER": False,
        "IP_TXT_SHOW_LATENCY": False,
    }

    try:
        with open(CONFIG_FILE, "r", encoding="utf-8") as f:
            config = json.load(f)
    except FileNotFoundError:
        log("警告", f"未找到配置文件 {CONFIG_FILE}，使用内置默认配置")
        return defaults
    except json.JSONDecodeError as e:
        log("错误", f"配置文件格式不正确 - {e}")
        sys.exit(1)

    missing = [k for k in defaults if k not in config]
    if missing:
        for k in missing:
            config[k] = defaults[k]
        log("警告", f"配置缺失 {len(missing)} 项：{', '.join(missing)}")

    return config

cfg = load_config()
USE_GLOBAL_MODE = cfg["USE_GLOBAL_MODE"]
GLOBAL_TOP_N = cfg["GLOBAL_TOP_N"]
PER_COUNTRY_TOP_N = cfg["PER_COUNTRY_TOP_N"]
BANDWIDTH_CANDIDATES = cfg["BANDWIDTH_CANDIDATES"]
TCP_PROBES = cfg["TCP_PROBES"]
MIN_SUCCESS_RATE = cfg["MIN_SUCCESS_RATE"]
TCP_LATENCY_WEIGHT = cfg["TCP_LATENCY_WEIGHT"]
TIMEOUT = cfg["TIMEOUT"]
SOCKET_DEFAULT_TIMEOUT = cfg["SOCKET_DEFAULT_TIMEOUT"]
PROGRESS_PRINT_INTERVAL = cfg["PROGRESS_PRINT_INTERVAL"]
FILTER_COUNTRIES_ENABLED = cfg["FILTER_COUNTRIES_ENABLED"]
ALLOWED_COUNTRIES = cfg["ALLOWED_COUNTRIES"]
PRE_FILTER_BLOCKED_ENABLED = cfg["PRE_FILTER_BLOCKED_ENABLED"]
PRE_FILTER_BLOCKED_COUNTRIES = [c.upper() for c in cfg["PRE_FILTER_BLOCKED_COUNTRIES"]]
PRE_FILTER_PORT_ENABLED = cfg["PRE_FILTER_PORT_ENABLED"]
PRE_FILTER_PORTS = [str(p) for p in cfg["PRE_FILTER_PORTS"]]
ENABLE_TG = cfg.get("ENABLE_TG", True)
TG_BOT_TOKEN = cfg.get("TG_BOT_TOKEN", "")
TG_CHAT_ID = cfg.get("TG_CHAT_ID", "")
TG_API_URL = cfg.get("TG_API_URL", "https://api.telegram.org").rstrip("/")
NOTIFY_TIMEOUT = cfg.get("NOTIFY_TIMEOUT", 3)
NOTIFY_CONNECT_TIMEOUT = cfg.get("NOTIFY_CONNECT_TIMEOUT", 3)
CF_ENABLED = cfg["CF_ENABLED"]
CF_API_TOKEN = cfg["CF_API_TOKEN"]
CF_ZONE_ID = cfg["CF_ZONE_ID"]
CF_DNS_RECORD_NAME = cfg["CF_DNS_RECORD_NAME"]
CF_TTL = cfg["CF_TTL"]
CF_PROXIED = cfg["CF_PROXIED"]
CF_DNS_CONNECT_TIMEOUT = cfg["CF_DNS_CONNECT_TIMEOUT"]
CF_DNS_READ_TIMEOUT = cfg["CF_DNS_READ_TIMEOUT"]
DNS_RECORD_TYPE = cfg["DNS_RECORD_TYPE"]
ADDITIONAL_SOURCES = cfg["ADDITIONAL_SOURCES"]
FETCH_MAX_RETRIES = cfg["FETCH_MAX_RETRIES"]
FETCH_RETRY_DELAY = cfg["FETCH_RETRY_DELAY"]
FETCH_TIMEOUT = cfg["FETCH_TIMEOUT"]
FETCH_CONNECT_TIMEOUT = cfg["FETCH_CONNECT_TIMEOUT"]
IP_CALIBRATION_ENABLED = cfg["IP_CALIBRATION_ENABLED"]
TOKEN_FAILURE_THRESHOLD = cfg["TOKEN_FAILURE_THRESHOLD"]
IP_CALIBRATION_MIN_INTERVAL = cfg["IP_CALIBRATION_MIN_INTERVAL"]
IP_CALIBRATION_REQUEST_TIMEOUT = cfg["IP_CALIBRATION_REQUEST_TIMEOUT"]
IP_CALIBRATION_TOKEN_FILE = cfg["IP_CALIBRATION_TOKEN_FILE"]
IP_CALIBRATION_CACHE_FILE = cfg["IP_CALIBRATION_CACHE_FILE"]
OUTPUT_FILE = cfg["OUTPUT_FILE"]
ENABLE_LOGGING = cfg["ENABLE_LOGGING"]
LOG_FILE = cfg["LOG_FILE"]
FORCE_DIRECT = cfg["FORCE_DIRECT"]
USER_AGENT = cfg["USER_AGENT"] or DEFAULT_UA
TEST_AVAILABILITY = cfg["TEST_AVAILABILITY"]
PROBE_HOST = cfg["PROBE_HOST"]
PROBE_PATH = cfg["PROBE_PATH"]
AVAILABILITY_TIMEOUT = cfg["AVAILABILITY_TIMEOUT"]
AVAILABILITY_CONNECT_TIMEOUT = cfg["AVAILABILITY_CONNECT_TIMEOUT"]
AVAILABILITY_RETRY_MAX = cfg["AVAILABILITY_RETRY_MAX"]
AVAILABILITY_RETRY_DELAY = cfg["AVAILABILITY_RETRY_DELAY"]
AVAILABILITY_INNER_RETRY_ENABLED = cfg["AVAILABILITY_INNER_RETRY_ENABLED"]
AVAILABILITY_INNER_RETRY_MAX = cfg["AVAILABILITY_INNER_RETRY_MAX"]
AVAILABILITY_INNER_RETRY_DELAY = cfg["AVAILABILITY_INNER_RETRY_DELAY"]
AVAILABILITY_TEST_ROUNDS = cfg["AVAILABILITY_TEST_ROUNDS"]
HTTP_LATENCY_ENABLED = cfg["HTTP_LATENCY_ENABLED"]
HTTP_LATENCY_WEIGHT = cfg["HTTP_LATENCY_WEIGHT"]
JITTER_WEIGHT = cfg["JITTER_WEIGHT"]
OUTPUT_FILTER_COUNTRIES_ENABLED = cfg["OUTPUT_FILTER_COUNTRIES_ENABLED"]
OUTPUT_ALLOWED_COUNTRIES = cfg["OUTPUT_ALLOWED_COUNTRIES"]
FILTER_IPV6_AVAILABILITY = cfg["FILTER_IPV6_AVAILABILITY"]
FILTER_BLOCKED_COUNTRIES_ENABLED = cfg["FILTER_BLOCKED_COUNTRIES_ENABLED"]
BLOCKED_COUNTRIES = cfg["BLOCKED_COUNTRIES"]
DNS_IP_RISK_FILTER_ENABLED = cfg["DNS_IP_RISK_FILTER_ENABLED"]
DNS_IP_RISK_MAX_LEVEL = cfg["DNS_IP_RISK_MAX_LEVEL"]
IP_RISK_TIMEOUT = cfg["IP_RISK_TIMEOUT"]
DNS_UPDATE_TARGET_COUNT = cfg["DNS_UPDATE_TARGET_COUNT"]
BANDWIDTH_SIZE_MB = cfg["BANDWIDTH_SIZE_MB"]
BANDWIDTH_TIMEOUT = cfg["BANDWIDTH_TIMEOUT"]
BANDWIDTH_RETRY_MAX = cfg["BANDWIDTH_RETRY_MAX"]
BANDWIDTH_RETRY_DELAY = cfg["BANDWIDTH_RETRY_DELAY"]
BANDWIDTH_URL_TEMPLATE = cfg["BANDWIDTH_URL_TEMPLATE"]
BANDWIDTH_PROCESS_BUFFER = cfg["BANDWIDTH_PROCESS_BUFFER"]
BANDWIDTH_CONNECT_TIMEOUT = cfg["BANDWIDTH_CONNECT_TIMEOUT"]
SPEED_WEIGHT = cfg["SPEED_WEIGHT"]
IP_CALIBRATION_CONCURRENCY = cfg["IP_CALIBRATION_CONCURRENCY"]
MAX_WORKERS = cfg["MAX_WORKERS"]
AVAILABILITY_WORKERS = cfg["AVAILABILITY_WORKERS"]
FALLBACK_WORKERS = cfg["FALLBACK_WORKERS"]
BANDWIDTH_WORKERS = cfg["BANDWIDTH_WORKERS"]
IP_RISK_WORKERS = cfg["IP_RISK_WORKERS"]
DNS_UPDATE_MAX_RETRIES = cfg["DNS_UPDATE_MAX_RETRIES"]
DNS_UPDATE_RETRY_DELAY = cfg["DNS_UPDATE_RETRY_DELAY"]
GITHUB_SYNC_MAX_RETRIES = cfg["GITHUB_SYNC_MAX_RETRIES"]
GITHUB_SYNC_RETRY_DELAY = cfg["GITHUB_SYNC_RETRY_DELAY"]
GIT_SYNC_PROCESS_TIMEOUT = cfg["GIT_SYNC_PROCESS_TIMEOUT"]
AD_HEADER_ENABLED = cfg["AD_HEADER_ENABLED"]
AD_HEADER_LINES = cfg["AD_HEADER_LINES"]
AD_FOOTER_ENABLED = cfg["AD_FOOTER_ENABLED"]
AD_FOOTER_LINES = cfg["AD_FOOTER_LINES"]
AD_PERLINE_ENABLED = cfg["AD_PERLINE_ENABLED"]
AD_PERLINE_TEXT = cfg["AD_PERLINE_TEXT"]
IP_TXT_SHOW_BANDWIDTH = cfg["IP_TXT_SHOW_BANDWIDTH"]
IP_TXT_SHOW_HTTP_LATENCY = cfg["IP_TXT_SHOW_HTTP_LATENCY"]
IP_TXT_SHOW_HTTP_JITTER = cfg["IP_TXT_SHOW_HTTP_JITTER"]
IP_TXT_SHOW_LATENCY = cfg["IP_TXT_SHOW_LATENCY"]

if FORCE_DIRECT:
    for key in ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy", "ALL_PROXY", "all_proxy"):
        os.environ.pop(key, None)
    os.environ["NO_PROXY"] = "*"

socket.setdefaulttimeout(SOCKET_DEFAULT_TIMEOUT)

class _PinnedHTTPSConnection(HTTPSConnection):
    pinned_ip = None
    pinned_port = 443

    def _new_conn(self):
        from urllib3.util.connection import create_connection
        timeout = self.timeout
        if hasattr(timeout, "connect_timeout"):
            timeout = timeout.connect_timeout
        return create_connection(
            (self.pinned_ip, self.pinned_port),
            timeout,
            getattr(self, "source_address", None),
            getattr(self, "socket_options", None),
        )

class PinnedAdapter(HTTPAdapter):
    def __init__(self, pinned_ip, pinned_port=443, **kwargs):
        self.pinned_ip = pinned_ip
        self.pinned_port = pinned_port
        super().__init__(**kwargs)

    def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
        from urllib3.poolmanager import PoolManager
        from urllib3.connectionpool import HTTPSConnectionPool
        tag = f"{self.pinned_ip}_{self.pinned_port}".replace(".", "_").replace(":", "_")
        conn_cls = type(
            f"PinnedConn_{tag}",
            (_PinnedHTTPSConnection,),
            {"pinned_ip": self.pinned_ip, "pinned_port": self.pinned_port},
        )
        pool_cls = type(
            f"PinnedPool_{tag}",
            (HTTPSConnectionPool,),
            {"ConnectionCls": conn_cls},
        )

        class _Manager(PoolManager):
            def __init__(self, *a, **kw):
                super().__init__(*a, **kw)
                self.pool_classes_by_scheme = dict(self.pool_classes_by_scheme)
                self.pool_classes_by_scheme["https"] = pool_cls

        self.poolmanager = _Manager(
            num_pools=connections,
            maxsize=maxsize,
            block=block,
            **pool_kwargs,
        )

def _parse_trace_text(text):
    out = {}
    for line in text.splitlines():
        line = line.strip()
        if "=" in line:
            k, v = line.split("=", 1)
            out[k.strip()] = v.strip()
    return out

def _local_probe(ip, port, timeout):
    s = requests.Session()
    s.trust_env = False
    adapter = PinnedAdapter(ip, port)
    s.mount(f"https://{PROBE_HOST}", adapter)
    try:
        start = time.time()
        r = s.get(
            f"https://{PROBE_HOST}{PROBE_PATH}",
            timeout=(AVAILABILITY_CONNECT_TIMEOUT, timeout),
            allow_redirects=False,
            headers={"User-Agent": USER_AGENT},
        )
        elapsed_ms = (time.time() - start) * 1000
        trace = _parse_trace_text(r.text) if r.status_code == 200 else {}
        return {
            "status": r.status_code,
            "headers": r.headers,
            "trace": trace,
            "ua": r.request.headers.get("User-Agent", ""),
            "error": None,
            "elapsed_ms": elapsed_ms,
        }
    except requests.exceptions.RequestException as e:
        return {"status": None, "headers": {}, "trace": {}, "ua": "", "error": str(e), "elapsed_ms": None}
    finally:
        s.close()

_LOCK_FILE_HANDLE = None
_LOCK_FILE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".run.lock")

def acquire_single_instance():
    global _LOCK_FILE_HANDLE
    f = None
    try:
        f = open(_LOCK_FILE_PATH, "a+")
        f.seek(0)
        if sys.platform == "win32":
            import msvcrt
            msvcrt.locking(f.fileno(), msvcrt.LK_NBLCK, 1)
        else:
            import fcntl
            fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except (OSError, IOError):
        if f is not None:
            try:
                f.close()
            except Exception:
                pass
        return False
    _LOCK_FILE_HANDLE = f
    return True

def release_single_instance():
    global _LOCK_FILE_HANDLE
    if _LOCK_FILE_HANDLE is None:
        return
    try:
        if sys.platform == "win32":
            import msvcrt
            _LOCK_FILE_HANDLE.seek(0)
            msvcrt.locking(_LOCK_FILE_HANDLE.fileno(), msvcrt.LK_UNLCK, 1)
        else:
            import fcntl
            fcntl.flock(_LOCK_FILE_HANDLE.fileno(), fcntl.LOCK_UN)
    except Exception:
        pass
    try:
        _LOCK_FILE_HANDLE.close()
    except Exception:
        pass
    _LOCK_FILE_HANDLE = None
    try:
        os.remove(_LOCK_FILE_PATH)
    except Exception:
        pass

# ====================================================

def send_tg_notification(content, summary):
    if not ENABLE_TG or not TG_BOT_TOKEN or not TG_CHAT_ID:
        return False

    try:
        tg_url = f"{TG_API_URL}/bot{TG_BOT_TOKEN}/sendMessage"
        tg_text = f"*{summary}*\n\n{content}"

        payload = {
            "chat_id": TG_CHAT_ID,
            "text": tg_text,
            "parse_mode": "Markdown"
        }

        headers = {"Content-Type": "application/json; charset=utf-8", "User-Agent": USER_AGENT}

        resp = requests.post(
            tg_url,
            json=payload,
            headers=headers,
            timeout=(NOTIFY_CONNECT_TIMEOUT, NOTIFY_TIMEOUT)
        )
        if resp.status_code == 200:
            print("Telegram 通知已发送")
            return True
        else:
            print(f"Telegram 通知发送失败: {resp.status_code}")
            return False
    except Exception as e:
        print(f"Telegram 通知异常: {e}")
        return False

# 这一行极其重要：做个别名映射，这样底下原本调用微信发送的代码就全都不用改了！
send_wxpusher_notification = send_tg_notification

RISK_LEVEL_ORDER = {
    "极度纯净": 0,
    "纯净": 1,
    "轻微风险": 2,
    "高风险": 3,
    "极度危险": 4,
}

def get_ip_risk_level(ip):
    url = f"https://api.090227.xyz/api/ipapi?ip={ip}"
    try:
        resp = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=IP_RISK_TIMEOUT)
        resp.raise_for_status()
        data = resp.json()
    except Exception:
        return "未知"

    company_score = data.get("company", {}).get("abuser_score")
    asn_score = data.get("asn", {}).get("abuser_score")
    security_flags = {
        "is_crawler": data.get("is_crawler", False),
        "is_proxy": data.get("is_proxy", False),
        "is_vpn": data.get("is_vpn", False),
        "is_tor": data.get("is_tor", False),
        "is_abuser": data.get("is_abuser", False),
        "is_bogon": data.get("is_bogon", False),
    }

    def extract_score(score_str):
        if not score_str:
            return 0.0
        match = re.match(r"([\d.]+)\s*\(([^)]+)\)", str(score_str).strip())
        if match:
            return float(match.group(1))
        try:
            return float(score_str)
        except (ValueError, TypeError):
            return 0.0

    company = extract_score(company_score)
    asn = extract_score(asn_score)
    base_score = ((company + asn) / 2) * 5
    risk_count = sum(1 for key in ["is_crawler", "is_proxy", "is_vpn", "is_tor", "is_abuser"]
                     if security_flags.get(key, False))
    final_score = base_score + risk_count * 0.15
    if security_flags.get("is_bogon", False):
        final_score += 1.0
    percentage = final_score * 100
    if percentage >= 100:
        return "极度危险"
    elif percentage >= 20:
        return "高风险"
    elif percentage >= 5:
        return "轻微风险"
    elif percentage >= 0.25:
        return "纯净"
    else:
        return "极度纯净"

def extract_country_code(label):
    label = label.strip()
    if not label:
        return None
    tokens = re.split(r'[\s,;|/]+', label)
    for token in tokens:
        token_cleaned = re.sub(r'^[\d\s\-_.|#]+', '', token.strip())
        m3 = re.match(r'^([A-Z]{3})(?![A-Za-z])', token_cleaned)
        if m3 and m3.group(1) in ALPHA3_TO_ALPHA2:
            return ALPHA3_TO_ALPHA2[m3.group(1)]
        m2 = re.match(r'^([A-Z]{2})(?![A-Za-z])', token_cleaned)
        if m2 and m2.group(1) in CODE_SET:
            return m2.group(1)
    for token in tokens:
        token_cleaned = re.sub(r'^[\d\s\-_.|#]+', '', token)
        cn_matches = re.findall(r'[\u4e00-\u9fff（）()]+', token_cleaned)
        for cn in cn_matches:
            code = CN_TO_CODE.get(cn)
            if code:
                return code
    emoji_chars = [c for c in label if '\U0001F1E6' <= c <= '\U0001F1FF']
    if len(emoji_chars) >= 2 and len(emoji_chars) % 2 == 0:
        first = ord(emoji_chars[0]) - 0x1F1E6
        second = ord(emoji_chars[1]) - 0x1F1E6
        if 0 <= first <= 25 and 0 <= second <= 25:
            return chr(first + ord('A')) + chr(second + ord('A'))
    return None

def _parse_json_nodes(data):
    nodes = []
    if isinstance(data, list):
        for item in data:
            nodes.extend(_parse_json_nodes(item))
    elif isinstance(data, dict):
        for key in ('nodes', 'data', 'result', 'list'):
            if key in data and isinstance(data[key], list):
                nodes.extend(_parse_json_nodes(data[key]))
                break
        ip = data.get('ip') or data.get('host')
        port = data.get('port')
        code = data.get('country') or data.get('cc')
        if ip and port and code:
            code_str = str(code).strip()
            normalized = extract_country_code(code_str) or code_str.upper()
            nodes.append(f"{ip}:{port}#{normalized}")
    elif isinstance(data, str):
        nodes.extend(_parse_text_nodes(data))
    return nodes

def _query_country(ip, port):
    r = _local_probe(ip, int(port), AVAILABILITY_TIMEOUT)
    if r["status"] == 200:
        loc = r["trace"].get("loc", "")
        if loc and len(loc) == 2:
            return loc.upper()
    return None

def _resolve_countries_batch(ipports):
    results = {}
    total = len(ipports)
    completed = 0
    last_print = time.time()

    def worker(ipport):
        ip, port = ipport.rsplit(':', 1)
        return ipport, _query_country(ip, port)

    with ThreadPoolExecutor(max_workers=FALLBACK_WORKERS) as executor:
        futures = {executor.submit(worker, ipp): ipp for ipp in ipports}
        for future in as_completed(futures):
            try:
                ipport, code = future.result()
                results[ipport] = code
            except Exception:
                results[futures[future]] = None
            completed += 1
            now = time.time()
            is_last = (completed == total)
            if is_last or now - last_print >= PROGRESS_PRINT_INTERVAL:
                log("进度", f"反代验证查询 {completed}/{total} ({(completed/total)*100:.1f}%)", end="\n" if is_last else "")
                last_print = now
    return results

def _parse_text_nodes(text):
    nodes = []
    pending = []
    tokens = text.split()
    for token in tokens:
        pure_match = re.match(r'^(\d+\.\d+\.\d+\.\d+:\d+)$', token)
        if pure_match:
            pending.append(pure_match.group(1))
            continue
        if '#' not in token:
            continue
        try:
            ipport, label = token.split('#', 1)
        except ValueError:
            continue
        ipport = ipport.strip()
        label = label.strip()
        if ipport.startswith('['):
            continue
        if not re.match(r'^\d+\.\d+\.\d+\.\d+:\d+$', ipport):
            continue
        code = extract_country_code(label)
        if code:
            nodes.append(f"{ipport}#{code}")
        else:
            pending.append(ipport)
    if pending:
        log("信息", f"{len(pending)} 个节点缺少国家，反代验证查询中")
        resolved = _resolve_countries_batch(pending)
        for ipport, code in resolved.items():
            if code:
                nodes.append(f"{ipport}#{code}")
    return nodes

def parse_adaptive(text):
    text = text.strip()
    if not text:
        return []
    if text.startswith('{') or text.startswith('['):
        try:
            data = json.loads(text)
            return _parse_json_nodes(data)
        except (json.JSONDecodeError, Exception):
            pass
    return _parse_text_nodes(text)

def fetch_additional_source(url):
    if not url:
        return []
    for attempt in range(1, FETCH_MAX_RETRIES + 1):
        try:
            log("信息", f"拉取数据源 第 {attempt}/{FETCH_MAX_RETRIES} 次（{url}）")
            headers = {"Accept-Encoding": "gzip, deflate, br", "User-Agent": USER_AGENT}
            resp = requests.get(url, timeout=(FETCH_CONNECT_TIMEOUT, FETCH_TIMEOUT), headers=headers)
            resp.raise_for_status()
            nodes = parse_adaptive(resp.text)
            log("信息", f"数据源解析完成（{len(nodes)} 节点，{url}）")
            return nodes
        except Exception as e:
            log("错误", f"数据源请求失败（{url}）: {e}")
            if attempt < FETCH_MAX_RETRIES:
                log("警告", f"等待 {FETCH_RETRY_DELAY}s 重试")
                time.sleep(FETCH_RETRY_DELAY)
            else:
                log("错误", f"已尝试 {FETCH_MAX_RETRIES} 次，放弃 {url}")
                return []

class IpInfoAsync:
    def __init__(self, token_list, concurrency, min_interval, trust_env, failure_threshold):
        self.token_list = token_list
        self.n_tokens = len(token_list)
        self.current_token_index = 0
        self.token_lock = asyncio.Lock()
        self.semaphore = asyncio.Semaphore(concurrency)
        self.min_interval = min_interval
        self.session = None
        self.trust_env = trust_env
        self.failure_threshold = max(2, failure_threshold)
        self.token_failures = [0] * self.n_tokens
        self.cooldown_until = [0.0] * self.n_tokens
        self.last_request_time = [0.0] * self.n_tokens
        self.rate_locks = [asyncio.Lock() for _ in range(self.n_tokens)]

    async def __aenter__(self):
        self.session = aiohttp.ClientSession(
            trust_env=self.trust_env,
            connector=aiohttp.TCPConnector(ssl=False),
            headers={"User-Agent": USER_AGENT},
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

    async def _acquire_token(self):
        async with self.token_lock:
            now = asyncio.get_event_loop().time()
            n = self.n_tokens
            for _ in range(n):
                self.current_token_index = (self.current_token_index + 1) % n
                idx = self.current_token_index
                if (self.token_failures[idx] < self.failure_threshold
                        and self.cooldown_until[idx] <= now):
                    return idx
            idx = min(range(n), key=lambda i: self.token_failures[i])
            self.current_token_index = idx
            return idx

    async def _rate_limit(self, idx):
        async with self.rate_locks[idx]:
            now = asyncio.get_event_loop().time()
            wait = self.last_request_time[idx] + self.min_interval - now
            if wait > 0:
                await asyncio.sleep(wait)
            self.last_request_time[idx] = asyncio.get_event_loop().time()

    async def get_ip_details(self, ip_address):
        max_attempts = max(3, min(self.n_tokens, 10))
        for _ in range(max_attempts):
            idx = await self._acquire_token()
            token = self.token_list[idx]
            url = (f"https://ipinfo.io/{ip_address}/json?token={token}"
                   if ip_address else
                   f"https://ipinfo.io/json?token={token}")
            await self._rate_limit(idx)
            async with self.semaphore:
                try:
                    async with self.session.get(
                        url, timeout=aiohttp.ClientTimeout(total=IP_CALIBRATION_REQUEST_TIMEOUT)
                    ) as resp:
                        if resp.status == 429:
                            self.cooldown_until[idx] = asyncio.get_event_loop().time() + 1.0
                            continue
                        if resp.status in (401, 403):
                            self.token_failures[idx] += 1
                            continue
                        resp.raise_for_status()
                        data = await resp.json()
                        self.token_failures[idx] = 0
                        country = data.get("country", "Unknown")
                        if country == "Unknown":
                            continue
                        return {
                            "CountryCode": country,
                            "Region": data.get("region", "Unknown"),
                            "City": data.get("city", "Unknown"),
                            "ASN": data.get("org", "").split(" ")[0] if data.get("org", "").startswith("AS") else "Unknown",
                            "ISP": data.get("org", "").split(" ", 1)[-1] if " " in data.get("org", "") else data.get("org", "Unknown"),
                        }
                except (asyncio.TimeoutError, aiohttp.ClientError):
                    self.cooldown_until[idx] = asyncio.get_event_loop().time() + 0.3
                    continue
        return None

def load_tokens(filepath):
    if not os.path.exists(filepath):
        return []
    with open(filepath, "r", encoding="utf-8") as f:
        return [line.strip() for line in f if line.strip()]

def load_ipinfo_cache(cache_file):
    if not os.path.exists(cache_file):
        return {}
    cache = {}
    with open(cache_file, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or '#' not in line:
                continue
            ipport, tag = line.split('#', 1)
            cache[ipport.strip()] = tag.strip()
    return cache

def save_ipinfo_cache(cache_file, new_records):
    with open(cache_file, "a", encoding="utf-8") as f:
        for ipport, tag in new_records:
            f.write(f"{ipport}#{tag}\n")

def sort_cache_file(cache_file):
    if not os.path.exists(cache_file):
        return
    with open(cache_file, "r", encoding="utf-8") as f:
        lines = f.readlines()
    parsed = []
    for line in lines:
        line = line.strip()
        if not line or '#' not in line:
            continue
        ipport, tag = line.split('#', 1)
        ip_str = ipport.split(':')[0]
        try:
            ip_obj = ipaddress.ip_address(ip_str)
        except ValueError:
            continue
        parsed.append((ip_obj, line))
    parsed.sort(key=lambda x: x[0])
    with open(cache_file, "w", encoding="utf-8") as f:
        for _, line in parsed:
            f.write(line + "\n")

async def validate_tokens(token_list, concurrency, min_interval, trust_env):
    valid = []
    sem = asyncio.Semaphore(concurrency)
    async with aiohttp.ClientSession(
        trust_env=trust_env,
        connector=aiohttp.TCPConnector(ssl=False),
        headers={"User-Agent": USER_AGENT},
    ) as session:
        async def check(token):
            url = f"https://ipinfo.io/json?token={token}"
            try:
                async with sem:
                    async with session.get(url, timeout=aiohttp.ClientTimeout(total=IP_CALIBRATION_REQUEST_TIMEOUT)) as resp:
                        if resp.status != 200:
                            return token, False
                        data = await resp.json()
                        return token, bool(data.get("country"))
            except Exception:
                return token, False

        tasks = [asyncio.ensure_future(check(t)) for t in token_list]
        total = len(tasks)
        completed = 0
        last_print = time.time()
        for coro in asyncio.as_completed(tasks):
            token, ok = await coro
            if ok:
                valid.append(token)
            completed += 1
            now = time.time()
            is_last = (completed == total)
            if is_last or now - last_print >= PROGRESS_PRINT_INTERVAL:
                log("进度", f"Token 校验 {completed}/{total}", end="\n" if is_last else "")
                last_print = now
    return valid

async def query_new_ips(new_ips, token_list, concurrency, min_interval, trust_env,
                        ipport_map=None, cache_file=None):
    result = {}
    if not new_ips or not token_list:
        return result
    log("信息", f"地区校准查询 {len(new_ips)} 个新 IP")
    async with IpInfoAsync(token_list, concurrency, min_interval, trust_env, TOKEN_FAILURE_THRESHOLD) as handler:
        tasks = []
        for ip in new_ips:
            task = asyncio.ensure_future(handler.get_ip_details(ip))
            task.my_ip = ip
            tasks.append(task)
        total = len(tasks)
        completed = 0
        failed = 0
        last_print = time.time()
        f = None
        if cache_file:
            f = open(cache_file, "a", encoding="utf-8")
        try:
            pending = set(tasks)
            while pending:
                done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
                for task in done:
                    ip = task.my_ip
                    try:
                        info = task.result()
                    except Exception:
                        info = None
                    completed += 1
                    if info and info.get("CountryCode") != "Unknown":
                        tag_parts = [info["CountryCode"]]
                        if info.get("Region") and info["Region"] != "Unknown":
                            tag_parts.append(info["Region"])
                        if info.get("City") and info["City"] != "Unknown":
                            tag_parts.append(info["City"])
                        if info.get("ASN") and info["ASN"] != "Unknown":
                            tag_parts.append(info["ASN"])
                        if info.get("ISP") and info["ISP"] != "Unknown":
                            tag_parts.append(info["ISP"])
                        tag = " ".join(tag_parts)
                        result[ip] = tag
                        if f and ipport_map and ip in ipport_map:
                            for ipport in ipport_map[ip]:
                                f.write(f"{ipport}#{tag}\n")
                                f.flush()
                    else:
                        failed += 1
                    now = time.time()
                    is_last = (completed == total)
                    if is_last or now - last_print >= PROGRESS_PRINT_INTERVAL:
                        log("进度", f"地区校准 {completed}/{total}", end="\n" if is_last else "")
                        last_print = now
        finally:
            if f:
                f.close()
    if total > 0 and failed == total:
        log("警告", "所有 IP 地区校准失败，可能 token 已失效")
        send_wxpusher_notification("IP地区校准：所有IP查询均失败，可能所有token均已失效。", "IP校准全失败")
    return result

def calibrate_regions(nodes, token_file, cache_file):
    if not IP_CALIBRATION_ENABLED:
        log("信息", "IP 地区校准已禁用")
        return
    cache = load_ipinfo_cache(cache_file)
    for i, node in enumerate(nodes):
        ipport = node.split('#')[0]
        tag = cache.get(ipport)
        if tag:
            nodes[i] = f"{ipport}#{tag.split()[0]}"
    token_list = load_tokens(token_file)
    if not token_list:
        log("警告", "valid_tokens.txt 为空，跳过校准")
        return
    trust_env = not FORCE_DIRECT
    valid_tokens = asyncio.run(validate_tokens(token_list, IP_CALIBRATION_CONCURRENCY, IP_CALIBRATION_MIN_INTERVAL, trust_env))
    if not valid_tokens:
        log("错误", "所有 token 已失效，跳过校准")
        send_wxpusher_notification("IP地区校准：所有token均已失效，本次校准跳过。", "IP校准 Token 耗尽")
        return
    log("信息", f"有效 token {len(valid_tokens)} 个")
    ipport_set = set()
    for node in nodes:
        ipport = node.split('#')[0]
        ipport_set.add(ipport)
    cached_ipports = set(cache.keys())
    new_ipports = ipport_set - cached_ipports
    if not new_ipports:
        log("信息", "所有 IP 已在缓存中")
    else:
        new_ips_set = set()
        ip_to_ipports = defaultdict(list)
        for ipport in new_ipports:
            ip = ipport.split(':')[0]
            new_ips_set.add(ip)
            ip_to_ipports[ip].append(ipport)
        log("信息", f"新 IP:端口 {len(new_ipports)} 个，唯一 IP {len(new_ips_set)} 个")
        ip_info = asyncio.run(query_new_ips(
            list(new_ips_set),
            valid_tokens,
            IP_CALIBRATION_CONCURRENCY,
            IP_CALIBRATION_MIN_INTERVAL,
            trust_env,
            ipport_map=ip_to_ipports,
            cache_file=cache_file
        ))
        for ip, tag in ip_info.items():
            for ipport in ip_to_ipports.get(ip, []):
                cache[ipport] = tag
    for i, node in enumerate(nodes):
        ipport = node.split('#')[0]
        tag = cache.get(ipport)
        if tag:
            country_code = tag.split()[0]
            nodes[i] = f"{ipport}#{country_code}"
    sort_cache_file(cache_file)

def test_tcp_latency(ip, port, timeout=TIMEOUT, probes=TCP_PROBES):
    max_latency = 0.0
    success = 0
    for _ in range(probes):
        try:
            start = time.time()
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                sock.settimeout(timeout)
                sock.connect((ip, int(port)))
            latency = time.time() - start
            if latency > max_latency:
                max_latency = latency
            success += 1
        except Exception:
            continue
    if success == 0:
        return float("inf"), 0
    return max_latency, success

def test_node(node_str):
    m = NODE_PATTERN.match(node_str)
    if not m:
        return None
    ip, port, country = m.groups()
    min_lat, success = test_tcp_latency(ip, port)
    if success == 0 or (success / TCP_PROBES) < MIN_SUCCESS_RATE:
        return None
    return (node_str, min_lat, country, success)

def check_availability(node_str, local_public_ip=None):
    m = IP_PORT_PATTERN.match(node_str)
    if not m:
        return (node_str, False, "unknown", {}, [])
    ip, port = m.group(1), int(m.group(2))
    test_rounds = max(3, AVAILABILITY_TEST_ROUNDS)
    total_attempts = (AVAILABILITY_INNER_RETRY_MAX + 1) if AVAILABILITY_INNER_RETRY_ENABLED else 1
    retry_delay = AVAILABILITY_INNER_RETRY_DELAY if AVAILABILITY_INNER_RETRY_ENABLED else 0
    best_trace = None
    families = set()
    latencies = []
    for _ in range(test_rounds):
        round_ok = False
        for attempt in range(total_attempts):
            r = _local_probe(ip, port, AVAILABILITY_TIMEOUT)
            if r["status"] is None:
                if attempt < total_attempts - 1 and retry_delay > 0:
                    time.sleep(retry_delay)
                continue
            if r["status"] != 200:
                return (node_str, False, "unknown", {}, [])
            server = r["headers"].get("Server", "").lower()
            trace = r["trace"]
            if not (server.startswith("cloudflare")
                    and r["headers"].get("CF-RAY")
                    and trace.get("uag") == r["ua"]
                    and trace.get("colo")
                    and trace.get("ip")
                    and trace.get("visit_scheme") == "https"
                    and trace.get("tls") != "off"
                    and trace.get("kex") != "none"):
                return (node_str, False, "unknown", {}, [])
            round_ok = True
            families.add("ipv6" if ":" in trace["ip"] else "ipv4")
            if best_trace is None:
                best_trace = trace
            if r.get("elapsed_ms") is not None:
                latencies.append(r["elapsed_ms"])
            break
        if not round_ok:
            return (node_str, False, "unknown", {}, [])
    if best_trace and families:
        if local_public_ip and best_trace.get("ip") == local_public_ip:
            return (node_str, False, "unknown", {}, [])
        if len(families) == 2:
            stack = "dual_stack"
        else:
            stack = f"{next(iter(families))}_only"
        return (node_str, True, stack, best_trace, latencies)
    return (node_str, False, "unknown", {}, [])

def availability_filter_candidates(candidates):
    if not TEST_AVAILABILITY or not candidates:
        return candidates, {}, {}, {}
    local_public_ip = None
    ip_hosts = ["www.cf-ns.com", "www.cf-ns.net", "www.cf-ns.tech"]
    for host in ip_hosts:
        try:
            r = requests.get(
                f"https://{host}/cdn-cgi/trace",
                timeout=3,
                proxies={"http": None, "https": None},
                headers={"User-Agent": USER_AGENT},
            )
            if r.status_code == 200:
                for line in r.text.splitlines():
                    if line.startswith("ip="):
                        local_public_ip = line[3:].strip()
                        break
            if local_public_ip:
                break
        except Exception:
            pass
    passed = []
    ip_info = {}
    avail_latency_map = {}
    avail_jitter_map = {}
    completed = 0
    total = len(candidates)
    last_print = time.time()
    with ThreadPoolExecutor(max_workers=AVAILABILITY_WORKERS) as executor:
        futures = {executor.submit(check_availability, node, local_public_ip): node for node in candidates}
        for future in as_completed(futures):
            completed += 1
            node_str, ok, stack, _, latencies = future.result()
            if ok:
                passed.append(node_str)
                ip_info[node_str] = stack
                if latencies:
                    avail_latency_map[node_str] = max(latencies)
                    mean = sum(latencies) / len(latencies)
                    variance = sum((l - mean) ** 2 for l in latencies) / len(latencies)
                    avail_jitter_map[node_str] = variance ** 0.5
            now = time.time()
            is_last = (completed == total)
            if is_last or now - last_print >= PROGRESS_PRINT_INTERVAL:
                log("进度", f"反代验证 {completed}/{total} ({(completed/total)*100:.1f}%)", end="\n" if is_last else "")
                last_print = now
    return passed, ip_info, avail_latency_map, avail_jitter_map

def availability_filter_with_retry(candidates):
    if not TEST_AVAILABILITY or not candidates:
        return candidates, {}, {}, {}
    passed = []
    ip_info = {}
    avail_latency_map = {}
    avail_jitter_map = {}
    for attempt in range(1, AVAILABILITY_RETRY_MAX + 1):
        log("信息", f"反代验证 第 {attempt}/{AVAILABILITY_RETRY_MAX} 轮（{len(candidates)} 候选）")
        passed, ip_info, avail_latency_map, avail_jitter_map = availability_filter_candidates(candidates)
        if passed:
            log("信息", f"反代验证通过 {len(passed)} 个节点")
            return passed, ip_info, avail_latency_map, avail_jitter_map
        if attempt < AVAILABILITY_RETRY_MAX:
            log("警告", f"本轮反代验证 0 通过，等待 {AVAILABILITY_RETRY_DELAY}s 重试")
            time.sleep(AVAILABILITY_RETRY_DELAY)
    log("错误", f"反代验证 {AVAILABILITY_RETRY_MAX} 轮后仍 0 通过")
    send_wxpusher_notification(
        content=f"反代验证经 {AVAILABILITY_RETRY_MAX} 轮重试后仍无节点通过，已跳过过滤，使用原候选列表继续。",
        summary="反代验证全部失败"
    )
    return candidates, {}, {}, {}

def measure_bandwidth_curl(node_str):
    m = IP_PORT_PATTERN.match(node_str)
    if not m:
        return (node_str, 0)
    ip, port = m.group(1), m.group(2)
    port_int = int(port)
    HTTP_PORTS = {80, 8080, 8880, 2052, 2082, 2086, 2095}
    HTTPS_PORTS = {443, 2053, 2083, 2087, 2096, 8443}
    if port_int in HTTP_PORTS:
        scheme = "http"
        insecure_flag = []
    elif port_int in HTTPS_PORTS:
        scheme = "https"
        insecure_flag = ["--insecure"]
    else:
        scheme = "https"
        insecure_flag = ["--insecure"]
    null_device = "NUL" if sys.platform == "win32" else "/dev/null"
    expected_size = BANDWIDTH_SIZE_MB * 1024 * 1024
    url = BANDWIDTH_URL_TEMPLATE.format(
        scheme=scheme,
        port=port,
        bytes=int(BANDWIDTH_SIZE_MB * 1024 * 1024)
    )
    _bw_host = urlparse(url).hostname
    if not _bw_host:
        _bw_host = "speed.cloudflare.com"
        url = f"{scheme}://speed.cloudflare.com:{port}/__down?bytes={int(BANDWIDTH_SIZE_MB * 1024 * 1024)}"
    curl_cmd = [
        "curl", "-s", "-o", null_device,
        "-w", "%{size_download} %{time_starttransfer} %{time_total}",
        "-L",
        "-H", f"User-Agent: {USER_AGENT}",
        "-H", "Accept: */*",
        "-H", "Accept-Language: zh-CN,zh;q=0.9",
        "-H", "Accept-Encoding: identity",
        #"--http2",
        "--noproxy", "*",
        "--resolve", f"{_bw_host}:{port}:{ip}",
        "--connect-timeout", str(BANDWIDTH_CONNECT_TIMEOUT),
        "--max-time", str(BANDWIDTH_TIMEOUT),
    ] + insecure_flag + [url]
    try:
        result = subprocess.run(curl_cmd, capture_output=True, text=True,
                                timeout=BANDWIDTH_TIMEOUT + BANDWIDTH_PROCESS_BUFFER)
        if result.returncode == 0 and result.stdout.strip():
            parts = result.stdout.strip().split()
            if len(parts) >= 3:
                size_bytes = float(parts[0])
                if size_bytes < expected_size:
                    return (node_str, 0)
                time_total = float(parts[2])
                if time_total > 0:
                    speed_mbps = (size_bytes * 8) / (time_total * 1000 * 1000)
                    return (node_str, speed_mbps)
    except Exception:
        pass
    return (node_str, 0)

def bandwidth_filter(candidates):
    if not candidates:
        return []
    if not shutil.which("curl"):
        log("错误", "未检测到 curl，跳过带宽测速")
        return []
    results = []
    completed = 0
    total = len(candidates)
    last_print = time.time()
    with ThreadPoolExecutor(max_workers=BANDWIDTH_WORKERS) as executor:
        futures = {executor.submit(measure_bandwidth_curl, node): node for node in candidates}
        for future in as_completed(futures):
            completed += 1
            node, speed = future.result()
            if speed > 0:
                results.append((node, speed))
            now = time.time()
            is_last = (completed == total)
            if is_last or now - last_print >= PROGRESS_PRINT_INTERVAL:
                log("进度", f"带宽测速 {completed}/{total} ({(completed/total)*100:.1f}%)", end="\n" if is_last else "")
                last_print = now
    results.sort(key=lambda x: x[1], reverse=True)
    return results

def batch_update_cloudflare_dns(ip_list, ip_info=None, full_bw_results=None, target_count=None, latency_map=None, http_latency_map=None, http_jitter_map=None, final_nodes=None):
    if not CF_ENABLED:
        log("信息", "CF DNS 更新未启用")
        return
    if target_count is None:
        target_count = DNS_UPDATE_TARGET_COUNT
    dns_content_list = []
    dns_node_list = []
    filtered_by_port = 0
    filtered_by_ipv6 = 0
    filtered_by_country = 0
    filtered_by_risk = 0
    risk_fallback_ip_list = []
    risk_fallback_node_list = []
    record_type = DNS_RECORD_TYPE.upper()
    if record_type not in ("A", "TXT"):
        log("错误", f"不支持 DNS_RECORD_TYPE: {record_type}，跳过")
        return
    risk_map = {}
    if DNS_IP_RISK_FILTER_ENABLED and full_bw_results:
        ip_set = set()
        for node_str, _ in full_bw_results:
            if ':' in node_str:
                ip_set.add(node_str.split(':')[0])
        if ip_set:
            workers = min(IP_RISK_WORKERS, len(ip_set))
            log("信息", f"IP 风险查询（{len(ip_set)} IP，并发 {workers}）")
            with ThreadPoolExecutor(max_workers=workers) as ex:
                futures = {ex.submit(get_ip_risk_level, ip): ip for ip in ip_set}
                for future in as_completed(futures):
                    ip = futures[future]
                    try:
                        risk_map[ip] = future.result()
                    except Exception:
                        risk_map[ip] = "未知"
    if full_bw_results:
        blocked_set = set()
        if FILTER_BLOCKED_COUNTRIES_ENABLED:
            blocked_set = {c.upper() for c in BLOCKED_COUNTRIES}
        for node_str, speed in full_bw_results:
            if ':' not in node_str:
                continue
            parts = node_str.split(':')
            if len(parts) < 2:
                continue
            pure_ip = parts[0]
            port = parts[1].split('#')[0]
            if record_type == "A" and port != '443':
                filtered_by_port += 1
                continue
            if FILTER_IPV6_AVAILABILITY:
                stack = (ip_info or {}).get(node_str, "unknown")
                if stack == "ipv6_only":
                    filtered_by_ipv6 += 1
                    continue
            if blocked_set and '#' in node_str:
                country = node_str.split('#')[-1].split()[0].upper()
                if country in blocked_set:
                    filtered_by_country += 1
                    continue
            if DNS_IP_RISK_FILTER_ENABLED:
                risk_fallback_ip_list.append(pure_ip)
                risk_fallback_node_list.append(node_str)
            if DNS_IP_RISK_FILTER_ENABLED:
                risk_level = risk_map.get(pure_ip, "未知")
                max_level = DNS_IP_RISK_MAX_LEVEL
                if risk_level == "未知" or RISK_LEVEL_ORDER.get(risk_level, 99) > RISK_LEVEL_ORDER.get(max_level, 2):
                    filtered_by_risk += 1
                    continue
            if record_type == "A":
                dns_content_list.append(pure_ip)
            else:
                dns_content_list.append(f"{pure_ip}:{port}")
            dns_node_list.append(node_str)
            if len(dns_content_list) >= target_count:
                break
        if DNS_IP_RISK_FILTER_ENABLED and not dns_content_list and filtered_by_risk > 0:
            send_wxpusher_notification(
                content="风险等级检测全部失败：所有候选节点均因风险等级过高或 API 查询失败被过滤，已回退到无风险等级过滤的候选列表。",
                summary="风险等级检测全部失败"
            )
            fallback_content = []
            fallback_nodes = []
            for i, (ip, node) in enumerate(zip(risk_fallback_ip_list, risk_fallback_node_list)):
                if record_type == "A":
                    fallback_content.append(ip)
                else:
                    ip_port = node.split('#')[0]
                    fallback_content.append(ip_port)
                fallback_nodes.append(node)
                if len(fallback_content) >= target_count:
                    break
            dns_content_list = fallback_content
            dns_node_list = fallback_nodes
        filter_parts = []
        if filtered_by_port > 0:
            filter_parts.append(f"非443端口过滤({filtered_by_port}个)")
        if FILTER_IPV6_AVAILABILITY:
            filter_parts.append(f"IPv6落地过滤({filtered_by_ipv6}个)")
        if FILTER_BLOCKED_COUNTRIES_ENABLED:
            filter_parts.append(f"DNS黑名单过滤({filtered_by_country}个)")
        if DNS_IP_RISK_FILTER_ENABLED and filtered_by_risk > 0:
            filter_parts.append(f"风险等级过滤({filtered_by_risk}个)")
        filter_str = " + ".join(filter_parts) if filter_parts else "无过滤"
        log("信息", f"DNS 筛选：{len(full_bw_results)} → {len(dns_content_list)}（{filter_str}）")
    if not dns_content_list:
        if final_nodes:
            print("未能从完整测速结果构建 DNS 列表，降级使用最终优选节点。")
            limited = list(final_nodes[:target_count])
            if record_type == "A":
                dns_content_list = [n.split(':')[0] for n in limited]
            else:
                dns_content_list = [n.split('#')[0] for n in limited]
            dns_node_list = limited
        elif ip_list:
            log("警告", "测速结果为空，降级使用 ip.txt 中的 IP")
            if record_type == "A":
                dns_content_list = ip_list[:target_count]
                dns_node_list = ip_list[:target_count]
            else:
                log("错误", "TXT 模式降级数据无端口，跳过 DNS")
                return
        else:
            log("错误", "没有可用 IP，跳过 DNS 更新")
            send_wxpusher_notification(content="没有可用的 IP 用于 DNS 更新，跳过。", summary="DNS 更新跳过")
            return
    seen = set()
    unique_content = []
    unique_nodes = []
    for content, node in zip(dns_content_list, dns_node_list):
        if content not in seen:
            seen.add(content)
            unique_content.append(content)
            unique_nodes.append(node)
    dns_content_list = unique_content
    dns_node_list = unique_nodes
    log("信息", f"DNS 更新（{len(dns_content_list)} 个 {record_type}）")
    speed_map = {}
    if full_bw_results:
        speed_map = {node: speed for node, speed in full_bw_results}
    for i, (content, node) in enumerate(zip(dns_content_list, dns_node_list), 1):
        speed = speed_map.get(node, 0)
        lat_ms = float('inf')
        http_lat_ms = None
        http_jitter_ms = None
        if latency_map and node in latency_map:
            lat_ms = latency_map[node] * 1000
        if http_latency_map and node in http_latency_map:
            http_lat_ms = http_latency_map[node]
        if http_jitter_map and node in http_jitter_map:
            http_jitter_ms = http_jitter_map[node]

        display_label = node if '#' in node else content
        line = f"{i}. {display_label} 速度 {speed:.2f} Mbps"
        if http_lat_ms is not None:
            line += f" 延迟 {http_lat_ms:.2f} ms"
        if http_jitter_ms is not None:
            line += f" 抖动 {http_jitter_ms:.2f} ms"
        if lat_ms != float('inf'):
            line += f" 延迟 {lat_ms:.2f} ms"
        log("信息", line)
    headers = {
        "Authorization": f"Bearer {CF_API_TOKEN}",
        "Content-Type": "application/json",
        "User-Agent": USER_AGENT,
    }
    if record_type == "A":
        for attempt in range(1, DNS_UPDATE_MAX_RETRIES + 1):
            log("信息", f"DNS 更新 第 {attempt}/{DNS_UPDATE_MAX_RETRIES} 次")
            try:
                list_url = f"https://api.cloudflare.com/client/v4/zones/{CF_ZONE_ID}/dns_records?type=A&name={CF_DNS_RECORD_NAME}"
                response = requests.get(list_url, headers=headers, timeout=(CF_DNS_CONNECT_TIMEOUT, CF_DNS_READ_TIMEOUT))
                response.raise_for_status()
                result = response.json()
                if not result.get('success'):
                    raise Exception(f"查询 DNS 记录失败: {result.get('errors')}")
                existing_records = result.get('result', [])
                deletes = [{"id": rec["id"]} for rec in existing_records]
                posts = [
                    {
                        "name": CF_DNS_RECORD_NAME,
                        "type": "A",
                        "content": ip,
                        "ttl": CF_TTL,
                        "proxied": CF_PROXIED
                    }
                    for ip in dns_content_list
                ]
                batch_url = f"https://api.cloudflare.com/client/v4/zones/{CF_ZONE_ID}/dns_records/batch"
                payload = {"deletes": deletes, "posts": posts}
                response = requests.post(batch_url, headers=headers, json=payload,
                                        timeout=(CF_DNS_CONNECT_TIMEOUT, CF_DNS_READ_TIMEOUT))
                response.raise_for_status()
                result = response.json()
                if not result.get('success'):
                    raise Exception(f"批量更新失败: {result.get('errors')}")

                success_msg = f"Cloudflare DNS 批量更新成功！已将 {CF_DNS_RECORD_NAME} 指向 {len(dns_content_list)} 个 IP。"
                print(success_msg)
                return len(dns_content_list)  # 返回实际成功提交的 IP 数量

            except Exception as e:
                error_msg = f"[尝试 {attempt}/{DNS_UPDATE_MAX_RETRIES}] DNS 更新出错: {e}"
                log("错误", error_msg)
                if attempt < DNS_UPDATE_MAX_RETRIES:
                    time.sleep(DNS_UPDATE_RETRY_DELAY)
                else:
                    final_error = f"Cloudflare DNS 更新失败，已重试 {DNS_UPDATE_MAX_RETRIES} 次，错误：{e}"
                    log("错误", final_error)
                    send_wxpusher_notification(content=final_error, summary="DNS 更新失败")
                    return False  # 彻底失败时汇报 False

    else:
        for attempt in range(1, DNS_UPDATE_MAX_RETRIES + 1):
            log("信息", f"TXT 更新 第 {attempt}/{DNS_UPDATE_MAX_RETRIES} 次")
            try:
                list_url = f"https://api.cloudflare.com/client/v4/zones/{CF_ZONE_ID}/dns_records?type=TXT&name={CF_DNS_RECORD_NAME}"
                resp = requests.get(list_url, headers=headers, timeout=(CF_DNS_CONNECT_TIMEOUT, CF_DNS_READ_TIMEOUT))
                resp.raise_for_status()
                existing = resp.json().get('result', [])
                deletes = [{"id": rec["id"]} for rec in existing]
                posts = [
                    {
                        "name": CF_DNS_RECORD_NAME,
                        "type": "TXT",
                        "content": content,
                        "ttl": CF_TTL
                    }
                    for content in dns_content_list
                ]
                batch_url = f"https://api.cloudflare.com/client/v4/zones/{CF_ZONE_ID}/dns_records/batch"
                payload = {"deletes": deletes, "posts": posts}
                batch_resp = requests.post(batch_url, headers=headers, json=payload,
                                           timeout=(CF_DNS_CONNECT_TIMEOUT, CF_DNS_READ_TIMEOUT))
                batch_resp.raise_for_status()
                result = batch_resp.json()
                if not result.get('success'):
                    raise Exception(f"批量更新失败: {result.get('errors')}")
                print(f"Cloudflare TXT 记录批量更新成功！共 {len(dns_content_list)} 条记录，每条内容为一个 IP:端口。")
                return len(dns_content_list)  # 返回实际成功提交的 IP 数量

            except Exception as e:
                error_msg = f"[尝试 {attempt}/{DNS_UPDATE_MAX_RETRIES}] TXT 更新出错: {e}"
                log("错误", error_msg)
                if attempt < DNS_UPDATE_MAX_RETRIES:
                    time.sleep(DNS_UPDATE_RETRY_DELAY)
                else:
                    final_error = f"Cloudflare TXT 记录更新失败，已重试 {DNS_UPDATE_MAX_RETRIES} 次，错误：{e}"
                    log("错误", final_error)
                    send_wxpusher_notification(content=final_error, summary="DNS 更新失败")
                    return False  # 彻底失败时汇报 False

def sync_to_github():
    print("\n[网络冷却] 测速结束，等待 10 秒释放系统底层 TCP 连接池...")
    time.sleep(10)
    script_dir = os.path.dirname(os.path.abspath(__file__))
    if sys.platform == "win32":
        script_name = "git_sync.ps1"
        interpreter = ["powershell", "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File"]
        creationflags = subprocess.CREATE_NO_WINDOW
    else:
        script_name = "git_sync.sh"
        interpreter = ["bash"]
        creationflags = 0
    script_path = os.path.join(script_dir, script_name)
    if not os.path.exists(script_path):
        log("警告", f"未找到 {script_name}，跳过 GitHub 同步")
        return
    if sys.platform != "win32":
        try:
            os.chmod(script_path, 0o755)
        except Exception:
            pass
    for attempt in range(1, GITHUB_SYNC_MAX_RETRIES + 1):
        log("信息", f"GitHub 同步 第 {attempt}/{GITHUB_SYNC_MAX_RETRIES} 次")
        try:
            cmd = interpreter + [script_path]
            process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                encoding="utf-8", errors="replace",
                creationflags=creationflags
            )
            try:
                stdout, stderr = process.communicate(timeout=GIT_SYNC_PROCESS_TIMEOUT)
                if process.returncode == 0:
                    print("已自动推送到 GitHub。")
                    # 强行把后台的 git 运行日志打印出来，让你看得明明白白
                    if stdout:
                        print(stdout.strip())
                    return True
                else:
                    log("错误", f"GitHub 推送失败（退出码 {process.returncode}）")
                    if stderr:
                        log("错误", f"GitHub stderr: {stderr.strip()}")
            except subprocess.TimeoutExpired:
                process.kill()
                process.communicate()
                log("错误", f"GitHub 推送超时（>{GIT_SYNC_PROCESS_TIMEOUT}s）")
        except Exception as e:
            log("错误", f"GitHub 推送异常: {e}")
        if attempt < GITHUB_SYNC_MAX_RETRIES:
            time.sleep(GITHUB_SYNC_RETRY_DELAY)
    send_wxpusher_notification(
        content=f"GitHub 推送失败，已重试 {GITHUB_SYNC_MAX_RETRIES} 次，请检查网络或仓库状态。",
        summary="GitHub 推送失败"
    )
    print(f"已尝试 {GITHUB_SYNC_MAX_RETRIES} 次推送，均失败，请检查网络或 GitHub 仓库状态。")
    return False

def write_ip_txt(final_nodes, output_file,
                 header_enabled, header_lines,
                 footer_enabled, footer_lines,
                 perline_enabled, perline_text,
                 speed_map=None, latency_map=None,
                 http_latency_map=None, http_jitter_map=None):
    with open(output_file, "w", encoding="utf-8") as f:
        if header_enabled:
            for line in header_lines:
                f.write(line + "\n")
        for i, node in enumerate(final_nodes, 1):
            # node 本身自带 #JP，我们在它后面加上带前导零的序号（01, 02...）
            line = f"{node} {i:02d}"

            if IP_TXT_SHOW_BANDWIDTH and speed_map and node in speed_map:
                line += f" {speed_map[node]:.2f} Mbps"
            if IP_TXT_SHOW_HTTP_LATENCY and http_latency_map and node in http_latency_map:
                line += f" {http_latency_map[node]:.2f} ms"
            if IP_TXT_SHOW_HTTP_JITTER and http_jitter_map and node in http_jitter_map:
                line += f" {http_jitter_map[node]:.2f} ms"
            if IP_TXT_SHOW_LATENCY and latency_map and node in latency_map:
                line += f" {latency_map[node]*1000:.2f} ms"
            if perline_enabled and perline_text:
                line += perline_text
            f.write(line + "\n")
        if footer_enabled:
            for line in footer_lines:
                f.write(line + "\n")

def main():
    mode_str = f"全局最优 {GLOBAL_TOP_N} 个" if USE_GLOBAL_MODE else f"每个国家最优 {PER_COUNTRY_TOP_N} 个"
    log("信息", f"模式 {mode_str}")
    if PRE_FILTER_PORT_ENABLED and PRE_FILTER_PORTS:
        log("信息", f"前置端口过滤 启用（{', '.join(PRE_FILTER_PORTS)}）")
    else:
        log("信息", "前置端口过滤 禁用")
    if PRE_FILTER_BLOCKED_ENABLED and PRE_FILTER_BLOCKED_COUNTRIES:
        log("信息", f"前置黑名单 启用（{', '.join(PRE_FILTER_BLOCKED_COUNTRIES)}）")
    else:
        log("信息", "前置黑名单 禁用")
    if FILTER_COUNTRIES_ENABLED and ALLOWED_COUNTRIES:
        log("信息", f"前置白名单 启用（{', '.join(ALLOWED_COUNTRIES)}）")
    else:
        log("信息", "前置白名单 禁用")
    log("信息", f"反代验证 {'启用' if TEST_AVAILABILITY else '禁用'}")
    log("信息", f"延迟/抖动排序 {'启用' if HTTP_LATENCY_ENABLED else '禁用'}")
    if OUTPUT_FILTER_COUNTRIES_ENABLED and OUTPUT_ALLOWED_COUNTRIES:
        log("信息", f"输出白名单 启用（{', '.join(OUTPUT_ALLOWED_COUNTRIES)}）")
    else:
        log("信息", "输出白名单 禁用")
    log("信息", f"DNS IPv6 过滤 {'启用' if FILTER_IPV6_AVAILABILITY else '禁用'}")
    if FILTER_BLOCKED_COUNTRIES_ENABLED and BLOCKED_COUNTRIES:
        log("信息", f"DNS 黑名单 启用（{', '.join(BLOCKED_COUNTRIES)}）")
    else:
        log("信息", "DNS 黑名单 禁用")
    log("信息", f"IP 风险过滤 {'启用' if DNS_IP_RISK_FILTER_ENABLED else '禁用'}（≤{DNS_IP_RISK_MAX_LEVEL}）")

    nodes = []
    for source in ADDITIONAL_SOURCES:
        if not source.get("enabled", True):
            continue
        url = source.get("url")
        if not url:
            continue
        v2_nodes = fetch_additional_source(url)
        if v2_nodes:
            seen = set()
            for n in nodes:
                seen.add(n.split('#')[0])
            for n in v2_nodes:
                key = n.split('#')[0]
                if key not in seen:
                    seen.add(key)
                    nodes.append(n)
    log("信息", f"合并后总计 {len(nodes)} 个节点")

    token_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), IP_CALIBRATION_TOKEN_FILE)
    cache_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), IP_CALIBRATION_CACHE_FILE)
    calibrate_regions(nodes, token_file, cache_file)

    if PRE_FILTER_PORT_ENABLED and PRE_FILTER_PORTS:
        before = len(nodes)
        nodes = [n for n in nodes if n.split(':')[1].split('#')[0] in PRE_FILTER_PORTS]
        after = len(nodes)
        log("信息", f"前置端口过滤：{before} → {after}")
        if not nodes:
            log("错误", "前置端口过滤后 0 节点，退出")
            sys.exit(0)
    if PRE_FILTER_BLOCKED_ENABLED and PRE_FILTER_BLOCKED_COUNTRIES:
        before = len(nodes)
        blocked_set = set(PRE_FILTER_BLOCKED_COUNTRIES)
        nodes = [n for n in nodes if n.split('#')[-1].split()[0].upper() not in blocked_set]
        after = len(nodes)
        log("信息", f"前置黑名单过滤：{before} → {after}")
        if not nodes:
            log("错误", "前置黑名单过滤后 0 节点，退出")
            sys.exit(0)
    if not nodes:
        log("错误", "没有有效节点，退出")
        sys.exit(1)
    if FILTER_COUNTRIES_ENABLED and ALLOWED_COUNTRIES:
        before = len(nodes)
        allowed_set = {c.upper() for c in ALLOWED_COUNTRIES}
        filtered_nodes = []
        for node in nodes:
            parts = node.split('#')
            if len(parts) == 2 and parts[1].split()[0].upper() in allowed_set:
                filtered_nodes.append(node)
        nodes = filtered_nodes
        after = len(nodes)
        log("信息", f"前置白名单过滤：{before} → {after}")
        if not nodes:
            log("错误", "前置白名单过滤后 0 节点，退出")
            sys.exit(0)

    total = len(nodes)
    log("信息", f"TCP 测试（{total} 候选）")
    results = []
    completed = 0
    last_print = time.time()
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = {executor.submit(test_node, node): node for node in nodes}
        for future in as_completed(futures):
            completed += 1
            res = future.result()
            if res:
                results.append(res)
            now = time.time()
            is_last = (completed == total)
            if is_last or now - last_print >= PROGRESS_PRINT_INTERVAL:
                log("进度", f"TCP 测试 {completed}/{total} ({(completed/total)*100:.1f}%)", end="\n" if is_last else "")
                last_print = now
    if not results:
        log("错误", "没有通过成功率筛选的节点")
        sys.exit(0)

    results.sort(key=lambda x: (-x[3], x[1]))
    latency_map = {node: lat for node, lat, _, _ in results}

    if USE_GLOBAL_MODE:
        candidates = [node for node, _, _, _ in results[:BANDWIDTH_CANDIDATES]]
        log("信息", f"TCP 候选池 {len(candidates)} 个节点")
    else:
        country_nodes = defaultdict(list)
        for node_str, lat, country, succ in results:
            country_nodes[country].append((node_str, lat, succ))
        total_countries = len(country_nodes)
        base_limit = max(1, BANDWIDTH_CANDIDATES // total_countries)
        candidates = []
        for country, nodes in country_nodes.items():
            nodes_sorted = sorted(nodes, key=lambda x: (-x[2], x[1]))
            limit = min(len(nodes_sorted), base_limit)
            for node_str, lat, succ in nodes_sorted[:limit]:
                candidates.append(node_str)
        log("信息", f"分国家候选池：{total_countries} 国，每国 ≤{base_limit}，合计 {len(candidates)}")

    if not candidates:
        log("错误", "没有候选节点，退出")
        sys.exit(0)

    candidates_after_availability, avail_ip_info, avail_latency_map, avail_jitter_map = availability_filter_with_retry(candidates)
    if HTTP_LATENCY_ENABLED:
        http_latency_map = avail_latency_map
        http_jitter_map = avail_jitter_map
    else:
        http_latency_map = {}
        http_jitter_map = {}

    bw_results = []
    for attempt in range(1, BANDWIDTH_RETRY_MAX + 1):
        log("信息", f"带宽测速 第 {attempt}/{BANDWIDTH_RETRY_MAX} 轮（{len(candidates_after_availability)} 候选）")
        bw_results = bandwidth_filter(candidates_after_availability)
        if bw_results:
            log("信息", f"带宽测速通过 {len(bw_results)} 个节点")
            break
        log("信息", "带宽测速通过 0 个节点")
        if attempt < BANDWIDTH_RETRY_MAX:
            log("警告", f"本轮测速 0 结果，等待 {BANDWIDTH_RETRY_DELAY}s 重试")
            time.sleep(BANDWIDTH_RETRY_DELAY)

    if not bw_results:
        log("错误", "带宽测速重试后仍 0 结果，降级用 TCP 排序")
        send_wxpusher_notification(
            content=f"带宽测速经 {BANDWIDTH_RETRY_MAX} 轮尝试后仍无有效结果，已降级使用 TCP 排序节点。",
            summary="带宽测速全部失败"
        )
        speed_map = {}
        if OUTPUT_FILTER_COUNTRIES_ENABLED and OUTPUT_ALLOWED_COUNTRIES:
            allowed_set = {c.upper() for c in OUTPUT_ALLOWED_COUNTRIES}
            before = len(results)
            filtered_results = [
                r for r in results
                if '#' in r[0] and r[0].split('#')[-1].split()[0].upper() in allowed_set
            ]
            after = len(filtered_results)
            if filtered_results:
                results = filtered_results
                if not USE_GLOBAL_MODE:
                    country_nodes = defaultdict(list)
                    for node_str, lat, country, succ in results:
                        country_nodes[country].append((node_str, lat, succ))
                log("信息", f"输出白名单过滤：{before} → {after}")
            else:
                log("警告", f"输出白名单过滤后 0 结果，降级用未过滤 {before} 节点")
                send_wxpusher_notification(
                    content=f"ip.txt 输出白名单过滤后仍无有效结果（允许国家：{', '.join(sorted(allowed_set))}），已降级使用未过滤的 {before} 个节点写入 ip.txt。",
                    summary="ip.txt 白名单过滤全部失败"
                )
        if USE_GLOBAL_MODE:
            final_selected = [node for node, _, _, _ in results[:GLOBAL_TOP_N]]
        else:
            final_selected = []
            for country, nodes in country_nodes.items():
                nodes_sorted = sorted(nodes, key=lambda x: (-x[2], x[1]))
                for node_str, _, _ in nodes_sorted[:PER_COUNTRY_TOP_N]:
                    final_selected.append(node_str)
    else:
        speed_map = {node: speed for node, speed in bw_results}
        scored_nodes = []
        for node, speed in bw_results:
            tcp_lat = latency_map.get(node, 999.0)
            http_lat = http_latency_map.get(node, 999999.0)
            http_jitter = http_jitter_map.get(node, 999999.0)
            http_lat_sec = http_lat / 1000.0
            http_jitter_sec = http_jitter / 1000.0
            penalty = 1.0 + TCP_LATENCY_WEIGHT * tcp_lat + HTTP_LATENCY_WEIGHT * http_lat_sec + JITTER_WEIGHT * http_jitter_sec
            score = (SPEED_WEIGHT * speed) / penalty
            scored_nodes.append((node, score, speed, tcp_lat, http_lat))
        scored_nodes.sort(key=lambda x: x[1], reverse=True)
        if OUTPUT_FILTER_COUNTRIES_ENABLED and OUTPUT_ALLOWED_COUNTRIES:
            allowed_set = {c.upper() for c in OUTPUT_ALLOWED_COUNTRIES}
            before = len(scored_nodes)
            filtered_scored = [
                it for it in scored_nodes
                if '#' in it[0] and it[0].split('#')[-1].split()[0].upper() in allowed_set
            ]
            after = len(filtered_scored)
            if filtered_scored:
                scored_nodes = filtered_scored
                log("信息", f"输出白名单过滤：{before} → {after}")
            else:
                log("警告", f"输出白名单过滤后 0 结果，降级用未过滤 {before} 节点")
                send_wxpusher_notification(
                    content=f"ip.txt 输出白名单过滤后仍无有效结果（允许国家：{', '.join(sorted(allowed_set))}），已降级使用未过滤的 {before} 个节点写入 ip.txt。",
                    summary="ip.txt 白名单过滤全部失败"
                )
        if USE_GLOBAL_MODE:
            final_selected = [item[0] for item in scored_nodes[:GLOBAL_TOP_N]]
        else:
            country_scored = defaultdict(list)
            for item in scored_nodes:
                node, score, speed, tcp_lat, http_lat = item
                country = node.split('#')[-1] if '#' in node else ''
                if country:
                    country_scored[country].append(item)
            final_selected = []
            for country, items in country_scored.items():
                items.sort(key=lambda x: x[1], reverse=True)
                for item in items[:PER_COUNTRY_TOP_N]:
                    final_selected.append(item[0])
            score_dict = {item[0]: item[1] for item in scored_nodes}
            final_selected.sort(key=lambda n: score_dict.get(n, 0), reverse=True)
        log("信息", "最终优选节点")
        for i, node in enumerate(final_selected, 1):
            speed = speed_map.get(node, 0)
            tcp_lat = latency_map.get(node, float('inf'))
            http_lat = http_latency_map.get(node, None)
            http_jitter = http_jitter_map.get(node, None)
            line = f"{i}. {node} 速度 {speed:.2f} Mbps"
            if http_lat is not None:
                line += f" 延迟 {http_lat:.2f} ms"
            if http_jitter is not None:
                line += f" 抖动 {http_jitter:.2f} ms"
            if tcp_lat != float('inf'):
                line += f" 延迟 {tcp_lat*1000:.2f} ms"
            log("信息", line)

    write_ip_txt(final_selected, OUTPUT_FILE,
                 AD_HEADER_ENABLED, AD_HEADER_LINES,
                 AD_FOOTER_ENABLED, AD_FOOTER_LINES,
                 AD_PERLINE_ENABLED, AD_PERLINE_TEXT,
                 speed_map=speed_map,
                 latency_map=latency_map,
                 http_latency_map=http_latency_map,
                 http_jitter_map=http_jitter_map)
    log("信息", f"ip.txt（{len(final_selected)} 个节点）")

    ip_list = [node.split(':')[0] for node in final_selected]

# 接收两个核心步骤的执行结果
    cf_status = batch_update_cloudflare_dns(
        ip_list,
        ip_info=avail_ip_info,
        full_bw_results=bw_results,
        target_count=None,
        latency_map=latency_map,
        http_latency_map=http_latency_map,
        http_jitter_map=http_jitter_map,
        final_nodes=final_selected
    )

    # 【零侵入拦截】：如果不推送，压根就不去调用原作者的函数，彻底避开内部的报错通知
    if GITHUB_SYNC_MAX_RETRIES > 0:
        git_status = sync_to_github()
    else:
        print("\n[GitHub 同步] 配置中已关闭 GitHub 推送 (重试次数设为 0)，主动跳过。")
        git_status = None

    # ================= 新增：大满贯战报通知 =================
    # 智能拦截逻辑：只要明确返回了 False，才视为执行报错
    is_git_failed = (git_status is False)
    is_cf_failed = CF_ENABLED and (cf_status is False)

    if not is_cf_failed and not is_git_failed:
        summary_title = "✅ CFNB_AZURE 节点优选同步完成"
        if final_selected:
            top_speed = speed_map.get(final_selected[0], 0)

            regions_list = []
            for node in final_selected:
                if '#' in node:
                    reg = node.split('#')[-1].split()[0].upper()
                    if reg not in regions_list:
                        regions_list.append(reg)
            regions_str = f"{'/'.join(regions_list)} " if regions_list else ""

            # 动态生成战报内容
            cf_text = f"🌐 *Cloudflare*：`{CF_DNS_RECORD_NAME}` 负载 {cf_status if type(cf_status) is int else len(ip_list)} 个优选 IP\n" if CF_ENABLED else "🌐 *Cloudflare*：(已跳过更新)\n"
            gh_text = f"📦 *GitHub*：`ProxyIP_A` 已成功同步更新" if GITHUB_SYNC_MAX_RETRIES > 0 else "📦 *GitHub*：(已跳过同步)"

            content = (
                f"👉 *已提取节点*：{regions_str}前 {len(final_selected)} 名\n"
                f"🚀 *最高速度*：{top_speed:.2f} Mbps\n"
                f"{cf_text}"
                f"{gh_text}"
            )
        else:
            content = "自动化任务执行完毕，但没有筛选出任何有效节点。"

        is_sent = send_tg_notification(content, summary_title)
        if is_sent:
            print("\n[TG 通知] 成功战报已发送！")
        else:
            print("\n[TG 通知] 战报发送失败，网络异常或超时！")
    else:
        err_msg = "检测到启用的关键步骤执行失败：\n"
        if is_cf_failed: err_msg += "❌ Cloudflare DNS 更新失败\n"
        if is_git_failed: err_msg += "❌ GitHub 同步失败\n"
        print(f"\n[TG 通知] {err_msg}已拦截取消成功战报。")
        send_tg_notification(err_msg, "⚠️ CFNB 关键步骤失败")
    # ========================================================

if __name__ == "__main__":
    import atexit

    if not acquire_single_instance():
        log("警告", "本程序已在运行，退出")
        sys.exit(0)
    atexit.register(release_single_instance)

    enable_log = ENABLE_LOGGING
    log_filename = LOG_FILE

    if enable_log:
        script_dir = os.path.dirname(os.path.abspath(__file__))
        log_path = os.path.join(script_dir, log_filename)
        try:
            log_f = open(log_path, "w", encoding="utf-8")
            log("信息", f"日志启用：{log_path}")
        except Exception as e:
            log("错误", f"无法打开日志文件 {log_path}: {e}")
            log_f = None
        else:
            class _Tee:
                def __init__(self, *files):
                    self.files = files
                def write(self, obj):
                    for f in self.files:
                        f.write(obj)
                        f.flush()
                def flush(self):
                    for f in self.files:
                        f.flush()
            sys.stdout = _Tee(sys.stdout, log_f)

            def _close_log():
                try:
                    sys.stdout = sys.__stdout__
                    log_f.close()
                except Exception:
                    pass
            atexit.register(_close_log)

    main()