From 007c9de345e4703edfb5464eb9e51f921d300ba6 Mon Sep 17 00:00:00 2001 From: siruoren Date: Fri, 17 Jul 2026 15:45:23 +0800 Subject: [PATCH 01/10] =?UTF-8?q?chore:=20=E6=B7=BB=E5=8A=A0resources/Subs?= =?UTF-8?q?cription.txt=E5=88=B0git=E5=BF=BD=E7=95=A5=E5=88=97=E8=A1=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将订阅文件目录加入git忽略,避免提交不必要的本地资源文件 --- .gitignore | 1 + 1 file changed, 1 insertion(+) create mode 100644 .gitignore diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..509f07a --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +resources/Subscription.txt From 909e71ebe9f4a89c5c10ab5356c8aab763d955ff Mon Sep 17 00:00:00 2001 From: siruoren Date: Fri, 17 Jul 2026 18:01:20 +0800 Subject: [PATCH 02/10] =?UTF-8?q?feat:=20=E5=88=9D=E5=A7=8B=E5=8C=96?= =?UTF-8?q?=E5=AE=8C=E6=95=B4=E7=9A=84=E4=BB=A3=E7=90=86=E6=B1=A0=E9=A1=B9?= =?UTF-8?q?=E7=9B=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 创建基础项目结构,包含FastAPI服务、数据库、检测、调度等核心模块 - 实现代理节点拉取、检测、入库、验证和清理全流程 - 提供Web管理界面和V2Ray/Clash格式订阅接口 - 支持Docker容器化部署和GitHub CI/CD - 添加.gitignore、配置文件、README文档等配套资源 --- .github/workflows/ci.yaml | 65 +++++++ .gitignore | 26 +++ Dockerfile | 22 +++ README.md | 134 ++++++++++++- app/__init__.py | 119 ++++++++++++ app/checker.py | 231 ++++++++++++++++++++++ app/config.py | 131 +++++++++++++ app/database.py | 207 ++++++++++++++++++++ app/generator.py | 326 ++++++++++++++++++++++++++++++++ app/main.py | 18 ++ app/models.py | 41 ++++ app/parser.py | 257 +++++++++++++++++++++++++ app/routers/__init__.py | 0 app/routers/api.py | 133 +++++++++++++ app/routers/web.py | 57 ++++++ app/scheduler.py | 193 +++++++++++++++++++ app/templates/base.html | 67 +++++++ app/templates/index.html | 179 ++++++++++++++++++ app/templates/subscription.html | 93 +++++++++ config.yaml | 25 +++ docker-compose.yaml | 18 ++ requirements.txt | 8 + resources/domain_check.txt | 2 + 23 files changed, 2350 insertions(+), 2 deletions(-) create mode 100644 .github/workflows/ci.yaml create mode 100644 Dockerfile create mode 100644 app/__init__.py create mode 100644 app/checker.py create mode 100644 app/config.py create mode 100644 app/database.py create mode 100644 app/generator.py create mode 100644 app/main.py create mode 100644 app/models.py create mode 100644 app/parser.py create mode 100644 app/routers/__init__.py create mode 100644 app/routers/api.py create mode 100644 app/routers/web.py create mode 100644 app/scheduler.py create mode 100644 app/templates/base.html create mode 100644 app/templates/index.html create mode 100644 app/templates/subscription.html create mode 100644 config.yaml create mode 100644 docker-compose.yaml create mode 100644 requirements.txt create mode 100644 resources/domain_check.txt diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml new file mode 100644 index 0000000..b4ae85e --- /dev/null +++ b/.github/workflows/ci.yaml @@ -0,0 +1,65 @@ +name: CI/CD + +on: + push: + branches: [main] + tags: ['v*'] + pull_request: + branches: [main] + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }} + +jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Install lint tools + run: pip install ruff + + - name: Lint + run: ruff check . + + build-and-push: + needs: lint + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + + steps: + - uses: actions/checkout@v4 + + - name: Login to GHCR + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Extract metadata + id: meta + uses: docker/metadata-action@v5 + with: + images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} + tags: | + type=ref,event=branch + type=semver,pattern={{version}} + type=sha + + - name: Build and push Docker image + uses: docker/build-push-action@v5 + with: + context: . + push: ${{ github.event_name != 'pull_request' }} + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + cache-from: type=gha + cache-to: type=gha,mode=max diff --git a/.gitignore b/.gitignore index 509f07a..88ddb90 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,27 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +dist/ +build/ +.eggs/ + +# 虚拟环境 +venv/ +.venv/ + +# 数据库 +data/ + +# IDE +.vscode/ +.idea/ + +# 资源文件(含敏感订阅地址) resources/Subscription.txt + +# OS +.DS_Store + +# 环境变量 +.env diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..36ff4b6 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,22 @@ +FROM python:3.11-slim + +WORKDIR /app + +# 安装系统依赖 +RUN apt-get update && apt-get install -y --no-install-recommends \ + gcc \ + && rm -rf /var/lib/apt/lists/* + +# 安装 Python 依赖 +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# 复制代码 +COPY . . + +# 创建数据目录 +RUN mkdir -p /app/data + +EXPOSE 8080 + +CMD ["python", "-m", "app.main"] diff --git a/README.md b/README.md index f2ecaba..b9f2793 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,132 @@ -# proxy_pool -proxy_pool +# ProxyPool - 代理池管理系统 + +自动获取、检测、维护代理节点池,提供 Web 界面和订阅链接。 + +## 功能特性 + +- 🔄 **定时拉取** - 自动从订阅源获取代理节点(vmess/vless/trojan/ss/hysteria2) +- ⚡ **延迟检测** - TCP/TLS 连通性检测,并发可配置 +- 📊 **自动入库** - 延迟低于阈值的代理自动入库,超标自动删除 +- ✅ **定时验证** - 定期验证已存代理可用性,连续失败自动清理 +- 🌐 **Web 界面** - 展示代理列表、延迟、状态和统计信息 +- 🔗 **订阅链接** - 提供 v2ray(base64)和 Clash(YAML)格式订阅 +- 🐳 **一键部署** - Docker Compose 部署,数据持久化 + +## 快速开始 + +### Docker Compose(推荐) + +```bash +# 克隆项目 +git clone https://github.com/your-username/proxy_pool.git +cd proxy_pool + +# 编辑订阅源 +vim resources/Subscription.txt + +# 按需修改配置 +vim config.yaml + +# 启动服务 +docker-compose up -d +``` + +访问 http://localhost:8080 查看 Web 界面。 + +### 本地运行 + +```bash +# 安装依赖 +pip install -r requirements.txt + +# 编辑订阅源 +vim resources/Subscription.txt + +# 启动服务 +python -m app.main +``` + +## 配置说明 + +编辑 `config.yaml` 进行配置: + +```yaml +server: + host: "0.0.0.0" + port: 8080 + +check: + timeout: 5.0 # 检测超时(秒) + max_concurrent: 50 # 并发检测数 + latency_threshold: 3000.0 # 延迟阈值(毫秒) + +scheduler: + fetch_interval: 3600 # 拉取订阅间隔(秒) + verify_interval: 1800 # 验证代理间隔(秒) + cleanup_interval: 7200 # 清理间隔(秒) + max_fail_count: 3 # 最大连续失败次数 +``` + +### 环境变量 + +| 变量 | 说明 | 默认值 | +|------|------|--------| +| `PROXY_POOL_PORT` | 服务端口 | 8080 | +| `PROXY_POOL_DB_PATH` | 数据库路径 | data/proxy_pool.db | + +## API 接口 + +| 方法 | 路径 | 说明 | +|------|------|------| +| GET | `/api/proxies` | 可用代理列表 | +| GET | `/api/proxies/all` | 所有代理 | +| DELETE | `/api/proxies/{id}` | 删除代理 | +| GET | `/api/subscription/v2ray` | V2Ray 订阅 | +| GET | `/api/subscription/clash` | Clash 订阅 | +| POST | `/api/fetch` | 手动拉取订阅 | +| POST | `/api/verify` | 手动验证代理 | +| GET | `/api/stats` | 统计信息 | +| GET | `/api/health` | 健康检查 | + +## 订阅链接使用 + +在代理客户端中添加以下订阅链接: + +- **V2Ray**: `http://your-server:8080/api/subscription/v2ray` +- **Clash**: `http://your-server:8080/api/subscription/clash` + +## 支持协议 + +| 协议 | 订阅解析 | Clash 生成 | +|------|----------|-----------| +| VMess | ✅ | ✅ | +| VLESS | ✅ | ✅ | +| Trojan | ✅ | ✅ | +| Shadowsocks | ✅ | ✅ | +| Hysteria2 | ✅ | ✅ | + +## 项目结构 + +``` +proxy_pool/ +├── app/ +│ ├── __init__.py # 应用工厂 +│ ├── main.py # 启动入口 +│ ├── config.py # 配置加载 +│ ├── database.py # 数据库操作 +│ ├── models.py # 数据模型 +│ ├── parser.py # 订阅解析 +│ ├── checker.py # 代理检测 +│ ├── generator.py # 订阅生成 +│ ├── scheduler.py # 定时任务 +│ ├── routers/ # 路由 +│ └── templates/ # 模板 +├── resources/ # 资源文件 +├── config.yaml # 配置文件 +├── docker-compose.yaml # Docker 编排 +└── Dockerfile # Docker 镜像 +``` + +## License + +MIT License diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..c440949 --- /dev/null +++ b/app/__init__.py @@ -0,0 +1,119 @@ +from __future__ import annotations +"""应用工厂 - 创建 FastAPI 实例并整合各模块""" + +import logging +from pathlib import Path + +from fastapi import FastAPI +from fastapi.templating import Jinja2Templates + +from app.checker import ProxyChecker +from app.config import AppConfig, load_config +from app.database import ProxyDatabase +from app.parser import load_check_urls +from app.scheduler import TaskScheduler + +logger = logging.getLogger(__name__) + +# 全局单例 +_config: AppConfig | None = None +_db: ProxyDatabase | None = None +_checker: ProxyChecker | None = None +_scheduler: TaskScheduler | None = None +_templates: Jinja2Templates | None = None + + +def get_config() -> AppConfig: + return _config + + +def get_db() -> ProxyDatabase: + return _db + + +def get_checker() -> ProxyChecker: + return _checker + + +def get_scheduler() -> TaskScheduler: + return _scheduler + + +def get_templates() -> Jinja2Templates: + return _templates + + +def create_app(config_path: str = "config.yaml") -> FastAPI: + """创建 FastAPI 应用实例""" + global _config, _db, _checker, _scheduler, _templates + + # 加载配置 + _config = load_config(config_path) + + # 配置日志 + log_level = logging.DEBUG if _config.server.debug else logging.INFO + logging.basicConfig( + level=log_level, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + # 初始化组件 + _db = ProxyDatabase(_config.database.path) + + # 初始化模板 + template_dir = Path(__file__).parent / "templates" + _templates = Jinja2Templates(directory=str(template_dir)) + + # 创建 FastAPI 实例 + app = FastAPI( + title="ProxyPool", + version="1.0.0", + description="代理池管理系统", + ) + + # 注册生命周期事件 + @app.on_event("startup") + async def startup(): + global _checker, _scheduler + + # 初始化数据库 + await _db.init() + logger.info("数据库初始化完成: %s", _config.database.path) + + # 加载检测 URL + check_urls = await load_check_urls(_config.resources.domain_check_file) + logger.info("检测 URL: %s", check_urls) + + # 初始化检测器 + _checker = ProxyChecker( + check_urls=check_urls, + timeout=_config.check.timeout, + max_concurrent=_config.check.max_concurrent, + ) + + # 初始化调度器 + _scheduler = TaskScheduler(_config, _db, _checker) + _scheduler.start() + + # 启动后立即执行一次拉取 + logger.info("启动首次拉取任务...") + import asyncio + asyncio.create_task(_scheduler.fetch_and_check()) + + @app.on_event("shutdown") + async def shutdown(): + if _scheduler: + _scheduler.shutdown() + if _db: + await _db.close() + logger.info("应用已关闭") + + # 注册路由 + from app.routers.api import router as api_router + from app.routers.web import router as web_router + + app.include_router(api_router) + app.include_router(web_router) + + return app diff --git a/app/checker.py b/app/checker.py new file mode 100644 index 0000000..5d2df74 --- /dev/null +++ b/app/checker.py @@ -0,0 +1,231 @@ +from __future__ import annotations +"""代理延迟检测模块 - 基于 asyncio TCP/TLS 连通性测试 + +检测策略: +- 使用 asyncio.open_connection 建立到 address:port 的 TCP 连接 +- 若协议需要 TLS,再通过 asyncio.start_tls 完成 TLS 握手 +- 总延迟 = TCP 时间 + (TLS 时间 if applicable) +- 使用 asyncio.Semaphore 控制并发数 +""" + +import asyncio +import json +import logging +import ssl +import time +import base64 +from urllib.parse import urlparse, parse_qs + +from app.models import ProxyInfo + +logger = logging.getLogger(__name__) + + +class ProxyChecker: + """基于 asyncio 的并发代理延迟检测器""" + + def __init__(self, check_urls: list[str], timeout: float, max_concurrent: int): + self.check_urls = check_urls + self.timeout = timeout + self.semaphore = asyncio.Semaphore(max_concurrent) + + async def check_proxy(self, link: str) -> float | None: + """检测单个代理的延迟 + + 解析分享链接获取 address/port/tls 信息, + 使用 TCP/TLS 连通性测试测量延迟。 + 返回延迟毫秒数,失败返回 None。 + """ + info = self._parse_link_for_check(link) + if not info: + return None + + address = info["address"] + port = info["port"] + use_tls = info["use_tls"] + + if not address or not port: + return None + + async with self.semaphore: + return await self._tcp_latency_check(address, int(port), use_tls, self.timeout) + + async def check_batch(self, links: list[str]) -> dict[str, float | None]: + """批量并发检测,返回 {link: latency_ms | None}""" + tasks = [self.check_proxy(link) for link in links] + results = await asyncio.gather(*tasks, return_exceptions=True) + + output = {} + for link, result in zip(links, results): + if isinstance(result, Exception): + logger.debug("检测异常 %s: %s", link[:50], result) + output[link] = None + else: + output[link] = result + return output + + async def check_proxy_infos(self, proxies: list[ProxyInfo]) -> dict[str, float | None]: + """批量检测 ProxyInfo 列表,返回 {link: latency_ms | None}""" + links = [p.link for p in proxies] + return await self.check_batch(links) + + async def _tcp_latency_check( + self, address: str, port: int, use_tls: bool, timeout: float + ) -> float | None: + """TCP 层延迟检测 + + 1. asyncio.open_connection 测 TCP 握手时间 + 2. 若需要 TLS,asyncio.start_tls 测 TLS 握手时间 + 3. 返回总延迟(毫秒) + """ + start = time.monotonic() + try: + reader, writer = await asyncio.wait_for( + asyncio.open_connection(address, port), timeout=timeout + ) + + if use_tls: + ssl_context = ssl.create_default_context() + ssl_context.check_hostname = False + ssl_context.verify_mode = ssl.CERT_NONE + + # 使用 asyncio.get_event_loop().start_tls 进行 TLS 握手 + loop = asyncio.get_event_loop() + transport = reader.transport + try: + new_transport = await asyncio.wait_for( + loop.start_tls(transport, transport.get_protocol(), ssl_context), + timeout=timeout, + ) + # 清理新 transport + new_transport.close() + except Exception: + pass + + elapsed = (time.monotonic() - start) * 1000 + writer.close() + try: + await writer.wait_closed() + except Exception: + pass + return round(elapsed, 1) + except Exception: + return None + + def _parse_link_for_check(self, link: str) -> dict | None: + """从分享链接中提取检测所需信息:address, port, use_tls""" + try: + if link.startswith("vmess://"): + return self._parse_vmess_for_check(link) + elif link.startswith("vless://"): + return self._parse_vless_for_check(link) + elif link.startswith("trojan://"): + return self._parse_trojan_for_check(link) + elif link.startswith("ss://"): + return self._parse_ss_for_check(link) + elif link.startswith("hysteria2://") or link.startswith("hy2://"): + return self._parse_hysteria2_for_check(link) + except Exception: + pass + return None + + def _parse_vmess_for_check(self, link: str) -> dict | None: + """解析 vmess 链接获取 address/port/tls""" + try: + config_b64 = link[8:] + 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) + use_tls = config.get("tls", "") == "tls" + return { + "address": config.get("add", ""), + "port": str(config.get("port", "")), + "use_tls": use_tls, + } + except Exception: + return None + + def _parse_vless_for_check(self, link: str) -> dict | None: + """解析 vless 链接获取 address/port/tls""" + try: + parsed = urlparse(link) + query = parse_qs(parsed.query) + security = query.get("security", ["none"])[0] + use_tls = security in ("tls", "reality") + return { + "address": parsed.hostname or "", + "port": str(parsed.port) if parsed.port else "", + "use_tls": use_tls, + } + except Exception: + return None + + def _parse_trojan_for_check(self, link: str) -> dict | None: + """解析 trojan 链接 - 始终使用 TLS""" + try: + parsed = urlparse(link) + return { + "address": parsed.hostname or "", + "port": str(parsed.port) if parsed.port else "", + "use_tls": True, + } + except Exception: + return None + + def _parse_ss_for_check(self, link: str) -> dict | None: + """解析 ss 链接 - 通常不使用 TLS""" + try: + # 去掉 fragment + line = link + if "#" in line: + line = line[: line.rindex("#")] + + ss_content = line[5:] # 去掉 'ss://' + address = "" + port = "" + + if "@" in ss_content: + at_idx = ss_content.rindex("@") + addr_port = ss_content[at_idx + 1:] + if addr_port.startswith("["): + bracket_end = addr_port.index("]") + address = addr_port[1:bracket_end] + port = addr_port[bracket_end + 2:] if bracket_end + 2 < len(addr_port) else "" + elif ":" in addr_port: + address, port = addr_port.rsplit(":", 1) + else: + try: + padding = 4 - len(ss_content) % 4 + if padding != 4: + ss_content += "=" * padding + decoded = base64.b64decode(ss_content).decode("utf-8") + if "@" in decoded: + _, addr_port = decoded.rsplit("@", 1) + if ":" in addr_port: + address, port = addr_port.rsplit(":", 1) + except Exception: + pass + + return { + "address": address, + "port": port, + "use_tls": False, + } + except Exception: + return None + + def _parse_hysteria2_for_check(self, link: str) -> dict | None: + """解析 hysteria2 链接 - 始终使用 TLS(QUIC)""" + try: + prefix_len = len("hysteria2://") if link.startswith("hysteria2://") else len("hy2://") + rest = link[prefix_len:] + parsed = urlparse("http://" + rest) + return { + "address": parsed.hostname or "", + "port": str(parsed.port) if parsed.port else "", + "use_tls": True, + } + except Exception: + return None diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..659a62c --- /dev/null +++ b/app/config.py @@ -0,0 +1,131 @@ +"""配置加载模块 - 从 YAML 文件加载配置""" + +import os +from dataclasses import dataclass, field + +import yaml + + +@dataclass +class ServerConfig: + host: str = "0.0.0.0" + port: int = 8080 + debug: bool = False + + +@dataclass +class DatabaseConfig: + path: str = "data/proxy_pool.db" + + +@dataclass +class CheckConfig: + urls: list = field(default_factory=lambda: [ + "http://www.google.com/generate_204", + "http://www.gstatic.com/generate_204", + ]) + timeout: float = 5.0 + max_concurrent: int = 50 + latency_threshold: float = 1500.0 + + +@dataclass +class SchedulerConfig: + fetch_interval: int = 3600 # 拉取订阅间隔(秒) + verify_interval: int = 1800 # 验证代理间隔(秒) + cleanup_interval: int = 7200 # 清理间隔(秒) + max_fail_count: int = 3 # 最大连续失败次数 + + +@dataclass +class ResourcesConfig: + subscription_file: str = "resources/Subscription.txt" + domain_check_file: str = "resources/domain_check.txt" + + +@dataclass +class AppConfig: + server: ServerConfig = field(default_factory=ServerConfig) + database: DatabaseConfig = field(default_factory=DatabaseConfig) + check: CheckConfig = field(default_factory=CheckConfig) + scheduler: SchedulerConfig = field(default_factory=SchedulerConfig) + resources: ResourcesConfig = field(default_factory=ResourcesConfig) + + +def _deep_merge(base: dict, override: dict) -> dict: + """深度合并字典,override 覆盖 base""" + result = base.copy() + for key, value in override.items(): + if key in result and isinstance(result[key], dict) and isinstance(value, dict): + result[key] = _deep_merge(result[key], value) + else: + result[key] = value + return result + + +def _dict_to_config(data: dict) -> AppConfig: + """将字典转为 AppConfig dataclass""" + server = ServerConfig(**data.get("server", {})) + database = DatabaseConfig(**data.get("database", {})) + check_data = data.get("check", {}) + default_check = CheckConfig() + check = CheckConfig( + urls=check_data.get("urls", default_check.urls), + timeout=check_data.get("timeout", default_check.timeout), + max_concurrent=check_data.get("max_concurrent", default_check.max_concurrent), + latency_threshold=check_data.get("latency_threshold", default_check.latency_threshold), + ) + scheduler = SchedulerConfig(**data.get("scheduler", {})) + resources = ResourcesConfig(**data.get("resources", {})) + return AppConfig( + server=server, + database=database, + check=check, + scheduler=scheduler, + resources=resources, + ) + + +def load_config(path: str = "config.yaml") -> AppConfig: + """从 YAML 文件加载配置,不存在则使用默认值""" + defaults = { + "server": {"host": "0.0.0.0", "port": 8080, "debug": False}, + "database": {"path": "data/proxy_pool.db"}, + "check": { + "urls": [ + "http://www.google.com/generate_204", + "http://www.gstatic.com/generate_204", + ], + "timeout": 5.0, + "max_concurrent": 50, + "latency_threshold": 3000.0, + }, + "scheduler": { + "fetch_interval": 3600, + "verify_interval": 1800, + "cleanup_interval": 7200, + "max_fail_count": 3, + }, + "resources": { + "subscription_file": "resources/Subscription.txt", + "domain_check_file": "resources/domain_check.txt", + }, + } + + if os.path.exists(path): + with open(path, "r", encoding="utf-8") as f: + file_data = yaml.safe_load(f) or {} + merged = _deep_merge(defaults, file_data) + else: + merged = defaults + + # 环境变量覆盖 + env_port = os.environ.get("PROXY_POOL_PORT") + if env_port: + merged["server"]["port"] = int(env_port) + + env_db_path = os.environ.get("PROXY_POOL_DB_PATH") + if env_db_path: + merged["database"]["path"] = env_db_path + + return _dict_to_config(merged) diff --git a/app/database.py b/app/database.py new file mode 100644 index 0000000..b298f15 --- /dev/null +++ b/app/database.py @@ -0,0 +1,207 @@ +from __future__ import annotations +"""数据库操作层 - aiosqlite 异步封装""" + +import os +from datetime import datetime, timezone + +import aiosqlite + +from app.models import ProxyDBRecord, ProxyInfo + + +class ProxyDatabase: + """代理数据库异步操作层""" + + def __init__(self, db_path: str): + self.db_path = db_path + self._db: aiosqlite.Connection | None = None + + async def init(self) -> None: + """初始化数据库连接和表结构""" + # 确保数据目录存在 + db_dir = os.path.dirname(self.db_path) + if db_dir: + os.makedirs(db_dir, exist_ok=True) + + self._db = await aiosqlite.connect(self.db_path) + self._db.row_factory = aiosqlite.Row + + 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 '' + ); + + 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); + """) + await self._db.commit() + + async def close(self) -> None: + """关闭数据库连接""" + if self._db: + await self._db.close() + self._db = None + + def _row_to_record(self, row: aiosqlite.Row) -> ProxyDBRecord: + """将数据库行转为 ProxyDBRecord""" + return ProxyDBRecord( + id=row["id"], + protocol=row["protocol"], + name=row["name"], + address=row["address"], + port=row["port"], + link=row["link"], + latency_ms=row["latency_ms"], + fail_count=row["fail_count"], + source=row["source"], + 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: + """插入新代理,link 唯一约束,重复则忽略。返回是否插入成功""" + now = datetime.now(timezone.utc).isoformat() + try: + await self._db.execute( + """INSERT OR IGNORE INTO proxies + (protocol, name, address, port, link, latency_ms, fail_count, source, + 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), + ) + await self._db.commit() + return self._db.changes > 0 + except aiosqlite.IntegrityError: + return False + + async def update_latency(self, proxy_id: int, latency_ms: float) -> None: + """更新延迟并重置 fail_count""" + now = datetime.now(timezone.utc).isoformat() + await self._db.execute( + """UPDATE proxies + SET latency_ms = ?, fail_count = 0, + last_check_time = ?, last_success_time = ? + WHERE id = ?""", + (latency_ms, now, now, proxy_id), + ) + 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 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,) + ) + 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" + ) + rows = await cursor.fetchall() + 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 + 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 查询代理""" + cursor = await self._db.execute( + "SELECT * FROM proxies WHERE link = ?", (link,) + ) + 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) + rows = await cursor.fetchall() + return [self._row_to_record(row) for row in rows] + + async def get_stats(self) -> dict: + """获取统计信息""" + cursor = await self._db.execute("SELECT COUNT(*) as total FROM proxies") + total = (await cursor.fetchone())["total"] + + cursor = await self._db.execute( + """SELECT COUNT(*) as available FROM proxies + WHERE latency_ms > 0 AND fail_count = 0""" + ) + available = (await cursor.fetchone())["available"] + + cursor = await self._db.execute( + "SELECT AVG(latency_ms) as avg_latency FROM proxies WHERE latency_ms > 0" + ) + avg_latency = (await cursor.fetchone())["avg_latency"] or 0 + + cursor = await self._db.execute( + "SELECT protocol, COUNT(*) as count FROM proxies GROUP BY protocol" + ) + rows = await cursor.fetchall() + protocol_dist = {row["protocol"]: row["count"] for row in rows} + + return { + "total": total, + "available": available, + "unavailable": total - available, + "avg_latency_ms": round(avg_latency, 1), + "protocol_distribution": protocol_dist, + } + + async def get_last_check_time(self) -> str: + """获取最近一次检测时间""" + cursor = await self._db.execute( + """SELECT MAX(last_check_time) as latest FROM proxies + WHERE last_check_time != ''""" + ) + row = await cursor.fetchone() + return row["latest"] or "" diff --git a/app/generator.py b/app/generator.py new file mode 100644 index 0000000..97ac984 --- /dev/null +++ b/app/generator.py @@ -0,0 +1,326 @@ +from __future__ import annotations +"""订阅链接生成模块 - 生成 v2ray 和 Clash 格式订阅内容""" + +import base64 +import json +import logging +from urllib.parse import urlparse, unquote, parse_qs + +import yaml + +from app.models import ProxyDBRecord + +logger = logging.getLogger(__name__) + + +def generate_v2ray_subscription(proxies: list[ProxyDBRecord]) -> str: + """生成 v2ray 格式订阅内容 + + 将所有代理的原始 link 换行拼接后 base64 编码 + """ + links = [p.link for p in proxies] + content = "\n".join(links) + return base64.b64encode(content.encode("utf-8")).decode("utf-8") + + +def generate_clash_subscription(proxies: list[ProxyDBRecord]) -> str: + """生成 Clash 格式订阅内容 + + 生成完整 Clash 配置:proxies + proxy-groups + """ + proxy_list = [] + proxy_names = [] + + for proxy in proxies: + clash_proxy = _link_to_clash_proxy(proxy.link, proxy.name) + if clash_proxy: + proxy_list.append(clash_proxy) + proxy_names.append(clash_proxy.get("name", proxy.name)) + + config = { + "mixed-port": 7890, + "allow-lan": False, + "mode": "rule", + "log-level": "info", + "proxies": proxy_list, + "proxy-groups": [ + { + "name": "ProxyPool", + "type": "url-test", + "proxies": proxy_names if proxy_names else ["DIRECT"], + "url": "http://www.gstatic.com/generate_204", + "interval": 300, + }, + { + "name": "Proxy", + "type": "select", + "proxies": ["ProxyPool", "DIRECT"] + proxy_names, + }, + ], + "rules": [ + "MATCH,Proxy", + ], + } + + return yaml.dump(config, default_flow_style=False, allow_unicode=True) + + +def _link_to_clash_proxy(link: str, fallback_name: str) -> dict | None: + """将分享链接转为 Clash proxy 字典""" + try: + if link.startswith("vmess://"): + return _vmess_link_to_clash(link, fallback_name) + elif link.startswith("vless://"): + return _vless_link_to_clash(link, fallback_name) + elif link.startswith("trojan://"): + return _trojan_link_to_clash(link, fallback_name) + elif link.startswith("ss://"): + return _ss_link_to_clash(link, fallback_name) + elif link.startswith("hysteria2://") or link.startswith("hy2://"): + return _hysteria2_link_to_clash(link, fallback_name) + except Exception as e: + logger.debug("转换 Clash 格式失败: %s - %s", link[:50], e) + return None + + +def _vmess_link_to_clash(link: str, fallback_name: str) -> dict: + """vmess:// 转 Clash proxy""" + 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")) + + proxy = { + "name": config.get("ps", fallback_name) or fallback_name, + "type": "vmess", + "server": config.get("add", ""), + "port": int(config.get("port", 443)), + "uuid": config.get("id", ""), + "alterId": int(config.get("aid", 0)), + "cipher": config.get("scy", "auto"), + } + + network = config.get("net", "tcp") + if network == "ws": + proxy["network"] = "ws" + ws_opts = {} + if config.get("path"): + ws_opts["path"] = config["path"] + if config.get("host"): + ws_opts["headers"] = {"Host": config["host"]} + if ws_opts: + proxy["ws-opts"] = ws_opts + elif network == "grpc": + proxy["network"] = "grpc" + if config.get("path"): + proxy["grpc-opts"] = {"grpc-service-name": config["path"]} + elif network == "h2": + proxy["network"] = "h2" + h2_opts = {} + if config.get("path"): + h2_opts["path"] = config["path"] + if config.get("host"): + h2_opts["host"] = [config["host"]] + if h2_opts: + proxy["h2-opts"] = h2_opts + + if config.get("tls") == "tls": + proxy["tls"] = True + if config.get("sni"): + proxy["servername"] = config["sni"] + + return proxy + + +def _vless_link_to_clash(link: str, fallback_name: str) -> dict: + """vless:// 转 Clash proxy""" + parsed = urlparse(link) + query = parse_qs(parsed.query) + + name = unquote(parsed.fragment) if parsed.fragment else fallback_name + uuid = unquote(parsed.username) if parsed.username else "" + + proxy = { + "name": name or fallback_name, + "type": "vless", + "server": parsed.hostname or "", + "port": parsed.port or 443, + "uuid": uuid, + } + + security = query.get("security", ["none"])[0] + if security in ("tls", "reality"): + proxy["tls"] = True + sni = query.get("sni", [None])[0] + if sni: + proxy["servername"] = sni + if security == "reality": + proxy["reality-opts"] = { + "public-key": query.get("pbk", [""])[0], + "short-id": query.get("sid", [""])[0], + } + if query.get("fp", [None])[0]: + proxy["client-fingerprint"] = query["fp"][0] + + network = query.get("type", ["tcp"])[0] + if network == "ws": + proxy["network"] = "ws" + ws_opts = {} + if query.get("path", [None])[0]: + ws_opts["path"] = query["path"][0] + if query.get("host", [None])[0]: + ws_opts["headers"] = {"Host": query["host"][0]} + if ws_opts: + proxy["ws-opts"] = ws_opts + elif network == "grpc": + proxy["network"] = "grpc" + if query.get("serviceName", [None])[0]: + proxy["grpc-opts"] = {"grpc-service-name": query["serviceName"][0]} + + flow = query.get("flow", [None])[0] + if flow: + proxy["flow"] = flow + + return proxy + + +def _trojan_link_to_clash(link: str, fallback_name: str) -> dict: + """trojan:// 转 Clash proxy""" + parsed = urlparse(link) + query = parse_qs(parsed.query) + + name = unquote(parsed.fragment) if parsed.fragment else fallback_name + password = unquote(parsed.username) if parsed.username else "" + + proxy = { + "name": name or fallback_name, + "type": "trojan", + "server": parsed.hostname or "", + "port": parsed.port or 443, + "password": password, + } + + sni = query.get("sni", [None])[0] + if sni: + proxy["sni"] = sni + + network = query.get("type", ["tcp"])[0] + if network == "ws": + proxy["network"] = "ws" + ws_opts = {} + if query.get("path", [None])[0]: + ws_opts["path"] = query["path"][0] + if query.get("host", [None])[0]: + ws_opts["headers"] = {"Host": query["host"][0]} + if ws_opts: + proxy["ws-opts"] = ws_opts + elif network == "grpc": + proxy["network"] = "grpc" + if query.get("serviceName", [None])[0]: + proxy["grpc-opts"] = {"grpc-service-name": query["serviceName"][0]} + + return proxy + + +def _ss_link_to_clash(link: str, fallback_name: str) -> dict: + """ss:// 转 Clash proxy""" + # 去掉 fragment + line = link + name = fallback_name + if "#" in line: + name = unquote(line[line.rindex("#") + 1:]) or fallback_name + line = line[: line.rindex("#")] + + ss_content = line[5:] # 去掉 'ss://' + cipher = "" + password = "" + address = "" + port = "" + + if "@" in ss_content: + # SIP002 格式 + 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 = addr_port[bracket_end + 2:] if bracket_end + 2 < len(addr_port) else "" + elif ":" in addr_port: + address, port = addr_port.rsplit(":", 1) + 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 = addr_port[bracket_end + 2:] if bracket_end + 2 < len(addr_port) else "" + elif ":" in addr_port: + address, port = addr_port.rsplit(":", 1) + except Exception: + pass + + return { + "name": name, + "type": "ss", + "server": address, + "port": int(port) if port else 443, + "cipher": cipher, + "password": password, + } + + +def _hysteria2_link_to_clash(link: str, fallback_name: str) -> dict: + """hysteria2:// 或 hy2:// 转 Clash proxy""" + prefix_len = len("hysteria2://") if link.startswith("hysteria2://") else len("hy2://") + rest = link[prefix_len:] + + name = fallback_name + if "#" in rest: + frag_start = rest.rindex("#") + name = unquote(rest[frag_start + 1:]) or fallback_name + rest = rest[:frag_start] + + # 构造标准 URL 以便解析 + 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 "" + + proxy = { + "name": name, + "type": "hysteria2", + "server": parsed.hostname or "", + "port": parsed.port or 443, + } + + if password: + proxy["password"] = password + + sni = query.get("sni", [None])[0] + if sni: + proxy["sni"] = sni + + return proxy diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..5276bcb --- /dev/null +++ b/app/main.py @@ -0,0 +1,18 @@ +"""ProxyPool 启动入口""" + +import uvicorn + +from app import create_app + +app = create_app() + +if __name__ == "__main__": + from app import get_config + + config = get_config() + uvicorn.run( + "app.main:app", + host=config.server.host, + port=config.server.port, + reload=config.server.debug, + ) diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..0f1c08e --- /dev/null +++ b/app/models.py @@ -0,0 +1,41 @@ +"""ProxyPool 数据模型""" + +from dataclasses import dataclass + + +@dataclass +class ProxyInfo: + """从订阅解析出的代理信息""" + + protocol: str # vmess / vless / trojan / ss / hysteria2 + name: str # 节点名称 + address: str # 服务器地址 + port: str # 端口 + link: str # 原始分享链接 + + +@dataclass +class ProxyDBRecord: + """数据库记录模型,与 proxies 表一一对应""" + + id: int + protocol: str + name: str + address: str + port: str + link: str + latency_ms: float # 延迟毫秒,-1=未检测 + fail_count: int # 连续失败次数 + source: str # 来源订阅 URL + last_check_time: str # ISO8601 + last_success_time: str # ISO8601 + created_at: str # ISO8601 + + @property + def status(self) -> str: + """代理状态:available / unavailable / unchecked""" + if self.latency_ms < 0: + return "unchecked" + if self.fail_count > 0: + return "unavailable" + return "available" diff --git a/app/parser.py b/app/parser.py new file mode 100644 index 0000000..a3a749a --- /dev/null +++ b/app/parser.py @@ -0,0 +1,257 @@ +from __future__ import annotations +"""订阅解析模块 - 拉取和解析代理订阅链接 + +核心解析逻辑移植自 Proxy_List/get_connected_proxies/get_connected_proxies.py +支持 vmess / vless / trojan / ss / hysteria2 五种协议 +""" + +import base64 +import json +import logging +from urllib.parse import urlparse, unquote + +import aiohttp + +from app.models import ProxyInfo + +logger = logging.getLogger(__name__) + + +async def fetch_subscription(url: str, timeout: float = 15.0) -> str: + """异步拉取订阅 URL 内容""" + async with aiohttp.ClientSession() as session: + async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as resp: + resp.raise_for_status() + return await resp.text() + + +def parse_subscription(content: str) -> list[ProxyInfo]: + """解析订阅内容,返回 ProxyInfo 列表 + + 支持 vmess / vless / trojan / ss / hysteria2 协议 + 自动检测 base64 编码的订阅内容并解码 + """ + share_links: list[ProxyInfo] = [] + lines = content.strip().split("\n") + + # 检测是否为 base64 编码的订阅内容 + has_protocol_prefix = any( + line.strip().startswith(( + "vmess://", "vless://", "trojan://", "ss://", + "hysteria2://", "hy2://", + )) + for line in lines if line.strip() + ) + + if not has_protocol_prefix: + try: + decoded = base64.b64decode(content.strip() + "===").decode("utf-8") + lines = decoded.strip().split("\n") + logger.info("检测到 base64 编码内容,已解码") + except Exception: + lines = content.strip().split("\n") + + for line in lines: + line = line.strip() + if not line: + continue + + if line.startswith("vmess://"): + info = _parse_vmess(line) + elif line.startswith("vless://"): + info = _parse_vless(line) + elif line.startswith("trojan://"): + info = _parse_trojan(line) + elif line.startswith("ss://"): + info = _parse_ss(line) + elif line.startswith("hysteria2://") or line.startswith("hy2://"): + info = _parse_hysteria2(line) + else: + info = None + + if info: + share_links.append(info) + + return share_links + + +def _parse_vmess(line: str) -> ProxyInfo | None: + """解析 vmess:// 链接""" + try: + config_b64 = line[8:] + 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) + return ProxyInfo( + protocol="vmess", + name=config.get("ps", ""), + address=config.get("add", ""), + port=str(config.get("port", "")), + link=line, + ) + except Exception: + return None + + +def _parse_vless(line: str) -> ProxyInfo | None: + """解析 vless:// 链接""" + try: + parsed = urlparse(line) + name = unquote(parsed.fragment) if parsed.fragment else "" + address = parsed.hostname or "" + port = str(parsed.port) if parsed.port else "" + return ProxyInfo( + protocol="vless", + name=name, + address=address, + port=port, + link=line, + ) + except Exception: + return None + + +def _parse_trojan(line: str) -> ProxyInfo | None: + """解析 trojan:// 链接""" + try: + parsed = urlparse(line) + name = unquote(parsed.fragment) if parsed.fragment else "" + address = parsed.hostname or "" + port = str(parsed.port) if parsed.port else "" + return ProxyInfo( + protocol="trojan", + name=name, + address=address, + port=port, + link=line, + ) + except Exception: + return None + + +def _parse_ss(line: str) -> ProxyInfo | None: + """解析 ss:// 链接(SIP002 和传统格式)""" + try: + name = "" + line_for_parse = line + if "#" in line: + frag_start = line.rindex("#") + name = unquote(line[frag_start + 1:]) + line_for_parse = line[:frag_start] + + ss_content = line_for_parse[5:] # 去掉 'ss://' + address = "" + port = "" + + if "@" in ss_content: + # SIP002 格式: ss://base64(method:password)@address:port + at_idx = ss_content.rindex("@") + addr_port = ss_content[at_idx + 1:] + if addr_port.startswith("["): + bracket_end = addr_port.index("]") + address = addr_port[1:bracket_end] + port = addr_port[bracket_end + 2:] if bracket_end + 2 < len(addr_port) else "" + elif ":" in addr_port: + address, port = addr_port.rsplit(":", 1) + else: + address = addr_port + else: + # 传统格式: ss://base64(method:password@address:port) + try: + padding = 4 - len(ss_content) % 4 + if padding != 4: + ss_content_padded = ss_content + "=" * padding + else: + ss_content_padded = ss_content + decoded = base64.b64decode(ss_content_padded).decode("utf-8") + if "@" in decoded: + _, addr_port = decoded.rsplit("@", 1) + if addr_port.startswith("["): + bracket_end = addr_port.index("]") + address = addr_port[1:bracket_end] + port = addr_port[bracket_end + 2:] if bracket_end + 2 < len(addr_port) else "" + elif ":" in addr_port: + address, port = addr_port.rsplit(":", 1) + else: + address = addr_port + except Exception: + pass + + return ProxyInfo( + protocol="ss", + name=name, + address=address, + port=port, + link=line, + ) + except Exception: + return None + + +def _parse_hysteria2(line: str) -> ProxyInfo | None: + """解析 hysteria2:// 或 hy2:// 链接""" + try: + prefix_len = len("hysteria2://") if line.startswith("hysteria2://") else len("hy2://") + rest = line[prefix_len:] + # 构造标准 URL 以便 urlparse 解析 + parsed = urlparse("http://" + rest) + name = "" + if "#" in rest: + name = unquote(rest.split("#")[-1]) + address = parsed.hostname or "" + port = str(parsed.port) if parsed.port else "" + return ProxyInfo( + protocol="hysteria2", + name=name, + address=address, + port=port, + link=line, + ) + except Exception: + return None + + +async def load_subscription_urls(file_path: str) -> list[str]: + """从 Subscription.txt 加载订阅 URL 列表 + + 支持两种格式: + - 纯 URL 格式(每行一个 URL) + - 行号\tURL 格式(忽略行号前缀) + """ + urls = [] + try: + with open(file_path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + # 支持 "行号\tURL" 格式 + if "\t" in line: + parts = line.split("\t", 1) + url = parts[-1].strip() + else: + url = line + if url.startswith("http://") or url.startswith("https://"): + urls.append(url) + except FileNotFoundError: + logger.warning("订阅文件不存在: %s", file_path) + return urls + + +async def load_check_urls(file_path: str) -> list[str]: + """从 domain_check.txt 加载检测 URL 列表""" + urls = [] + try: + with open(file_path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + if line.startswith("http://") or line.startswith("https://"): + urls.append(line) + except FileNotFoundError: + logger.warning("检测 URL 文件不存在: %s,使用默认值", file_path) + urls = ["http://www.google.com/generate_204"] + return urls diff --git a/app/routers/__init__.py b/app/routers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/routers/api.py b/app/routers/api.py new file mode 100644 index 0000000..fca3f6c --- /dev/null +++ b/app/routers/api.py @@ -0,0 +1,133 @@ +"""REST API 路由 - JSON 格式的代理数据接口""" + +import asyncio +import logging + +from fastapi import APIRouter, HTTPException + +from app import get_checker, get_config, get_db, get_scheduler +from app.generator import generate_clash_subscription, generate_v2ray_subscription + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api") + + +@router.get("/proxies") +async def get_available_proxies(): + """获取所有可用代理列表(latency_ms > 0 且 fail_count=0)""" + db = get_db() + config = get_config() + proxies = await db.get_available_proxies(config.check.latency_threshold) + return { + "total": len(proxies), + "proxies": [_proxy_to_dict(p) for p in proxies], + } + + +@router.get("/proxies/all") +async def get_all_proxies(): + """获取所有代理(含不可用)""" + db = get_db() + proxies = await db.get_all_proxies() + return { + "total": len(proxies), + "proxies": [_proxy_to_dict(p) for p in 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"} + + +@router.get("/subscription/v2ray") +async def v2ray_subscription(): + """获取 v2ray 格式订阅(base64 编码)""" + db = get_db() + config = get_config() + proxies = await db.get_available_proxies(config.check.latency_threshold) + content = generate_v2ray_subscription(proxies) + return _subscription_response(content, "text/plain") + + +@router.get("/subscription/clash") +async def clash_subscription(): + """获取 Clash 格式订阅(YAML)""" + db = get_db() + config = get_config() + proxies = await db.get_available_proxies(config.check.latency_threshold) + content = generate_clash_subscription(proxies) + return _subscription_response(content, "text/yaml") + + +@router.post("/fetch") +async def manual_fetch(): + """手动触发拉取订阅""" + scheduler = get_scheduler() + if scheduler._fetching: + return {"message": "拉取任务正在进行中"} + asyncio.create_task(scheduler.fetch_and_check()) + return {"message": "已触发拉取任务"} + + +@router.post("/verify") +async def manual_verify(): + """手动触发验证代理""" + scheduler = get_scheduler() + if scheduler._verifying: + return {"message": "验证任务正在进行中"} + asyncio.create_task(scheduler.verify_stored_proxies()) + return {"message": "已触发验证任务"} + + +@router.get("/stats") +async def get_stats(): + """获取统计信息""" + db = get_db() + scheduler = get_scheduler() + stats = await db.get_stats() + stats["last_fetch_time"] = scheduler.last_fetch_time + stats["last_verify_time"] = scheduler.last_verify_time + return stats + + +@router.get("/health") +async def health_check(): + """健康检查""" + return {"status": "ok"} + + +def _proxy_to_dict(proxy) -> dict: + """将 ProxyDBRecord 转为 API 响应字典""" + return { + "id": proxy.id, + "protocol": proxy.protocol, + "name": proxy.name, + "address": proxy.address, + "port": proxy.port, + "latency_ms": proxy.latency_ms, + "fail_count": proxy.fail_count, + "source": proxy.source, + "last_check_time": proxy.last_check_time, + "last_success_time": proxy.last_success_time, + "status": proxy.status, + } + + +def _subscription_response(content: str, content_type: str): + """生成订阅响应,添加通用 headers""" + from fastapi.responses import Response + + return Response( + content=content, + media_type=f"{content_type}; charset=utf-8", + headers={ + "Content-Disposition": 'attachment; filename="subscription"', + "Profile-Update-Interval": "24", + "Profile-Title": "ProxyPool", + }, + ) diff --git a/app/routers/web.py b/app/routers/web.py new file mode 100644 index 0000000..50598d4 --- /dev/null +++ b/app/routers/web.py @@ -0,0 +1,57 @@ +"""Web UI 页面路由 - Jinja2 模板渲染""" + +import logging + +from fastapi import APIRouter, Request + +from app import get_config, get_db, get_scheduler, get_templates + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +@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() + + return templates.TemplateResponse( + "index.html", + { + "request": request, + "proxies": proxies, + "stats": stats, + "latency_threshold": config.check.latency_threshold, + "last_fetch_time": scheduler.last_fetch_time, + "last_verify_time": scheduler.last_verify_time, + }, + ) + + +@router.get("/subscription") +async def subscription_page(request: Request): + """订阅链接页面""" + templates = get_templates() + + # 基于请求 host 动态生成订阅链接 + host = request.headers.get("host", f"localhost:{get_config().server.port}") + scheme = request.url.scheme + + v2ray_url = f"{scheme}://{host}/api/subscription/v2ray" + clash_url = f"{scheme}://{host}/api/subscription/clash" + + return templates.TemplateResponse( + "subscription.html", + { + "request": request, + "v2ray_url": v2ray_url, + "clash_url": clash_url, + }, + ) diff --git a/app/scheduler.py b/app/scheduler.py new file mode 100644 index 0000000..7769923 --- /dev/null +++ b/app/scheduler.py @@ -0,0 +1,193 @@ +"""定时任务调度模块 - 使用 APScheduler 管理拉取/验证/清理任务""" + +import logging +from datetime import datetime, timezone + +from apscheduler.schedulers.asyncio import AsyncIOScheduler + +from app.checker import ProxyChecker +from app.config import AppConfig +from app.database import ProxyDatabase +from app.parser import fetch_subscription, parse_subscription, load_subscription_urls + +logger = logging.getLogger(__name__) + + +class TaskScheduler: + """代理池定时任务调度器""" + + def __init__(self, config: AppConfig, db: ProxyDatabase, checker: ProxyChecker): + self.config = config + self.db = db + self.checker = checker + self.scheduler = AsyncIOScheduler() + self._last_fetch_time = "" + self._last_verify_time = "" + self._fetching = False + self._verifying = False + + def start(self) -> None: + """注册所有定时任务并启动调度器""" + cfg = self.config.scheduler + + self.scheduler.add_job( + self.fetch_and_check, + "interval", + seconds=cfg.fetch_interval, + id="fetch_subscriptions", + name="拉取订阅并检测", + ) + self.scheduler.add_job( + self.verify_stored_proxies, + "interval", + seconds=cfg.verify_interval, + id="verify_proxies", + name="验证已存代理", + ) + self.scheduler.add_job( + self.cleanup_proxies, + "interval", + seconds=cfg.cleanup_interval, + id="cleanup_proxies", + name="清理不合格代理", + ) + self.scheduler.start() + logger.info("调度器已启动: fetch=%ds, verify=%ds, cleanup=%ds", + cfg.fetch_interval, cfg.verify_interval, cfg.cleanup_interval) + + def shutdown(self) -> None: + """关闭调度器""" + self.scheduler.shutdown(wait=False) + logger.info("调度器已关闭") + + async def fetch_and_check(self) -> None: + """任务1:拉取订阅 + 检测 + 入库 + + 1. 从 Subscription.txt 读取所有订阅 URL + 2. 逐个 URL 拉取内容并解析为 ProxyInfo 列表 + 3. 对所有代理并发检测延迟 + 4. 延迟低于 threshold 的代理插入/更新到数据库 + """ + if self._fetching: + logger.info("上一次拉取任务尚未完成,跳过本次") + return + + self._fetching = True + try: + logger.info("开始拉取订阅...") + urls = await load_subscription_urls(self.config.resources.subscription_file) + if not urls: + logger.warning("未找到订阅 URL") + return + + all_proxies = [] + for url in urls: + try: + content = await fetch_subscription(url) + proxies = parse_subscription(content) + logger.info("订阅 %s: 解析到 %d 个节点", url[:50], len(proxies)) + # 标记来源 + for p in proxies: + all_proxies.append((p, url)) + except Exception as e: + logger.error("拉取订阅失败 %s: %s", url[:50], e) + + if not all_proxies: + logger.warning("未解析到任何代理节点") + return + + logger.info("共解析到 %d 个代理节点,开始并发检测...", len(all_proxies)) + + # 并发检测延迟 + links = [p.link for p, _ in all_proxies] + results = await self.checker.check_batch(links) + + # 处理检测结果 + added = 0 + updated = 0 + threshold = self.config.check.latency_threshold + + for proxy, source in all_proxies: + latency = results.get(proxy.link) + if latency is not None and latency <= threshold: + # 延迟达标,尝试入库 + existing = await self.db.get_proxy_by_link(proxy.link) + if existing: + await self.db.update_latency(existing.id, latency) + updated += 1 + else: + success = await self.db.insert_proxy(proxy, latency, source) + if success: + added += 1 + elif latency is not None and latency > threshold: + # 延迟超标,如果已在库中则增加失败计数 + existing = await self.db.get_proxy_by_link(proxy.link) + if existing: + await self.db.increment_fail(existing.id) + + self._last_fetch_time = datetime.now(timezone.utc).isoformat() + logger.info("拉取完成: 新增 %d, 更新 %d, 总解析 %d", added, updated, len(all_proxies)) + except Exception as e: + logger.error("拉取任务异常: %s", e, exc_info=True) + finally: + self._fetching = False + + async def verify_stored_proxies(self) -> None: + """任务2:验证已存代理可用性 + + 1. 从数据库获取所有代理 + 2. 并发检测延迟 + 3. 成功的更新 latency + 重置 fail_count + 4. 失败的 fail_count + 1 + """ + if self._verifying: + logger.info("上一次验证任务尚未完成,跳过本次") + return + + self._verifying = True + try: + proxies = await self.db.get_all_proxies() + if not proxies: + logger.info("数据库中没有代理,跳过验证") + return + + 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 + + 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 + else: + await self.db.increment_fail(proxy.id) + fail_count += 1 + + self._last_verify_time = datetime.now(timezone.utc).isoformat() + logger.info("验证完成: 成功 %d, 失败 %d", success_count, fail_count) + except Exception as e: + logger.error("验证任务异常: %s", e, exc_info=True) + finally: + self._verifying = False + + async def cleanup_proxies(self) -> None: + """任务3:清理不合格代理 + + 删除 fail_count >= max_fail_count 的代理 + """ + max_fail = self.config.scheduler.max_fail_count + deleted = await self.db.delete_proxies_by_fail_count(max_fail) + if deleted > 0: + logger.info("清理完成: 删除 %d 个不合格代理 (fail_count >= %d)", deleted, max_fail) + + @property + def last_fetch_time(self) -> str: + return self._last_fetch_time + + @property + def last_verify_time(self) -> str: + return self._last_verify_time diff --git a/app/templates/base.html b/app/templates/base.html new file mode 100644 index 0000000..821e55e --- /dev/null +++ b/app/templates/base.html @@ -0,0 +1,67 @@ + + + + + + {% block title %}ProxyPool{% endblock %} + + + + {% block extra_head %}{% endblock %} + + + + +
+ {% block content %}{% endblock %} +
+ +
+ ProxyPool - 代理池管理系统 +
+ + + {% block extra_script %}{% endblock %} + + diff --git a/app/templates/index.html b/app/templates/index.html new file mode 100644 index 0000000..a8c7652 --- /dev/null +++ b/app/templates/index.html @@ -0,0 +1,179 @@ +{% extends "base.html" %} + +{% block title %}代理列表 - ProxyPool{% endblock %} + +{% block content %} + +
+
+
+
+
总代理数
+

