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 new file mode 100644 index 0000000..40a8b45 --- /dev/null +++ b/.gitignore @@ -0,0 +1,29 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +dist/ +build/ +.eggs/ + +# 虚拟环境 +venv/ +.venv/ + +# 数据库 +data/ + +# IDE +.vscode/ +.idea/ + +# 资源文件(含敏感订阅地址) +resources/Subscription.txt + +# OS +.DS_Store + +# 环境变量 +.env +logs/** +logs/proxy_pool.log diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..6486f70 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,66 @@ +# Changelog + +All notable changes to this project will be documented in this file. + +## [1.0.0] - 2026-07-18 + +### 核心功能 + +- **订阅源管理** - 数据库驱动的订阅源增删改查,支持独立配置 Crontab、延迟阈值、重试次数和最大并发数 +- **Crontab 定时拉取** - 每个订阅源支持独立的 Crontab 表达式,通过 APScheduler CronTrigger 实现精准调度 +- **HTTP 延迟检测** - 通过代理发起 HTTP 请求到多个目标 URL 检测延迟,取多目标最大延迟值作为结果 +- **检测目标动态配置** - 检测目标 URL 存入数据库,页面可增删,修改后即时生效 +- **多协议解析** - 支持 vmess / vless / trojan / ss / hysteria2 五种协议的订阅解析和 Clash 配置生成 +- **对外订阅输出** - 提供 V2Ray(base64)和 Clash(YAML)格式订阅,仅包含当前可用代理 +- **自动添加订阅 API** - `POST /api/subscriptions/auto` 接口支持仅传 URL 自动创建订阅并触发拉取验证 + +### 代理管理 + +- **独立延迟检测** - 每个代理独立测试延迟,报错自动跳过 +- **延迟阈值过滤** - 延迟低于阈值自动入库,超标代理累加失败计数 +- **连续失败清理** - 连续 3 次验证失败的代理自动移除 +- **空订阅清理** - 连续 30 天代理数为 0 的订阅源自动删除 +- **按订阅源分组** - 可用代理按订阅源分组,Tab 切换查看 +- **延迟与入库时间排序** - 每个订阅源的代理列表支持延迟和入库时间升降序排序 + +### Web 界面 + +- **单页面管理** - 订阅源管理和可用代理列表合并为一个页面,无跳转 +- **协议分布饼状图** - Chart.js 环形图动态展示各协议代理数量和百分比 +- **订阅源状态** - 实时显示每个订阅源的拉取状态(更新中/成功/失败/待更新) +- **启用/禁用切换** - 状态列点击切换启用与禁用,启用后自动拉取并验证 +- **编辑弹窗** - 操作列仅保留编辑按钮,删除功能移至编辑弹窗内 +- **订阅地址提示** - 鼠标悬停订阅 ID 行或 Tab 标签时通过 Tooltip 显示订阅地址 +- **可用代理 Tab 过滤** - 仅显示有可用代理的订阅源标签页 + +### 运维特性 + +- **日志按天归档** - TimedRotatingFileHandler 实现日志午夜自动轮转 +- **日志自动清理** - 自动清理 7 天以前的归档日志 +- **双输出日志** - 同时输出到控制台和文件 +- **Docker Compose 部署** - 一键启动,数据持久化,时区配置 + +### 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/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/proxies/grouped` | 按订阅来源分组的可用代理 | 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..dbdbbae 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,198 @@ -# proxy_pool -proxy_pool +# ProxyPool v1.0.0 + +自动获取、检测、维护代理节点池,提供 Web 管理界面和订阅链接输出。 + +## 功能特性 + +- **订阅源管理** - 数据库驱动的增删改查,每个订阅源独立配置 Crontab、延迟阈值、重试次数、并发数 +- **Crontab 定时拉取** - 每个订阅源支持 5 位 Crontab 表达式精准调度 +- **HTTP 延迟检测** - 模拟通过代理访问目标网站,测量完整请求延迟(DNS + TCP + TLS + HTTP),多目标取最大值 +- **检测目标动态配置** - 检测目标 URL 存入数据库,页面可增删,修改即时生效,无需外部文件 +- **多协议支持** - vmess / vless / trojan / ss / hysteria2 解析与 Clash 配置生成 +- **自动清理** - 连续 3 次验证失败的代理自动移除;连续 30 天无代理的订阅源自动删除 +- **单页面管理** - 订阅源管理 + 可用代理列表在同一页面,按订阅源 Tab 切换 +- **协议分布图** - 饼状图动态展示各协议代理数量和百分比 +- **订阅输出** - 仅输出当前可用代理,支持 V2Ray(base64)和 Clash(YAML)格式 +- **日志归档** - 按天自动归档,自动清理 7 天前的日志 +- **Docker 部署** - Docker Compose 一键启动 + +## 快速开始 + +### Docker Compose(推荐) + +```bash +git clone https://github.com/your-username/proxy_pool.git +cd proxy_pool + +# 按需修改配置 +vim config.yaml + +# 启动服务 +docker-compose up -d +``` + +访问 http://localhost:8080 查看 Web 界面。 + +### 本地运行 + +```bash +# 安装依赖 +pip install -r requirements.txt + +# 启动服务 +python -m app.main +``` + +## 配置说明 + +编辑 `config.yaml`: + +```yaml +server: + host: "0.0.0.0" + port: 8080 + debug: false + +database: + path: "data/proxy_pool.db" + +check: + timeout: 5.0 # 检测超时(秒) + max_concurrent: 50 # 全局并发检测数 + latency_threshold: 1500.0 # 全局延迟阈值(毫秒) + +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`,首次启动自动写入数据库,后续在页面「检测目标」中管理,修改即时生效。 + +### 环境变量 + +| 变量 | 说明 | 默认值 | +|------|------|--------| +| `PROXY_POOL_PORT` | 服务端口 | 8080 | +| `PROXY_POOL_DB_PATH` | 数据库路径 | data/proxy_pool.db | + +## API 接口 + +### 代理 + +| 方法 | 路径 | 说明 | +|------|------|------| +| GET | `/api/proxies` | 可用代理列表 | +| GET | `/api/proxies/all` | 所有代理 | +| GET | `/api/proxies/grouped` | 按订阅来源分组的可用代理 | +| DELETE | `/api/proxies/{id}` | 删除代理 | + +### 订阅输出 + +| 方法 | 路径 | 说明 | +|------|------|------| +| GET | `/api/subscription/v2ray` | V2Ray 格式订阅(base64) | +| GET | `/api/subscription/clash` | Clash 格式订阅(YAML) | + +### 订阅源管理 + +| 方法 | 路径 | 说明 | +|------|------|------| +| GET | `/api/subscriptions` | 订阅源列表 | +| POST | `/api/subscriptions` | 添加订阅源 | +| POST | `/api/subscriptions/auto` | 自动添加(仅 URL 必填,自动拉取验证) | +| PUT | `/api/subscriptions/{sub_id}` | 更新订阅源 | +| DELETE | `/api/subscriptions/{sub_id}` | 删除订阅源 | + +### 拉取与验证 + +| 方法 | 路径 | 说明 | +|------|------|------| +| POST | `/api/fetch` | 拉取所有订阅 | +| POST | `/api/fetch/{sub_id}` | 拉取指定订阅 | +| POST | `/api/verify` | 验证所有代理 | +| POST | `/api/verify/{sub_id}` | 验证指定订阅代理 | + +### 检测目标 + +| 方法 | 路径 | 说明 | +|------|------|------| +| GET | `/api/check-urls` | 检测目标 URL 列表 | +| POST | `/api/check-urls` | 添加检测目标 URL | +| DELETE | `/api/check-urls/{url_id}` | 删除检测目标 URL | + +### 其他 + +| 方法 | 路径 | 说明 | +|------|------|------| +| 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 | ✅ | ✅ | + +## 延迟检测原理 + +延迟检测模拟真实上网场景:通过代理服务器访问目标检测 URL,测量完整请求延迟。 + +``` +客户端 → 代理服务器 → 目标网站 + ├── DNS 解析 + ├── TCP 连接建立 + ├── TLS 握手(HTTPS 目标) + └── HTTP 请求/响应 +``` + +- 检测目标默认为 `https://www.google.com/generate_204` 和 `https://www.gstatic.com/generate_204` +- 多个目标取最大延迟值,确保所有目标均可达 +- 相比仅测试 TCP/TLS 连通性,HTTP 请求延迟更贴近真实上网体验 + +## 项目结构 + +``` +proxy_pool/ +├── app/ +│ ├── __init__.py # 应用工厂 & 全局单例 +│ ├── main.py # 启动入口 +│ ├── config.py # YAML 配置加载 +│ ├── database.py # aiosqlite 异步数据库操作 +│ ├── models.py # 数据模型(ProxyInfo / ProxyDBRecord / SubscriptionRecord) +│ ├── parser.py # 订阅拉取 & 解析(5 协议) +│ ├── checker.py # HTTP 延迟检测(多目标取最大值) +│ ├── generator.py # V2Ray / Clash 订阅生成 +│ ├── scheduler.py # APScheduler 定时任务调度 +│ ├── routers/ +│ │ ├── api.py # REST API 路由 +│ │ └── web.py # Web 页面路由 +│ └── templates/ +│ ├── base.html # 基础模板 +│ ├── index.html # 主页面(管理 + 代理列表) +│ └── subscription.html # 订阅链接页 +├── logs/ # 日志目录(自动归档) +├── config.yaml # 配置文件 +├── docker-compose.yaml # Docker 编排 +├── Dockerfile # Docker 镜像 +├── CHANGELOG.md # 变更日志 +└── requirements.txt # Python 依赖 +``` + +## License + +MIT License diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..f8384a4 --- /dev/null +++ b/app/__init__.py @@ -0,0 +1,145 @@ +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, DEFAULT_CHECK_URLS +from app.config import AppConfig, load_config +from app.database import ProxyDatabase +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) + + # 配置日志(控制台 + 按天归档文件,保留7天) + log_level = logging.DEBUG if _config.server.debug else logging.INFO + 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) + + # 初始化模板 + template_dir = Path(__file__).parent / "templates" + _templates = Jinja2Templates(directory=str(template_dir)) + # 禁用 Jinja2 模板缓存,避免新版 Starlette 缓存兼容性问题 + _templates.env.cache_size = 0 + + # 创建 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:仅当数据库为空时插入默认值 + await _db.init_check_urls(DEFAULT_CHECK_URLS) + + # 从数据库加载检测 URL + db_check_urls = await _db.get_check_urls() + check_urls = [u["url"] for u in db_check_urls] or DEFAULT_CHECK_URLS + 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..d953833 --- /dev/null +++ b/app/checker.py @@ -0,0 +1,251 @@ +from __future__ import annotations +"""代理延迟检测模块 - 通过 TCP/TLS 连接检测代理服务器响应速度 + +检测策略: +- 对代理服务器发起 TCP 连接,测量连接建立耗时 +- 若协议使用 TLS(vmess+tls / vless+tls / trojan / hysteria2), + 则在 TCP 连接基础上完成 TLS 握手,测量总耗时 +- TCP+TLS 延迟与实际上网体验高度相关:它反映到代理服务器的网络往返时间 + 和服务端处理能力,是真实代理请求延迟的主要组成部分 +- 使用 asyncio.Semaphore 控制并发数 +- 检测目标 URL 存入数据库供页面展示和未来扩展 +""" + +import asyncio +import json +import logging +import ssl +import time +import base64 +from dataclasses import dataclass +from urllib.parse import urlparse, unquote + +from app.models import ProxyInfo + +logger = logging.getLogger(__name__) + +# 默认检测目标 URL(数据库为空时使用,页面展示用) +DEFAULT_CHECK_URLS = [ + "https://www.google.com/generate_204", + "https://www.gstatic.com/generate_204", +] + + +@dataclass +class ProxyConnInfo: + """代理服务器连接信息""" + host: str + port: int + use_tls: bool + protocol: str + + +class ProxyChecker: + """基于 asyncio 的并发代理延迟检测器 + + 通过 TCP/TLS 连接测试代理服务器响应速度。 + 延迟值 = TCP 连接耗时 (+ TLS 握手耗时),单位毫秒。 + 该值与实际通过代理浏览网页的延迟高度正相关。 + """ + + def __init__(self, check_urls: list[str], timeout: float, max_concurrent: int): + self.check_urls = check_urls or DEFAULT_CHECK_URLS + self.timeout = timeout + self.semaphore = asyncio.Semaphore(max_concurrent) + + async def check_proxy(self, link: str) -> float | None: + """检测单个代理的延迟 + + 通过 TCP/TLS 连接测试代理服务器响应速度。 + 返回连接建立耗时(毫秒),失败返回 None。 + """ + conn_info = self._parse_link(link) + if not conn_info: + return None + + async with self.semaphore: + return await self._check_connectivity(conn_info) + + async def _check_connectivity(self, info: ProxyConnInfo) -> float | None: + """测试代理服务器 TCP/TLS 连通性并测量延迟""" + start = time.monotonic() + try: + if info.use_tls: + # TLS 协议:TCP 连接 + TLS 握手,更贴近真实代理请求体验 + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + _, writer = await asyncio.wait_for( + asyncio.open_connection( + info.host, info.port, ssl=ctx, server_hostname=info.host + ), + timeout=self.timeout, + ) + else: + # 非 TLS 协议:仅 TCP 连接 + _, writer = await asyncio.wait_for( + asyncio.open_connection(info.host, info.port), + timeout=self.timeout, + ) + + elapsed = (time.monotonic() - start) * 1000 + writer.close() + try: + await writer.wait_closed() + except Exception: + pass + return round(elapsed, 1) + except Exception: + return None + + 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) + + # ---- 协议解析 ---- + + def _parse_link(self, link: str) -> ProxyConnInfo | None: + """解析分享链接,提取连接信息(地址、端口、是否 TLS)""" + try: + if link.startswith("vmess://"): + return self._parse_vmess(link) + elif link.startswith("vless://"): + return self._parse_vless(link) + elif link.startswith("trojan://"): + return self._parse_trojan(link) + elif link.startswith("ss://"): + return self._parse_ss(link) + elif link.startswith("hysteria2://") or link.startswith("hy2://"): + return self._parse_hysteria2(link) + except Exception: + pass + return None + + def _parse_vmess(self, link: str) -> ProxyConnInfo | None: + 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) + address = config.get("add", "") + port = config.get("port", 0) + if not address or not port: + return None + # vmess 的 tls 字段为 "tls" 或 "reality" 时使用 TLS + use_tls = config.get("tls", "") in ("tls", "reality") + return ProxyConnInfo( + host=address, port=int(port), + use_tls=use_tls, protocol="vmess", + ) + except Exception: + return None + + def _parse_vless(self, link: str) -> ProxyConnInfo | None: + try: + parsed = urlparse(link) + address = parsed.hostname or "" + port = parsed.port or 0 + if not address or not port: + return None + params = dict(pair.split("=", 1) for pair in parsed.query.split("&") if "=" in pair) + security = params.get("security", "none") + use_tls = security in ("tls", "reality") + return ProxyConnInfo( + host=address, port=int(port), + use_tls=use_tls, protocol="vless", + ) + except Exception: + return None + + def _parse_trojan(self, link: str) -> ProxyConnInfo | None: + try: + parsed = urlparse(link) + address = parsed.hostname or "" + port = parsed.port or 0 + if not address or not port: + return None + # trojan 默认使用 TLS + return ProxyConnInfo( + host=address, port=int(port), + use_tls=True, protocol="trojan", + ) + except Exception: + return None + + def _parse_ss(self, link: str) -> ProxyConnInfo | None: + try: + line = link + if "#" in line: + line = line[:line.rindex("#")] + + ss_content = line[5:] + address = "" + port = 0 + + 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 = int(addr_port[bracket_end + 2:]) if bracket_end + 2 < len(addr_port) else 0 + elif ":" in addr_port: + address, port_str = addr_port.rsplit(":", 1) + port = int(port_str) + else: + try: + padding = 4 - len(ss_content) % 4 + if padding != 4: + ss_content += "=" * padding + decoded = base64.b64decode(ss_content).decode("utf-8") + if "@" in decoded: + _, addr_port = decoded.rsplit("@", 1) + if ":" in addr_port: + address, port_str = addr_port.rsplit(":", 1) + port = int(port_str) + except Exception: + pass + + if not address or not port: + return None + return ProxyConnInfo( + host=address, port=port, + use_tls=False, protocol="ss", + ) + except Exception: + return None + + def _parse_hysteria2(self, link: str) -> ProxyConnInfo | None: + try: + prefix_len = len("hysteria2://") if link.startswith("hysteria2://") else len("hy2://") + rest = link[prefix_len:] + # 构造标准 URL 以便 urlparse 解析 + parsed = urlparse("http://" + rest) + address = parsed.hostname or "" + port = parsed.port or 0 + if not address or not port: + return None + # hysteria2 默认使用 TLS + return ProxyConnInfo( + host=address, port=int(port), + use_tls=True, protocol="hysteria2", + ) + except Exception: + return None diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..322b4f1 --- /dev/null +++ b/app/config.py @@ -0,0 +1,103 @@ +"""配置加载模块 - 从 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: + 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 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) + + +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 = CheckConfig(**data.get("check", {})) + scheduler = SchedulerConfig(**data.get("scheduler", {})) + return AppConfig( + server=server, + database=database, + check=check, + scheduler=scheduler, + ) + + +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": { + "timeout": 5.0, + "max_concurrent": 50, + "latency_threshold": 1500.0, + }, + "scheduler": { + "fetch_interval": 3600, + "verify_interval": 1800, + "cleanup_interval": 7200, + "max_fail_count": 3, + }, + } + + 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..c409fb3 --- /dev/null +++ b/app/database.py @@ -0,0 +1,479 @@ +from __future__ import annotations +"""数据库操作层 - aiosqlite 异步封装""" + +import os +from datetime import datetime, timezone + +import aiosqlite + +from app.models import ProxyDBRecord, ProxyInfo, SubscriptionRecord + + +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); + + 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); + + CREATE TABLE IF NOT EXISTS check_urls ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT NOT NULL UNIQUE + ); + """) + await self._db.commit() + + # 兼容旧表:添加 check_urls 表(如不存在) + try: + await self._db.execute( + "CREATE TABLE IF NOT EXISTS check_urls (id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT NOT NULL UNIQUE)" + ) + await self._db.commit() + except Exception: + pass + + # 迁移:将旧的 http:// 检测 URL 升级为 https:// + try: + await self._db.execute( + "UPDATE check_urls SET url = REPLACE(url, 'http://', 'https://') WHERE url LIKE 'http://%'" + ) + await self._db.commit() + except Exception: + pass + + # 兼容旧表:添加 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: + 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: + 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) + 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 cursor.rowcount > 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] + + 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") + 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 "" + + # ---- 检测目标 URL ---- + + async def get_check_urls(self) -> list[dict]: + """获取所有检测目标 URL""" + cursor = await self._db.execute("SELECT id, url FROM check_urls ORDER BY id ASC") + rows = await cursor.fetchall() + return [{"id": row["id"], "url": row["url"]} for row in rows] + + async def add_check_url(self, url: str) -> dict | None: + """添加检测目标 URL,重复则忽略""" + try: + cursor = await self._db.execute( + "INSERT INTO check_urls (url) VALUES (?)", (url,) + ) + await self._db.commit() + return {"id": cursor.lastrowid, "url": url} + except aiosqlite.IntegrityError: + return None + + async def delete_check_url(self, url_id: int) -> bool: + """删除检测目标 URL""" + cursor = await self._db.execute("DELETE FROM check_urls WHERE id = ?", (url_id,)) + await self._db.commit() + return cursor.rowcount > 0 + + async def init_check_urls(self, urls: list[str]) -> None: + """初始化检测 URL(仅在表为空时插入)""" + cursor = await self._db.execute("SELECT COUNT(*) as cnt FROM check_urls") + row = await cursor.fetchone() + if row["cnt"] == 0 and urls: + for url in urls: + try: + await self._db.execute("INSERT INTO check_urls (url) VALUES (?)", (url,)) + except aiosqlite.IntegrityError: + pass + await self._db.commit() 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..df1fb45 --- /dev/null +++ b/app/models.py @@ -0,0 +1,57 @@ +"""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" + + +@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/parser.py b/app/parser.py new file mode 100644 index 0000000..cac96c1 --- /dev/null +++ b/app/parser.py @@ -0,0 +1,213 @@ +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 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..ebc9f49 --- /dev/null +++ b/app/routers/api.py @@ -0,0 +1,371 @@ +"""REST API 路由 - JSON 格式的代理数据接口""" + +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 + +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("/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(): + """手动触发验证代理""" + scheduler = get_scheduler() + if scheduler._verifying: + return {"message": "验证任务正在进行中"} + asyncio.create_task(scheduler.verify_stored_proxies()) + 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(): + """获取统计信息""" + 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"} + + +# ---- 订阅管理 ---- + +@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 { + "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", + }, + ) + + +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, + } + + +# ---- 检测目标 URL ---- + +@router.get("/check-urls") +async def get_check_urls(): + """获取所有检测目标 URL""" + db = get_db() + urls = await db.get_check_urls() + return {"total": len(urls), "urls": urls} + + +@router.post("/check-urls") +async def add_check_url(url: str = ""): + """添加检测目标 URL""" + from fastapi import HTTPException + if not url: + raise HTTPException(status_code=400, detail="URL 不能为空") + db = get_db() + result = await db.add_check_url(url) + if not result: + raise HTTPException(status_code=409, detail="URL 已存在") + # 更新 checker 的 check_urls + checker = get_checker() + if checker: + checker.check_urls = [u["url"] for u in await db.get_check_urls()] + return result + + +@router.delete("/check-urls/{url_id}") +async def delete_check_url(url_id: int): + """删除检测目标 URL""" + from fastapi import HTTPException + db = get_db() + success = await db.delete_check_url(url_id) + if not success: + raise HTTPException(status_code=404, detail="URL 不存在") + # 更新 checker 的 check_urls + checker = get_checker() + if checker: + checker.check_urls = [u["url"] for u in await db.get_check_urls()] + return {"message": "deleted"} diff --git a/app/routers/web.py b/app/routers/web.py new file mode 100644 index 0000000..e1e6f9f --- /dev/null +++ b/app/routers/web.py @@ -0,0 +1,78 @@ +from __future__ import annotations +"""Web UI 页面路由 - Jinja2 模板渲染""" + +import json +import logging +from dataclasses import asdict + +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() + subscriptions = await db.get_all_subscriptions() + # 按来源分组的可用代理 + grouped = await db.get_proxies_grouped_by_source(config.check.latency_threshold) + # 检测目标 URL + check_urls = await db.get_check_urls() + # 计算每个订阅源的可用代理数量 + 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( + request, + "index.html", + { + "proxies": proxies, + "stats": stats, + "subscriptions": subscriptions, + "subscriptions_json": subscriptions_json, + "sub_available_counts": sub_available_counts, + "check_urls": check_urls, + "grouped": grouped, + "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( + request, + "subscription.html", + { + "v2ray_url": v2ray_url, + "clash_url": clash_url, + }, + ) diff --git a/app/scheduler.py b/app/scheduler.py new file mode 100644 index 0000000..7ea61b4 --- /dev/null +++ b/app/scheduler.py @@ -0,0 +1,307 @@ +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 + +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.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("调度器已启动: 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_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 + + try: + 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 + + logger.info("订阅 #%d: 解析到 %d 个节点,开始逐个检测...", sub.id, len(proxies)) + + # 创建该订阅专用的 checker(使用订阅自己的并发数和超时) + sub_checker = ProxyChecker( + check_urls=self.checker.check_urls, + timeout=self.config.check.timeout, + max_concurrent=sub.max_concurrent, + ) + + threshold = sub.latency_threshold + added = 0 + updated = 0 + skipped = 0 + + for proxy in proxies: + # 每个代理独立检测延迟,报错自动跳过 + try: + latency = await sub_checker.check_proxy(proxy.link) + except Exception as e: + logger.debug("检测代理 %s 异常,跳过: %s", proxy.name[:30], e) + skipped += 1 + continue + + 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, sub.url) + if success: + added += 1 + 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, 跳过 %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: + """验证已存代理可用性 - 失败则累加 fail_count,连续3次失败由清理任务移除""" + 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 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)) + + 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 个连续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: + 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..3489b57 --- /dev/null +++ b/app/templates/base.html @@ -0,0 +1,69 @@ + + +
+ + +| # | +Crontab | +总数 | +可用数 | +延迟 | +并发 | +状态 | +更新 | +操作 | +
|---|---|---|---|---|---|---|---|---|
| {{ sub.id }} | +{{ sub.crontab }} |
+ {{ sub.total_count }} | +{{ sub_available_counts.get(sub.url, 0) }} | +{{ sub.latency_threshold }}ms | +{{ sub.max_concurrent }} | ++ + {{ '启用' if sub.enabled else '禁用' }} + + | ++ {% if sub.fetch_status == 'updating' %} + 更新中 + {% elif sub.fetch_status == 'success' %} + 成功 + {% elif sub.fetch_status == 'failed' %} + 失败 + {% else %} + 待更新 + {% 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 '-' }} | ++ + | +
| 暂无可用代理,请先拉取此订阅 | |||||||
将以下链接复制到 v2rayN、Clash 等客户端中使用。
+ + +