diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6486f70..3d1dd02 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,30 +8,30 @@ All notable changes to this project will be documented in this file.
- **订阅源管理** - 数据库驱动的订阅源增删改查，支持独立配置 Crontab、延迟阈值、重试次数和最大并发数
- **Crontab 定时拉取** - 每个订阅源支持独立的 Crontab 表达式，通过 APScheduler CronTrigger 实现精准调度
-- **HTTP 延迟检测** - 通过代理发起 HTTP 请求到多个目标 URL 检测延迟，取多目标最大延迟值作为结果
+- **HTTP 延迟检测** - 通过节点发起 HTTP 请求到多个目标 URL 检测延迟，取多目标最大延迟值作为结果
- **检测目标动态配置** - 检测目标 URL 存入数据库，页面可增删，修改后即时生效
- **多协议解析** - 支持 vmess / vless / trojan / ss / hysteria2 五种协议的订阅解析和 Clash 配置生成
-- **对外订阅输出** - 提供 V2Ray（base64）和 Clash（YAML）格式订阅，仅包含当前可用代理
+- **对外订阅输出** - 提供核心（base64）和 Clash（YAML）格式订阅，仅包含当前可用节点
- **自动添加订阅 API** - `POST /api/subscriptions/auto` 接口支持仅传 URL 自动创建订阅并触发拉取验证
-### 代理管理
+### 节点管理
-- **独立延迟检测** - 每个代理独立测试延迟，报错自动跳过
-- **延迟阈值过滤** - 延迟低于阈值自动入库，超标代理累加失败计数
-- **连续失败清理** - 连续 3 次验证失败的代理自动移除
-- **空订阅清理** - 连续 30 天代理数为 0 的订阅源自动删除
-- **按订阅源分组** - 可用代理按订阅源分组，Tab 切换查看
-- **延迟与入库时间排序** - 每个订阅源的代理列表支持延迟和入库时间升降序排序
+- **独立延迟检测** - 每个节点独立测试延迟，报错自动跳过
+- **延迟阈值过滤** - 延迟低于阈值自动入库，超标节点累加失败计数
+- **连续失败清理** - 连续 3 次验证失败的节点自动移除
+- **空订阅清理** - 连续 30 天节点数为 0 的订阅源自动删除
+- **按订阅源分组** - 可用节点按订阅源分组，Tab 切换查看
+- **延迟与入库时间排序** - 每个订阅源的节点列表支持延迟和入库时间升降序排序
### Web 界面
-- **单页面管理** - 订阅源管理和可用代理列表合并为一个页面，无跳转
-- **协议分布饼状图** - Chart.js 环形图动态展示各协议代理数量和百分比
+- **单页面管理** - 订阅源管理和可用节点列表合并为一个页面，无跳转
+- **协议分布饼状图** - Chart.js 环形图动态展示各协议节点数量和百分比
- **订阅源状态** - 实时显示每个订阅源的拉取状态（更新中/成功/失败/待更新）
- **启用/禁用切换** - 状态列点击切换启用与禁用，启用后自动拉取并验证
- **编辑弹窗** - 操作列仅保留编辑按钮，删除功能移至编辑弹窗内
- **订阅地址提示** - 鼠标悬停订阅 ID 行或 Tab 标签时通过 Tooltip 显示订阅地址
-- **可用代理 Tab 过滤** - 仅显示有可用代理的订阅源标签页
+- **可用节点 Tab 过滤** - 仅显示有可用节点的订阅源标签页
### 运维特性
@@ -44,15 +44,15 @@ All notable changes to this project will be documented in this file.
| 方法 | 路径 | 说明 |
|------|------|------|
-| GET | `/api/proxies` | 可用代理列表 |
-| GET | `/api/proxies/all` | 所有代理 |
-| DELETE | `/api/proxies/{id}` | 删除代理 |
-| GET | `/api/subscription/v2ray` | V2Ray 订阅 |
+| GET | `/api/proxies` | 可用节点列表 |
+| GET | `/api/proxies/all` | 所有节点 |
+| DELETE | `/api/proxies/{id}` | 删除节点 |
+| GET | `/api/subscription/v2ray` | 核心订阅 |
| GET | `/api/subscription/clash` | Clash 订阅 |
| POST | `/api/fetch` | 拉取所有订阅 |
| POST | `/api/fetch/{sub_id}` | 拉取指定订阅 |
-| POST | `/api/verify` | 验证所有代理 |
-| POST | `/api/verify/{sub_id}` | 验证指定订阅代理 |
+| POST | `/api/verify` | 验证所有节点 |
+| POST | `/api/verify/{sub_id}` | 验证指定订阅节点 |
| GET | `/api/stats` | 统计信息 |
| GET | `/api/health` | 健康检查 |
| GET | `/api/subscriptions` | 订阅源列表 |
@@ -63,4 +63,4 @@ All notable changes to this project will be documented in this file.
| GET | `/api/check-urls` | 检测目标 URL 列表 |
| POST | `/api/check-urls` | 添加检测目标 URL |
| DELETE | `/api/check-urls/{url_id}` | 删除检测目标 URL |
-| GET | `/api/proxies/grouped` | 按订阅来源分组的可用代理 |
+| GET | `/api/proxies/grouped` | 按订阅来源分组的可用节点 |
diff --git a/Dockerfile b/Dockerfile
index 36ff4b6..70a6715 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,16 +1,45 @@
FROM python:3.11-slim
+# GitHub 下载镜像前缀（国内构建可设为 https://ghgo.xyz/ 等）
+ARG GH_MIRROR="https://github.geekery.cn/"
+
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
gcc \
+ curl \
+ unzip \
&& rm -rf /var/lib/apt/lists/*
+
# 安装 Python 依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
+
+# 下载核心（固定版本 + 超时重试）
+# v2ray-core v5.23.0
+RUN curl -fsSL --connect-timeout 30 --max-time 180 --retry 3 --retry-delay 5 \
+ "${GH_MIRROR}https://github.com/v2fly/v2ray-core/releases/download/v5.51.2/v2ray-linux-64.zip" \
+ -o /tmp/core.zip \
+ && unzip -o /tmp/core.zip -d /usr/local/bin/ v2ray \
+ && chmod +x /usr/local/bin/v2ray \
+ && rm -f /tmp/core.zip
+
+# Xray-core v25.7.16
+RUN curl -fsSL --connect-timeout 30 --max-time 180 --retry 3 --retry-delay 5 \
+ "${GH_MIRROR}https://github.com/XTLS/Xray-core/releases/download/v26.7.11/Xray-macos-64.zip" \
+ -o /tmp/xray.zip \
+ && unzip -o /tmp/xray.zip -d /usr/local/bin/ xray \
+ && chmod +x /usr/local/bin/xray \
+ && rm -f /tmp/xray.zip
+
+# 下载 geosite.dat / geoip.dat 数据文件
+RUN curl -fsSL --connect-timeout 30 "https://ghp.keleyaa.com/https://github.com/v2fly/domain-list-community/releases/download/20260717032527/dlc.dat" -o /usr/local/bin/geosite.dat
+RUN curl -fsSL --connect-timeout 30 "https://gitproxy.mrhjx.cn/https://github.com/v2fly/geoip/releases/download/202607171233/geoip.dat" -o /usr/local/bin/geoip.dat
+
+
# 复制代码
COPY . .
diff --git a/README.md b/README.md
index dbdbbae..b2e7173 100644
--- a/README.md
+++ b/README.md
@@ -1,18 +1,18 @@
-# ProxyPool v1.0.0
+# NetHub v1.0.0
-自动获取、检测、维护代理节点池，提供 Web 管理界面和订阅链接输出。
+自动获取、检测、维护节点池，提供 Web 管理界面和订阅链接输出。
## 功能特性
- **订阅源管理** - 数据库驱动的增删改查，每个订阅源独立配置 Crontab、延迟阈值、重试次数、并发数
- **Crontab 定时拉取** - 每个订阅源支持 5 位 Crontab 表达式精准调度
-- **HTTP 延迟检测** - 模拟通过代理访问目标网站，测量完整请求延迟（DNS + TCP + TLS + HTTP），多目标取最大值
+- **HTTP 延迟检测** - 模拟通过节点访问目标网站，测量完整请求延迟（DNS + TCP + TLS + HTTP），多目标取最大值
- **检测目标动态配置** - 检测目标 URL 存入数据库，页面可增删，修改即时生效，无需外部文件
- **多协议支持** - vmess / vless / trojan / ss / hysteria2 解析与 Clash 配置生成
-- **自动清理** - 连续 3 次验证失败的代理自动移除；连续 30 天无代理的订阅源自动删除
-- **单页面管理** - 订阅源管理 + 可用代理列表在同一页面，按订阅源 Tab 切换
-- **协议分布图** - 饼状图动态展示各协议代理数量和百分比
-- **订阅输出** - 仅输出当前可用代理，支持 V2Ray（base64）和 Clash（YAML）格式
+- **自动清理** - 连续 3 次验证失败的节点自动移除；连续 30 天无节点的订阅源自动删除
+- **单页面管理** - 订阅源管理 + 可用节点列表在同一页面，按订阅源 Tab 切换
+- **协议分布图** - 饼状图动态展示各协议节点数量和百分比
+- **订阅输出** - 仅输出当前可用节点，支持核心格式（base64）和 Clash（YAML）格式
- **日志归档** - 按天自动归档，自动清理 7 天前的日志
- **Docker 部署** - Docker Compose 一键启动
@@ -63,7 +63,7 @@ check:
scheduler:
fetch_interval: 3600 # 拉取订阅间隔（秒）
- verify_interval: 1800 # 验证代理间隔（秒）
+ verify_interval: 1800 # 验证节点间隔（秒）
cleanup_interval: 7200 # 清理间隔（秒）
max_fail_count: 3 # 最大连续失败次数
```
@@ -79,20 +79,20 @@ scheduler:
## API 接口
-### 代理
+### 节点
| 方法 | 路径 | 说明 |
|------|------|------|
-| GET | `/api/proxies` | 可用代理列表 |
-| GET | `/api/proxies/all` | 所有代理 |
-| GET | `/api/proxies/grouped` | 按订阅来源分组的可用代理 |
-| DELETE | `/api/proxies/{id}` | 删除代理 |
+| GET | `/api/proxies` | 可用节点列表 |
+| GET | `/api/proxies/all` | 所有节点 |
+| GET | `/api/proxies/grouped` | 按订阅来源分组的可用节点 |
+| DELETE | `/api/proxies/{id}` | 删除节点 |
### 订阅输出
| 方法 | 路径 | 说明 |
|------|------|------|
-| GET | `/api/subscription/v2ray` | V2Ray 格式订阅（base64） |
+| GET | `/api/subscription/v2ray` | 核心格式订阅（base64） |
| GET | `/api/subscription/clash` | Clash 格式订阅（YAML） |
### 订阅源管理
@@ -111,8 +111,8 @@ scheduler:
|------|------|------|
| POST | `/api/fetch` | 拉取所有订阅 |
| POST | `/api/fetch/{sub_id}` | 拉取指定订阅 |
-| POST | `/api/verify` | 验证所有代理 |
-| POST | `/api/verify/{sub_id}` | 验证指定订阅代理 |
+| POST | `/api/verify` | 验证所有节点 |
+| POST | `/api/verify/{sub_id}` | 验证指定订阅节点 |
### 检测目标
@@ -131,12 +131,12 @@ scheduler:
## 订阅链接使用
-在代理客户端中添加以下订阅链接：
+在节点客户端中添加以下订阅链接：
-- **V2Ray**: `http://your-server:8080/api/subscription/v2ray`
+- **核心**: `http://your-server:8080/api/subscription/v2ray`
- **Clash**: `http://your-server:8080/api/subscription/clash`
-> 订阅内容仅包含延迟低于阈值的可用代理，随代理池自动更新。
+> 订阅内容仅包含延迟低于阈值的可用节点，随节点池自动更新。
## 支持协议
@@ -150,10 +150,10 @@ scheduler:
## 延迟检测原理
-延迟检测模拟真实上网场景：通过代理服务器访问目标检测 URL，测量完整请求延迟。
+延迟检测模拟真实上网场景：通过转发服务访问目标检测 URL，测量完整请求延迟。
```
-客户端 → 代理服务器 → 目标网站
+客户端 → 转发服务 → 目标网站
├── DNS 解析
├── TCP 连接建立
├── TLS 握手（HTTPS 目标）
@@ -176,14 +176,14 @@ proxy_pool/
│ ├── models.py # 数据模型（ProxyInfo / ProxyDBRecord / SubscriptionRecord）
│ ├── parser.py # 订阅拉取 & 解析（5 协议）
│ ├── checker.py # HTTP 延迟检测（多目标取最大值）
-│ ├── generator.py # V2Ray / Clash 订阅生成
+│ ├── generator.py # 核心 / Clash 订阅生成
│ ├── scheduler.py # APScheduler 定时任务调度
│ ├── routers/
│ │ ├── api.py # REST API 路由
│ │ └── web.py # Web 页面路由
│ └── templates/
│ ├── base.html # 基础模板
-│ ├── index.html # 主页面（管理 + 代理列表）
+│ ├── index.html # 主页面（管理 + 节点列表）
│ └── subscription.html # 订阅链接页
├── logs/ # 日志目录（自动归档）
├── config.yaml # 配置文件
diff --git a/app/__init__.py b/app/__init__.py
index f8384a4..f213a10 100644
--- a/app/__init__.py
+++ b/app/__init__.py
@@ -89,9 +89,9 @@ def create_app(config_path: str = "config.yaml") -> FastAPI:
# 创建 FastAPI 实例
app = FastAPI(
- title="ProxyPool",
+ title="NetHub",
version="1.0.0",
- description="代理池管理系统",
+ description="节点池管理系统",
)
# 注册生命周期事件
@@ -116,6 +116,10 @@ async def startup():
check_urls=check_urls,
timeout=_config.check.timeout,
max_concurrent=_config.check.max_concurrent,
+ socks_port=_config.check.socks_port,
+ http_port=_config.check.http_port,
+ check_mode=_config.check.check_mode,
+ kernel_path=_config.check.kernel_path,
)
# 初始化调度器
diff --git a/app/checker.py b/app/checker.py
index d953833..4e920b3 100644
--- a/app/checker.py
+++ b/app/checker.py
@@ -1,39 +1,54 @@
from __future__ import annotations
-"""代理延迟检测模块 - 通过 TCP/TLS 连接检测代理服务器响应速度
-
-检测策略：
-- 对代理服务器发起 TCP 连接，测量连接建立耗时
-- 若协议使用 TLS（vmess+tls / vless+tls / trojan / hysteria2），
- 则在 TCP 连接基础上完成 TLS 握手，测量总耗时
-- TCP+TLS 延迟与实际上网体验高度相关：它反映到代理服务器的网络往返时间
- 和服务端处理能力，是真实代理请求延迟的主要组成部分
-- 使用 asyncio.Semaphore 控制并发数
-- 检测目标 URL 存入数据库供页面展示和未来扩展
+"""节点连通性检测模块 - 通过启动本地内核实例转发流量检测延迟
+
+检测策略（优先级从高到低）：
+1. 内核转发检测：为每条节点启动一个内核实例，通过本地 HTTP 转发端口
+ 发送 HTTP 请求到检测目标 URL，测量完整往返时间（含协议握手+传输），
+ 最贴近真实上网体验
+2. TCP/TLS 直连检测：直接 TCP 连接到节点服务器地址，测量连接建立耗时；
+ 若协议使用 TLS，则额外完成 TLS 握手（作为回退方案）
+3. DNS 预解析：非 IP 地址先做 DNS 解析，解析失败直接判定不可达
+
+参照逻辑：
+- 连接被拒绝（refused）仍返回延迟值（端口可达）
+- 超时则返回 None 表示不可达
"""
import asyncio
import json
import logging
+import os
+import socket
import ssl
+import tempfile
import time
import base64
+import shutil
from dataclasses import dataclass
-from urllib.parse import urlparse, unquote
+from urllib.parse import urlparse, unquote, parse_qs
+
+import aiohttp
from app.models import ProxyInfo
logger = logging.getLogger(__name__)
-# 默认检测目标 URL（数据库为空时使用，页面展示用）
+# 默认检测目标 URL（数据库为空时使用）
DEFAULT_CHECK_URLS = [
"https://www.google.com/generate_204",
"https://www.gstatic.com/generate_204",
+ "https://cp.cloudflare.com/",
+ "https://www.apple.com/library/test/success.html",
]
+# 本地转发端口默认值
+DEFAULT_SOCKS_PORT = 1080
+DEFAULT_HTTP_PORT = 1081
+
@dataclass
class ProxyConnInfo:
- """代理服务器连接信息"""
+ """节点服务器连接信息"""
host: str
port: int
use_tls: bool
@@ -41,50 +56,606 @@ class ProxyConnInfo:
class ProxyChecker:
- """基于 asyncio 的并发代理延迟检测器
+ """基于 asyncio 的并发节点连通性检测器
- 通过 TCP/TLS 连接测试代理服务器响应速度。
- 延迟值 = TCP 连接耗时 (+ TLS 握手耗时)，单位毫秒。
- 该值与实际通过代理浏览网页的延迟高度正相关。
+ 检测模式：
+ - http: 为每条节点启动本地内核实例，通过转发端口发送 HTTP 请求（最真实）
+ - tcp: 直接 TCP/TLS 连接到节点服务器地址（回退方案）
+ - auto: 优先内核转发，失败则回退 TCP
"""
- def __init__(self, check_urls: list[str], timeout: float, max_concurrent: int):
+ def __init__(self, check_urls: list[str], timeout: float, max_concurrent: int,
+ socks_port: int = 0, http_port: int = 0,
+ check_mode: str = "auto", kernel_path: str = "xray"):
+ """
+ Args:
+ check_urls: HTTP 检测目标 URL 列表
+ timeout: 单次检测超时秒数
+ max_concurrent: 最大并发数
+ socks_port: 本地 SOCKS5 转发端口（ConnectivityMonitor 使用）
+ http_port: 本地 HTTP 转发端口（ConnectivityMonitor 使用）
+ check_mode: 检测模式 "http" / "tcp" / "auto"
+ kernel_path: 内核可执行文件路径
+ """
self.check_urls = check_urls or DEFAULT_CHECK_URLS
self.timeout = timeout
self.semaphore = asyncio.Semaphore(max_concurrent)
+ self.socks_port = socks_port if socks_port > 0 else DEFAULT_SOCKS_PORT
+ self.http_port = http_port if http_port > 0 else DEFAULT_HTTP_PORT
+ self.check_mode = check_mode
+ self.kernel_path = self._resolve_kernel_path(kernel_path)
+
+ @staticmethod
+ def _resolve_kernel_path(kernel_path: str) -> str:
+ """解析内核路径，返回绝对路径或空字符串"""
+ resolved = shutil.which(kernel_path)
+ if resolved:
+ return resolved
+ if os.path.isfile(kernel_path) and os.access(kernel_path, os.X_OK):
+ return kernel_path
+ logger.warning("内核可执行文件未找到: %s，内核转发检测不可用", kernel_path)
+ return ""
async def check_proxy(self, link: str) -> float | None:
- """检测单个代理的延迟
+ """检测单个节点的延迟
- 通过 TCP/TLS 连接测试代理服务器响应速度。
- 返回连接建立耗时（毫秒），失败返回 None。
+ 根据 check_mode 选择检测方式：
+ - auto: 优先内核转发，失败则回退 TCP/TLS
+ - http: 仅通过内核转发检测
+ - tcp: 仅直连 TCP/TLS 检测
"""
- conn_info = self._parse_link(link)
- if not conn_info:
+ async with self.semaphore:
+ if self.check_mode == "tcp":
+ conn_info = self._parse_link(link)
+ if not conn_info:
+ return None
+ return await self._check_tcp_tls(conn_info)
+
+ # http / auto 模式：优先内核转发
+ if self.kernel_path:
+ latency = await self._check_via_kernel(link)
+ if latency is not None:
+ return latency
+
+ # auto 回退或 http 模式内核不可用时回退 TCP
+ if self.check_mode == "auto" or not self.kernel_path:
+ conn_info = self._parse_link(link)
+ if conn_info:
+ return await self._check_tcp_tls(conn_info)
+
return None
- async with self.semaphore:
- return await self._check_connectivity(conn_info)
+ # ---- 内核转发检测 ----
+
+ async def _check_via_kernel(self, link: str) -> float | None:
+ """为单条节点启动内核实例，通过本地 HTTP 转发端口检测延迟
+
+ 流程：
+ 1. 将分享链接转换为内核 outbound 配置
+ 2. 分配一个空闲端口作为 HTTP inbound
+ 3. 生成完整内核配置并写入临时文件
+ 4. 启动内核进程
+ 5. 等待端口就绪
+ 6. 通过本地 HTTP 转发端口发送 HTTP 请求
+ 7. 测量延迟
+ 8. 终止内核进程并清理临时文件
+ """
+ outbound = self._link_to_xray_outbound(link)
+ if not outbound:
+ return None
+
+ port = self._find_free_port()
+ config = {
+ "log": {"loglevel": "warning"},
+ "inbounds": [{
+ "tag": "http-in",
+ "protocol": "http",
+ "port": port,
+ "listen": "127.0.0.1",
+ }],
+ "outbounds": [outbound],
+ }
+
+ config_path = ""
+ proc = None
+ try:
+ # 写入临时配置文件
+ fd, config_path = tempfile.mkstemp(suffix=".json", prefix="nethub_")
+ with os.fdopen(fd, "w") as f:
+ json.dump(config, f, ensure_ascii=False)
+
+ # 启动内核进程
+ proc = await asyncio.create_subprocess_exec(
+ self.kernel_path, "run", "-c", config_path,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
+ )
+
+ # 等待端口就绪
+ ready = await self._wait_for_port(port, timeout=5.0)
+ if not ready:
+ return None
+
+ # 通过本地 HTTP 转发端口发送请求
+ return await self._http_request_via_proxy(port)
+
+ except Exception:
+ return None
+ finally:
+ # 终止内核进程
+ if proc and proc.returncode is None:
+ try:
+ proc.terminate()
+ await asyncio.wait_for(proc.wait(), timeout=3.0)
+ except Exception:
+ try:
+ proc.kill()
+ except Exception:
+ pass
+ # 清理临时文件
+ if config_path:
+ try:
+ os.unlink(config_path)
+ except Exception:
+ pass
+
+ async def _http_request_via_proxy(self, port: int) -> float | None:
+ """通过本地 HTTP 转发端口发送 HTTP 请求检测延迟"""
+ check_url = self.check_urls[0] if self.check_urls else DEFAULT_CHECK_URLS[0]
+ start = time.monotonic()
+ try:
+ connector = aiohttp.TCPConnector(limit=1, force_close=True)
+ proxy_url = f"http://127.0.0.1:{port}"
+ async with aiohttp.ClientSession(connector=connector) as session:
+ async with session.head(
+ check_url,
+ proxy=proxy_url,
+ timeout=aiohttp.ClientTimeout(total=self.timeout),
+ allow_redirects=True,
+ ssl=False,
+ ) as resp:
+ elapsed = (time.monotonic() - start) * 1000
+ if resp.status < 500:
+ return round(elapsed, 1)
+ return None
+ except Exception:
+ return None
+
+ @staticmethod
+ def _find_free_port() -> int:
+ """查找一个空闲 TCP 端口"""
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+ s.bind(("127.0.0.1", 0))
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ return s.getsockname()[1]
+
+ @staticmethod
+ async def _wait_for_port(port: int, timeout: float = 5.0) -> bool:
+ """等待端口可连接"""
+ deadline = time.monotonic() + timeout
+ while time.monotonic() < deadline:
+ try:
+ _, writer = await asyncio.wait_for(
+ asyncio.open_connection("127.0.0.1", port),
+ timeout=1.0,
+ )
+ writer.close()
+ try:
+ await writer.wait_closed()
+ except Exception:
+ pass
+ return True
+ except Exception:
+ await asyncio.sleep(0.1)
+ return False
+
+ # ---- 内核配置生成 ----
+
+ @staticmethod
+ def _link_to_xray_outbound(link: str) -> dict | None:
+ """将分享链接转换为内核 outbound 配置"""
+ try:
+ if link.startswith("vmess://"):
+ return ProxyChecker._vmess_to_xray(link)
+ elif link.startswith("vless://"):
+ return ProxyChecker._vless_to_xray(link)
+ elif link.startswith("trojan://"):
+ return ProxyChecker._trojan_to_xray(link)
+ elif link.startswith("ss://"):
+ return ProxyChecker._ss_to_xray(link)
+ elif link.startswith("hysteria2://") or link.startswith("hy2://"):
+ return ProxyChecker._hysteria2_to_xray(link)
+ except Exception:
+ pass
+ return None
+
+ @staticmethod
+ def _vmess_to_xray(link: str) -> dict | None:
+ """vmess:// 分享链接转内核 outbound"""
+ try:
+ config_b64 = link[8:]
+ padding = 4 - len(config_b64) % 4
+ if padding != 4:
+ config_b64 += "=" * padding
+ config = json.loads(base64.b64decode(config_b64).decode("utf-8"))
+
+ address = config.get("add", "")
+ port = int(config.get("port", 0))
+ if not address or not port:
+ return None
+
+ outbound = {
+ "protocol": "vmess",
+ "settings": {
+ "vnext": [{
+ "address": address,
+ "port": port,
+ "users": [{
+ "id": config.get("id", ""),
+ "alterId": int(config.get("aid", 0)),
+ "security": config.get("scy", "auto"),
+ }]
+ }]
+ }
+ }
+
+ network = config.get("net", "tcp")
+ tls = config.get("tls", "")
+ stream = {"network": network}
+
+ if tls == "tls":
+ stream["security"] = "tls"
+ tls_settings = {}
+ if config.get("sni"):
+ tls_settings["serverName"] = config["sni"]
+ if config.get("alpn"):
+ tls_settings["alpn"] = config["alpn"].split(",")
+ if config.get("fp"):
+ tls_settings["fingerprint"] = config["fp"]
+ if tls_settings:
+ stream["tlsSettings"] = tls_settings
+ elif tls == "reality":
+ stream["security"] = "reality"
+ reality_settings = {}
+ if config.get("sni"):
+ reality_settings["serverName"] = config["sni"]
+ if config.get("pbk"):
+ reality_settings["publicKey"] = config["pbk"]
+ if config.get("sid"):
+ reality_settings["shortId"] = config["sid"]
+ if config.get("fp"):
+ reality_settings["fingerprint"] = config["fp"]
+ if reality_settings:
+ stream["realitySettings"] = reality_settings
+ else:
+ stream["security"] = "none"
+
+ if network == "ws":
+ ws_settings = {}
+ if config.get("path"):
+ ws_settings["path"] = config["path"]
+ if config.get("host"):
+ ws_settings["headers"] = {"Host": config["host"]}
+ stream["wsSettings"] = ws_settings
+ elif network == "grpc":
+ grpc_settings = {}
+ if config.get("path"):
+ grpc_settings["serviceName"] = config["path"]
+ stream["grpcSettings"] = grpc_settings
+ elif network in ("h2", "http"):
+ h2_settings = {}
+ if config.get("path"):
+ h2_settings["path"] = config["path"]
+ if config.get("host"):
+ h2_settings["host"] = [config["host"]]
+ stream["httpSettings"] = h2_settings
+
+ outbound["streamSettings"] = stream
+ return outbound
+ except Exception:
+ return None
+
+ @staticmethod
+ def _vless_to_xray(link: str) -> dict | None:
+ """vless:// 分享链接转内核 outbound"""
+ try:
+ parsed = urlparse(link)
+ query = parse_qs(parsed.query)
+
+ uuid = unquote(parsed.username) if parsed.username else ""
+ address = parsed.hostname or ""
+ port = parsed.port or 0
+ if not address or not port:
+ return None
+
+ users = [{"id": uuid, "encryption": "none"}]
+ flow = query.get("flow", [None])[0]
+ if flow:
+ users[0]["flow"] = flow
+
+ outbound = {
+ "protocol": "vless",
+ "settings": {
+ "vnext": [{
+ "address": address,
+ "port": port,
+ "users": users,
+ }]
+ }
+ }
+
+ network = query.get("type", ["tcp"])[0]
+ security = query.get("security", ["none"])[0]
+ stream = {"network": network}
+
+ if security == "tls":
+ stream["security"] = "tls"
+ tls_settings = {}
+ sni = query.get("sni", [None])[0]
+ if sni:
+ tls_settings["serverName"] = sni
+ alpn = query.get("alpn", [None])[0]
+ if alpn:
+ tls_settings["alpn"] = alpn.split(",")
+ fp = query.get("fp", [None])[0]
+ if fp:
+ tls_settings["fingerprint"] = fp
+ if tls_settings:
+ stream["tlsSettings"] = tls_settings
+ elif security == "reality":
+ stream["security"] = "reality"
+ reality_settings = {}
+ sni = query.get("sni", [None])[0]
+ if sni:
+ reality_settings["serverName"] = sni
+ pbk = query.get("pbk", [None])[0]
+ if pbk:
+ reality_settings["publicKey"] = pbk
+ sid = query.get("sid", [None])[0]
+ if sid:
+ reality_settings["shortId"] = sid
+ fp = query.get("fp", [None])[0]
+ if fp:
+ reality_settings["fingerprint"] = fp
+ if reality_settings:
+ stream["realitySettings"] = reality_settings
+ else:
+ stream["security"] = "none"
+
+ if network == "ws":
+ ws_settings = {}
+ path = query.get("path", [None])[0]
+ if path:
+ ws_settings["path"] = path
+ host = query.get("host", [None])[0]
+ if host:
+ ws_settings["headers"] = {"Host": host}
+ stream["wsSettings"] = ws_settings
+ elif network == "grpc":
+ grpc_settings = {}
+ service_name = query.get("serviceName", [None])[0]
+ if service_name:
+ grpc_settings["serviceName"] = service_name
+ stream["grpcSettings"] = grpc_settings
+ elif network in ("h2", "http"):
+ h2_settings = {}
+ path = query.get("path", [None])[0]
+ if path:
+ h2_settings["path"] = path
+ host = query.get("host", [None])[0]
+ if host:
+ h2_settings["host"] = [host]
+ stream["httpSettings"] = h2_settings
+
+ outbound["streamSettings"] = stream
+ return outbound
+ except Exception:
+ return None
+
+ @staticmethod
+ def _trojan_to_xray(link: str) -> dict | None:
+ """trojan:// 分享链接转内核 outbound"""
+ try:
+ parsed = urlparse(link)
+ query = parse_qs(parsed.query)
+
+ password = unquote(parsed.username) if parsed.username else ""
+ address = parsed.hostname or ""
+ port = parsed.port or 443
+ if not address or not port:
+ return None
+
+ outbound = {
+ "protocol": "trojan",
+ "settings": {
+ "servers": [{
+ "address": address,
+ "port": port,
+ "password": password,
+ }]
+ }
+ }
+
+ network = query.get("type", ["tcp"])[0]
+ security = query.get("security", ["tls"])[0]
+ stream = {"network": network, "security": "tls"}
+
+ tls_settings = {}
+ sni = query.get("sni", [None])[0]
+ if sni:
+ tls_settings["serverName"] = sni
+ alpn = query.get("alpn", [None])[0]
+ if alpn:
+ tls_settings["alpn"] = alpn.split(",")
+ fp = query.get("fp", [None])[0]
+ if fp:
+ tls_settings["fingerprint"] = fp
+ if tls_settings:
+ stream["tlsSettings"] = tls_settings
+
+ if network == "ws":
+ ws_settings = {}
+ path = query.get("path", [None])[0]
+ if path:
+ ws_settings["path"] = path
+ host = query.get("host", [None])[0]
+ if host:
+ ws_settings["headers"] = {"Host": host}
+ stream["wsSettings"] = ws_settings
+ elif network == "grpc":
+ grpc_settings = {}
+ service_name = query.get("serviceName", [None])[0]
+ if service_name:
+ grpc_settings["serviceName"] = service_name
+ stream["grpcSettings"] = grpc_settings
+
+ outbound["streamSettings"] = stream
+ return outbound
+ except Exception:
+ return None
+
+ @staticmethod
+ def _ss_to_xray(link: str) -> dict | None:
+ """ss:// 分享链接转内核 outbound"""
+ try:
+ line = link
+ if "#" in line:
+ line = line[:line.rindex("#")]
+
+ ss_content = line[5:]
+ cipher = ""
+ password = ""
+ address = ""
+ port = 0
+
+ if "@" in ss_content:
+ at_idx = ss_content.rindex("@")
+ addr_port = ss_content[at_idx + 1:]
+ user_info_b64 = ss_content[:at_idx]
+
+ try:
+ padding = 4 - len(user_info_b64) % 4
+ if padding != 4:
+ user_info_b64 += "=" * padding
+ decoded = base64.b64decode(user_info_b64).decode("utf-8")
+ cipher, password = decoded.split(":", 1)
+ except Exception:
+ pass
+
+ if addr_port.startswith("["):
+ bracket_end = addr_port.index("]")
+ address = addr_port[1:bracket_end]
+ port = int(addr_port[bracket_end + 2:]) if bracket_end + 2 < len(addr_port) else 0
+ elif ":" in addr_port:
+ address, port_str = addr_port.rsplit(":", 1)
+ port = int(port_str)
+ else:
+ try:
+ padding = 4 - len(ss_content) % 4
+ if padding != 4:
+ ss_content += "=" * padding
+ decoded = base64.b64decode(ss_content).decode("utf-8")
+ user_info, addr_port = decoded.rsplit("@", 1)
+ cipher, password = user_info.split(":", 1)
+ if addr_port.startswith("["):
+ bracket_end = addr_port.index("]")
+ address = addr_port[1:bracket_end]
+ port = int(addr_port[bracket_end + 2:]) if bracket_end + 2 < len(addr_port) else 0
+ elif ":" in addr_port:
+ address, port_str = addr_port.rsplit(":", 1)
+ port = int(port_str)
+ except Exception:
+ pass
+
+ if not address or not port:
+ return None
+
+ return {
+ "protocol": "shadowsocks",
+ "settings": {
+ "servers": [{
+ "address": address,
+ "port": port,
+ "method": cipher,
+ "password": password,
+ }]
+ }
+ }
+ except Exception:
+ return None
+
+ @staticmethod
+ def _hysteria2_to_xray(link: str) -> dict | None:
+ """hysteria2:// 分享链接转内核 outbound"""
+ try:
+ prefix_len = len("hysteria2://") if link.startswith("hysteria2://") else len("hy2://")
+ rest = link[prefix_len:]
+
+ if "#" in rest:
+ rest = rest[:rest.rindex("#")]
+
+ if "?" in rest:
+ host_part, query_part = rest.split("?", 1)
+ parsed = urlparse("http://" + host_part)
+ query = parse_qs(query_part)
+ else:
+ parsed = urlparse("http://" + rest)
+ query = {}
+
+ password = unquote(parsed.username) if parsed.username else ""
+ address = parsed.hostname or ""
+ port = parsed.port or 443
+ if not address or not port:
+ return None
+
+ outbound = {
+ "protocol": "hysteria2",
+ "settings": {
+ "servers": [{
+ "address": address,
+ "port": port,
+ "password": password,
+ }]
+ }
+ }
+
+ stream = {"network": "h2", "security": "tls"}
+ sni = query.get("sni", [None])[0]
+ if sni:
+ stream["tlsSettings"] = {"serverName": sni}
+
+ outbound["streamSettings"] = stream
+ return outbound
+ except Exception:
+ return None
+
+ # ---- TCP/TLS 直连检测（回退方案） ----
+
+ async def _check_tcp_tls(self, info: ProxyConnInfo) -> float | None:
+ """直接 TCP/TLS 连接到节点服务器检测延迟（回退方案）"""
+ host = info.host
+ if not self._is_ip(host):
+ resolved = await self._resolve_host(host)
+ if not resolved:
+ logger.debug("DNS 解析失败: %s", host)
+ return None
+ host = resolved
- async def _check_connectivity(self, info: ProxyConnInfo) -> float | None:
- """测试代理服务器 TCP/TLS 连通性并测量延迟"""
start = time.monotonic()
try:
if info.use_tls:
- # TLS 协议：TCP 连接 + TLS 握手，更贴近真实代理请求体验
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
_, writer = await asyncio.wait_for(
asyncio.open_connection(
- info.host, info.port, ssl=ctx, server_hostname=info.host
+ host, info.port, ssl=ctx, server_hostname=host
),
timeout=self.timeout,
)
else:
- # 非 TLS 协议：仅 TCP 连接
_, writer = await asyncio.wait_for(
- asyncio.open_connection(info.host, info.port),
+ asyncio.open_connection(host, info.port),
timeout=self.timeout,
)
@@ -95,6 +666,37 @@ async def _check_connectivity(self, info: ProxyConnInfo) -> float | None:
except Exception:
pass
return round(elapsed, 1)
+ except ConnectionRefusedError:
+ elapsed = (time.monotonic() - start) * 1000
+ return round(elapsed, 1)
+ except Exception:
+ return None
+
+ @staticmethod
+ def _is_ip(host: str) -> bool:
+ """判断是否为 IP 地址"""
+ try:
+ socket.inet_pton(socket.AF_INET, host)
+ return True
+ except (OSError, socket.error):
+ pass
+ try:
+ socket.inet_pton(socket.AF_INET6, host)
+ return True
+ except (OSError, socket.error):
+ pass
+ return False
+
+ @staticmethod
+ async def _resolve_host(host: str) -> str | None:
+ """异步 DNS 解析，返回第一个 IPv4/IPv6 地址"""
+ try:
+ loop = asyncio.get_running_loop()
+ addrs = await loop.getaddrinfo(host, None)
+ for family, _, _, _, sockaddr in addrs:
+ if family in (socket.AF_INET, socket.AF_INET6):
+ return sockaddr[0]
+ return None
except Exception:
return None
@@ -117,7 +719,7 @@ async def check_proxy_infos(self, proxies: list[ProxyInfo]) -> dict[str, float |
links = [p.link for p in proxies]
return await self.check_batch(links)
- # ---- 协议解析 ----
+ # ---- 协议解析（TCP/TLS 回退方案使用） ----
def _parse_link(self, link: str) -> ProxyConnInfo | None:
"""解析分享链接，提取连接信息（地址、端口、是否 TLS）"""
@@ -142,13 +744,11 @@ def _parse_vmess(self, link: str) -> ProxyConnInfo | None:
padding = 4 - len(config_b64) % 4
if padding != 4:
config_b64 += "=" * padding
- config_json = base64.b64decode(config_b64).decode("utf-8")
- config = json.loads(config_json)
+ config = json.loads(base64.b64decode(config_b64).decode("utf-8"))
address = config.get("add", "")
port = config.get("port", 0)
if not address or not port:
return None
- # vmess 的 tls 字段为 "tls" 或 "reality" 时使用 TLS
use_tls = config.get("tls", "") in ("tls", "reality")
return ProxyConnInfo(
host=address, port=int(port),
@@ -181,7 +781,6 @@ def _parse_trojan(self, link: str) -> ProxyConnInfo | None:
port = parsed.port or 0
if not address or not port:
return None
- # trojan 默认使用 TLS
return ProxyConnInfo(
host=address, port=int(port),
use_tls=True, protocol="trojan",
@@ -236,16 +835,54 @@ def _parse_hysteria2(self, link: str) -> ProxyConnInfo | None:
try:
prefix_len = len("hysteria2://") if link.startswith("hysteria2://") else len("hy2://")
rest = link[prefix_len:]
- # 构造标准 URL 以便 urlparse 解析
parsed = urlparse("http://" + rest)
address = parsed.hostname or ""
port = parsed.port or 0
if not address or not port:
return None
- # hysteria2 默认使用 TLS
return ProxyConnInfo(
host=address, port=int(port),
use_tls=True, protocol="hysteria2",
)
except Exception:
return None
+
+
+class ConnectivityMonitor:
+ """连通性监控器 - 定期检测本地转发端口是否可达"""
+
+ def __init__(self, socks_port: int = DEFAULT_SOCKS_PORT,
+ check_interval: float = 15.0,
+ timeout: float = 5.0,
+ backoff_base: float = 30.0,
+ backoff_max: float = 120.0):
+ self.socks_port = socks_port if socks_port > 0 else DEFAULT_SOCKS_PORT
+ self.check_interval = check_interval
+ self.timeout = timeout
+ self.backoff_base = backoff_base
+ self.backoff_max = backoff_max
+
+ async def probe(self) -> bool:
+ """检测本地转发端口是否可达"""
+ try:
+ _, writer = await asyncio.wait_for(
+ asyncio.open_connection("127.0.0.1", self.socks_port),
+ timeout=self.timeout,
+ )
+ writer.close()
+ try:
+ await writer.wait_closed()
+ except Exception:
+ pass
+ return True
+ except Exception:
+ return False
+
+ def backoff_delay(self, consecutive_failures: int) -> float:
+ """计算失败后的退避等待时间"""
+ delay = self.backoff_base
+ for _ in range(1, consecutive_failures):
+ delay *= 2
+ if delay >= self.backoff_max:
+ return self.backoff_max
+ return delay
diff --git a/app/config.py b/app/config.py
index 322b4f1..9d6f107 100644
--- a/app/config.py
+++ b/app/config.py
@@ -23,12 +23,16 @@ class CheckConfig:
timeout: float = 5.0
max_concurrent: int = 50
latency_threshold: float = 1500.0
+ check_mode: str = "auto" # 检测模式: "auto" / "http" / "tcp"
+ socks_port: int = 1080 # 本地转发端口（ConnectivityMonitor 使用）
+ http_port: int = 1081 # 本地 HTTP 转发端口
+ kernel_path: str = "xray" # 内核可执行文件路径
@dataclass
class SchedulerConfig:
fetch_interval: int = 3600 # 拉取订阅间隔（秒）
- verify_interval: int = 1800 # 验证代理间隔（秒）
+ verify_interval: int = 1800 # 验证节点间隔（秒）
cleanup_interval: int = 7200 # 清理间隔（秒）
max_fail_count: int = 3 # 最大连续失败次数
@@ -75,6 +79,10 @@ def load_config(path: str = "config.yaml") -> AppConfig:
"timeout": 5.0,
"max_concurrent": 50,
"latency_threshold": 1500.0,
+ "check_mode": "auto",
+ "socks_port": 1080,
+ "http_port": 1081,
+ "kernel_path": "xray",
},
"scheduler": {
"fetch_interval": 3600,
diff --git a/app/database.py b/app/database.py
index c409fb3..766f651 100644
--- a/app/database.py
+++ b/app/database.py
@@ -2,15 +2,15 @@
"""数据库操作层 - aiosqlite 异步封装"""
import os
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
import aiosqlite
-from app.models import ProxyDBRecord, ProxyInfo, SubscriptionRecord
+from app.models import ProxyDBRecord, ProxyInfo, SubscriptionRecord, InstanceSourceRecord
class ProxyDatabase:
- """代理数据库异步操作层"""
+ """节点数据库异步操作层"""
def __init__(self, db_path: str):
self.db_path = db_path
@@ -67,6 +67,22 @@ async def init(self) -> None:
id INTEGER PRIMARY KEY AUTOINCREMENT,
url TEXT NOT NULL UNIQUE
);
+
+ CREATE TABLE IF NOT EXISTS instance_sources (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ base_url TEXT NOT NULL UNIQUE,
+ username TEXT NOT NULL DEFAULT '',
+ password TEXT NOT NULL DEFAULT '',
+ crontab TEXT NOT NULL DEFAULT '*/10 * * * *',
+ latency_threshold REAL DEFAULT 1500.0,
+ max_concurrent INTEGER DEFAULT 50,
+ enabled INTEGER DEFAULT 1,
+ created_at TEXT DEFAULT '',
+ total_count INTEGER DEFAULT 0,
+ fetch_status TEXT DEFAULT 'idle'
+ );
+
+ CREATE INDEX IF NOT EXISTS idx_instance_sources_base_url ON instance_sources(base_url);
""")
await self._db.commit()
@@ -129,7 +145,7 @@ def _row_to_record(self, row: aiosqlite.Row) -> ProxyDBRecord:
)
async def insert_proxy(self, proxy: ProxyInfo, latency_ms: float, source: str) -> bool:
- """插入新代理，link 唯一约束，重复则忽略。返回是否插入成功"""
+ """插入新节点，link 唯一约束，重复则忽略。返回是否插入成功"""
now = datetime.now(timezone.utc).isoformat()
try:
cursor = await self._db.execute(
@@ -157,6 +173,37 @@ async def update_latency(self, proxy_id: int, latency_ms: float) -> None:
)
await self._db.commit()
+ async def batch_update_latency(self, updates: list[tuple[int, float]]) -> None:
+ """批量更新延迟并重置 fail_count
+
+ updates: [(proxy_id, latency_ms), ...]
+ 单次 commit，避免逐条提交的性能开销
+ """
+ if not updates:
+ return
+ now = datetime.now(timezone.utc).isoformat()
+ await self._db.executemany(
+ """UPDATE proxies
+ SET latency_ms = ?, fail_count = 0,
+ last_check_time = ?, last_success_time = ?
+ WHERE id = ?""",
+ [(lat, now, now, pid) for pid, lat in updates],
+ )
+ await self._db.commit()
+
+ async def batch_increment_fail(self, proxy_ids: list[int]) -> None:
+ """批量 fail_count + 1，单次 commit"""
+ if not proxy_ids:
+ return
+ now = datetime.now(timezone.utc).isoformat()
+ await self._db.executemany(
+ """UPDATE proxies
+ SET fail_count = fail_count + 1, last_check_time = ?
+ WHERE id = ?""",
+ [(now, pid) for pid in proxy_ids],
+ )
+ await self._db.commit()
+
async def increment_fail(self, proxy_id: int) -> int:
"""fail_count + 1，返回当前值"""
now = datetime.now(timezone.utc).isoformat()
@@ -174,21 +221,55 @@ async def increment_fail(self, proxy_id: int) -> int:
row = await cursor.fetchone()
return row["fail_count"] if row else 0
+ async def update_proxy_source(self, proxy_id: int, source: str) -> None:
+ """更新节点的 source 字段（用于去重：将实例源节点归属转移到订阅源）"""
+ await self._db.execute(
+ "UPDATE proxies SET source = ? WHERE id = ?",
+ (source, proxy_id),
+ )
+ await self._db.commit()
+
async def delete_proxy(self, proxy_id: int) -> None:
- """删除指定代理"""
+ """删除指定节点"""
await self._db.execute("DELETE FROM proxies WHERE id = ?", (proxy_id,))
await self._db.commit()
async def delete_proxies_by_fail_count(self, max_fail: int) -> int:
- """删除 fail_count >= max_fail 的代理，返回删除数量"""
+ """删除 fail_count >= max_fail 的节点，返回删除数量"""
cursor = await self._db.execute(
"DELETE FROM proxies WHERE fail_count >= ?", (max_fail,)
)
await self._db.commit()
return cursor.rowcount
+ async def delete_sub_proxies_by_fail_count(self, max_fail: int) -> int:
+ """删除订阅源中 fail_count >= max_fail 的节点（source 不以 instance: 开头），返回删除数量"""
+ cursor = await self._db.execute(
+ "DELETE FROM proxies WHERE fail_count >= ? AND source NOT LIKE 'instance:%'",
+ (max_fail,),
+ )
+ await self._db.commit()
+ return cursor.rowcount
+
+ async def delete_instance_proxies_stale(self, days: int) -> int:
+ """删除实例源中连续 days 天无成功的节点
+
+ 条件：source 以 'instance:' 开头 且 fail_count > 0 且
+ (last_success_time 为空 或 last_success_time 距今超过 days 天)
+ """
+ cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
+ cursor = await self._db.execute(
+ """DELETE FROM proxies
+ WHERE source LIKE 'instance:%'
+ AND fail_count > 0
+ AND (last_success_time = '' OR last_success_time < ?)""",
+ (cutoff,),
+ )
+ await self._db.commit()
+ return cursor.rowcount
+
async def get_all_proxies(self) -> list[ProxyDBRecord]:
- """获取所有代理"""
+ """获取所有节点"""
cursor = await self._db.execute(
"SELECT * FROM proxies ORDER BY latency_ms ASC"
)
@@ -196,7 +277,7 @@ async def get_all_proxies(self) -> list[ProxyDBRecord]:
return [self._row_to_record(row) for row in rows]
async def get_available_proxies(self, max_latency: float) -> list[ProxyDBRecord]:
- """获取延迟低于阈值且未失败的可用代理"""
+ """获取延迟低于阈值且未失败的可用节点"""
cursor = await self._db.execute(
"""SELECT * FROM proxies
WHERE latency_ms > 0 AND latency_ms <= ? AND fail_count = 0
@@ -206,8 +287,24 @@ async def get_available_proxies(self, max_latency: float) -> list[ProxyDBRecord]
rows = await cursor.fetchall()
return [self._row_to_record(row) for row in rows]
+ async def get_subscription_output_proxies(self, max_latency: float) -> list[ProxyDBRecord]:
+ """获取对外订阅输出节点列表
+
+ 输出所有延迟达标且未失败的节点，包含订阅源和检测通过的实例源节点
+ """
+ cursor = await self._db.execute(
+ """SELECT * FROM proxies
+ WHERE latency_ms > 0
+ AND latency_ms <= ?
+ AND fail_count = 0
+ ORDER BY latency_ms ASC""",
+ (max_latency,),
+ )
+ rows = await cursor.fetchall()
+ return [self._row_to_record(row) for row in rows]
+
async def get_proxy_by_link(self, link: str) -> ProxyDBRecord | None:
- """根据 link 查询代理"""
+ """根据 link 查询节点"""
cursor = await self._db.execute(
"SELECT * FROM proxies WHERE link = ?", (link,)
)
@@ -215,7 +312,7 @@ async def get_proxy_by_link(self, link: str) -> ProxyDBRecord | None:
return self._row_to_record(row) if row else None
async def get_proxies_needing_verify(self, limit: int = 0) -> list[ProxyDBRecord]:
- """获取需要验证的代理（按 last_check_time 排序，最旧的优先）"""
+ """获取需要验证的节点（按 last_check_time 排序，最旧的优先）"""
query = "SELECT * FROM proxies ORDER BY last_check_time ASC"
if limit > 0:
query += f" LIMIT {limit}"
@@ -319,7 +416,7 @@ async def delete_subscription(self, sub_id: int) -> bool:
return cursor.rowcount > 0
async def increment_empty_days(self, sub_id: int) -> None:
- """订阅源连续空代理天数 +1"""
+ """订阅源连续空节点天数 +1"""
await self._db.execute(
"UPDATE subscriptions SET empty_days = empty_days + 1 WHERE id = ?",
(sub_id,),
@@ -327,7 +424,7 @@ async def increment_empty_days(self, sub_id: int) -> None:
await self._db.commit()
async def reset_empty_days(self, sub_id: int) -> None:
- """订阅源有代理时重置空天数为0"""
+ """订阅源有节点时重置空天数为0"""
await self._db.execute(
"UPDATE subscriptions SET empty_days = 0 WHERE id = ?",
(sub_id,),
@@ -335,7 +432,7 @@ async def reset_empty_days(self, sub_id: int) -> None:
await self._db.commit()
async def get_subscriptions_with_empty_days(self, min_days: int) -> list[SubscriptionRecord]:
- """获取连续空代理天数 >= min_days 的订阅源"""
+ """获取连续空节点天数 >= min_days 的订阅源"""
cursor = await self._db.execute(
"SELECT * FROM subscriptions WHERE empty_days >= ?", (min_days,)
)
@@ -343,7 +440,7 @@ async def get_subscriptions_with_empty_days(self, min_days: int) -> list[Subscri
return [self._row_to_subscription(row) for row in rows]
async def get_proxy_count_by_source(self, source: str) -> int:
- """获取指定来源的代理总数"""
+ """获取指定来源的节点总数"""
cursor = await self._db.execute(
"SELECT COUNT(*) as cnt FROM proxies WHERE source = ?", (source,)
)
@@ -351,7 +448,7 @@ async def get_proxy_count_by_source(self, source: str) -> int:
return row["cnt"] if row else 0
async def update_total_count(self, sub_id: int, count: int) -> None:
- """更新订阅源最新一次拉取的代理总数"""
+ """更新订阅源最新一次拉取的节点总数"""
await self._db.execute(
"UPDATE subscriptions SET total_count = ? WHERE id = ?",
(count, sub_id),
@@ -367,7 +464,7 @@ async def update_fetch_status(self, sub_id: int, status: str) -> None:
await self._db.commit()
async def get_proxies_by_source(self, source: str) -> list[ProxyDBRecord]:
- """根据来源订阅 URL 获取代理，按入库时间正序"""
+ """根据来源订阅 URL 获取节点，按入库时间正序"""
cursor = await self._db.execute(
"""SELECT * FROM proxies WHERE source = ?
ORDER BY created_at ASC""",
@@ -377,7 +474,7 @@ async def get_proxies_by_source(self, source: str) -> list[ProxyDBRecord]:
return [self._row_to_record(row) for row in rows]
async def get_available_proxies_by_source(self, source: str, max_latency: float) -> list[ProxyDBRecord]:
- """根据来源订阅 URL 获取可用代理，按入库时间正序"""
+ """根据来源订阅 URL 获取可用节点，按入库时间正序"""
cursor = await self._db.execute(
"""SELECT * FROM proxies
WHERE source = ? AND latency_ms > 0 AND latency_ms <= ? AND fail_count = 0
@@ -388,7 +485,7 @@ async def get_available_proxies_by_source(self, source: str, max_latency: float)
return [self._row_to_record(row) for row in rows]
async def get_proxies_grouped_by_source(self, max_latency: float) -> dict[str, list[ProxyDBRecord]]:
- """获取按订阅来源分组的可用代理，按入库时间正序"""
+ """获取按订阅来源分组的可用节点，按入库时间正序"""
cursor = await self._db.execute(
"""SELECT * FROM proxies
WHERE latency_ms > 0 AND latency_ms <= ? AND fail_count = 0
@@ -477,3 +574,107 @@ async def init_check_urls(self, urls: list[str]) -> None:
except aiosqlite.IntegrityError:
pass
await self._db.commit()
+
+ # ---- 服务实例源管理 ----
+
+ def _row_to_instance_source(self, row: aiosqlite.Row) -> InstanceSourceRecord:
+ """将数据库行转为 InstanceSourceRecord"""
+ return InstanceSourceRecord(
+ id=row["id"],
+ base_url=row["base_url"],
+ username=row["username"],
+ password=row["password"],
+ crontab=row["crontab"],
+ latency_threshold=row["latency_threshold"],
+ max_concurrent=row["max_concurrent"],
+ enabled=bool(row["enabled"]),
+ created_at=row["created_at"],
+ total_count=row["total_count"] if "total_count" in row.keys() else 0,
+ fetch_status=row["fetch_status"] if "fetch_status" in row.keys() else "idle",
+ )
+
+ async def add_instance_source(self, base_url: str, username: str, password: str,
+ crontab: str = "*/10 * * * *",
+ latency_threshold: float = 1500.0,
+ max_concurrent: int = 50,
+ enabled: bool = True) -> InstanceSourceRecord | None:
+ """添加服务实例源"""
+ now = datetime.now(timezone.utc).isoformat()
+ try:
+ cursor = await self._db.execute(
+ """INSERT INTO instance_sources
+ (base_url, username, password, crontab, latency_threshold, max_concurrent, enabled, created_at)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
+ (base_url, username, password, crontab, latency_threshold, max_concurrent, int(enabled), now),
+ )
+ await self._db.commit()
+ return await self.get_instance_source_by_id(cursor.lastrowid)
+ except aiosqlite.IntegrityError:
+ return None
+
+ async def get_instance_source_by_id(self, source_id: int) -> InstanceSourceRecord | None:
+ """根据 ID 获取服务实例源"""
+ cursor = await self._db.execute("SELECT * FROM instance_sources WHERE id = ?", (source_id,))
+ row = await cursor.fetchone()
+ return self._row_to_instance_source(row) if row else None
+
+ async def get_instance_source_by_url(self, base_url: str) -> InstanceSourceRecord | None:
+ """根据 base_url 获取服务实例源"""
+ cursor = await self._db.execute("SELECT * FROM instance_sources WHERE base_url = ?", (base_url,))
+ row = await cursor.fetchone()
+ return self._row_to_instance_source(row) if row else None
+
+ async def get_all_instance_sources(self) -> list[InstanceSourceRecord]:
+ """获取所有服务实例源"""
+ cursor = await self._db.execute("SELECT * FROM instance_sources ORDER BY id ASC")
+ rows = await cursor.fetchall()
+ return [self._row_to_instance_source(row) for row in rows]
+
+ async def get_enabled_instance_sources(self) -> list[InstanceSourceRecord]:
+ """获取所有启用的服务实例源"""
+ cursor = await self._db.execute("SELECT * FROM instance_sources WHERE enabled = 1 ORDER BY id ASC")
+ rows = await cursor.fetchall()
+ return [self._row_to_instance_source(row) for row in rows]
+
+ async def update_instance_source(self, source_id: int, **kwargs) -> bool:
+ """更新服务实例源，支持部分字段更新"""
+ allowed = {"base_url", "username", "password", "crontab", "latency_threshold", "max_concurrent", "enabled"}
+ updates = {}
+ for k, v in kwargs.items():
+ if k in allowed:
+ if k == "enabled":
+ updates[k] = int(v)
+ else:
+ updates[k] = v
+ if not updates:
+ return False
+
+ set_clause = ", ".join(f"{k} = ?" for k in updates)
+ values = list(updates.values()) + [source_id]
+ cursor = await self._db.execute(
+ f"UPDATE instance_sources SET {set_clause} WHERE id = ?", values
+ )
+ await self._db.commit()
+ return cursor.rowcount > 0
+
+ async def delete_instance_source(self, source_id: int) -> bool:
+ """删除服务实例源"""
+ cursor = await self._db.execute("DELETE FROM instance_sources WHERE id = ?", (source_id,))
+ await self._db.commit()
+ return cursor.rowcount > 0
+
+ async def update_instance_total_count(self, source_id: int, count: int) -> None:
+ """更新服务实例源最新一次获取的节点总数"""
+ await self._db.execute(
+ "UPDATE instance_sources SET total_count = ? WHERE id = ?",
+ (count, source_id),
+ )
+ await self._db.commit()
+
+ async def update_instance_fetch_status(self, source_id: int, status: str) -> None:
+ """更新服务实例源拉取状态: idle / updating / success / failed"""
+ await self._db.execute(
+ "UPDATE instance_sources SET fetch_status = ? WHERE id = ?",
+ (status, source_id),
+ )
+ await self._db.commit()
diff --git a/app/generator.py b/app/generator.py
index 97ac984..6add092 100644
--- a/app/generator.py
+++ b/app/generator.py
@@ -1,5 +1,5 @@
from __future__ import annotations
-"""订阅链接生成模块 - 生成 v2ray 和 Clash 格式订阅内容"""
+"""订阅链接生成模块 - 生成纯文本/Clash 格式订阅内容"""
import base64
import json
@@ -13,14 +13,21 @@
logger = logging.getLogger(__name__)
-def generate_v2ray_subscription(proxies: list[ProxyDBRecord]) -> str:
- """生成 v2ray 格式订阅内容
+def generate_plain_subscription(proxies: list[ProxyDBRecord]) -> str:
+ """生成纯文本格式订阅内容（参照 subdom.txt 格式）
- 将所有代理的原始 link 换行拼接后 base64 编码
+ 每行一条原始代理 URI
"""
links = [p.link for p in proxies]
- content = "\n".join(links)
- return base64.b64encode(content.encode("utf-8")).decode("utf-8")
+ return "\n".join(links)
+
+
+def generate_v2ray_subscription(proxies: list[ProxyDBRecord]) -> str:
+ """生成 v2ray 格式订阅内容（纯文本，每行一条原始链接）
+
+ 兼容旧名，实际输出与 generate_plain_subscription 一致
+ """
+ return generate_plain_subscription(proxies)
def generate_clash_subscription(proxies: list[ProxyDBRecord]) -> str:
@@ -45,7 +52,7 @@ def generate_clash_subscription(proxies: list[ProxyDBRecord]) -> str:
"proxies": proxy_list,
"proxy-groups": [
{
- "name": "ProxyPool",
+ "name": "NetHub",
"type": "url-test",
"proxies": proxy_names if proxy_names else ["DIRECT"],
"url": "http://www.gstatic.com/generate_204",
@@ -54,7 +61,7 @@ def generate_clash_subscription(proxies: list[ProxyDBRecord]) -> str:
{
"name": "Proxy",
"type": "select",
- "proxies": ["ProxyPool", "DIRECT"] + proxy_names,
+ "proxies": ["NetHub", "DIRECT"] + proxy_names,
},
],
"rules": [
diff --git a/app/main.py b/app/main.py
index 5276bcb..5f4c2d9 100644
--- a/app/main.py
+++ b/app/main.py
@@ -1,4 +1,4 @@
-"""ProxyPool 启动入口"""
+"""NetHub 启动入口"""
import uvicorn
diff --git a/app/models.py b/app/models.py
index df1fb45..cf020a2 100644
--- a/app/models.py
+++ b/app/models.py
@@ -1,11 +1,11 @@
-"""ProxyPool 数据模型"""
+"""NetHub 数据模型"""
from dataclasses import dataclass
@dataclass
class ProxyInfo:
- """从订阅解析出的代理信息"""
+ """从订阅解析出的节点信息"""
protocol: str # vmess / vless / trojan / ss / hysteria2
name: str # 节点名称
@@ -33,7 +33,7 @@ class ProxyDBRecord:
@property
def status(self) -> str:
- """代理状态：available / unavailable / unchecked"""
+ """节点状态：available / unavailable / unchecked"""
if self.latency_ms < 0:
return "unchecked"
if self.fail_count > 0:
@@ -52,6 +52,22 @@ class SubscriptionRecord:
max_concurrent: int # 异步最大并发数
enabled: bool # 是否启用
created_at: str # ISO8601
- empty_days: int # 连续代理数为0的天数
- total_count: int # 最新一次拉取的代理地址总数
+ empty_days: int # 连续节点数为0的天数
+ total_count: int # 最新一次拉取的节点地址总数
+ fetch_status: str # 拉取状态: idle / updating / success / failed
+
+
+@dataclass
+class InstanceSourceRecord:
+ """服务实例源记录 - 从已连接的转发服务实例获取可用节点配置"""
+ id: int
+ base_url: str # 服务实例地址
+ username: str # 登录用户名
+ password: str # 登录密码
+ crontab: str # crontab 格式的定时表达式
+ latency_threshold: float # 延迟阈值(ms)
+ max_concurrent: int # 异步最大并发数
+ enabled: bool # 是否启用
+ created_at: str # ISO8601
+ total_count: int # 最新一次获取的节点总数
fetch_status: str # 拉取状态: idle / updating / success / failed
diff --git a/app/parser.py b/app/parser.py
index cac96c1..28224f2 100644
--- a/app/parser.py
+++ b/app/parser.py
@@ -1,11 +1,12 @@
from __future__ import annotations
-"""订阅解析模块 - 拉取和解析代理订阅链接
+"""订阅解析模块 - 拉取和解析节点订阅链接
核心解析逻辑移植自 Proxy_List/get_connected_proxies/get_connected_proxies.py
支持 vmess / vless / trojan / ss / hysteria2 五种协议
"""
import base64
+import difflib
import json
import logging
from urllib.parse import urlparse, unquote
@@ -211,3 +212,216 @@ def _parse_hysteria2(line: str) -> ProxyInfo | None:
)
except Exception:
return None
+
+
+# ---- 服务实例 API 获取 ----
+
+async def instance_login(base_url: str, username: str, password: str,
+ timeout: float = 10.0) -> tuple[aiohttp.ClientSession, dict]:
+ """登录服务实例，返回 (session, headers)"""
+ session = aiohttp.ClientSession()
+ headers = {"User-Agent": "Mozilla/5.0", "Content-Type": "application/json"}
+ try:
+ async with session.post(
+ f"{base_url}/api/login",
+ json={"username": username, "password": password},
+ headers=headers,
+ timeout=aiohttp.ClientTimeout(total=timeout),
+ ) as resp:
+ data = await resp.json()
+ if data.get("code") != "SUCCESS":
+ await session.close()
+ raise Exception(f"登录失败: {data}")
+ token = data["data"]["token"]
+ headers["Authorization"] = token
+ return session, headers
+ except Exception:
+ await session.close()
+ raise
+
+
+async def get_instance_connected_nodes(
+ session: aiohttp.ClientSession, headers: dict, base_url: str,
+ timeout: float = 10.0,
+) -> tuple[list[dict], list[dict]]:
+ """获取服务实例已连接节点和订阅列表
+
+ 返回 (connected_nodes, subscriptions)
+ connected_nodes: [{"id", "sub_index", "name", "address", "net", "outbound"}, ...]
+ subscriptions: 原始订阅列表
+ """
+ async with session.get(
+ f"{base_url}/api/touch", headers=headers,
+ timeout=aiohttp.ClientTimeout(total=timeout),
+ ) as resp:
+ data = await resp.json()
+
+ connected = data["data"]["touch"]["connectedServer"]
+ subscriptions = data["data"]["touch"]["subscriptions"]
+
+ connected_nodes = []
+ for conn in connected:
+ node_id = conn["id"]
+ sub_index = conn["sub"]
+ outbound = conn.get("outbound", "proxy")
+
+ if sub_index < 0:
+ logger.debug("跳过手动添加的服务器 id=%d（无订阅源）", node_id)
+ continue
+
+ if sub_index < len(subscriptions):
+ sub = subscriptions[sub_index]
+ for server in sub.get("servers", []):
+ if server["id"] == node_id:
+ connected_nodes.append({
+ "id": node_id,
+ "sub_index": sub_index,
+ "name": server.get("name"),
+ "address": server.get("address"),
+ "net": server.get("net"),
+ "outbound": outbound,
+ })
+ break
+ return connected_nodes, subscriptions
+
+
+def _normalize_string(s: str) -> str:
+ """标准化字符串用于模糊匹配"""
+ if not s:
+ return ""
+ return s.replace(" ", "").replace("-", "").replace("_", "").lower()
+
+
+def _strings_similar(s1: str, s2: str, threshold: float = 0.5) -> bool:
+ """判断两个字符串是否相似"""
+ if not s1 or not s2:
+ return False
+ s1_norm = _normalize_string(s1)
+ s2_norm = _normalize_string(s2)
+ if not s1_norm or not s2_norm:
+ return False
+ if s1_norm in s2_norm or s2_norm in s1_norm:
+ return True
+ ratio = difflib.SequenceMatcher(None, s1_norm, s2_norm).ratio()
+ return ratio >= threshold
+
+
+async def fetch_connected_proxies(
+ base_url: str, username: str, password: str,
+) -> tuple[list[ProxyInfo], list[str]]:
+ """从服务实例获取所有已连接节点的分享链接和订阅地址列表
+
+ 返回: (matched_proxies, subscription_urls)
+ - matched_proxies: 匹配成功的 ProxyInfo 列表
+ - subscription_urls: 服务实例中所有订阅源的地址 URL 列表
+
+ 流程：
+ 1. 登录服务实例
+ 2. 获取已连接节点列表和订阅列表
+ 3. 提取所有订阅源地址 URL
+ 4. 遍历每个订阅源，拉取并解析节点列表
+ 5. 将已连接节点与订阅中的节点进行匹配（名称+地址模糊匹配）
+ """
+ session, headers = await instance_login(base_url, username, password)
+ try:
+ connected_nodes, subscriptions = await get_instance_connected_nodes(
+ session, headers, base_url,
+ )
+ logger.info("服务实例 %s: 已连接 %d 个节点, 共 %d 个订阅源",
+ base_url, len(connected_nodes), len(subscriptions))
+
+ # 提取所有订阅源地址 URL
+ subscription_urls = [
+ sub.get("address", "")
+ for sub in subscriptions
+ if sub.get("address")
+ ]
+
+ matched_proxies: list[ProxyInfo] = []
+ matched_node_ids: set[int] = set()
+
+ for sub_index, sub in enumerate(subscriptions):
+ sub_address = sub.get("address")
+ if not sub_address:
+ continue
+
+ # 拉取订阅内容
+ try:
+ async with session.get(
+ sub_address, headers=headers,
+ timeout=aiohttp.ClientTimeout(total=15.0),
+ ) as sub_resp:
+ content = await sub_resp.text()
+ share_links = parse_subscription(content)
+ logger.info("订阅源 %d: 解析到 %d 个节点", sub_index, len(share_links))
+ except Exception as e:
+ logger.warning("订阅源 %d 加载失败: %s", sub_index, e)
+ continue
+
+ # 该订阅源下的已连接节点
+ sub_connected = [
+ n for n in connected_nodes
+ if n["sub_index"] == sub_index and n["id"] not in matched_node_ids
+ ]
+
+ # 第一次匹配：名称+地址模糊匹配
+ unmatched = []
+ for conn_node in sub_connected:
+ if conn_node["id"] in matched_node_ids:
+ continue
+ best_match = None
+ best_quality = 0
+ for link_info in share_links:
+ quality = 0
+ name_sim = _strings_similar(conn_node["name"], link_info.name)
+ addr_sim = _strings_similar(conn_node["address"], link_info.address)
+ if name_sim and addr_sim:
+ quality = 3
+ elif name_sim:
+ quality = 2
+ elif addr_sim:
+ quality = 1
+ if quality > best_quality:
+ best_quality = quality
+ best_match = link_info
+
+ if best_match and best_quality >= 1:
+ matched_node_ids.add(conn_node["id"])
+ matched_proxies.append(best_match)
+ else:
+ unmatched.append(conn_node)
+
+ # 第二次匹配：地址精确匹配 或 地址模糊+端口匹配
+ for conn_node in unmatched:
+ if conn_node["id"] in matched_node_ids:
+ continue
+ best_match = None
+ best_quality = 0
+ for link_info in share_links:
+ quality = 0
+ addr_exact = (conn_node["address"] and link_info.address
+ and conn_node["address"] == link_info.address)
+ addr_fuzzy = _strings_similar(conn_node["address"], link_info.address, 0.7)
+ port_match = (conn_node.get("net") and link_info.port
+ and str(conn_node["net"]) == str(link_info.port))
+ if addr_exact:
+ quality = 4
+ elif addr_fuzzy and port_match:
+ quality = 3
+ elif addr_fuzzy:
+ quality = 2
+ elif port_match and _strings_similar(conn_node["name"], link_info.name, 0.3):
+ quality = 1
+ if quality > best_quality:
+ best_quality = quality
+ best_match = link_info
+
+ if best_match and best_quality >= 1:
+ matched_node_ids.add(conn_node["id"])
+ matched_proxies.append(best_match)
+
+ logger.info("服务实例 %s: 共匹配到 %d 个已连接节点配置, 发现 %d 个订阅源",
+ base_url, len(matched_proxies), len(subscription_urls))
+ return matched_proxies, subscription_urls
+ finally:
+ await session.close()
diff --git a/app/routers/api.py b/app/routers/api.py
index ebc9f49..7f02f15 100644
--- a/app/routers/api.py
+++ b/app/routers/api.py
@@ -1,4 +1,4 @@
-"""REST API 路由 - JSON 格式的代理数据接口"""
+"""REST API 路由 - JSON 格式的节点数据接口"""
import asyncio
import logging
@@ -8,7 +8,7 @@
from pydantic import BaseModel
from app import get_checker, get_config, get_db, get_scheduler
-from app.generator import generate_clash_subscription, generate_v2ray_subscription
+from app.generator import generate_clash_subscription, generate_plain_subscription, generate_v2ray_subscription
logger = logging.getLogger(__name__)
@@ -17,7 +17,7 @@
@router.get("/proxies")
async def get_available_proxies():
- """获取所有可用代理列表（latency_ms > 0 且 fail_count=0）"""
+ """获取所有可用节点列表（latency_ms > 0 且 fail_count=0）"""
db = get_db()
config = get_config()
proxies = await db.get_available_proxies(config.check.latency_threshold)
@@ -29,7 +29,7 @@ async def get_available_proxies():
@router.get("/proxies/all")
async def get_all_proxies():
- """获取所有代理（含不可用）"""
+ """获取所有节点（含不可用）"""
db = get_db()
proxies = await db.get_all_proxies()
return {
@@ -40,7 +40,7 @@ async def get_all_proxies():
@router.delete("/proxies/{proxy_id}")
async def delete_proxy(proxy_id: int):
- """删除指定代理"""
+ """删除指定节点"""
db = get_db()
await db.delete_proxy(proxy_id)
return {"message": "deleted"}
@@ -48,20 +48,39 @@ async def delete_proxy(proxy_id: int):
@router.get("/subscription/v2ray")
async def v2ray_subscription():
- """获取 v2ray 格式订阅（base64 编码）"""
+ """获取纯文本格式订阅（每行一条原始代理 URI）
+
+ 输出所有延迟达标且未失败的节点（含检测通过的实例源节点）
+ """
db = get_db()
config = get_config()
- proxies = await db.get_available_proxies(config.check.latency_threshold)
+ proxies = await db.get_subscription_output_proxies(config.check.latency_threshold)
content = generate_v2ray_subscription(proxies)
return _subscription_response(content, "text/plain")
+@router.get("/subscription/plain")
+async def plain_subscription():
+ """获取纯文本格式订阅（每行一条原始代理 URI，参照 subdom.txt 格式）
+
+ 输出所有延迟达标且未失败的节点（含检测通过的实例源节点）
+ """
+ db = get_db()
+ config = get_config()
+ proxies = await db.get_subscription_output_proxies(config.check.latency_threshold)
+ content = generate_plain_subscription(proxies)
+ return _subscription_response(content, "text/plain")
+
+
@router.get("/subscription/clash")
async def clash_subscription():
- """获取 Clash 格式订阅（YAML）"""
+ """获取 Clash 格式订阅（YAML）
+
+ 输出所有延迟达标且未失败的节点（含检测通过的实例源节点）
+ """
db = get_db()
config = get_config()
- proxies = await db.get_available_proxies(config.check.latency_threshold)
+ proxies = await db.get_subscription_output_proxies(config.check.latency_threshold)
content = generate_clash_subscription(proxies)
return _subscription_response(content, "text/yaml")
@@ -91,7 +110,7 @@ async def manual_fetch_subscription(sub_id: int):
@router.post("/verify")
async def manual_verify():
- """手动触发验证代理"""
+ """手动触发验证节点"""
scheduler = get_scheduler()
if scheduler._verifying:
return {"message": "验证任务正在进行中"}
@@ -101,7 +120,7 @@ async def manual_verify():
@router.post("/verify/{sub_id}")
async def manual_verify_subscription(sub_id: int):
- """手动触发验证指定订阅源的代理"""
+ """手动触发验证指定订阅源的节点"""
db = get_db()
sub = await db.get_subscription_by_id(sub_id)
if not sub:
@@ -109,7 +128,7 @@ async def manual_verify_subscription(sub_id: int):
raise HTTPException(status_code=404, detail="订阅不存在")
scheduler = get_scheduler()
asyncio.create_task(scheduler.verify_subscription_proxies(sub_id))
- return {"message": f"已触发订阅 #{sub_id} 的代理验证"}
+ return {"message": f"已触发订阅 #{sub_id} 的节点验证"}
@router.get("/stats")
@@ -133,9 +152,10 @@ async def health_check():
@router.get("/subscriptions")
async def get_subscriptions():
- """获取所有订阅源"""
+ """获取所有订阅源（过滤 nethub 内部地址）"""
db = get_db()
subs = await db.get_all_subscriptions()
+ subs = [s for s in subs if "nethub" not in s.url.lower()]
return {
"total": len(subs),
"subscriptions": [_subscription_to_dict(s) for s in subs],
@@ -208,7 +228,7 @@ async def fetch_and_verify():
asyncio.create_task(fetch_and_verify())
return {
"status": "added",
- "message": "订阅已添加，正在拉取并验证代理",
+ "message": "订阅已添加，正在拉取并验证节点",
"subscription": _subscription_to_dict(sub),
}
@@ -270,7 +290,7 @@ async def delete_subscription(sub_id: int):
@router.get("/proxies/grouped")
async def get_proxies_grouped():
- """获取按订阅来源分组的可用代理"""
+ """获取按订阅来源分组的可用节点"""
db = get_db()
config = get_config()
grouped = await db.get_proxies_grouped_by_source(config.check.latency_threshold)
@@ -307,7 +327,7 @@ def _subscription_response(content: str, content_type: str):
headers={
"Content-Disposition": 'attachment; filename="subscription"',
"Profile-Update-Interval": "24",
- "Profile-Title": "ProxyPool",
+ "Profile-Title": "NetHub",
},
)
@@ -369,3 +389,149 @@ async def delete_check_url(url_id: int):
if checker:
checker.check_urls = [u["url"] for u in await db.get_check_urls()]
return {"message": "deleted"}
+
+
+# ---- 服务实例源管理 ----
+
+@router.get("/instance-sources")
+async def get_instance_sources():
+ """获取所有服务实例源"""
+ db = get_db()
+ sources = await db.get_all_instance_sources()
+ return {
+ "total": len(sources),
+ "sources": [_instance_source_to_dict(s) for s in sources],
+ }
+
+
+class InstanceSourceCreate(BaseModel):
+ """添加服务实例源请求体"""
+ base_url: str
+ username: str
+ password: str
+ crontab: Optional[str] = "*/10 * * * *"
+ latency_threshold: Optional[float] = 1500.0
+ max_concurrent: Optional[int] = 50
+
+
+@router.post("/instance-sources")
+async def add_instance_source(req: InstanceSourceCreate):
+ """添加服务实例源"""
+ from fastapi import HTTPException
+ if not req.base_url:
+ raise HTTPException(status_code=400, detail="服务实例地址不能为空")
+ db = get_db()
+ source = await db.add_instance_source(
+ base_url=req.base_url,
+ username=req.username,
+ password=req.password,
+ crontab=req.crontab or "*/10 * * * *",
+ latency_threshold=req.latency_threshold or 1500.0,
+ max_concurrent=req.max_concurrent or 50,
+ enabled=True,
+ )
+ if not source:
+ raise HTTPException(status_code=409, detail="服务实例地址已存在")
+ # 注册定时任务
+ scheduler = get_scheduler()
+ scheduler._add_instance_source_job(source)
+ # 自动触发首次获取
+ asyncio.create_task(scheduler._fetch_single_instance_source(source.id))
+ return _instance_source_to_dict(source)
+
+
+@router.put("/instance-sources/{source_id}")
+async def update_instance_source(source_id: int, base_url: str = None, username: str = None,
+ password: str = None, crontab: str = None,
+ latency_threshold: float = None, max_concurrent: int = None,
+ enabled: bool = None):
+ """更新服务实例源"""
+ from fastapi import HTTPException
+ db = get_db()
+ kwargs = {}
+ if base_url is not None:
+ kwargs["base_url"] = base_url
+ if username is not None:
+ kwargs["username"] = username
+ if password is not None:
+ kwargs["password"] = password
+ if crontab is not None:
+ kwargs["crontab"] = crontab
+ if latency_threshold is not None:
+ kwargs["latency_threshold"] = latency_threshold
+ if max_concurrent is not None:
+ kwargs["max_concurrent"] = max_concurrent
+ if enabled is not None:
+ kwargs["enabled"] = enabled
+
+ success = await db.update_instance_source(source_id, **kwargs)
+ if not success:
+ raise HTTPException(status_code=404, detail="服务实例源不存在或无更新")
+ # 刷新定时任务
+ source = await db.get_instance_source_by_id(source_id)
+ scheduler = get_scheduler()
+ scheduler.refresh_instance_source_job(source)
+
+ if source.enabled:
+ # 启用时自动获取
+ asyncio.create_task(scheduler._fetch_single_instance_source(source.id))
+ else:
+ await db.update_instance_fetch_status(source_id, "idle")
+ return _instance_source_to_dict(source)
+
+
+@router.delete("/instance-sources/{source_id}")
+async def delete_instance_source(source_id: int):
+ """删除服务实例源"""
+ from fastapi import HTTPException
+ db = get_db()
+ success = await db.delete_instance_source(source_id)
+ if not success:
+ raise HTTPException(status_code=404, detail="服务实例源不存在")
+ # 移除定时任务
+ scheduler = get_scheduler()
+ scheduler.remove_instance_source_job(source_id)
+ return {"message": "deleted"}
+
+
+@router.post("/instance-sources/{source_id}/fetch")
+async def manual_fetch_instance_source(source_id: int):
+ """手动触发获取指定服务实例源的已连接节点"""
+ db = get_db()
+ source = await db.get_instance_source_by_id(source_id)
+ if not source:
+ from fastapi import HTTPException
+ raise HTTPException(status_code=404, detail="服务实例源不存在")
+ scheduler = get_scheduler()
+ asyncio.create_task(scheduler._fetch_single_instance_source(source_id))
+ return {"message": f"已触发实例源 #{source_id} 的获取任务"}
+
+
+@router.post("/instance-sources/{source_id}/import-subs")
+async def import_instance_subscriptions(source_id: int):
+ """手工导入服务实例中的订阅源到本地订阅源表"""
+ db = get_db()
+ source = await db.get_instance_source_by_id(source_id)
+ if not source:
+ from fastapi import HTTPException
+ raise HTTPException(status_code=404, detail="服务实例源不存在")
+ scheduler = get_scheduler()
+ count = await scheduler.import_instance_subscriptions(source_id)
+ return {"message": f"已导入 {count} 个订阅源", "imported": count}
+
+
+def _instance_source_to_dict(source) -> dict:
+ """将 InstanceSourceRecord 转为 API 响应字典"""
+ return {
+ "id": source.id,
+ "base_url": source.base_url,
+ "username": source.username,
+ "password": source.password,
+ "crontab": source.crontab,
+ "latency_threshold": source.latency_threshold,
+ "max_concurrent": source.max_concurrent,
+ "enabled": source.enabled,
+ "created_at": source.created_at,
+ "total_count": source.total_count,
+ "fetch_status": source.fetch_status,
+ }
diff --git a/app/routers/web.py b/app/routers/web.py
index e1e6f9f..1695119 100644
--- a/app/routers/web.py
+++ b/app/routers/web.py
@@ -1,6 +1,7 @@
from __future__ import annotations
"""Web UI 页面路由 - Jinja2 模板渲染"""
+import asyncio
import json
import logging
from dataclasses import asdict
@@ -16,23 +17,40 @@
@router.get("/")
async def index(request: Request):
- """代理列表主页"""
+ """节点列表主页"""
db = get_db()
config = get_config()
templates = get_templates()
scheduler = get_scheduler()
- proxies = await db.get_all_proxies()
- stats = await db.get_stats()
- subscriptions = await db.get_all_subscriptions()
- # 按来源分组的可用代理
- grouped = await db.get_proxies_grouped_by_source(config.check.latency_threshold)
- # 检测目标 URL
- check_urls = await db.get_check_urls()
- # 计算每个订阅源的可用代理数量
+ # 并行查询所有数据
+ results = await asyncio.gather(
+ db.get_all_proxies(),
+ db.get_stats(),
+ db.get_all_subscriptions(),
+ db.get_proxies_grouped_by_source(config.check.latency_threshold),
+ db.get_check_urls(),
+ db.get_all_instance_sources(),
+ )
+ proxies, stats, subscriptions, grouped, check_urls, instance_sources = results
+
+ # 过滤包含 nethub 的订阅源（内部地址不在管理界面展示）
+ subscriptions = [s for s in subscriptions if "nethub" not in s.url.lower()]
+
+ # 序列化
+ instance_sources_json = json.dumps(
+ [dict(asdict(s)) for s in instance_sources],
+ ensure_ascii=False,
+ )
+ # 计算每个订阅源的可用节点数量
sub_available_counts = {}
for sub in subscriptions:
sub_available_counts[sub.url] = len(grouped.get(sub.url, []))
+ # 计算每个服务实例源的可用节点数量
+ inst_available_counts = {}
+ for inst in instance_sources:
+ source_tag = f"instance:{inst.base_url}"
+ inst_available_counts[inst.id] = len(grouped.get(source_tag, []))
subscriptions_json = json.dumps(
[dict(asdict(s), available_count=sub_available_counts.get(s.url, 0)) for s in subscriptions],
ensure_ascii=False,
@@ -48,6 +66,9 @@ async def index(request: Request):
"subscriptions_json": subscriptions_json,
"sub_available_counts": sub_available_counts,
"check_urls": check_urls,
+ "instance_sources": instance_sources,
+ "instance_sources_json": instance_sources_json,
+ "inst_available_counts": inst_available_counts,
"grouped": grouped,
"latency_threshold": config.check.latency_threshold,
"last_fetch_time": scheduler.last_fetch_time,
@@ -65,6 +86,7 @@ async def subscription_page(request: Request):
host = request.headers.get("host", f"localhost:{get_config().server.port}")
scheme = request.url.scheme
+ plain_url = f"{scheme}://{host}/api/subscription/plain"
v2ray_url = f"{scheme}://{host}/api/subscription/v2ray"
clash_url = f"{scheme}://{host}/api/subscription/clash"
@@ -72,6 +94,7 @@ async def subscription_page(request: Request):
request,
"subscription.html",
{
+ "plain_url": plain_url,
"v2ray_url": v2ray_url,
"clash_url": clash_url,
},
diff --git a/app/scheduler.py b/app/scheduler.py
index 7ea61b4..40888d0 100644
--- a/app/scheduler.py
+++ b/app/scheduler.py
@@ -11,13 +11,13 @@
from app.checker import ProxyChecker
from app.config import AppConfig
from app.database import ProxyDatabase
-from app.parser import fetch_subscription, parse_subscription
+from app.parser import fetch_subscription, parse_subscription, fetch_connected_proxies
logger = logging.getLogger(__name__)
class TaskScheduler:
- """代理池定时任务调度器"""
+ """节点池定时任务调度器"""
def __init__(self, config: AppConfig, db: ProxyDatabase, checker: ProxyChecker):
self.config = config
@@ -28,6 +28,7 @@ def __init__(self, config: AppConfig, db: ProxyDatabase, checker: ProxyChecker):
self._last_verify_time = ""
self._fetching = False
self._verifying = False
+ self._last_instance_sub_urls: list[str] = [] # 最近一次实例源获取的订阅地址缓存
def start(self) -> None:
"""注册默认定时任务并启动调度器"""
@@ -39,14 +40,14 @@ def start(self) -> None:
"interval",
seconds=cfg.verify_interval,
id="verify_proxies",
- name="验证已存代理",
+ name="验证已存节点",
)
self.scheduler.add_job(
self.cleanup_proxies,
"interval",
seconds=cfg.cleanup_interval,
id="cleanup_proxies",
- name="清理不合格代理",
+ name="清理不合格节点",
)
self.scheduler.start()
logger.info("调度器已启动: verify=%ds, cleanup=%ds",
@@ -54,6 +55,7 @@ def start(self) -> None:
# 启动后注册订阅任务
asyncio.create_task(self._register_subscription_jobs())
+ asyncio.create_task(self._register_instance_source_jobs())
async def _register_subscription_jobs(self) -> None:
"""从数据库加载订阅源并注册 crontab 定时任务"""
@@ -62,6 +64,13 @@ async def _register_subscription_jobs(self) -> None:
self._add_subscription_job(sub)
logger.info("已注册 %d 个订阅拉取任务", len(subs))
+ async def _register_instance_source_jobs(self) -> None:
+ """从数据库加载服务实例源并注册 crontab 定时任务"""
+ sources = await self.db.get_enabled_instance_sources()
+ for source in sources:
+ self._add_instance_source_job(source)
+ logger.info("已注册 %d 个实例源获取任务", len(sources))
+
def _add_subscription_job(self, sub) -> None:
"""为单个订阅注册 crontab 定时任务"""
job_id = f"fetch_sub_{sub.id}"
@@ -108,7 +117,7 @@ def shutdown(self) -> None:
logger.info("调度器已关闭")
async def _fetch_single_subscription(self, sub_id: int) -> None:
- """拉取单个订阅并检测入库"""
+ """拉取单个订阅并检测入库（并发检测 + 批量数据库写入）"""
sub = await self.db.get_subscription_by_id(sub_id)
if not sub or not sub.enabled:
return
@@ -125,42 +134,51 @@ async def _fetch_single_subscription(self, sub_id: int) -> None:
await self.db.update_fetch_status(sub_id, "success")
return
- logger.info("订阅 #%d: 解析到 %d 个节点，开始逐个检测...", sub.id, len(proxies))
+ logger.info("订阅 #%d: 解析到 %d 个节点，并发检测中...", sub.id, len(proxies))
- # 创建该订阅专用的 checker（使用订阅自己的并发数和超时）
+ # 并发检测所有节点延迟
sub_checker = ProxyChecker(
check_urls=self.checker.check_urls,
timeout=self.config.check.timeout,
max_concurrent=sub.max_concurrent,
+ socks_port=self.checker.socks_port,
+ http_port=self.checker.http_port,
+ check_mode=self.checker.check_mode,
+ kernel_path=self.checker.kernel_path,
)
+ links = [p.link for p in proxies]
+ results = await sub_checker.check_batch(links)
+ # 批量处理检测结果
threshold = sub.latency_threshold
+ latency_updates = [] # [(proxy_id, latency), ...]
+ fail_ids = [] # [proxy_id, ...]
added = 0
updated = 0
skipped = 0
for proxy in proxies:
- # 每个代理独立检测延迟，报错自动跳过
- try:
- latency = await sub_checker.check_proxy(proxy.link)
- except Exception as e:
- logger.debug("检测代理 %s 异常，跳过: %s", proxy.name[:30], e)
- skipped += 1
- continue
+ latency = results.get(proxy.link)
if latency is None:
- # 检测失败，跳过
+ # 检测失败
skipped += 1
existing = await self.db.get_proxy_by_link(proxy.link)
if existing:
- await self.db.increment_fail(existing.id)
+ # 去重：若该节点原属于实例源，将 source 转移到订阅源
+ if existing.source.startswith("instance:"):
+ await self.db.update_proxy_source(existing.id, sub.url)
+ fail_ids.append(existing.id)
continue
if latency <= threshold:
# 延迟达标
existing = await self.db.get_proxy_by_link(proxy.link)
if existing:
- await self.db.update_latency(existing.id, latency)
+ # 去重：若该节点原属于实例源，将 source 转移到订阅源
+ if existing.source.startswith("instance:"):
+ await self.db.update_proxy_source(existing.id, sub.url)
+ latency_updates.append((existing.id, latency))
updated += 1
else:
success = await self.db.insert_proxy(proxy, latency, sub.url)
@@ -168,16 +186,25 @@ async def _fetch_single_subscription(self, sub_id: int) -> None:
added += 1
else:
# 延迟超标
+ skipped += 1
existing = await self.db.get_proxy_by_link(proxy.link)
if existing:
- await self.db.increment_fail(existing.id)
- skipped += 1
-
- # 有代理入库则重置空天数
+ # 去重：若该节点原属于实例源，将 source 转移到订阅源
+ if existing.source.startswith("instance:"):
+ await self.db.update_proxy_source(existing.id, sub.url)
+ fail_ids.append(existing.id)
+
+ # 批量写入数据库
+ if latency_updates:
+ await self.db.batch_update_latency(latency_updates)
+ if fail_ids:
+ await self.db.batch_increment_fail(fail_ids)
+
+ # 有节点入库则重置空天数
if added > 0:
await self.db.reset_empty_days(sub_id)
- # 更新订阅源的代理总数
+ # 更新订阅源的节点总数
await self.db.update_total_count(sub_id, len(proxies))
await self.db.update_fetch_status(sub_id, "success")
@@ -189,7 +216,7 @@ async def _fetch_single_subscription(self, sub_id: int) -> None:
logger.error("拉取订阅 #%d 异常: %s", sub.id, e, exc_info=True)
async def fetch_and_check(self) -> None:
- """手动触发：拉取所有启用的订阅并检测入库"""
+ """手动触发：并行拉取所有启用的订阅和实例源并检测入库"""
if self._fetching:
logger.info("上一次拉取任务尚未完成，跳过本次")
return
@@ -197,19 +224,30 @@ async def fetch_and_check(self) -> None:
self._fetching = True
try:
subs = await self.db.get_enabled_subscriptions()
- if not subs:
- logger.warning("没有启用的订阅源")
+ sources = await self.db.get_enabled_instance_sources()
+
+ if not subs and not sources:
+ logger.warning("没有启用的订阅源或实例源")
return
+ # 并行拉取所有订阅和实例源
+ tasks = []
for sub in subs:
- await self._fetch_single_subscription(sub.id)
+ tasks.append(self._fetch_single_subscription(sub.id))
+ for source in sources:
+ tasks.append(self._fetch_single_instance_source(source.id))
+
+ results = await asyncio.gather(*tasks, return_exceptions=True)
+ for i, r in enumerate(results):
+ if isinstance(r, Exception):
+ logger.error("并行拉取任务 %d 异常: %s", i, r)
except Exception as e:
logger.error("拉取任务异常: %s", e, exc_info=True)
finally:
self._fetching = False
async def verify_stored_proxies(self) -> None:
- """验证已存代理可用性 - 失败则累加 fail_count，连续3次失败由清理任务移除"""
+ """验证已存节点可用性（并发检测 + 批量数据库写入）"""
if self._verifying:
logger.info("上一次验证任务尚未完成，跳过本次")
return
@@ -218,34 +256,38 @@ async def verify_stored_proxies(self) -> None:
try:
proxies = await self.db.get_all_proxies()
if not proxies:
- logger.info("数据库中没有代理，跳过验证")
+ logger.info("数据库中没有节点，跳过验证")
return
- logger.info("开始验证 %d 个代理...", len(proxies))
+ logger.info("开始验证 %d 个节点...", len(proxies))
links = [p.link for p in proxies]
results = await self.checker.check_batch(links)
- success_count = 0
- fail_count = 0
+ latency_updates = []
+ fail_ids = []
for proxy in proxies:
latency = results.get(proxy.link)
if latency is not None:
- await self.db.update_latency(proxy.id, latency)
- success_count += 1
+ latency_updates.append((proxy.id, latency))
else:
- await self.db.increment_fail(proxy.id)
- fail_count += 1
+ fail_ids.append(proxy.id)
+
+ # 批量写入
+ if latency_updates:
+ await self.db.batch_update_latency(latency_updates)
+ if fail_ids:
+ await self.db.batch_increment_fail(fail_ids)
self._last_verify_time = datetime.now(timezone.utc).isoformat()
- logger.info("验证完成: 成功 %d, 失败 %d", success_count, fail_count)
+ logger.info("验证完成: 成功 %d, 失败 %d", len(latency_updates), len(fail_ids))
except Exception as e:
logger.error("验证任务异常: %s", e, exc_info=True)
finally:
self._verifying = False
async def verify_subscription_proxies(self, sub_id: int) -> None:
- """验证指定订阅源的代理可用性 - 失败则累加 fail_count，连续3次失败由清理任务移除"""
+ """验证指定订阅源的节点可用性（并发检测 + 批量数据库写入）"""
sub = await self.db.get_subscription_by_id(sub_id)
if not sub:
logger.warning("订阅 #%d 不存在", sub_id)
@@ -253,50 +295,250 @@ async def verify_subscription_proxies(self, sub_id: int) -> None:
proxies = await self.db.get_proxies_by_source(sub.url)
if not proxies:
- logger.info("订阅 #%d 没有代理，跳过验证", sub_id)
+ logger.info("订阅 #%d 没有节点，跳过验证", sub_id)
return
- logger.info("开始验证订阅 #%d 的 %d 个代理...", sub_id, len(proxies))
+ logger.info("开始验证订阅 #%d 的 %d 个节点...", sub_id, len(proxies))
sub_checker = ProxyChecker(
check_urls=self.checker.check_urls,
timeout=self.config.check.timeout,
max_concurrent=sub.max_concurrent,
+ socks_port=self.checker.socks_port,
+ http_port=self.checker.http_port,
+ check_mode=self.checker.check_mode,
+ kernel_path=self.checker.kernel_path,
)
links = [p.link for p in proxies]
results = await sub_checker.check_batch(links)
- success_count = 0
- fail_count = 0
+ latency_updates = []
+ fail_ids = []
for proxy in proxies:
latency = results.get(proxy.link)
if latency is not None:
- await self.db.update_latency(proxy.id, latency)
- success_count += 1
+ latency_updates.append((proxy.id, latency))
else:
- await self.db.increment_fail(proxy.id)
- fail_count += 1
+ fail_ids.append(proxy.id)
+
+ if latency_updates:
+ await self.db.batch_update_latency(latency_updates)
+ if fail_ids:
+ await self.db.batch_increment_fail(fail_ids)
self._last_verify_time = datetime.now(timezone.utc).isoformat()
- logger.info("订阅 #%d 验证完成: 成功 %d, 失败 %d", sub_id, success_count, fail_count)
+ logger.info("订阅 #%d 验证完成: 成功 %d, 失败 %d", sub_id, len(latency_updates), len(fail_ids))
+
+ # ---- 服务实例源管理 ----
+
+ def _add_instance_source_job(self, source) -> None:
+ """为单个服务实例源注册 crontab 定时任务"""
+ job_id = f"fetch_instance_{source.id}"
+ try:
+ if self.scheduler.get_job(job_id):
+ self.scheduler.remove_job(job_id)
+
+ parts = source.crontab.strip().split()
+ if len(parts) != 5:
+ logger.error("服务实例源 %d crontab 格式错误: %s", source.id, source.crontab)
+ return
+
+ trigger = CronTrigger(
+ minute=parts[0], hour=parts[1], day=parts[2],
+ month=parts[3], day_of_week=parts[4],
+ )
+ self.scheduler.add_job(
+ self._fetch_single_instance_source,
+ trigger=trigger,
+ id=job_id,
+ name=f"获取实例源 #{source.id}",
+ args=[source.id],
+ )
+ logger.info("注册实例源任务 #%d: crontab=%s", source.id, source.crontab)
+ except Exception as e:
+ logger.error("注册实例源任务 #%d 失败: %s", source.id, e)
+
+ def remove_instance_source_job(self, source_id: int) -> None:
+ """移除服务实例源定时任务"""
+ job_id = f"fetch_instance_{source_id}"
+ if self.scheduler.get_job(job_id):
+ self.scheduler.remove_job(job_id)
+
+ def refresh_instance_source_job(self, source) -> None:
+ """刷新服务实例源定时任务（更新 crontab 后调用）"""
+ self.remove_instance_source_job(source.id)
+ if source.enabled:
+ self._add_instance_source_job(source)
+
+ async def _fetch_single_instance_source(self, source_id: int) -> None:
+ """从服务实例获取已连接节点配置，先全部入库再并发检测延迟
+
+ 实例源的节点：先入库（延迟=-1），再并发检测更新延迟。
+ 检测不通过也保留在库中，仅累加 fail_count，由定时清理任务
+ 在连续 7 天无成功后才移除。
+ """
+ source = await self.db.get_instance_source_by_id(source_id)
+ if not source or not source.enabled:
+ return
+
+ # 实例源的 source 标识
+ source_tag = f"instance:{source.base_url}"
+
+ try:
+ logger.info("开始获取实例源 #%d: %s", source.id, source.base_url)
+ await self.db.update_instance_fetch_status(source_id, "updating")
+
+ proxies, subscription_urls = await fetch_connected_proxies(
+ source.base_url, source.username, source.password,
+ )
+
+ # 缓存订阅地址列表，供手工导入时使用
+ self._last_instance_sub_urls = subscription_urls
+
+ if not proxies:
+ logger.warning("实例源 #%d 未获取到任何已连接节点", source.id)
+ await self.db.update_instance_total_count(source_id, 0)
+ await self.db.update_instance_fetch_status(source_id, "success")
+ return
+
+ logger.info("实例源 #%d: 获取到 %d 个节点，先入库再并发检测...",
+ source.id, len(proxies))
+
+ # ---- 第一步：全部入库（延迟=-1 表示未检测） ----
+ added = 0
+ existed = 0
+ for proxy in proxies:
+ existing = await self.db.get_proxy_by_link(proxy.link)
+ if existing:
+ existed += 1
+ else:
+ success = await self.db.insert_proxy(proxy, -1, source_tag)
+ if success:
+ added += 1
+
+ logger.info("实例源 #%d: 入库完成, 新增 %d, 已存在 %d, 并发检测中...",
+ source.id, added, existed)
+
+ # ---- 第二步：并发检测所有节点延迟 ----
+ sub_checker = ProxyChecker(
+ check_urls=self.checker.check_urls,
+ timeout=self.config.check.timeout,
+ max_concurrent=source.max_concurrent,
+ socks_port=self.checker.socks_port,
+ http_port=self.checker.http_port,
+ check_mode=self.checker.check_mode,
+ kernel_path=self.checker.kernel_path,
+ )
+ links = [p.link for p in proxies]
+ results = await sub_checker.check_batch(links)
+
+ # 批量处理检测结果
+ latency_updates = [] # [(proxy_id, latency), ...]
+ fail_ids = [] # [proxy_id, ...]
+
+ for proxy in proxies:
+ latency = results.get(proxy.link)
+ db_record = await self.db.get_proxy_by_link(proxy.link)
+ if not db_record:
+ continue
+
+ if latency is not None and latency > 0:
+ latency_updates.append((db_record.id, latency))
+ else:
+ fail_ids.append(db_record.id)
+
+ # 批量写入数据库
+ if latency_updates:
+ await self.db.batch_update_latency(latency_updates)
+ if fail_ids:
+ await self.db.batch_increment_fail(fail_ids)
+
+ await self.db.update_instance_total_count(source_id, len(proxies))
+ await self.db.update_instance_fetch_status(source_id, "success")
+
+ self._last_fetch_time = datetime.now(timezone.utc).isoformat()
+ logger.info("实例源 #%d 获取完成: 入库 %d(新增%d), 检测成功 %d, 失败 %d, 总 %d",
+ source.id, added + existed, added,
+ len(latency_updates), len(fail_ids), len(proxies))
+ except Exception as e:
+ await self.db.update_instance_fetch_status(source_id, "failed")
+ logger.error("获取实例源 #%d 异常: %s", source.id, e, exc_info=True)
+
+ async def fetch_all_instance_sources(self) -> None:
+ """手动触发：获取所有启用的服务实例源的已连接节点"""
+ sources = await self.db.get_enabled_instance_sources()
+ if not sources:
+ logger.warning("没有启用的服务实例源")
+ return
+ for source in sources:
+ await self._fetch_single_instance_source(source.id)
+
+ async def import_instance_subscriptions(self, source_id: int) -> int:
+ """手工导入服务实例中的订阅源到本地订阅源表
+
+ 先重新获取实例源以拿到最新订阅地址列表，然后逐个新增。
+ 返回新增数量。
+ """
+ source = await self.db.get_instance_source_by_id(source_id)
+ if not source or not source.enabled:
+ return 0
+
+ # 获取实例源的订阅地址列表
+ _, subscription_urls = await fetch_connected_proxies(
+ source.base_url, source.username, source.password,
+ )
+
+ if not subscription_urls:
+ logger.info("实例源 #%d 没有订阅源可导入", source_id)
+ return 0
+
+ new_sub_count = 0
+ for sub_url in subscription_urls:
+ existing = await self.db.get_subscription_by_url(sub_url)
+ if not existing:
+ sub_record = await self.db.add_subscription(
+ url=sub_url,
+ crontab="0 * * * *",
+ latency_threshold=source.latency_threshold,
+ max_retries=3,
+ max_concurrent=source.max_concurrent,
+ enabled=True,
+ )
+ if sub_record:
+ self._add_subscription_job(sub_record)
+ asyncio.create_task(self._fetch_single_subscription(sub_record.id))
+ new_sub_count += 1
+ logger.info("导入订阅源: %s (来自实例源 #%d)", sub_url[:50], source_id)
+
+ logger.info("实例源 #%d: 导入完成, 新增 %d 个订阅源", source_id, new_sub_count)
+ return new_sub_count
async def cleanup_proxies(self) -> None:
- """清理不合格代理和空订阅源"""
- # 清理连续3次验证失败的代理
- deleted = await self.db.delete_proxies_by_fail_count(3)
+ """清理不合格节点和空订阅源
+
+ - 订阅源节点：连续 3 次验证失败则移除（原逻辑不变）
+ - 实例源节点：last_success_time 为空或距今超过 7 天且 fail_count > 0 才移除
+ """
+ # 清理订阅源节点：连续3次验证失败
+ deleted = await self.db.delete_sub_proxies_by_fail_count(3)
if deleted > 0:
- logger.info("清理代理: 删除 %d 个连续3次不可用的代理", deleted)
+ logger.info("清理订阅源节点: 删除 %d 个连续3次不可用的节点", deleted)
+
+ # 清理实例源节点：连续7天无成功
+ deleted_inst = await self.db.delete_instance_proxies_stale(7)
+ if deleted_inst > 0:
+ logger.info("清理实例源节点: 删除 %d 个连续7天无成功的节点", deleted_inst)
- # 清理连续30天代理数为0的订阅源
- empty_subs = await self.db.get_subscriptions_with_empty_days(30)
+ # 清理连续7天节点数为0的订阅源
+ empty_subs = await self.db.get_subscriptions_with_empty_days(7)
for sub in empty_subs:
- # 再次确认该订阅源下确实没有代理
+ # 再次确认该订阅源下确实没有节点
count = await self.db.get_proxy_count_by_source(sub.url)
if count == 0:
await self.db.delete_subscription(sub.id)
self.remove_subscription_job(sub.id)
- logger.info("清理订阅: 删除 #%d 连续30天无代理 (%s)", sub.id, sub.url[:50])
+ logger.info("清理订阅: 删除 #%d 连续7天无节点 (%s)", sub.id, sub.url[:50])
@property
def last_fetch_time(self) -> str:
diff --git a/app/templates/base.html b/app/templates/base.html
index 3489b57..c224014 100644
--- a/app/templates/base.html
+++ b/app/templates/base.html
@@ -3,7 +3,7 @@
- {% block title %}ProxyPool{% endblock %}
+ {% block title %}NetHub{% endblock %}