{{ stats.total }}

+
+
+
+
+
+
+
可用数
+

{{ stats.available }}

+
+
+
+
+
+
+
不可用
+

{{ stats.unavailable }}

+
+
+
+
+
+
+
平均延迟
+

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

+
+
+
+
+ + +
+
+
代理列表
+ + 延迟阈值: {{ latency_threshold }} ms + {% if last_fetch_time %} + | 上次拉取: {{ last_fetch_time[:19] }} + {% endif %} + {% if last_verify_time %} + | 上次验证: {{ last_verify_time[:19] }} + {% endif %} + +
+
+ + + +
+
+ + +
+
+ + + + + + + + + + + + + + + + {% for proxy in proxies %} + + + + + + + + + + + + {% endfor %} + {% if not proxies %} + + + + {% endif %} + +
#名称协议地址端口延迟状态失败次数操作
{{ loop.index }}{{ proxy.name[:30] }}{% if proxy.name|length > 30 %}...{% endif %}{{ proxy.protocol }}{{ proxy.address }}{{ proxy.port }} + {% if proxy.latency_ms >= 0 %} + + {{ "%.0f"|format(proxy.latency_ms) }} ms + + {% else %} + - + {% endif %} + + + {% if proxy.status == 'available' %}可用{% elif proxy.status == 'unavailable' %}不可用{% else %}未检测{% endif %} + + {{ proxy.fail_count }} + +
暂无代理数据,请点击"拉取订阅"获取
+
+
+ + +{% if stats.protocol_distribution %} +
+
+
协议分布
+
+ {% for proto, count in stats.protocol_distribution.items() %} + {{ proto }}: {{ count }} + {% endfor %} +
+
+
+{% endif %} +{% endblock %} + +{% block extra_script %} + +{% endblock %} diff --git a/app/templates/subscription.html b/app/templates/subscription.html new file mode 100644 index 0000000..80f8f43 --- /dev/null +++ b/app/templates/subscription.html @@ -0,0 +1,93 @@ +{% extends "base.html" %} + +{% block title %}订阅链接 - ProxyPool{% endblock %} + +{% block content %} +
+
+

