diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d1dd02..250a2df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,99 @@ All notable changes to this project will be documented in this file. +## [2.0.0] - 2026-07-20 + +### 核心重构 + +- **节点-订阅关联重构** - 代理节点使用 `subscription_id`（整数）替代 `source`（字符串）绑定所属订阅源，已存在于其他订阅的节点不重复入库 +- **纯文本订阅格式** - 对外订阅输出改为纯文本格式（每行一条原始代理 URI），参照 `subdom.txt` 样式；移除 base64 编码输出 +- **内核转发检测** - 新增 Xray 内核转发检测能力，支持 HTTP/SOCKS 代理转发后检测连通性；TCP/TLS 直接检测作为回退 +- **检测失败直接删除** - 取消 `fail_count` 累积逻辑，检测不通过的节点直接从数据库删除，不再保留 +- **实例源节点不入库** - 服务实例获取的节点仅统计已连接数量，不再写入代理数据库 +- **socks4 协议移除** - 不再支持 socks4/socks4a 协议，拉取订阅时自动移除已有的 socks4 节点 + +### 多协议解析扩展 + +- **socks5/http 代理支持** - 新增 socks5:// 和 http(s):// 格式的解析、检测和 Clash 配置生成 +- **http/https 仅带 #fragment 时视为节点** - 避免 URL 误判为代理节点 +- **链接规范化** - socks5/http 代理对外输出确保格式为 `protocol://host:port#host-port`，保留认证信息 + +### 检测优化 + +- **多目标 URL 轮询** - 新增 5 个检测目标（Google 204、Gstatic 204、Cloudflare、Apple、华为连通性检测） +- **响应体验证** - 新增 `_validate_check_response` 排除劫持页面和空响应 +- **GET + 4KB body 读取** - 替代 HEAD 请求，提升服务器兼容性 +- **检测重试机制** - `check_retries` 参数（默认 2），单次检测失败后自动重试 +- **ConnectionRefusedError/ResetError** - 不再返回延迟值（视为不可用） +- **TLS 回退检测** - SSL 对象状态验证，`server_hostname` 使用原始域名 +- **华为连通性检测** - 新增 `connectivitycheck.platform.hicloud.com/generate_204` + +### 调度优化 + +- **CronTrigger 随机延迟** - 所有 crontab 任务加入 `jitter`（0~600 秒），避免多订阅源同时更新 +- **订阅验证防重入** - `_verifying_subs` 集合跟踪正在验证的订阅 ID，防止并发重复验证 +- **拉取后自动验证** - `_fetch_single_subscription` 完成后自动触发该订阅的已入库节点验证 + +### 时间与日志 + +- **UTC+8 统一** - 所有服务时间统一为东八区（UTC+8），包括数据库时间戳和日志时间 +- **单文件日志** - 日志写入 `logs/proxy_pool.log` 单文件，不归档、不保留历史日志 +- **内核日志合并** - 内核 stdout/stderr 合并写入 `logs/proxy-core.log` + +### Web 界面 + +- **统计面板重构** - "总节点数"→"总订阅条目数"（显示订阅源数量），"可用数"→"可用节点数"，移除"不可用"统计卡片 +- **分页显示** - 每个订阅源可用节点列表分页，每页 10 条 +- **操作按钮更新** - "验证所有订阅"→"验证所有节点"，新增"清除所有节点"、"导出配置"、"导入配置"按钮 +- **实例源表头** - "总数"→"已连接"，显示已连接节点数量而非数据库条目数 + +### 配置管理 + +- **一键导出/导入** - 导出订阅源和实例源配置为 JSON 文件（含时间戳），导入时自动去重 +- **导出文件名时间戳** - 格式 `nethub_config_YYYYMMDD_HHMMSS.json` + +### API 接口 + +| 方法 | 路径 | 说明 | +|------|------|------| +| GET | `/api/proxies` | 可用节点列表 | +| GET | `/api/proxies/all` | 所有节点 | +| GET | `/api/proxies/grouped` | 按 subscription_id 分组的可用节点 | +| DELETE | `/api/proxies/{id}` | 删除节点 | +| DELETE | `/api/proxies` | 一键清除所有节点 | +| GET | `/api/subscription/plain` | 纯文本格式订阅 | +| GET | `/api/subscription/v2ray` | 纯文本格式订阅（同 plain） | +| GET | `/api/subscription/clash` | Clash 格式订阅（YAML） | +| POST | `/api/fetch` | 拉取所有订阅 | +| POST | `/api/fetch/{sub_id}` | 拉取指定订阅 | +| POST | `/api/verify` | 验证所有节点 | +| POST | `/api/verify/{sub_id}` | 验证指定订阅节点 | +| GET | `/api/stats` | 统计信息 | +| GET | `/api/health` | 健康检查 | +| GET | `/api/subscriptions` | 订阅源列表 | +| POST | `/api/subscriptions` | 添加订阅源 | +| POST | `/api/subscriptions/auto` | 自动添加订阅源（仅 URL 必填） | +| PUT | `/api/subscriptions/{sub_id}` | 更新订阅源 | +| DELETE | `/api/subscriptions/{sub_id}` | 删除订阅源及其下所有节点 | +| GET | `/api/check-urls` | 检测目标 URL 列表 | +| POST | `/api/check-urls` | 添加检测目标 URL | +| DELETE | `/api/check-urls/{url_id}` | 删除检测目标 URL | +| GET | `/api/config/export` | 导出配置（JSON） | +| POST | `/api/config/import` | 导入配置（JSON） | +| GET | `/api/instance-sources` | 服务实例源列表 | +| POST | `/api/instance-sources` | 添加服务实例源 | +| PUT | `/api/instance-sources/{source_id}` | 更新服务实例源 | +| DELETE | `/api/instance-sources/{source_id}` | 删除服务实例源 | +| POST | `/api/instance-sources/{source_id}/fetch` | 获取实例源已连接节点数 | +| POST | `/api/instance-sources/{source_id}/import` | 导入实例源订阅 | + +### Docker 构建 + +- **固定版本下载** - 内核数据文件使用固定版本标签，替代 `latest` 避免超时 +- **curl 重试参数** - `--connect-timeout 30 --max-time 180 --retry 3 --retry-delay 5` + +--- + ## [1.0.0] - 2026-07-18 ### 核心功能 @@ -47,7 +140,7 @@ 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` | 核心订阅 | +| GET | `/api/subscription/v2ray` | 核心订阅（base64） | | GET | `/api/subscription/clash` | Clash 订阅 | | POST | `/api/fetch` | 拉取所有订阅 | | POST | `/api/fetch/{sub_id}` | 拉取指定订阅 | diff --git a/README.md b/README.md index b2e7173..dd29a49 100644 --- a/README.md +++ b/README.md @@ -1,19 +1,20 @@ -# NetHub v1.0.0 +# NetHub v2.0.0 自动获取、检测、维护节点池，提供 Web 管理界面和订阅链接输出。 ## 功能特性 -- **订阅源管理** - 数据库驱动的增删改查，每个订阅源独立配置 Crontab、延迟阈值、重试次数、并发数 -- **Crontab 定时拉取** - 每个订阅源支持 5 位 Crontab 表达式精准调度 -- **HTTP 延迟检测** - 模拟通过节点访问目标网站，测量完整请求延迟（DNS + TCP + TLS + HTTP），多目标取最大值 -- **检测目标动态配置** - 检测目标 URL 存入数据库，页面可增删，修改即时生效，无需外部文件 -- **多协议支持** - vmess / vless / trojan / ss / hysteria2 解析与 Clash 配置生成 -- **自动清理** - 连续 3 次验证失败的节点自动移除；连续 30 天无节点的订阅源自动删除 -- **单页面管理** - 订阅源管理 + 可用节点列表在同一页面，按订阅源 Tab 切换 -- **协议分布图** - 饼状图动态展示各协议节点数量和百分比 -- **订阅输出** - 仅输出当前可用节点，支持核心格式（base64）和 Clash（YAML）格式 -- **日志归档** - 按天自动归档，自动清理 7 天前的日志 +- **订阅源管理** - 数据库驱动的增删改查，每个订阅源独立配置 Crontab、延迟阈值、并发数；删除订阅源时自动清除其下所有节点 +- **Crontab 定时拉取** - 每个订阅源支持 5 位 Crontab 表达式，内置随机延迟（0~10 分钟）避免多源同时更新 +- **内核转发检测** - Xray 内核转发后检测连通性，TCP/TLS 直接检测作为回退；多目标 URL 轮询 + 响应体验证 + 检测重试 +- **检测失败直接删除** - 取消失败计数累积，检测不通过的节点直接从数据库删除 +- **节点-订阅绑定** - 每个节点绑定所属 `subscription_id`，已存在于其他订阅的节点不重复入库 +- **纯文本订阅输出** - 每行一条原始代理 URI，参照 `subdom.txt` 格式；同时提供 Clash（YAML）格式 +- **多协议支持** - vmess / vless / trojan / ss / hysteria2 / socks5 / http(s) 解析、检测与 Clash 配置生成 +- **服务实例源** - 获取已连接节点数量统计（不入库），支持手工导入实例源中的订阅地址 +- **配置导出/导入** - 一键导出订阅源和实例源配置为 JSON 文件（含时间戳），导入时自动去重 +- **UTC+8 时区统一** - 所有服务时间统一为东八区 +- **单文件日志** - 不归档、不保留历史日志 - **Docker 部署** - Docker Compose 一键启动 ## 快速开始 @@ -31,7 +32,7 @@ vim config.yaml docker-compose up -d ``` -访问 http://localhost:8080 查看 Web 界面。 +访问 http://localhost:2020 查看 Web 界面。 ### 本地运行 @@ -60,15 +61,19 @@ check: timeout: 5.0 # 检测超时（秒） max_concurrent: 50 # 全局并发检测数 latency_threshold: 1500.0 # 全局延迟阈值（毫秒） + check_mode: "auto" # 检测模式: auto(优先内核转发回退TCP) / http(仅内核转发) / tcp(仅TCP/TLS) + socks_port: 1080 # 本地 SOCKS 转发端口 + http_port: 1081 # 本地 HTTP 转发端口 + kernel_path: "xray" # 内核可执行文件路径 + check_retries: 2 # 单次检测失败后重试次数 scheduler: fetch_interval: 3600 # 拉取订阅间隔（秒） verify_interval: 1800 # 验证节点间隔（秒） cleanup_interval: 7200 # 清理间隔（秒） - max_fail_count: 3 # 最大连续失败次数 ``` -> 检测目标 URL 默认为 `https://www.google.com/generate_204` 和 `https://www.gstatic.com/generate_204`，首次启动自动写入数据库，后续在页面「检测目标」中管理，修改即时生效。 +> 检测目标 URL 默认包含 Google 204、Gstatic 204、Cloudflare、Apple、华为连通性检测等，首次启动自动写入数据库，后续在页面「检测目标」中管理，修改即时生效。 ### 环境变量 @@ -85,15 +90,17 @@ scheduler: |------|------|------| | GET | `/api/proxies` | 可用节点列表 | | GET | `/api/proxies/all` | 所有节点 | -| GET | `/api/proxies/grouped` | 按订阅来源分组的可用节点 | +| GET | `/api/proxies/grouped` | 按 subscription_id 分组的可用节点 | | DELETE | `/api/proxies/{id}` | 删除节点 | +| DELETE | `/api/proxies` | 一键清除所有节点 | ### 订阅输出 | 方法 | 路径 | 说明 | |------|------|------| -| GET | `/api/subscription/v2ray` | 核心格式订阅（base64） | -| GET | `/api/subscription/clash` | Clash 格式订阅（YAML） | +| GET | `/api/subscription/plain` | 纯文本格式（每行一条 URI） | +| GET | `/api/subscription/v2ray` | 纯文本格式（同 plain） | +| GET | `/api/subscription/clash` | Clash 格式（YAML） | ### 订阅源管理 @@ -103,7 +110,7 @@ scheduler: | POST | `/api/subscriptions` | 添加订阅源 | | POST | `/api/subscriptions/auto` | 自动添加（仅 URL 必填，自动拉取验证） | | PUT | `/api/subscriptions/{sub_id}` | 更新订阅源 | -| DELETE | `/api/subscriptions/{sub_id}` | 删除订阅源 | +| DELETE | `/api/subscriptions/{sub_id}` | 删除订阅源及其下所有节点 | ### 拉取与验证 @@ -114,10 +121,28 @@ scheduler: | POST | `/api/verify` | 验证所有节点 | | POST | `/api/verify/{sub_id}` | 验证指定订阅节点 | -### 检测目标 +### 服务实例源 | 方法 | 路径 | 说明 | |------|------|------| +| GET | `/api/instance-sources` | 服务实例源列表 | +| POST | `/api/instance-sources` | 添加服务实例源 | +| PUT | `/api/instance-sources/{source_id}` | 更新服务实例源 | +| DELETE | `/api/instance-sources/{source_id}` | 删除服务实例源 | +| POST | `/api/instance-sources/{source_id}/fetch` | 获取实例源已连接节点数 | +| POST | `/api/instance-sources/{source_id}/import` | 导入实例源订阅 | + +### 配置管理 + +| 方法 | 路径 | 说明 | +|------|------|------| +| GET | `/api/config/export` | 导出配置（JSON） | +| POST | `/api/config/import` | 导入配置（JSON，自动去重） | + +### 检测目标 + +| 方法 | 路径 | 说明 | +|------|------|--------| | GET | `/api/check-urls` | 检测目标 URL 列表 | | POST | `/api/check-urls` | 添加检测目标 URL | | DELETE | `/api/check-urls/{url_id}` | 删除检测目标 URL | @@ -125,16 +150,16 @@ scheduler: ### 其他 | 方法 | 路径 | 说明 | -|------|------|------| -| GET | `/api/stats` | 统计信息 | +|------|------|--------| +| GET | `/api/stats` | 统计信息（总订阅条目数、可用节点数、平均延迟） | | GET | `/api/health` | 健康检查 | ## 订阅链接使用 在节点客户端中添加以下订阅链接： -- **核心**: `http://your-server:8080/api/subscription/v2ray` -- **Clash**: `http://your-server:8080/api/subscription/clash` +- **纯文本**: `http://your-server:2020/api/subscription/plain` +- **Clash**: `http://your-server:2020/api/subscription/clash` > 订阅内容仅包含延迟低于阈值的可用节点，随节点池自动更新。 @@ -147,22 +172,28 @@ scheduler: | Trojan | ✅ | ✅ | | Shadowsocks | ✅ | ✅ | | Hysteria2 | ✅ | ✅ | +| SOCKS5 | ✅ | ✅ | +| HTTP/HTTPS | ✅ | ✅ | + +> socks4/socks4a 协议不再支持，拉取时自动移除。 ## 延迟检测原理 -延迟检测模拟真实上网场景：通过转发服务访问目标检测 URL，测量完整请求延迟。 +延迟检测模拟真实上网场景：通过 Xray 内核转发访问目标检测 URL，测量完整请求延迟。 ``` -客户端 → 转发服务 → 目标网站 +客户端 → Xray 内核转发 → 目标网站 ├── DNS 解析 ├── TCP 连接建立 ├── TLS 握手（HTTPS 目标） └── HTTP 请求/响应 ``` -- 检测目标默认为 `https://www.google.com/generate_204` 和 `https://www.gstatic.com/generate_204` +- 检测目标包含 Google 204、Gstatic 204、Cloudflare、Apple、华为连通性检测等 - 多个目标取最大延迟值，确保所有目标均可达 -- 相比仅测试 TCP/TLS 连通性，HTTP 请求延迟更贴近真实上网体验 +- 响应体验证排除劫持页面和空响应 +- 单次检测失败后自动重试（默认 2 次） +- 检测失败节点直接删除，不保留 ## 项目结构 @@ -174,9 +205,9 @@ proxy_pool/ │ ├── config.py # YAML 配置加载 │ ├── database.py # aiosqlite 异步数据库操作 │ ├── models.py # 数据模型（ProxyInfo / ProxyDBRecord / SubscriptionRecord） -│ ├── parser.py # 订阅拉取 & 解析（5 协议） -│ ├── checker.py # HTTP 延迟检测（多目标取最大值） -│ ├── generator.py # 核心 / Clash 订阅生成 +│ ├── parser.py # 订阅拉取 & 解析（7 协议） +│ ├── checker.py # 内核转发检测 + TCP/TLS 回退检测 +│ ├── generator.py # 纯文本 / Clash 订阅生成 │ ├── scheduler.py # APScheduler 定时任务调度 │ ├── routers/ │ │ ├── api.py # REST API 路由 @@ -185,7 +216,8 @@ proxy_pool/ │ ├── base.html # 基础模板 │ ├── index.html # 主页面（管理 + 节点列表） │ └── subscription.html # 订阅链接页 -├── logs/ # 日志目录（自动归档） +├── logs/ # 日志目录（单文件，不归档） +├── data/ # 数据库目录 ├── config.yaml # 配置文件 ├── docker-compose.yaml # Docker 编排 ├── Dockerfile # Docker 镜像 diff --git a/app/__init__.py b/app/__init__.py index f213a10..0371706 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -63,19 +63,14 @@ def create_app(config_path: str = "config.yaml") -> FastAPI: console_handler.setFormatter(log_fmt) root_logger.addHandler(console_handler) - # 文件：按天归档，保留7天 + # 文件：只保留当前日志，不归档 log_dir = Path("logs") log_dir.mkdir(exist_ok=True) - from logging.handlers import TimedRotatingFileHandler - file_handler = TimedRotatingFileHandler( + file_handler = logging.FileHandler( filename=str(log_dir / "proxy_pool.log"), - when="midnight", - interval=1, - backupCount=7, encoding="utf-8", ) file_handler.setFormatter(log_fmt) - file_handler.suffix = "%Y-%m-%d" root_logger.addHandler(file_handler) # 初始化组件 @@ -120,6 +115,7 @@ async def startup(): http_port=_config.check.http_port, check_mode=_config.check.check_mode, kernel_path=_config.check.kernel_path, + check_retries=_config.check.check_retries, ) # 初始化调度器 diff --git a/app/checker.py b/app/checker.py index 5968631..7898b6e 100644 --- a/app/checker.py +++ b/app/checker.py @@ -39,8 +39,33 @@ "https://www.gstatic.com/generate_204", "https://cp.cloudflare.com/", "https://www.apple.com/library/test/success.html", + "https://connectivitycheck.platform.hicloud.com/generate_204", ] +# 检测 URL 预期响应验证规则 +CHECK_URL_VALIDATORS = { + "generate_204": lambda url, status, body: status == 204 or (status == 200 and len(body) == 0), + "cp.cloudflare.com": lambda url, status, body: status == 200 and len(body) > 0, + "success.html": lambda url, status, body: status == 200 and len(body) > 0, + "connectivitycheck": lambda url, status, body: status == 204 or status == 200, +} + + +def _validate_check_response(url: str, status: int, body: bytes) -> bool: + """验证检测响应是否为有效响应（排除劫持页面/认证门户） + + 默认规则：2xx 状态码即通过 + 特定 URL 有额外验证规则 + """ + if not (200 <= status < 300): + return False + # 逐条匹配验证规则 + for keyword, validator in CHECK_URL_VALIDATORS.items(): + if keyword in url: + return validator(url, status, body) + # 通用规则：2xx + 有响应体或 204 + return status == 204 or len(body) > 0 + # 本地转发端口默认值 DEFAULT_SOCKS_PORT = 1080 DEFAULT_HTTP_PORT = 1081 @@ -66,7 +91,8 @@ class ProxyChecker: 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"): + check_mode: str = "auto", kernel_path: str = "xray", + check_retries: int = 2): """ Args: check_urls: HTTP 检测目标 URL 列表 @@ -76,6 +102,7 @@ def __init__(self, check_urls: list[str], timeout: float, max_concurrent: int, http_port: 本地 HTTP 转发端口（ConnectivityMonitor 使用） check_mode: 检测模式 "http" / "tcp" / "auto" kernel_path: 内核可执行文件路径 + check_retries: 单次检测失败后重试次数（提升有效性） """ self.check_urls = check_urls or DEFAULT_CHECK_URLS self.timeout = timeout @@ -84,6 +111,7 @@ def __init__(self, check_urls: list[str], timeout: float, max_concurrent: int, 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) + self.check_retries = max(check_retries, 0) @staticmethod def _resolve_kernel_path(kernel_path: str) -> str: @@ -103,27 +131,50 @@ async def check_proxy(self, link: str) -> float | None: - auto: 优先内核转发，失败则回退 TCP/TLS - http: 仅通过内核转发检测 - tcp: 仅直连 TCP/TLS 检测 + + 支持重试：首次失败后重试 check_retries 次，取最快成功的延迟值 """ 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) + best_latency = None - # http / auto 模式：优先内核转发 - if self.kernel_path: - latency = await self._check_via_kernel(link) + for attempt in range(1 + self.check_retries): + latency = await self._check_once(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) + # 成功则取最快延迟 + if best_latency is None or latency < best_latency: + best_latency = latency + # 首次成功就不再重试 + if attempt == 0: + break + # 重试成功也直接返回（已有可用延迟） + break + # 首次失败，短暂等待后重试 + if attempt < self.check_retries: + await asyncio.sleep(0.5) + + return best_latency + + async def _check_once(self, link: str) -> float | None: + """单次检测节点延迟（不含重试逻辑）""" + 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 + return None # ---- 内核转发检测 ---- @@ -200,26 +251,36 @@ async def _check_via_kernel(self, link: str) -> float | None: 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 + """通过本地 HTTP 转发端口发送 HTTP 请求检测延迟 + + 轮询多个检测 URL，使用 GET 请求并验证响应内容； + 确保节点能真正访问目标页面，排除劫持页面和认证门户 + """ + urls = self.check_urls if self.check_urls else DEFAULT_CHECK_URLS + proxy_url = f"http://127.0.0.1:{port}" + + for check_url in urls: + start = time.monotonic() + try: + connector = aiohttp.TCPConnector(limit=1, force_close=True) + async with aiohttp.ClientSession(connector=connector) as session: + async with session.get( + check_url, + proxy=proxy_url, + timeout=aiohttp.ClientTimeout(total=self.timeout), + allow_redirects=True, + ssl=False, + ) as resp: + # 读取响应体（限制最大 4KB 防止下载大文件） + body = await resp.content.read(4096) + elapsed = (time.monotonic() - start) * 1000 + + if _validate_check_response(check_url, resp.status, body): + return round(elapsed, 1) + # 验证不通过，尝试下一个 URL + except Exception: + continue + return None @staticmethod def _find_free_port() -> int: @@ -265,7 +326,7 @@ def _link_to_xray_outbound(link: str) -> dict | None: return ProxyChecker._ss_to_xray(link) elif link.startswith("hysteria2://") or link.startswith("hy2://"): return ProxyChecker._hysteria2_to_xray(link) - elif link.startswith(("socks5://", "socks4://", "socks4a://")): + elif link.startswith("socks5://"): return ProxyChecker._socks_to_xray(link) elif link.startswith(("http://", "https://")) and "#" in link: return ProxyChecker._http_to_xray(link) @@ -635,7 +696,7 @@ def _hysteria2_to_xray(link: str) -> dict | None: @staticmethod def _socks_to_xray(link: str) -> dict | None: - """socks5:// / socks4:// / socks4a:// 分享链接转内核 outbound""" + """socks5:// 分享链接转内核 outbound""" try: parsed = urlparse(link) address = parsed.hostname or "" @@ -682,7 +743,12 @@ def _http_to_xray(link: str) -> dict | None: # ---- TCP/TLS 直连检测（回退方案） ---- async def _check_tcp_tls(self, info: ProxyConnInfo) -> float | None: - """直接 TCP/TLS 连接到节点服务器检测延迟（回退方案）""" + """直接 TCP/TLS 连接到节点服务器检测延迟（回退方案） + + 注意：此方法仅验证服务器端口可达和 TLS 握手成功， + 不代表节点能真正转发流量到目标网站。 + 连接被拒绝视为不可用，不返回延迟值。 + """ host = info.host if not self._is_ip(host): resolved = await self._resolve_host(host) @@ -697,14 +763,19 @@ async def _check_tcp_tls(self, info: ProxyConnInfo) -> float | None: ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE - _, writer = await asyncio.wait_for( + reader, writer = await asyncio.wait_for( asyncio.open_connection( - host, info.port, ssl=ctx, server_hostname=host + host, info.port, ssl=ctx, server_hostname=info.host ), timeout=self.timeout, ) + # 验证 TLS 握手完成：检查 SSL 对象状态 + ssl_obj = writer.get_extra_info("ssl_object") + if ssl_obj is None: + writer.close() + return None else: - _, writer = await asyncio.wait_for( + reader, writer = await asyncio.wait_for( asyncio.open_connection(host, info.port), timeout=self.timeout, ) @@ -717,8 +788,11 @@ async def _check_tcp_tls(self, info: ProxyConnInfo) -> float | None: pass return round(elapsed, 1) except ConnectionRefusedError: - elapsed = (time.monotonic() - start) * 1000 - return round(elapsed, 1) + # 连接被拒绝 = 不可用，不返回延迟值 + return None + except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError): + # 连接被重置/中断 = 不可用 + return None except Exception: return None @@ -784,7 +858,7 @@ def _parse_link(self, link: str) -> ProxyConnInfo | None: return self._parse_ss(link) elif link.startswith("hysteria2://") or link.startswith("hy2://"): return self._parse_hysteria2(link) - elif link.startswith(("socks5://", "socks4://", "socks4a://")): + elif link.startswith("socks5://"): return self._parse_socks(link) elif link.startswith(("http://", "https://")) and "#" in link: return self._parse_http_proxy(link) @@ -903,15 +977,17 @@ def _parse_hysteria2(self, link: str) -> ProxyConnInfo | None: def _parse_socks(self, link: str) -> ProxyConnInfo | None: try: + # socks4/socks4a 不再支持 + if link.lower().startswith(("socks4://", "socks4a://")): + return None parsed = urlparse(link) address = parsed.hostname or "" port = parsed.port or 0 if not address or not port: return None - protocol = "socks5" if link.lower().startswith("socks5://") else "socks4" return ProxyConnInfo( host=address, port=int(port), - use_tls=False, protocol=protocol, + use_tls=False, protocol="socks5", ) except Exception: return None diff --git a/app/config.py b/app/config.py index 9d6f107..30c874e 100644 --- a/app/config.py +++ b/app/config.py @@ -27,6 +27,7 @@ class CheckConfig: socks_port: int = 1080 # 本地转发端口（ConnectivityMonitor 使用） http_port: int = 1081 # 本地 HTTP 转发端口 kernel_path: str = "xray" # 内核可执行文件路径 + check_retries: int = 2 # 单次检测失败后重试次数 @dataclass @@ -83,6 +84,7 @@ def load_config(path: str = "config.yaml") -> AppConfig: "socks_port": 1080, "http_port": 1081, "kernel_path": "xray", + "check_retries": 2, }, "scheduler": { "fetch_interval": 3600, diff --git a/app/database.py b/app/database.py index 766f651..71cf42f 100644 --- a/app/database.py +++ b/app/database.py @@ -28,24 +28,24 @@ async def init(self) -> None: await self._db.executescript(""" CREATE TABLE IF NOT EXISTS proxies ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - protocol TEXT NOT NULL, - name TEXT NOT NULL DEFAULT '', - address TEXT NOT NULL DEFAULT '', - port TEXT NOT NULL DEFAULT '', - link TEXT NOT NULL UNIQUE, - latency_ms REAL DEFAULT -1, - fail_count INTEGER DEFAULT 0, - source TEXT NOT NULL DEFAULT '', - last_check_time TEXT DEFAULT '', - last_success_time TEXT DEFAULT '', - created_at TEXT DEFAULT '' + id INTEGER PRIMARY KEY AUTOINCREMENT, + protocol TEXT NOT NULL, + name TEXT NOT NULL DEFAULT '', + address TEXT NOT NULL DEFAULT '', + port TEXT NOT NULL DEFAULT '', + link TEXT NOT NULL UNIQUE, + latency_ms REAL DEFAULT -1, + fail_count INTEGER DEFAULT 0, + subscription_id INTEGER NOT NULL DEFAULT 0, + last_check_time TEXT DEFAULT '', + last_success_time TEXT DEFAULT '', + created_at TEXT DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_proxies_protocol ON proxies(protocol); CREATE INDEX IF NOT EXISTS idx_proxies_latency ON proxies(latency_ms); - CREATE INDEX IF NOT EXISTS idx_proxies_fail_count ON proxies(fail_count); CREATE INDEX IF NOT EXISTS idx_proxies_link ON proxies(link); + CREATE INDEX IF NOT EXISTS idx_proxies_sub_id ON proxies(subscription_id); CREATE TABLE IF NOT EXISTS subscriptions ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -86,6 +86,13 @@ async def init(self) -> None: """) await self._db.commit() + # 迁移：为旧表添加 subscription_id 列（如不存在） + try: + await self._db.execute("ALTER TABLE proxies ADD COLUMN subscription_id INTEGER NOT NULL DEFAULT 0") + await self._db.commit() + except Exception: + pass + # 兼容旧表：添加 check_urls 表（如不存在） try: await self._db.execute( @@ -138,23 +145,23 @@ def _row_to_record(self, row: aiosqlite.Row) -> ProxyDBRecord: link=row["link"], latency_ms=row["latency_ms"], fail_count=row["fail_count"], - source=row["source"], + subscription_id=row["subscription_id"], last_check_time=row["last_check_time"], last_success_time=row["last_success_time"], created_at=row["created_at"], ) - async def insert_proxy(self, proxy: ProxyInfo, latency_ms: float, source: str) -> bool: + async def insert_proxy(self, proxy: ProxyInfo, latency_ms: float, subscription_id: int) -> bool: """插入新节点，link 唯一约束，重复则忽略。返回是否插入成功""" - now = datetime.now(timezone.utc).isoformat() + now = datetime.now(timezone(timedelta(hours=8))).isoformat() try: cursor = await self._db.execute( """INSERT OR IGNORE INTO proxies - (protocol, name, address, port, link, latency_ms, fail_count, source, + (protocol, name, address, port, link, latency_ms, fail_count, subscription_id, last_check_time, last_success_time, created_at) VALUES (?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?)""", (proxy.protocol, proxy.name, proxy.address, proxy.port, proxy.link, - latency_ms, source, now, now if latency_ms >= 0 else "", now), + latency_ms, subscription_id, now, now if latency_ms >= 0 else "", now), ) await self._db.commit() return cursor.rowcount > 0 @@ -163,7 +170,7 @@ async def insert_proxy(self, proxy: ProxyInfo, latency_ms: float, source: str) - async def update_latency(self, proxy_id: int, latency_ms: float) -> None: """更新延迟并重置 fail_count""" - now = datetime.now(timezone.utc).isoformat() + now = datetime.now(timezone(timedelta(hours=8))).isoformat() await self._db.execute( """UPDATE proxies SET latency_ms = ?, fail_count = 0, @@ -181,7 +188,7 @@ async def batch_update_latency(self, updates: list[tuple[int, float]]) -> None: """ if not updates: return - now = datetime.now(timezone.utc).isoformat() + now = datetime.now(timezone(timedelta(hours=8))).isoformat() await self._db.executemany( """UPDATE proxies SET latency_ms = ?, fail_count = 0, @@ -191,41 +198,13 @@ async def batch_update_latency(self, updates: list[tuple[int, float]]) -> None: ) await self._db.commit() - async def batch_increment_fail(self, proxy_ids: list[int]) -> None: - """批量 fail_count + 1，单次 commit""" + async def batch_delete_proxies(self, proxy_ids: list[int]) -> None: + """批量删除节点（检测失败直接删除）""" 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() - await self._db.execute( - """UPDATE proxies - SET fail_count = fail_count + 1, last_check_time = ? - WHERE id = ?""", - (now, proxy_id), - ) - await self._db.commit() - - cursor = await self._db.execute( - "SELECT fail_count FROM proxies WHERE id = ?", (proxy_id,) - ) - 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), + "DELETE FROM proxies WHERE id = ?", + [(pid,) for pid in proxy_ids], ) await self._db.commit() @@ -234,36 +213,16 @@ 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 的节点，返回删除数量""" - cursor = await self._db.execute( - "DELETE FROM proxies WHERE fail_count >= ?", (max_fail,) - ) + async def delete_all_proxies(self) -> int: + """删除所有节点，返回删除数量""" + cursor = await self._db.execute("DELETE FROM proxies") 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() + async def delete_proxies_by_subscription_id(self, subscription_id: int) -> int: + """删除指定订阅源 ID 下的所有节点，返回删除数量""" 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,), + "DELETE FROM proxies WHERE subscription_id = ?", (subscription_id,) ) await self._db.commit() return cursor.rowcount @@ -290,7 +249,7 @@ async def get_available_proxies(self, max_latency: float) -> list[ProxyDBRecord] async def get_subscription_output_proxies(self, max_latency: float) -> list[ProxyDBRecord]: """获取对外订阅输出节点列表 - 输出所有延迟达标且未失败的节点，包含订阅源和检测通过的实例源节点 + 输出所有延迟达标且未失败的节点 """ cursor = await self._db.execute( """SELECT * FROM proxies @@ -311,15 +270,24 @@ async def get_proxy_by_link(self, link: str) -> ProxyDBRecord | None: row = await cursor.fetchone() 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 排序，最旧的优先）""" - query = "SELECT * FROM proxies ORDER BY last_check_time ASC" - if limit > 0: - query += f" LIMIT {limit}" - cursor = await self._db.execute(query) + async def get_proxies_by_subscription_id(self, subscription_id: int) -> list[ProxyDBRecord]: + """根据订阅源 ID 获取节点，按入库时间正序""" + cursor = await self._db.execute( + """SELECT * FROM proxies WHERE subscription_id = ? + ORDER BY created_at ASC""", + (subscription_id,), + ) rows = await cursor.fetchall() return [self._row_to_record(row) for row in rows] + async def get_proxy_count_by_subscription_id(self, subscription_id: int) -> int: + """获取指定订阅源的节点总数""" + cursor = await self._db.execute( + "SELECT COUNT(*) as cnt FROM proxies WHERE subscription_id = ?", (subscription_id,) + ) + row = await cursor.fetchone() + return row["cnt"] if row else 0 + def _row_to_subscription(self, row: aiosqlite.Row) -> SubscriptionRecord: """将数据库行转为 SubscriptionRecord""" return SubscriptionRecord( @@ -343,7 +311,7 @@ async def add_subscription(self, url: str, crontab: str = "0 * * * *", max_retries: int = 3, max_concurrent: int = 50, enabled: bool = True) -> SubscriptionRecord | None: """添加订阅源""" - now = datetime.now(timezone.utc).isoformat() + now = datetime.now(timezone(timedelta(hours=8))).isoformat() try: cursor = await self._db.execute( """INSERT INTO subscriptions (url, crontab, latency_threshold, max_retries, max_concurrent, enabled, created_at) @@ -380,7 +348,7 @@ async def get_enabled_subscriptions(self) -> list[SubscriptionRecord]: return [self._row_to_subscription(row) for row in rows] async def update_subscription(self, sub_id: int, **kwargs) -> bool: - """更新订阅源，支持部分字段更新。若 URL 变更则同步更新 proxies.source""" + """更新订阅源，支持部分字段更新""" allowed = {"url", "crontab", "latency_threshold", "max_retries", "max_concurrent", "enabled"} updates = {} for k, v in kwargs.items(): @@ -392,15 +360,6 @@ async def update_subscription(self, sub_id: int, **kwargs) -> bool: if not updates: return False - # 如果更新了 URL，同步更新 proxies 表的 source - if "url" in updates: - old_sub = await self.get_subscription_by_id(sub_id) - if old_sub and old_sub.url != updates["url"]: - await self._db.execute( - "UPDATE proxies SET source = ? WHERE source = ?", - (updates["url"], old_sub.url), - ) - set_clause = ", ".join(f"{k} = ?" for k in updates) values = list(updates.values()) + [sub_id] cursor = await self._db.execute( @@ -410,7 +369,9 @@ async def update_subscription(self, sub_id: int, **kwargs) -> bool: return cursor.rowcount > 0 async def delete_subscription(self, sub_id: int) -> bool: - """删除订阅源""" + """删除订阅源及其下所有节点""" + # 先删除该订阅下的所有节点 + await self.delete_proxies_by_subscription_id(sub_id) cursor = await self._db.execute("DELETE FROM subscriptions WHERE id = ?", (sub_id,)) await self._db.commit() return cursor.rowcount > 0 @@ -439,14 +400,6 @@ async def get_subscriptions_with_empty_days(self, min_days: int) -> list[Subscri rows = await cursor.fetchall() 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,) - ) - row = await cursor.fetchone() - return row["cnt"] if row else 0 - async def update_total_count(self, sub_id: int, count: int) -> None: """更新订阅源最新一次拉取的节点总数""" await self._db.execute( @@ -463,50 +416,32 @@ 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 获取节点，按入库时间正序""" - cursor = await self._db.execute( - """SELECT * FROM proxies WHERE source = ? - ORDER BY created_at ASC""", - (source,), - ) - rows = await cursor.fetchall() - 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 获取可用节点，按入库时间正序""" - cursor = await self._db.execute( - """SELECT * FROM proxies - WHERE source = ? AND latency_ms > 0 AND latency_ms <= ? AND fail_count = 0 - ORDER BY created_at ASC""", - (source, max_latency), - ) - rows = await cursor.fetchall() - 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]]: - """获取按订阅来源分组的可用节点，按入库时间正序""" + async def get_proxies_grouped_by_subscription(self, max_latency: float) -> dict[int, list[ProxyDBRecord]]: + """获取按订阅源 ID 分组的可用节点，按入库时间正序""" cursor = await self._db.execute( """SELECT * FROM proxies WHERE latency_ms > 0 AND latency_ms <= ? AND fail_count = 0 - ORDER BY source ASC, created_at ASC""", + ORDER BY subscription_id ASC, created_at ASC""", (max_latency,), ) rows = await cursor.fetchall() grouped = {} for row in rows: record = self._row_to_record(row) - grouped.setdefault(record.source, []).append(record) + grouped.setdefault(record.subscription_id, []).append(record) return grouped async def get_stats(self) -> dict: - """获取统计信息""" - cursor = await self._db.execute("SELECT COUNT(*) as total FROM proxies") + """获取统计信息 + + total: 订阅源地址数量（有多少个订阅源URL） + available: 数据库中检测通过的可用节点数 + """ + cursor = await self._db.execute("SELECT COUNT(*) as total FROM subscriptions") total = (await cursor.fetchone())["total"] cursor = await self._db.execute( - """SELECT COUNT(*) as available FROM proxies - WHERE latency_ms > 0 AND fail_count = 0""" + "SELECT COUNT(*) as available FROM proxies WHERE latency_ms > 0" ) available = (await cursor.fetchone())["available"] @@ -524,7 +459,6 @@ async def get_stats(self) -> dict: return { "total": total, "available": available, - "unavailable": total - available, "avg_latency_ms": round(avg_latency, 1), "protocol_distribution": protocol_dist, } @@ -599,7 +533,7 @@ async def add_instance_source(self, base_url: str, username: str, password: str, max_concurrent: int = 50, enabled: bool = True) -> InstanceSourceRecord | None: """添加服务实例源""" - now = datetime.now(timezone.utc).isoformat() + now = datetime.now(timezone(timedelta(hours=8))).isoformat() try: cursor = await self._db.execute( """INSERT INTO instance_sources @@ -664,7 +598,7 @@ async def delete_instance_source(self, source_id: int) -> bool: 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), diff --git a/app/generator.py b/app/generator.py index 40f6ba4..dd2e1b3 100644 --- a/app/generator.py +++ b/app/generator.py @@ -16,12 +16,52 @@ def generate_plain_subscription(proxies: list[ProxyDBRecord]) -> str: """生成纯文本格式订阅内容（参照 subdom.txt 格式） - 每行一条原始代理 URI + 每行一条原始代理 URI，socks/http 代理确保格式为 protocol://host:port#host-port """ - links = [p.link for p in proxies] + links = [_normalize_link(p) for p in proxies] return "\n".join(links) +def _normalize_link(proxy: ProxyDBRecord) -> str: + """规范化分享链接，确保 socks/http 代理带有 #host-port 名称""" + link = proxy.link + if link.startswith(("socks5://", "socks4://", "socks4a://")): + # socks4/socks4a 不再支持，跳过 + if link.lower().startswith(("socks4://", "socks4a://")): + return link + parsed = urlparse(link) + host = parsed.hostname or proxy.address + port = parsed.port or int(proxy.port) if proxy.port.isdigit() else 0 + if parsed.fragment: + name = unquote(parsed.fragment) + else: + name = f"{host}-{port}" + auth = "" + if parsed.username: + auth = unquote(parsed.username) + if parsed.password: + auth += f":{unquote(parsed.password)}" + auth += "@" + return f"socks5://{auth}{host}:{port}#{name}" + elif link.startswith(("http://", "https://")) and ("#" in link or proxy.protocol in ("http", "https")): + parsed = urlparse(link) + host = parsed.hostname or proxy.address + port = parsed.port or int(proxy.port) if proxy.port.isdigit() else 8080 + protocol = "https" if link.lower().startswith("https://") else "http" + if parsed.fragment: + name = unquote(parsed.fragment) + else: + name = f"{host}-{port}" + auth = "" + if parsed.username: + auth = unquote(parsed.username) + if parsed.password: + auth += f":{unquote(parsed.password)}" + auth += "@" + return f"{protocol}://{auth}{host}:{port}#{name}" + return link + + def generate_v2ray_subscription(proxies: list[ProxyDBRecord]) -> str: """生成 v2ray 格式订阅内容（纯文本，每行一条原始链接） @@ -85,7 +125,7 @@ def _link_to_clash_proxy(link: str, fallback_name: str) -> dict | None: return _ss_link_to_clash(link, fallback_name) elif link.startswith("hysteria2://") or link.startswith("hy2://"): return _hysteria2_link_to_clash(link, fallback_name) - elif link.startswith(("socks5://", "socks4://", "socks4a://")): + elif link.startswith("socks5://"): return _socks_link_to_clash(link, fallback_name) elif link.startswith(("http://", "https://")) and "#" in link: return _http_link_to_clash(link, fallback_name) @@ -95,7 +135,7 @@ def _link_to_clash_proxy(link: str, fallback_name: str) -> dict | None: def _socks_link_to_clash(link: str, fallback_name: str) -> dict: - """socks5:// / socks4:// / socks4a:// 转 Clash proxy""" + """socks5:// 转 Clash proxy""" parsed = urlparse(link) name = unquote(parsed.fragment) if parsed.fragment else fallback_name proxy = { diff --git a/app/models.py b/app/models.py index cf020a2..e74709d 100644 --- a/app/models.py +++ b/app/models.py @@ -26,7 +26,7 @@ class ProxyDBRecord: link: str latency_ms: float # 延迟毫秒，-1=未检测 fail_count: int # 连续失败次数 - source: str # 来源订阅 URL + subscription_id: int # 所属订阅源 ID last_check_time: str # ISO8601 last_success_time: str # ISO8601 created_at: str # ISO8601 diff --git a/app/parser.py b/app/parser.py index 14ce7e5..a7a69cb 100644 --- a/app/parser.py +++ b/app/parser.py @@ -225,21 +225,22 @@ def _parse_hysteria2(line: str) -> ProxyInfo | None: def _parse_socks(line: str) -> ProxyInfo | None: - """解析 socks5:// / socks4:// / socks4a:// 链接 + """解析 socks5:// 链接（socks4/socks4a 不再支持，直接跳过） 格式: socks5://[user:pass@]host:port[#name] """ try: + # socks4/socks4a 协议不再支持，跳过 + if line.lower().startswith(("socks4://", "socks4a://")): + return None parsed = urlparse(line) address = parsed.hostname or "" port = str(parsed.port) if parsed.port else "" if not address or not port: return None name = unquote(parsed.fragment) if parsed.fragment else f"{address}:{port}" - # 统一协议名 - protocol = "socks5" if line.lower().startswith("socks5://") else "socks4" return ProxyInfo( - protocol=protocol, + protocol="socks5", name=name, address=address, port=port, diff --git a/app/routers/api.py b/app/routers/api.py index 7f02f15..4d691b4 100644 --- a/app/routers/api.py +++ b/app/routers/api.py @@ -46,6 +46,14 @@ async def delete_proxy(proxy_id: int): return {"message": "deleted"} +@router.delete("/proxies") +async def delete_all_proxies(): + """一键清除数据库内所有节点""" + db = get_db() + count = await db.delete_all_proxies() + return {"message": "deleted", "count": count} + + @router.get("/subscription/v2ray") async def v2ray_subscription(): """获取纯文本格式订阅（每行一条原始代理 URI） @@ -127,6 +135,8 @@ async def manual_verify_subscription(sub_id: int): from fastapi import HTTPException raise HTTPException(status_code=404, detail="订阅不存在") scheduler = get_scheduler() + if sub_id in scheduler._verifying_subs: + return {"message": f"订阅 #{sub_id} 正在验证中，请勿重复操作"} asyncio.create_task(scheduler.verify_subscription_proxies(sub_id)) return {"message": f"已触发订阅 #{sub_id} 的节点验证"} @@ -221,11 +231,8 @@ async def auto_add_subscription(req: AutoSubRequest): # 注册定时任务 scheduler = get_scheduler() scheduler._add_subscription_job(sub) - # 自动触发拉取，拉取完成后自动验证 - async def fetch_and_verify(): - await scheduler._fetch_single_subscription(sub.id) - await scheduler.verify_subscription_proxies(sub.id) - asyncio.create_task(fetch_and_verify()) + # 自动触发拉取（拉取完成后会自动验证该订阅下所有已入库节点） + asyncio.create_task(scheduler._fetch_single_subscription(sub.id)) return { "status": "added", "message": "订阅已添加，正在拉取并验证节点", @@ -263,11 +270,8 @@ async def update_subscription(sub_id: int, url: str = None, crontab: str = None, scheduler.refresh_subscription_job(sub) if sub.enabled: - # 启用时自动拉取并验证该订阅源 - async def fetch_and_verify(): - await scheduler._fetch_single_subscription(sub.id) - await scheduler.verify_subscription_proxies(sub.id) - asyncio.create_task(fetch_and_verify()) + # 启用时自动拉取（拉取完成后会自动验证该订阅下所有已入库节点） + asyncio.create_task(scheduler._fetch_single_subscription(sub.id)) else: # 禁用时重置拉取状态 await db.update_fetch_status(sub_id, "idle") @@ -293,10 +297,10 @@ async def get_proxies_grouped(): """获取按订阅来源分组的可用节点""" db = get_db() config = get_config() - grouped = await db.get_proxies_grouped_by_source(config.check.latency_threshold) + grouped = await db.get_proxies_grouped_by_subscription(config.check.latency_threshold) result = {} - for source, proxies in grouped.items(): - result[source] = [_proxy_to_dict(p) for p in proxies] + for sub_id, proxies in grouped.items(): + result[str(sub_id)] = [_proxy_to_dict(p) for p in proxies] return {"grouped": result} @@ -310,7 +314,7 @@ def _proxy_to_dict(proxy) -> dict: "port": proxy.port, "latency_ms": proxy.latency_ms, "fail_count": proxy.fail_count, - "source": proxy.source, + "subscription_id": proxy.subscription_id, "last_check_time": proxy.last_check_time, "last_success_time": proxy.last_success_time, "status": proxy.status, @@ -535,3 +539,122 @@ def _instance_source_to_dict(source) -> dict: "total_count": source.total_count, "fetch_status": source.fetch_status, } + + +# ---- 配置导出/导入 ---- + +@router.get("/config/export") +async def export_config(): + """导出订阅源和服务实例源配置为 JSON 文件""" + db = get_db() + subs = await db.get_all_subscriptions() + sources = await db.get_all_instance_sources() + + config = { + "version": 1, + "subscriptions": [ + { + "url": s.url, + "crontab": s.crontab, + "latency_threshold": s.latency_threshold, + "max_retries": s.max_retries, + "max_concurrent": s.max_concurrent, + "enabled": s.enabled, + } + for s in subs + ], + "instance_sources": [ + { + "base_url": s.base_url, + "username": s.username, + "password": s.password, + "crontab": s.crontab, + "latency_threshold": s.latency_threshold, + "max_concurrent": s.max_concurrent, + "enabled": s.enabled, + } + for s in sources + ], + } + + from fastapi.responses import Response + from datetime import datetime, timedelta, timezone + import json + content = json.dumps(config, ensure_ascii=False, indent=2) + ts = datetime.now(timezone(timedelta(hours=8))).strftime("%Y%m%d_%H%M%S") + return Response( + content=content, + media_type="application/json; charset=utf-8", + headers={ + "Content-Disposition": f'attachment; filename="nethub_config_{ts}.json"', + }, + ) + + +class ConfigImportRequest(BaseModel): + """配置导入请求体""" + config: dict + + +@router.post("/config/import") +async def import_config(req: ConfigImportRequest): + """从 JSON 导入订阅源和服务实例源配置（去重，不覆盖已有）""" + db = get_db() + scheduler = get_scheduler() + + config = req.config + sub_added = 0 + sub_dup = 0 + inst_added = 0 + inst_dup = 0 + + # 导入订阅源 + for item in config.get("subscriptions", []): + url = item.get("url", "").strip() + if not url: + continue + existing = await db.get_subscription_by_url(url) + if existing: + sub_dup += 1 + continue + sub = await db.add_subscription( + url=url, + crontab=item.get("crontab", "0 * * * *"), + latency_threshold=item.get("latency_threshold", 1500.0), + max_retries=item.get("max_retries", 3), + max_concurrent=item.get("max_concurrent", 50), + enabled=item.get("enabled", True), + ) + if sub: + scheduler._add_subscription_job(sub) + sub_added += 1 + + # 导入服务实例源 + for item in config.get("instance_sources", []): + base_url = item.get("base_url", "").strip() + if not base_url: + continue + existing = await db.get_instance_source_by_url(base_url) + if existing: + inst_dup += 1 + continue + source = await db.add_instance_source( + base_url=base_url, + username=item.get("username", ""), + password=item.get("password", ""), + crontab=item.get("crontab", "*/10 * * * *"), + latency_threshold=item.get("latency_threshold", 1500.0), + max_concurrent=item.get("max_concurrent", 50), + enabled=item.get("enabled", True), + ) + if source: + scheduler._add_instance_source_job(source) + inst_added += 1 + + return { + "message": f"导入完成: 订阅源新增 {sub_added} 个(跳过 {sub_dup} 个重复), 实例源新增 {inst_added} 个(跳过 {inst_dup} 个重复)", + "subscription_added": sub_added, + "subscription_duplicate": sub_dup, + "instance_added": inst_added, + "instance_duplicate": inst_dup, + } diff --git a/app/routers/web.py b/app/routers/web.py index 1695119..2d15d6a 100644 --- a/app/routers/web.py +++ b/app/routers/web.py @@ -28,7 +28,7 @@ async def index(request: Request): db.get_all_proxies(), db.get_stats(), db.get_all_subscriptions(), - db.get_proxies_grouped_by_source(config.check.latency_threshold), + db.get_proxies_grouped_by_subscription(config.check.latency_threshold), db.get_check_urls(), db.get_all_instance_sources(), ) @@ -42,17 +42,14 @@ async def index(request: Request): [dict(asdict(s)) for s in instance_sources], ensure_ascii=False, ) - # 计算每个订阅源的可用节点数量 + + # 计算每个订阅源的可用节点数量（grouped 按 subscription_id 分组） 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, [])) + sub_available_counts[sub.id] = len(grouped.get(sub.id, [])) + subscriptions_json = json.dumps( - [dict(asdict(s), available_count=sub_available_counts.get(s.url, 0)) for s in subscriptions], + [dict(asdict(s), available_count=sub_available_counts.get(s.id, 0)) for s in subscriptions], ensure_ascii=False, ) @@ -68,7 +65,6 @@ async def index(request: Request): "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, diff --git a/app/scheduler.py b/app/scheduler.py index 40888d0..2e1377d 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -3,7 +3,8 @@ import asyncio import logging -from datetime import datetime, timezone +import random +from datetime import datetime, timedelta, timezone from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger @@ -28,6 +29,7 @@ def __init__(self, config: AppConfig, db: ProxyDatabase, checker: ProxyChecker): self._last_verify_time = "" self._fetching = False self._verifying = False + self._verifying_subs: set[int] = set() # 正在验证的订阅 ID 集合，防止并发 self._last_instance_sub_urls: list[str] = [] # 最近一次实例源获取的订阅地址缓存 def start(self) -> None: @@ -87,6 +89,7 @@ def _add_subscription_job(self, sub) -> None: trigger = CronTrigger( minute=parts[0], hour=parts[1], day=parts[2], month=parts[3], day_of_week=parts[4], + jitter=random.randint(0, 600), # 随机延迟0~600秒(10分钟内) ) self.scheduler.add_job( self._fetch_single_subscription, @@ -117,7 +120,11 @@ def shutdown(self) -> None: logger.info("调度器已关闭") async def _fetch_single_subscription(self, sub_id: int) -> None: - """拉取单个订阅并检测入库（并发检测 + 批量数据库写入）""" + """拉取单个订阅并检测入库（并发检测 + 批量数据库写入） + + 节点绑定 subscription_id，已存在于其他订阅则跳过不更新。 + 拉取完成后同时验证该订阅下所有已入库节点。 + """ sub = await self.db.get_subscription_by_id(sub_id) if not sub or not sub.enabled: return @@ -136,6 +143,13 @@ async def _fetch_single_subscription(self, sub_id: int) -> None: logger.info("订阅 #%d: 解析到 %d 个节点，并发检测中...", sub.id, len(proxies)) + # 自动移除该订阅下已有的 socks4 节点（不再支持） + existing_proxies = await self.db.get_proxies_by_subscription_id(sub_id) + socks4_ids = [p.id for p in existing_proxies if p.protocol == "socks4"] + if socks4_ids: + await self.db.batch_delete_proxies(socks4_ids) + logger.info("订阅 #%d: 自动移除 %d 个 socks4 节点", sub_id, len(socks4_ids)) + # 并发检测所有节点延迟 sub_checker = ProxyChecker( check_urls=self.checker.check_urls, @@ -145,6 +159,7 @@ async def _fetch_single_subscription(self, sub_id: int) -> None: http_port=self.checker.http_port, check_mode=self.checker.check_mode, kernel_path=self.checker.kernel_path, + check_retries=self.config.check.check_retries, ) links = [p.link for p in proxies] results = await sub_checker.check_batch(links) @@ -152,53 +167,43 @@ async def _fetch_single_subscription(self, sub_id: int) -> None: # 批量处理检测结果 threshold = sub.latency_threshold latency_updates = [] # [(proxy_id, latency), ...] - fail_ids = [] # [proxy_id, ...] + delete_ids = [] # [proxy_id, ...] 已入库但延迟超标需删除的节点 added = 0 - updated = 0 skipped = 0 for proxy in proxies: latency = results.get(proxy.link) if latency is None: - # 检测失败 + # 检测失败 - 不入库，跳过 skipped += 1 - existing = await self.db.get_proxy_by_link(proxy.link) - if existing: - # 去重：若该节点原属于实例源，将 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: - # 去重：若该节点原属于实例源，将 source 转移到订阅源 - if existing.source.startswith("instance:"): - await self.db.update_proxy_source(existing.id, sub.url) + existing = await self.db.get_proxy_by_link(proxy.link) + if existing: + # 节点已存在于数据库中（可能属于其他订阅） + # 不更新、不转移，仅更新延迟 + if latency <= threshold: latency_updates.append((existing.id, latency)) - updated += 1 else: - success = await self.db.insert_proxy(proxy, latency, sub.url) + # 延迟超标 - 如果属于当前订阅则删除，否则仅跳过 + if existing.subscription_id == sub_id: + delete_ids.append(existing.id) + skipped += 1 + else: + # 新节点 + if latency <= threshold: + success = await self.db.insert_proxy(proxy, latency, sub_id) if success: added += 1 - else: - # 延迟超标 - skipped += 1 - existing = await self.db.get_proxy_by_link(proxy.link) - if existing: - # 去重：若该节点原属于实例源，将 source 转移到订阅源 - if existing.source.startswith("instance:"): - await self.db.update_proxy_source(existing.id, sub.url) - fail_ids.append(existing.id) + else: + skipped += 1 # 批量写入数据库 if latency_updates: await self.db.batch_update_latency(latency_updates) - if fail_ids: - await self.db.batch_increment_fail(fail_ids) + if delete_ids: + await self.db.batch_delete_proxies(delete_ids) # 有节点入库则重置空天数 if added > 0: @@ -208,9 +213,13 @@ async def _fetch_single_subscription(self, sub_id: int) -> None: await self.db.update_total_count(sub_id, len(proxies)) await self.db.update_fetch_status(sub_id, "success") - self._last_fetch_time = datetime.now(timezone.utc).isoformat() - logger.info("订阅 #%d 拉取完成: 新增 %d, 更新 %d, 跳过 %d, 总解析 %d", - sub.id, added, updated, skipped, len(proxies)) + self._last_fetch_time = datetime.now(timezone(timedelta(hours=8))).isoformat() + logger.info("订阅 #%d 拉取完成: 新增 %d, 更新延迟 %d, 跳过 %d, 总解析 %d", + sub.id, added, len(latency_updates), skipped, len(proxies)) + + # 拉取完成后，同时验证该订阅下所有已入库节点 + await self.verify_subscription_proxies(sub_id) + except Exception as e: await self.db.update_fetch_status(sub_id, "failed") logger.error("拉取订阅 #%d 异常: %s", sub.id, e, exc_info=True) @@ -264,71 +273,85 @@ async def verify_stored_proxies(self) -> None: results = await self.checker.check_batch(links) latency_updates = [] - fail_ids = [] + delete_ids = [] for proxy in proxies: latency = results.get(proxy.link) if latency is not None: latency_updates.append((proxy.id, latency)) else: - fail_ids.append(proxy.id) + # 检测失败直接删除 + delete_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) + if delete_ids: + await self.db.batch_delete_proxies(delete_ids) - self._last_verify_time = datetime.now(timezone.utc).isoformat() - logger.info("验证完成: 成功 %d, 失败 %d", len(latency_updates), len(fail_ids)) + self._last_verify_time = datetime.now(timezone(timedelta(hours=8))).isoformat() + logger.info("验证完成: 成功 %d, 删除 %d", len(latency_updates), len(delete_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: - """验证指定订阅源的节点可用性（并发检测 + 批量数据库写入）""" - sub = await self.db.get_subscription_by_id(sub_id) - if not sub: - logger.warning("订阅 #%d 不存在", sub_id) + """验证指定订阅源 ID 下所有已入库节点的可用性""" + if sub_id in self._verifying_subs: + logger.info("订阅 #%d 正在验证中，跳过本次", sub_id) return - proxies = await self.db.get_proxies_by_source(sub.url) - if not proxies: - logger.info("订阅 #%d 没有节点，跳过验证", sub_id) - return + self._verifying_subs.add(sub_id) + try: + sub = await self.db.get_subscription_by_id(sub_id) + if not sub: + logger.warning("订阅 #%d 不存在", sub_id) + return - logger.info("开始验证订阅 #%d 的 %d 个节点...", sub_id, len(proxies)) + # 按 subscription_id 查询该订阅下所有已入库节点 + proxies = await self.db.get_proxies_by_subscription_id(sub_id) + if not proxies: + logger.info("订阅 #%d 没有已入库节点，跳过验证", sub_id) + return - 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) + 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, + check_retries=self.config.check.check_retries, + ) + links = [p.link for p in proxies] + results = await sub_checker.check_batch(links) - latency_updates = [] - fail_ids = [] + latency_updates = [] + delete_ids = [] - for proxy in proxies: - latency = results.get(proxy.link) - if latency is not None: - latency_updates.append((proxy.id, latency)) - else: - fail_ids.append(proxy.id) + for proxy in proxies: + latency = results.get(proxy.link) + if latency is not None: + latency_updates.append((proxy.id, latency)) + else: + # 检测失败直接删除 + delete_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) + if latency_updates: + await self.db.batch_update_latency(latency_updates) + if delete_ids: + await self.db.batch_delete_proxies(delete_ids) - self._last_verify_time = datetime.now(timezone.utc).isoformat() - logger.info("订阅 #%d 验证完成: 成功 %d, 失败 %d", sub_id, len(latency_updates), len(fail_ids)) + self._last_verify_time = datetime.now(timezone(timedelta(hours=8))).isoformat() + logger.info("订阅 #%d 验证完成: 成功 %d, 删除 %d", sub_id, len(latency_updates), len(delete_ids)) + except Exception as e: + logger.error("订阅 #%d 验证异常: %s", sub_id, e, exc_info=True) + finally: + self._verifying_subs.discard(sub_id) # ---- 服务实例源管理 ---- @@ -347,6 +370,7 @@ def _add_instance_source_job(self, source) -> None: trigger = CronTrigger( minute=parts[0], hour=parts[1], day=parts[2], month=parts[3], day_of_week=parts[4], + jitter=random.randint(0, 600), # 随机延迟0~600秒(10分钟内) ) self.scheduler.add_job( self._fetch_single_instance_source, @@ -372,19 +396,11 @@ def refresh_instance_source_job(self, source) -> None: 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") @@ -396,77 +412,21 @@ async def _fetch_single_instance_source(self, source_id: int) -> None: # 缓存订阅地址列表，供手工导入时使用 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) + connected_count = len(proxies) + logger.info("实例源 #%d: 已连接 %d 个节点, 发现 %d 个订阅源", + source.id, connected_count, len(subscription_urls)) - # ---- 第二步：并发检测所有节点延迟 ---- - 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_total_count(source_id, connected_count) 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)) + self._last_fetch_time = datetime.now(timezone(timedelta(hours=8))).isoformat() 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("没有启用的服务实例源") @@ -515,26 +475,15 @@ async def import_instance_subscriptions(self, source_id: int) -> int: return new_sub_count async def cleanup_proxies(self) -> None: - """清理不合格节点和空订阅源 + """清理空订阅源 - - 订阅源节点：连续 3 次验证失败则移除（原逻辑不变） - - 实例源节点：last_success_time 为空或距今超过 7 天且 fail_count > 0 才移除 + 检测失败的节点已在验证时直接删除，无需按 fail_count 清理。 + 仅清理连续7天节点数为0的订阅源。 """ - # 清理订阅源节点：连续3次验证失败 - deleted = await self.db.delete_sub_proxies_by_fail_count(3) - if deleted > 0: - 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) - - # 清理连续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) + count = await self.db.get_proxy_count_by_subscription_id(sub.id) if count == 0: await self.db.delete_subscription(sub.id) self.remove_subscription_job(sub.id) diff --git a/app/templates/index.html b/app/templates/index.html index ecce0fa..5631331 100644 --- a/app/templates/index.html +++ b/app/templates/index.html @@ -5,34 +5,26 @@ {% block content %}
-
+
-
总节点数
+
订阅条目数