订阅链接

+

将以下链接复制到 v2rayN、Clash 等客户端中使用。

+ + +
+
+ V2Ray 订阅 + +
+
+
+ + + + +
+ Base64 编码格式,适用于 v2rayN / v2rayNG / Shadowrocket 等客户端 +
+
+ + +
+
+ Clash 订阅 + +
+
+
+ + + + +
+ YAML 格式,适用于 Clash for Windows / ClashX / Clash for Android 等客户端 +
+
+ + +
+
+ 使用说明 +
+
+
    +
  1. 点击上方复制按钮获取订阅链接
  2. +
  3. 在代理客户端中添加订阅,粘贴链接
  4. +
  5. 更新订阅即可获取当前可用的代理节点
  6. +
  7. 订阅内容会随代理池自动更新,无需手动更换链接
  8. +
+
+ 提示:订阅链接中仅包含延迟低于阈值的可用代理。代理池会定时拉取、检测和清理,确保订阅质量。 +
+
+
+
+
+{% endblock %} + +{% block extra_script %} + +{% endblock %} diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..6959754 --- /dev/null +++ b/config.yaml @@ -0,0 +1,25 @@ +server: + host: "0.0.0.0" + port: 8080 + debug: false + +database: + path: "data/proxy_pool.db" + +check: + urls: + - "http://www.google.com/generate_204" + - "http://www.gstatic.com/generate_204" + timeout: 5.0 + max_concurrent: 50 + latency_threshold: 1500.0 + +scheduler: + fetch_interval: 3600 + verify_interval: 1800 + cleanup_interval: 7200 + max_fail_count: 3 + +resources: + subscription_file: "resources/Subscription.txt" + domain_check_file: "resources/domain_check.txt" diff --git a/docker-compose.yaml b/docker-compose.yaml new file mode 100644 index 0000000..05f8bab --- /dev/null +++ b/docker-compose.yaml @@ -0,0 +1,18 @@ +version: "3.8" + +services: + proxy-pool: + build: . + container_name: proxy-pool + ports: + - "8080:8080" + volumes: + - ./config.yaml:/app/config.yaml:ro + - ./resources:/app/resources:ro + - proxy-pool-data:/app/data + environment: + - TZ=Asia/Shanghai + restart: unless-stopped + +volumes: + proxy-pool-data: diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f95ca32 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,8 @@ +fastapi>=0.104.0 +uvicorn[standard]>=0.24.0 +jinja2>=3.1.2 +aiohttp>=3.9.0 +aiosqlite>=0.19.0 +apscheduler>=3.10.0 +pyyaml>=6.0.1 +python-multipart>=0.0.6 diff --git a/resources/domain_check.txt b/resources/domain_check.txt new file mode 100644 index 0000000..8e9a64b --- /dev/null +++ b/resources/domain_check.txt @@ -0,0 +1,2 @@ +http://www.google.com/generate_204 +http://www.gstatic.com/generate_204 From 5ff4af32b7d49cadc6cbb2b68661813a4bbca319 Mon Sep 17 00:00:00 2001 From: siruoren Date: Fri, 17 Jul 2026 19:52:47 +0800 Subject: [PATCH 03/10] feat: add subscription management feature and optimize UI This commit introduces full subscription management capabilities: 1. Add SubscriptionRecord data model and SQLite subscription table 2. Implement complete subscription CRUD API endpoints 3. Add per-subscription fetch/verify tasks with crontab scheduling 4. Redesign index page with subscription management UI, tabbed proxy list and protocol chart 5. Refactor logging config to add daily rotating file logs 6. Update UI copy and styling for better user experience --- app/__init__.py | 29 +- app/database.py | 218 ++++++++++++- app/models.py | 16 + app/routers/api.py | 198 +++++++++++- app/routers/web.py | 18 ++ app/scheduler.py | 256 ++++++++++----- app/templates/base.html | 4 +- app/templates/index.html | 536 ++++++++++++++++++++++++++------ app/templates/subscription.html | 11 +- logs/proxy_pool.log | 343 ++++++++++++++++++++ 10 files changed, 1449 insertions(+), 180 deletions(-) create mode 100644 logs/proxy_pool.log diff --git a/app/__init__.py b/app/__init__.py index c440949..5ee4e24 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -50,13 +50,34 @@ def create_app(config_path: str = "config.yaml") -> FastAPI: # 加载配置 _config = load_config(config_path) - # 配置日志 + # 配置日志(控制台 + 按天归档文件,保留7天) log_level = logging.DEBUG if _config.server.debug else logging.INFO - logging.basicConfig( - level=log_level, - format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + log_fmt = logging.Formatter( + fmt="%(asctime)s [%(levelname)s] %(name)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) + root_logger = logging.getLogger() + root_logger.setLevel(log_level) + + # 控制台 + console_handler = logging.StreamHandler() + 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( + 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) # 初始化组件 _db = ProxyDatabase(_config.database.path) diff --git a/app/database.py b/app/database.py index b298f15..6d34a6f 100644 --- a/app/database.py +++ b/app/database.py @@ -6,7 +6,7 @@ import aiosqlite -from app.models import ProxyDBRecord, ProxyInfo +from app.models import ProxyDBRecord, ProxyInfo, SubscriptionRecord class ProxyDatabase: @@ -46,9 +46,42 @@ async def init(self) -> None: 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 TABLE IF NOT EXISTS subscriptions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT NOT NULL UNIQUE, + crontab TEXT NOT NULL DEFAULT '0 * * * *', + latency_threshold REAL DEFAULT 1500.0, + max_retries INTEGER DEFAULT 3, + max_concurrent INTEGER DEFAULT 50, + enabled INTEGER DEFAULT 1, + created_at TEXT DEFAULT '', + empty_days INTEGER DEFAULT 0, + total_count INTEGER DEFAULT 0, + fetch_status TEXT DEFAULT 'idle' + ); + + CREATE INDEX IF NOT EXISTS idx_subscriptions_url ON subscriptions(url); """) await self._db.commit() + # 兼容旧表:添加 empty_days 和 total_count 列(如不存在) + try: + await self._db.execute("ALTER TABLE subscriptions ADD COLUMN empty_days INTEGER DEFAULT 0") + await self._db.commit() + except Exception: + pass + try: + await self._db.execute("ALTER TABLE subscriptions ADD COLUMN total_count INTEGER DEFAULT 0") + await self._db.commit() + except Exception: + pass + try: + await self._db.execute("ALTER TABLE subscriptions ADD COLUMN fetch_status TEXT DEFAULT 'idle'") + await self._db.commit() + except Exception: + pass + async def close(self) -> None: """关闭数据库连接""" if self._db: @@ -76,7 +109,7 @@ async def insert_proxy(self, proxy: ProxyInfo, latency_ms: float, source: str) - """插入新代理,link 唯一约束,重复则忽略。返回是否插入成功""" now = datetime.now(timezone.utc).isoformat() try: - await self._db.execute( + cursor = await self._db.execute( """INSERT OR IGNORE INTO proxies (protocol, name, address, port, link, latency_ms, fail_count, source, last_check_time, last_success_time, created_at) @@ -85,7 +118,7 @@ async def insert_proxy(self, proxy: ProxyInfo, latency_ms: float, source: str) - latency_ms, source, now, now if latency_ms >= 0 else "", now), ) await self._db.commit() - return self._db.changes > 0 + return cursor.rowcount > 0 except aiosqlite.IntegrityError: return False @@ -167,6 +200,185 @@ async def get_proxies_needing_verify(self, limit: int = 0) -> list[ProxyDBRecord rows = await cursor.fetchall() return [self._row_to_record(row) for row in rows] + def _row_to_subscription(self, row: aiosqlite.Row) -> SubscriptionRecord: + """将数据库行转为 SubscriptionRecord""" + return SubscriptionRecord( + id=row["id"], + url=row["url"], + crontab=row["crontab"], + latency_threshold=row["latency_threshold"], + max_retries=row["max_retries"], + max_concurrent=row["max_concurrent"], + enabled=bool(row["enabled"]), + created_at=row["created_at"], + empty_days=row["empty_days"] if "empty_days" in row.keys() else 0, + 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_subscription(self, url: str, crontab: str = "0 * * * *", + latency_threshold: float = 1500.0, + max_retries: int = 3, max_concurrent: int = 50, + enabled: bool = True) -> SubscriptionRecord | None: + """添加订阅源""" + now = datetime.now(timezone.utc).isoformat() + try: + cursor = await self._db.execute( + """INSERT INTO subscriptions (url, crontab, latency_threshold, max_retries, max_concurrent, enabled, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + (url, crontab, latency_threshold, max_retries, max_concurrent, int(enabled), now), + ) + await self._db.commit() + return await self.get_subscription_by_id(cursor.lastrowid) + except aiosqlite.IntegrityError: + return None + + async def get_subscription_by_id(self, sub_id: int) -> SubscriptionRecord | None: + """根据 ID 获取订阅""" + cursor = await self._db.execute("SELECT * FROM subscriptions WHERE id = ?", (sub_id,)) + row = await cursor.fetchone() + return self._row_to_subscription(row) if row else None + + async def get_subscription_by_url(self, url: str) -> SubscriptionRecord | None: + """根据 URL 获取订阅""" + cursor = await self._db.execute("SELECT * FROM subscriptions WHERE url = ?", (url,)) + row = await cursor.fetchone() + return self._row_to_subscription(row) if row else None + + async def get_all_subscriptions(self) -> list[SubscriptionRecord]: + """获取所有订阅源""" + cursor = await self._db.execute("SELECT * FROM subscriptions ORDER BY id ASC") + rows = await cursor.fetchall() + return [self._row_to_subscription(row) for row in rows] + + async def get_enabled_subscriptions(self) -> list[SubscriptionRecord]: + """获取所有启用的订阅源""" + cursor = await self._db.execute("SELECT * FROM subscriptions WHERE enabled = 1 ORDER BY id ASC") + rows = await cursor.fetchall() + 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(): + if k in allowed: + if k == "enabled": + updates[k] = int(v) + else: + updates[k] = v + 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( + f"UPDATE subscriptions SET {set_clause} WHERE id = ?", values + ) + await self._db.commit() + return cursor.rowcount > 0 + + async def delete_subscription(self, sub_id: int) -> bool: + """删除订阅源""" + cursor = await self._db.execute("DELETE FROM subscriptions WHERE id = ?", (sub_id,)) + await self._db.commit() + return cursor.rowcount > 0 + + async def increment_empty_days(self, sub_id: int) -> None: + """订阅源连续空代理天数 +1""" + await self._db.execute( + "UPDATE subscriptions SET empty_days = empty_days + 1 WHERE id = ?", + (sub_id,), + ) + await self._db.commit() + + async def reset_empty_days(self, sub_id: int) -> None: + """订阅源有代理时重置空天数为0""" + await self._db.execute( + "UPDATE subscriptions SET empty_days = 0 WHERE id = ?", + (sub_id,), + ) + await self._db.commit() + + async def get_subscriptions_with_empty_days(self, min_days: int) -> list[SubscriptionRecord]: + """获取连续空代理天数 >= min_days 的订阅源""" + cursor = await self._db.execute( + "SELECT * FROM subscriptions WHERE empty_days >= ?", (min_days,) + ) + 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( + "UPDATE subscriptions SET total_count = ? WHERE id = ?", + (count, sub_id), + ) + await self._db.commit() + + async def update_fetch_status(self, sub_id: int, status: str) -> None: + """更新订阅源拉取状态: idle / updating / success / failed""" + await self._db.execute( + "UPDATE subscriptions SET fetch_status = ? WHERE id = ?", + (status, sub_id), + ) + 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]]: + """获取按订阅来源分组的可用代理,按入库时间正序""" + 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""", + (max_latency,), + ) + rows = await cursor.fetchall() + grouped = {} + for row in rows: + record = self._row_to_record(row) + grouped.setdefault(record.source, []).append(record) + return grouped + async def get_stats(self) -> dict: """获取统计信息""" cursor = await self._db.execute("SELECT COUNT(*) as total FROM proxies") diff --git a/app/models.py b/app/models.py index 0f1c08e..df1fb45 100644 --- a/app/models.py +++ b/app/models.py @@ -39,3 +39,19 @@ def status(self) -> str: if self.fail_count > 0: return "unavailable" return "available" + + +@dataclass +class SubscriptionRecord: + """订阅源记录""" + id: int + url: str + crontab: str # crontab 格式的定时表达式 + latency_threshold: float # 该订阅的延迟阈值(ms) + max_retries: int # 最大重试次数 + max_concurrent: int # 异步最大并发数 + enabled: bool # 是否启用 + created_at: str # ISO8601 + empty_days: int # 连续代理数为0的天数 + total_count: int # 最新一次拉取的代理地址总数 + fetch_status: str # 拉取状态: idle / updating / success / failed diff --git a/app/routers/api.py b/app/routers/api.py index fca3f6c..2bea311 100644 --- a/app/routers/api.py +++ b/app/routers/api.py @@ -2,8 +2,10 @@ import asyncio import logging +from typing import Optional from fastapi import APIRouter, HTTPException +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 @@ -66,7 +68,7 @@ async def clash_subscription(): @router.post("/fetch") async def manual_fetch(): - """手动触发拉取订阅""" + """手动触发拉取所有订阅""" scheduler = get_scheduler() if scheduler._fetching: return {"message": "拉取任务正在进行中"} @@ -74,6 +76,19 @@ async def manual_fetch(): return {"message": "已触发拉取任务"} +@router.post("/fetch/{sub_id}") +async def manual_fetch_subscription(sub_id: int): + """手动触发拉取指定订阅""" + db = get_db() + sub = await db.get_subscription_by_id(sub_id) + if not sub: + from fastapi import HTTPException + raise HTTPException(status_code=404, detail="订阅不存在") + scheduler = get_scheduler() + asyncio.create_task(scheduler._fetch_single_subscription(sub_id)) + return {"message": f"已触发订阅 #{sub_id} 的拉取任务"} + + @router.post("/verify") async def manual_verify(): """手动触发验证代理""" @@ -84,6 +99,19 @@ async def manual_verify(): return {"message": "已触发验证任务"} +@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: + from fastapi import HTTPException + raise HTTPException(status_code=404, detail="订阅不存在") + scheduler = get_scheduler() + asyncio.create_task(scheduler.verify_subscription_proxies(sub_id)) + return {"message": f"已触发订阅 #{sub_id} 的代理验证"} + + @router.get("/stats") async def get_stats(): """获取统计信息""" @@ -101,6 +129,157 @@ async def health_check(): return {"status": "ok"} +# ---- 订阅管理 ---- + +@router.get("/subscriptions") +async def get_subscriptions(): + """获取所有订阅源""" + db = get_db() + subs = await db.get_all_subscriptions() + return { + "total": len(subs), + "subscriptions": [_subscription_to_dict(s) for s in subs], + } + + +@router.post("/subscriptions") +async def add_subscription(url: str = "", crontab: str = "0 * * * *", + latency_threshold: float = 1500.0, + max_retries: int = 3, max_concurrent: int = 50, + enabled: bool = True): + """添加订阅源""" + from fastapi import HTTPException + if not url: + raise HTTPException(status_code=400, detail="URL 不能为空") + db = get_db() + sub = await db.add_subscription(url, crontab, latency_threshold, max_retries, max_concurrent, enabled) + if not sub: + raise HTTPException(status_code=409, detail="订阅地址已存在") + # 注册定时任务 + scheduler = get_scheduler() + if sub.enabled: + scheduler._add_subscription_job(sub) + # 自动触发该订阅的延迟检测 + asyncio.create_task(scheduler._fetch_single_subscription(sub.id)) + return _subscription_to_dict(sub) + + +class AutoSubRequest(BaseModel): + """自动添加订阅请求体""" + url: str + crontab: Optional[str] = None + latency_threshold: Optional[float] = None + max_retries: Optional[int] = None + max_concurrent: Optional[int] = None + + +@router.post("/subscriptions/auto") +async def auto_add_subscription(req: AutoSubRequest): + """自动添加订阅源 - 仅 URL 必填,其余可选,支持去重,新增后自动拉取并验证""" + if not req.url: + raise HTTPException(status_code=400, detail="URL 不能为空") + db = get_db() + # 去重:检查是否已存在 + existing = await db.get_subscription_by_url(req.url) + if existing: + return { + "status": "duplicate", + "message": "订阅地址已存在", + "subscription": _subscription_to_dict(existing), + } + # 创建新订阅,未填字段使用默认值 + sub = await db.add_subscription( + url=req.url, + crontab=req.crontab or "0 * * * *", + latency_threshold=req.latency_threshold or 1500.0, + max_retries=req.max_retries or 3, + max_concurrent=req.max_concurrent or 50, + enabled=True, + ) + if not sub: + raise HTTPException(status_code=500, detail="添加订阅失败") + # 注册定时任务 + 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()) + return { + "status": "added", + "message": "订阅已添加,正在拉取并验证代理", + "subscription": _subscription_to_dict(sub), + } + + +@router.put("/subscriptions/{sub_id}") +async def update_subscription(sub_id: int, url: str = None, crontab: str = None, + latency_threshold: float = None, max_retries: int = None, + max_concurrent: int = None, enabled: bool = None): + """更新订阅源""" + from fastapi import HTTPException + db = get_db() + kwargs = {} + if url is not None: + kwargs["url"] = url + if crontab is not None: + kwargs["crontab"] = crontab + if latency_threshold is not None: + kwargs["latency_threshold"] = latency_threshold + if max_retries is not None: + kwargs["max_retries"] = max_retries + if max_concurrent is not None: + kwargs["max_concurrent"] = max_concurrent + if enabled is not None: + kwargs["enabled"] = enabled + + success = await db.update_subscription(sub_id, **kwargs) + if not success: + raise HTTPException(status_code=404, detail="订阅不存在或无更新") + # 刷新定时任务 + sub = await db.get_subscription_by_id(sub_id) + scheduler = get_scheduler() + 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()) + else: + # 禁用时重置拉取状态 + await db.update_fetch_status(sub_id, "idle") + return _subscription_to_dict(sub) + + +@router.delete("/subscriptions/{sub_id}") +async def delete_subscription(sub_id: int): + """删除订阅源""" + db = get_db() + success = await db.delete_subscription(sub_id) + if not success: + from fastapi import HTTPException + raise HTTPException(status_code=404, detail="订阅不存在") + # 移除定时任务 + scheduler = get_scheduler() + scheduler.remove_subscription_job(sub_id) + return {"message": "deleted"} + + +@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) + result = {} + for source, proxies in grouped.items(): + result[source] = [_proxy_to_dict(p) for p in proxies] + return {"grouped": result} + + def _proxy_to_dict(proxy) -> dict: """将 ProxyDBRecord 转为 API 响应字典""" return { @@ -131,3 +310,20 @@ def _subscription_response(content: str, content_type: str): "Profile-Title": "ProxyPool", }, ) + + +def _subscription_to_dict(sub) -> dict: + """将 SubscriptionRecord 转为 API 响应字典""" + return { + "id": sub.id, + "url": sub.url, + "crontab": sub.crontab, + "latency_threshold": sub.latency_threshold, + "max_retries": sub.max_retries, + "max_concurrent": sub.max_concurrent, + "enabled": sub.enabled, + "created_at": sub.created_at, + "empty_days": sub.empty_days, + "total_count": sub.total_count, + "fetch_status": sub.fetch_status, + } diff --git a/app/routers/web.py b/app/routers/web.py index 50598d4..4ce889c 100644 --- a/app/routers/web.py +++ b/app/routers/web.py @@ -1,6 +1,9 @@ +from __future__ import annotations """Web UI 页面路由 - Jinja2 模板渲染""" +import json import logging +from dataclasses import asdict from fastapi import APIRouter, Request @@ -21,6 +24,17 @@ async def index(request: Request): 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) + # 计算每个订阅源的可用代理数量 + sub_available_counts = {} + for sub in subscriptions: + sub_available_counts[sub.url] = len(grouped.get(sub.url, [])) + subscriptions_json = json.dumps( + [dict(asdict(s), available_count=sub_available_counts.get(s.url, 0)) for s in subscriptions], + ensure_ascii=False, + ) return templates.TemplateResponse( "index.html", @@ -28,6 +42,10 @@ async def index(request: Request): "request": request, "proxies": proxies, "stats": stats, + "subscriptions": subscriptions, + "subscriptions_json": subscriptions_json, + "sub_available_counts": sub_available_counts, + "grouped": grouped, "latency_threshold": config.check.latency_threshold, "last_fetch_time": scheduler.last_fetch_time, "last_verify_time": scheduler.last_verify_time, diff --git a/app/scheduler.py b/app/scheduler.py index 7769923..7ea61b4 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -1,14 +1,17 @@ -"""定时任务调度模块 - 使用 APScheduler 管理拉取/验证/清理任务""" +from __future__ import annotations +"""定时任务调度模块 - 使用 APScheduler 管理拉取/验证/清理任务,支持 crontab 配置""" +import asyncio import logging from datetime import datetime, timezone from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.cron import CronTrigger from app.checker import ProxyChecker from app.config import AppConfig from app.database import ProxyDatabase -from app.parser import fetch_subscription, parse_subscription, load_subscription_urls +from app.parser import fetch_subscription, parse_subscription logger = logging.getLogger(__name__) @@ -27,16 +30,10 @@ def __init__(self, config: AppConfig, db: ProxyDatabase, checker: ProxyChecker): self._verifying = False def start(self) -> None: - """注册所有定时任务并启动调度器""" + """注册默认定时任务并启动调度器""" cfg = self.config.scheduler - self.scheduler.add_job( - self.fetch_and_check, - "interval", - seconds=cfg.fetch_interval, - id="fetch_subscriptions", - name="拉取订阅并检测", - ) + # 默认验证和清理任务(使用全局间隔配置) self.scheduler.add_job( self.verify_stored_proxies, "interval", @@ -52,94 +49,167 @@ def start(self) -> None: name="清理不合格代理", ) self.scheduler.start() - logger.info("调度器已启动: fetch=%ds, verify=%ds, cleanup=%ds", - cfg.fetch_interval, cfg.verify_interval, cfg.cleanup_interval) + logger.info("调度器已启动: verify=%ds, cleanup=%ds", + cfg.verify_interval, cfg.cleanup_interval) + + # 启动后注册订阅任务 + asyncio.create_task(self._register_subscription_jobs()) + + async def _register_subscription_jobs(self) -> None: + """从数据库加载订阅源并注册 crontab 定时任务""" + subs = await self.db.get_enabled_subscriptions() + for sub in subs: + self._add_subscription_job(sub) + logger.info("已注册 %d 个订阅拉取任务", len(subs)) + + def _add_subscription_job(self, sub) -> None: + """为单个订阅注册 crontab 定时任务""" + job_id = f"fetch_sub_{sub.id}" + try: + # 移除已有同 ID 任务 + if self.scheduler.get_job(job_id): + self.scheduler.remove_job(job_id) + + parts = sub.crontab.strip().split() + if len(parts) != 5: + logger.error("订阅 %d crontab 格式错误: %s", sub.id, sub.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_subscription, + trigger=trigger, + id=job_id, + name=f"拉取订阅 #{sub.id}", + args=[sub.id], + ) + logger.info("注册订阅任务 #%d: crontab=%s", sub.id, sub.crontab) + except Exception as e: + logger.error("注册订阅任务 #%d 失败: %s", sub.id, e) + + def remove_subscription_job(self, sub_id: int) -> None: + """移除订阅定时任务""" + job_id = f"fetch_sub_{sub_id}" + if self.scheduler.get_job(job_id): + self.scheduler.remove_job(job_id) + + def refresh_subscription_job(self, sub) -> None: + """刷新订阅定时任务(更新 crontab 后调用)""" + self.remove_subscription_job(sub.id) + if sub.enabled: + self._add_subscription_job(sub) def shutdown(self) -> None: """关闭调度器""" self.scheduler.shutdown(wait=False) logger.info("调度器已关闭") - async def fetch_and_check(self) -> None: - """任务1:拉取订阅 + 检测 + 入库 - - 1. 从 Subscription.txt 读取所有订阅 URL - 2. 逐个 URL 拉取内容并解析为 ProxyInfo 列表 - 3. 对所有代理并发检测延迟 - 4. 延迟低于 threshold 的代理插入/更新到数据库 - """ - if self._fetching: - 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 - self._fetching = True try: - logger.info("开始拉取订阅...") - urls = await load_subscription_urls(self.config.resources.subscription_file) - if not urls: - logger.warning("未找到订阅 URL") + logger.info("开始拉取订阅 #%d: %s", sub.id, sub.url[:50]) + await self.db.update_fetch_status(sub_id, "updating") + content = await fetch_subscription(sub.url) + proxies = parse_subscription(content) + if not proxies: + logger.warning("订阅 #%d 未解析到任何节点", sub.id) + await self.db.increment_empty_days(sub_id) + await self.db.update_total_count(sub_id, 0) + await self.db.update_fetch_status(sub_id, "success") return - all_proxies = [] - for url in urls: - try: - content = await fetch_subscription(url) - proxies = parse_subscription(content) - logger.info("订阅 %s: 解析到 %d 个节点", url[:50], len(proxies)) - # 标记来源 - for p in proxies: - all_proxies.append((p, url)) - except Exception as e: - logger.error("拉取订阅失败 %s: %s", url[:50], e) - - if not all_proxies: - logger.warning("未解析到任何代理节点") - return + logger.info("订阅 #%d: 解析到 %d 个节点,开始逐个检测...", sub.id, len(proxies)) - logger.info("共解析到 %d 个代理节点,开始并发检测...", len(all_proxies)) + # 创建该订阅专用的 checker(使用订阅自己的并发数和超时) + sub_checker = ProxyChecker( + check_urls=self.checker.check_urls, + timeout=self.config.check.timeout, + max_concurrent=sub.max_concurrent, + ) - # 并发检测延迟 - links = [p.link for p, _ in all_proxies] - results = await self.checker.check_batch(links) - - # 处理检测结果 + threshold = sub.latency_threshold added = 0 updated = 0 - threshold = self.config.check.latency_threshold + skipped = 0 - for proxy, source in all_proxies: - latency = results.get(proxy.link) - if latency is not None and latency <= threshold: - # 延迟达标,尝试入库 + 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 + + 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) + continue + + if latency <= threshold: + # 延迟达标 existing = await self.db.get_proxy_by_link(proxy.link) if existing: await self.db.update_latency(existing.id, latency) updated += 1 else: - success = await self.db.insert_proxy(proxy, latency, source) + success = await self.db.insert_proxy(proxy, latency, sub.url) if success: added += 1 - elif latency is not None and latency > threshold: - # 延迟超标,如果已在库中则增加失败计数 + else: + # 延迟超标 existing = await self.db.get_proxy_by_link(proxy.link) if existing: await self.db.increment_fail(existing.id) + skipped += 1 + + # 有代理入库则重置空天数 + 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") self._last_fetch_time = datetime.now(timezone.utc).isoformat() - logger.info("拉取完成: 新增 %d, 更新 %d, 总解析 %d", added, updated, len(all_proxies)) + logger.info("订阅 #%d 拉取完成: 新增 %d, 更新 %d, 跳过 %d, 总解析 %d", + sub.id, added, updated, skipped, len(proxies)) + except Exception as e: + await self.db.update_fetch_status(sub_id, "failed") + logger.error("拉取订阅 #%d 异常: %s", sub.id, e, exc_info=True) + + async def fetch_and_check(self) -> None: + """手动触发:拉取所有启用的订阅并检测入库""" + if self._fetching: + logger.info("上一次拉取任务尚未完成,跳过本次") + return + + self._fetching = True + try: + subs = await self.db.get_enabled_subscriptions() + if not subs: + logger.warning("没有启用的订阅源") + return + + for sub in subs: + await self._fetch_single_subscription(sub.id) except Exception as e: logger.error("拉取任务异常: %s", e, exc_info=True) finally: self._fetching = False async def verify_stored_proxies(self) -> None: - """任务2:验证已存代理可用性 - - 1. 从数据库获取所有代理 - 2. 并发检测延迟 - 3. 成功的更新 latency + 重置 fail_count - 4. 失败的 fail_count + 1 - """ + """验证已存代理可用性 - 失败则累加 fail_count,连续3次失败由清理任务移除""" if self._verifying: logger.info("上一次验证任务尚未完成,跳过本次") return @@ -174,15 +244,59 @@ async def verify_stored_proxies(self) -> None: finally: self._verifying = False - async def cleanup_proxies(self) -> None: - """任务3:清理不合格代理 + 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) + return + + proxies = await self.db.get_proxies_by_source(sub.url) + if not proxies: + logger.info("订阅 #%d 没有代理,跳过验证", sub_id) + return + + logger.info("开始验证订阅 #%d 的 %d 个代理...", sub_id, len(proxies)) - 删除 fail_count >= max_fail_count 的代理 - """ - max_fail = self.config.scheduler.max_fail_count - deleted = await self.db.delete_proxies_by_fail_count(max_fail) + sub_checker = ProxyChecker( + check_urls=self.checker.check_urls, + timeout=self.config.check.timeout, + max_concurrent=sub.max_concurrent, + ) + links = [p.link for p in proxies] + results = await sub_checker.check_batch(links) + + success_count = 0 + fail_count = 0 + + 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 + else: + await self.db.increment_fail(proxy.id) + fail_count += 1 + + self._last_verify_time = datetime.now(timezone.utc).isoformat() + logger.info("订阅 #%d 验证完成: 成功 %d, 失败 %d", sub_id, success_count, fail_count) + + async def cleanup_proxies(self) -> None: + """清理不合格代理和空订阅源""" + # 清理连续3次验证失败的代理 + deleted = await self.db.delete_proxies_by_fail_count(3) if deleted > 0: - logger.info("清理完成: 删除 %d 个不合格代理 (fail_count >= %d)", deleted, max_fail) + logger.info("清理代理: 删除 %d 个连续3次不可用的代理", deleted) + + # 清理连续30天代理数为0的订阅源 + empty_subs = await self.db.get_subscriptions_with_empty_days(30) + 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]) @property def last_fetch_time(self) -> str: diff --git a/app/templates/base.html b/app/templates/base.html index 821e55e..3489b57 100644 --- a/app/templates/base.html +++ b/app/templates/base.html @@ -17,6 +17,8 @@ .status-unchecked { background-color: #fff3cd; color: #664d03; } .stat-card { transition: transform 0.2s; } .stat-card:hover { transform: translateY(-2px); } + th.sortable { cursor: pointer; user-select: none; } + th.sortable:hover { background-color: #e2e6ea; } {% block extra_head %}{% endblock %} @@ -33,7 +35,7 @@