{{ stats.total }}

-
+
-
可用数
+
可用节点数

{{ stats.available }}

-
-
-
-
不可用
-

{{ stats.unavailable }}

-
-
-
-
+
-
平均延迟
+
平均延迟数

{{ "%.0f"|format(stats.avg_latency_ms) }} ms

@@ -142,7 +134,7 @@
协议分布
{{ sub.id }} {{ sub.crontab }} {{ sub.total_count }} - {{ sub_available_counts.get(sub.url, 0) }} + {{ sub_available_counts.get(sub.id, 0) }} {{ sub.latency_threshold }}ms {{ sub.max_concurrent }} @@ -226,7 +218,7 @@
协议分布
# 地址 - 总数 + 已连接 Crontab 状态 更新 @@ -290,8 +282,20 @@
可用节点
拉取所有订阅 + + + + + +
- + {% set available_subs = [] %} {% for sub in subscriptions %} -{% if sub_available_counts.get(sub.url, 0) > 0 %} +{% if sub_available_counts.get(sub.id, 0) > 0 %} {% if available_subs.append(sub) %}{% endif %} {% endif %} {% endfor %} -{% set available_insts = [] %} -{% for inst in instance_sources %} -{% if inst_available_counts.get(inst.id, 0) > 0 %} -{% if available_insts.append(inst) %}{% endif %} -{% endif %} -{% endfor %} -{% set has_any = available_subs or available_insts %} +{% set has_any = available_subs %} {% if has_any %}
{% for sub in available_subs %} -
+
-
- {% endfor %} - {% for inst in available_insts %} - {% set inst_source_tag = 'instance:' + inst.base_url %} -
-
- -
-
- - - - - - - - - - - - - - - {% for proxy in grouped.get(inst_source_tag, []) %} - - - - - - - - - - - {% endfor %} - {% if not grouped.get(inst_source_tag) %} - - {% endif %} - -
#名称协议地址端口延迟 入库时间 操作
{{ loop.index }}{{ proxy.name[:30] }}{% if proxy.name|length > 30 %}...{% endif %}{{ proxy.protocol }}{{ proxy.address }}{{ proxy.port }} - - {{ "%.0f"|format(proxy.latency_ms) }} ms - - {{ proxy.created_at[:19] if proxy.created_at else '-' }} - -
暂无可用节点，请先获取此实例源
-
+
{% endfor %}
@@ -603,8 +544,7 @@
编辑服务实例源
var cards = document.querySelectorAll('.stat-card h3'); if (cards[0]) cards[0].textContent = stats.total; if (cards[1]) cards[1].textContent = stats.available; - if (cards[2]) cards[2].textContent = stats.unavailable; - if (cards[3]) cards[3].textContent = Math.round(stats.avg_latency_ms) + ' ms'; + if (cards[2]) cards[2].textContent = Math.round(stats.avg_latency_ms) + ' ms'; }).catch(() => {}); }, 30000); @@ -738,8 +678,120 @@
编辑服务实例源
} catch (e) { alert('删除失败: ' + e.message); } } +async function deleteAllProxies() { + if (!confirm('确定清除数据库内所有已保存节点？此操作不可恢复！')) return; + try { + const resp = await fetch('/api/proxies', { method: 'DELETE' }); + const data = await resp.json(); + if (resp.ok) { + alert('已清除 ' + data.count + ' 个节点'); + location.reload(); + } else { + alert('清除失败'); + } + } catch (e) { alert('请求失败: ' + e.message); } +} + +function exportConfig() { + window.location.href = '/api/config/export'; +} + +async function importConfig(input) { + const file = input.files[0]; + if (!file) return; + try { + const text = await file.text(); + const config = JSON.parse(text); + const resp = await fetch('/api/config/import', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ config: config }), + }); + const data = await resp.json(); + if (resp.ok) { + alert(data.message); + location.reload(); + } else { + alert('导入失败: ' + (data.detail || '未知错误')); + } + } catch (e) { + alert('导入失败: ' + e.message); + } + // 重置 file input 以便重复选择同一文件 + input.value = ''; +} + function refreshPage() { location.reload(); } +// ---- 分页逻辑 ---- +const PAGE_SIZE = 10; +const pageState = {}; // { subId: currentPage } + +function initPagination() { + document.querySelectorAll('nav[id^="pager-"]').forEach(function(nav) { + var subId = nav.dataset.sub; + pageState[subId] = 1; + applyPagination(subId); + }); +} + +function applyPagination(subId) { + var tbody = document.querySelector('#tbl-' + subId + ' tbody'); + if (!tbody) return; + var rows = Array.from(tbody.querySelectorAll('tr[data-sub="' + subId + '"]')); + var total = rows.length; + var totalPages = Math.max(1, Math.ceil(total / PAGE_SIZE)); + var page = pageState[subId] || 1; + if (page > totalPages) page = totalPages; + pageState[subId] = page; + + // 显示/隐藏行 + rows.forEach(function(row, i) { + row.style.display = (i >= (page - 1) * PAGE_SIZE && i < page * PAGE_SIZE) ? '' : 'none'; + }); + + // 更新序号（仅当前页可见行） + var idx = (page - 1) * PAGE_SIZE; + rows.forEach(function(row) { + if (row.style.display !== 'none') { + row.children[0].textContent = ++idx; + } + }); + + // 渲染分页按钮 + var nav = document.getElementById('pager-' + subId); + if (!nav) return; + var ul = nav.querySelector('ul'); + if (total <= PAGE_SIZE) { ul.innerHTML = ''; return; } + + var html = ''; + html += '
  • ' + + '«
    • '; + + // 显示页码（最多显示5页，中间页码） + var startPage = Math.max(1, page - 2); + var endPage = Math.min(totalPages, startPage + 4); + if (endPage - startPage < 4) startPage = Math.max(1, endPage - 4); + + for (var p = startPage; p <= endPage; p++) { + html += '
  • ' + + '' + p + '
    • '; + } + + html += '
  • ' + + '»
    • '; + + ul.innerHTML = html; +} + +function goPage(subId, page) { + pageState[subId] = page; + applyPagination(subId); +} + +// 页面加载后初始化分页 +document.addEventListener('DOMContentLoaded', initPagination); + // 表格排序 function sortTable(tableId, key, th) { var table = document.getElementById(tableId); @@ -767,10 +819,10 @@
    编辑服务实例源
    } }); rows.forEach(function(row) { tbody.appendChild(row); }); - // 更新序号 - tbody.querySelectorAll('tr').forEach(function(row, i) { - row.children[0].textContent = i + 1; - }); + // 排序后重置到第1页并重新分页 + var subId = tableId.replace('tbl-', ''); + pageState[subId] = 1; + applyPagination(subId); } // ---- 检测目标 URL ---- diff --git a/config.yaml b/config.yaml index a51ac21..456a387 100644 --- a/config.yaml +++ b/config.yaml @@ -14,6 +14,7 @@ check: socks_port: 1080 # 本地转发端口（ConnectivityMonitor使用） http_port: 1081 # 本地HTTP转发端口 kernel_path: "xray" # 内核可执行文件路径 + check_retries: 2 # 单次检测失败后重试次数 scheduler: fetch_interval: 3600