From effb9506eb5cb5a1081e8a1a588e346841cca9ce Mon Sep 17 00:00:00 2001 From: siruoren Date: Sun, 19 Jul 2026 00:36:24 +0800 Subject: [PATCH 01/19] =?UTF-8?q?refactor:=20=E5=B0=86=E9=A1=B9=E7=9B=AE?= =?UTF-8?q?=E4=BB=8EProxyPool=E9=87=8D=E5=91=BD=E5=90=8D=E4=B8=BANetHub?= =?UTF-8?q?=E5=B9=B6=E6=96=B0=E5=A2=9E=E6=9C=8D=E5=8A=A1=E5=AE=9E=E4=BE=8B?= =?UTF-8?q?=E6=BA=90=E7=AE=A1=E7=90=86=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 全量替换项目名称、描述等元数据，将代理池重命名为节点池 2. 新增InstanceSourceRecord数据模型与数据库表 3. 实现服务实例源的增删改查API与管理页面 4. 添加从远程服务实例拉取节点的定时任务与匹配逻辑 5. 更新前端页面与模板，适配新的节点池管理界面 --- app/__init__.py | 4 +- app/database.py | 122 +++++++++++++++- app/generator.py | 4 +- app/main.py | 2 +- app/models.py | 18 ++- app/parser.py | 203 ++++++++++++++++++++++++++ app/routers/api.py | 135 +++++++++++++++++- app/routers/web.py | 8 ++ app/scheduler.py | 142 ++++++++++++++++++- app/templates/base.html | 6 +- app/templates/index.html | 244 +++++++++++++++++++++++++++++++- app/templates/subscription.html | 6 +- 12 files changed, 873 insertions(+), 21 deletions(-) diff --git a/app/__init__.py b/app/__init__.py index f8384a4..ecd7543 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -89,9 +89,9 @@ def create_app(config_path: str = "config.yaml") -> FastAPI: # 创建 FastAPI 实例 app = FastAPI( - title="ProxyPool", + title="NetHub", version="1.0.0", - description="代理池管理系统", + description="节点池管理系统", ) # 注册生命周期事件 diff --git a/app/database.py b/app/database.py index c409fb3..27b5e28 100644 --- a/app/database.py +++ b/app/database.py @@ -6,7 +6,7 @@ import aiosqlite -from app.models import ProxyDBRecord, ProxyInfo, SubscriptionRecord +from app.models import ProxyDBRecord, ProxyInfo, SubscriptionRecord, InstanceSourceRecord class ProxyDatabase: @@ -67,6 +67,22 @@ async def init(self) -> None: id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT NOT NULL UNIQUE ); + + CREATE TABLE IF NOT EXISTS instance_sources ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + base_url TEXT NOT NULL UNIQUE, + username TEXT NOT NULL DEFAULT '', + password TEXT NOT NULL DEFAULT '', + crontab TEXT NOT NULL DEFAULT '*/10 * * * *', + latency_threshold REAL DEFAULT 1500.0, + max_concurrent INTEGER DEFAULT 50, + enabled INTEGER DEFAULT 1, + created_at TEXT DEFAULT '', + total_count INTEGER DEFAULT 0, + fetch_status TEXT DEFAULT 'idle' + ); + + CREATE INDEX IF NOT EXISTS idx_instance_sources_base_url ON instance_sources(base_url); """) await self._db.commit() @@ -477,3 +493,107 @@ async def init_check_urls(self, urls: list[str]) -> None: except aiosqlite.IntegrityError: pass await self._db.commit() + + # ---- 服务实例源管理 ---- + + def _row_to_instance_source(self, row: aiosqlite.Row) -> InstanceSourceRecord: + """将数据库行转为 InstanceSourceRecord""" + return InstanceSourceRecord( + id=row["id"], + base_url=row["base_url"], + username=row["username"], + password=row["password"], + crontab=row["crontab"], + latency_threshold=row["latency_threshold"], + max_concurrent=row["max_concurrent"], + enabled=bool(row["enabled"]), + created_at=row["created_at"], + total_count=row["total_count"] if "total_count" in row.keys() else 0, + fetch_status=row["fetch_status"] if "fetch_status" in row.keys() else "idle", + ) + + async def add_instance_source(self, base_url: str, username: str, password: str, + crontab: str = "*/10 * * * *", + latency_threshold: float = 1500.0, + max_concurrent: int = 50, + enabled: bool = True) -> InstanceSourceRecord | None: + """添加服务实例源""" + now = datetime.now(timezone.utc).isoformat() + try: + cursor = await self._db.execute( + """INSERT INTO instance_sources + (base_url, username, password, crontab, latency_threshold, max_concurrent, enabled, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + (base_url, username, password, crontab, latency_threshold, max_concurrent, int(enabled), now), + ) + await self._db.commit() + return await self.get_instance_source_by_id(cursor.lastrowid) + except aiosqlite.IntegrityError: + return None + + async def get_instance_source_by_id(self, source_id: int) -> InstanceSourceRecord | None: + """根据 ID 获取服务实例源""" + cursor = await self._db.execute("SELECT * FROM instance_sources WHERE id = ?", (source_id,)) + row = await cursor.fetchone() + return self._row_to_instance_source(row) if row else None + + async def get_instance_source_by_url(self, base_url: str) -> InstanceSourceRecord | None: + """根据 base_url 获取服务实例源""" + cursor = await self._db.execute("SELECT * FROM instance_sources WHERE base_url = ?", (base_url,)) + row = await cursor.fetchone() + return self._row_to_instance_source(row) if row else None + + async def get_all_instance_sources(self) -> list[InstanceSourceRecord]: + """获取所有服务实例源""" + cursor = await self._db.execute("SELECT * FROM instance_sources ORDER BY id ASC") + rows = await cursor.fetchall() + return [self._row_to_instance_source(row) for row in rows] + + async def get_enabled_instance_sources(self) -> list[InstanceSourceRecord]: + """获取所有启用的服务实例源""" + cursor = await self._db.execute("SELECT * FROM instance_sources WHERE enabled = 1 ORDER BY id ASC") + rows = await cursor.fetchall() + return [self._row_to_instance_source(row) for row in rows] + + async def update_instance_source(self, source_id: int, **kwargs) -> bool: + """更新服务实例源，支持部分字段更新""" + allowed = {"base_url", "username", "password", "crontab", "latency_threshold", "max_concurrent", "enabled"} + updates = {} + for k, v in kwargs.items(): + if k in allowed: + if k == "enabled": + updates[k] = int(v) + else: + updates[k] = v + if not updates: + return False + + set_clause = ", ".join(f"{k} = ?" for k in updates) + values = list(updates.values()) + [source_id] + cursor = await self._db.execute( + f"UPDATE instance_sources SET {set_clause} WHERE id = ?", values + ) + await self._db.commit() + return cursor.rowcount > 0 + + async def delete_instance_source(self, source_id: int) -> bool: + """删除服务实例源""" + cursor = await self._db.execute("DELETE FROM instance_sources WHERE id = ?", (source_id,)) + await self._db.commit() + return cursor.rowcount > 0 + + async def update_instance_total_count(self, source_id: int, count: int) -> None: + """更新服务实例源最新一次获取的节点总数""" + await self._db.execute( + "UPDATE instance_sources SET total_count = ? WHERE id = ?", + (count, source_id), + ) + await self._db.commit() + + async def update_instance_fetch_status(self, source_id: int, status: str) -> None: + """更新服务实例源拉取状态: idle / updating / success / failed""" + await self._db.execute( + "UPDATE instance_sources SET fetch_status = ? WHERE id = ?", + (status, source_id), + ) + await self._db.commit() diff --git a/app/generator.py b/app/generator.py index 97ac984..b58f31a 100644 --- a/app/generator.py +++ b/app/generator.py @@ -45,7 +45,7 @@ def generate_clash_subscription(proxies: list[ProxyDBRecord]) -> str: "proxies": proxy_list, "proxy-groups": [ { - "name": "ProxyPool", + "name": "NetHub", "type": "url-test", "proxies": proxy_names if proxy_names else ["DIRECT"], "url": "http://www.gstatic.com/generate_204", @@ -54,7 +54,7 @@ def generate_clash_subscription(proxies: list[ProxyDBRecord]) -> str: { "name": "Proxy", "type": "select", - "proxies": ["ProxyPool", "DIRECT"] + proxy_names, + "proxies": ["NetHub", "DIRECT"] + proxy_names, }, ], "rules": [ diff --git a/app/main.py b/app/main.py index 5276bcb..5f4c2d9 100644 --- a/app/main.py +++ b/app/main.py @@ -1,4 +1,4 @@ -"""ProxyPool 启动入口""" +"""NetHub 启动入口""" import uvicorn diff --git a/app/models.py b/app/models.py index df1fb45..81bab22 100644 --- a/app/models.py +++ b/app/models.py @@ -1,4 +1,4 @@ -"""ProxyPool 数据模型""" +"""NetHub 数据模型""" from dataclasses import dataclass @@ -55,3 +55,19 @@ class SubscriptionRecord: empty_days: int # 连续代理数为0的天数 total_count: int # 最新一次拉取的代理地址总数 fetch_status: str # 拉取状态: idle / updating / success / failed + + +@dataclass +class InstanceSourceRecord: + """服务实例源记录 - 从已连接的转发服务实例获取可用节点配置""" + id: int + base_url: str # 服务实例地址 + username: str # 登录用户名 + password: str # 登录密码 + crontab: str # crontab 格式的定时表达式 + latency_threshold: float # 延迟阈值(ms) + max_concurrent: int # 异步最大并发数 + enabled: bool # 是否启用 + created_at: str # ISO8601 + total_count: int # 最新一次获取的节点总数 + fetch_status: str # 拉取状态: idle / updating / success / failed diff --git a/app/parser.py b/app/parser.py index cac96c1..e378b53 100644 --- a/app/parser.py +++ b/app/parser.py @@ -6,6 +6,7 @@ """ import base64 +import difflib import json import logging from urllib.parse import urlparse, unquote @@ -211,3 +212,205 @@ def _parse_hysteria2(line: str) -> ProxyInfo | None: ) except Exception: return None + + +# ---- 服务实例 API 获取 ---- + +async def instance_login(base_url: str, username: str, password: str, + timeout: float = 10.0) -> tuple[aiohttp.ClientSession, dict]: + """登录服务实例，返回 (session, headers)""" + session = aiohttp.ClientSession() + headers = {"User-Agent": "Mozilla/5.0", "Content-Type": "application/json"} + try: + async with session.post( + f"{base_url}/api/login", + json={"username": username, "password": password}, + headers=headers, + timeout=aiohttp.ClientTimeout(total=timeout), + ) as resp: + data = await resp.json() + if data.get("code") != "SUCCESS": + await session.close() + raise Exception(f"登录失败: {data}") + token = data["data"]["token"] + headers["Authorization"] = token + return session, headers + except Exception: + await session.close() + raise + + +async def get_instance_connected_nodes( + session: aiohttp.ClientSession, headers: dict, base_url: str, + timeout: float = 10.0, +) -> tuple[list[dict], list[dict]]: + """获取服务实例已连接节点和订阅列表 + + 返回 (connected_nodes, subscriptions) + connected_nodes: [{"id", "sub_index", "name", "address", "net", "outbound"}, ...] + subscriptions: 原始订阅列表 + """ + async with session.get( + f"{base_url}/api/touch", headers=headers, + timeout=aiohttp.ClientTimeout(total=timeout), + ) as resp: + data = await resp.json() + + connected = data["data"]["touch"]["connectedServer"] + subscriptions = data["data"]["touch"]["subscriptions"] + + connected_nodes = [] + for conn in connected: + node_id = conn["id"] + sub_index = conn["sub"] + outbound = conn.get("outbound", "proxy") + + if sub_index < 0: + logger.debug("跳过手动添加的服务器 id=%d（无订阅源）", node_id) + continue + + if sub_index < len(subscriptions): + sub = subscriptions[sub_index] + for server in sub.get("servers", []): + if server["id"] == node_id: + connected_nodes.append({ + "id": node_id, + "sub_index": sub_index, + "name": server.get("name"), + "address": server.get("address"), + "net": server.get("net"), + "outbound": outbound, + }) + break + return connected_nodes, subscriptions + + +def _normalize_string(s: str) -> str: + """标准化字符串用于模糊匹配""" + if not s: + return "" + return s.replace(" ", "").replace("-", "").replace("_", "").lower() + + +def _strings_similar(s1: str, s2: str, threshold: float = 0.5) -> bool: + """判断两个字符串是否相似""" + if not s1 or not s2: + return False + s1_norm = _normalize_string(s1) + s2_norm = _normalize_string(s2) + if not s1_norm or not s2_norm: + return False + if s1_norm in s2_norm or s2_norm in s1_norm: + return True + ratio = difflib.SequenceMatcher(None, s1_norm, s2_norm).ratio() + return ratio >= threshold + + +async def fetch_connected_proxies( + base_url: str, username: str, password: str, +) -> list[ProxyInfo]: + """从服务实例获取所有已连接节点的分享链接 + + 流程： + 1. 登录服务实例 + 2. 获取已连接节点列表和订阅列表 + 3. 遍历每个订阅源，拉取并解析节点列表 + 4. 将已连接节点与订阅中的节点进行匹配（名称+地址模糊匹配） + 5. 返回匹配成功的 ProxyInfo 列表 + """ + session, headers = await instance_login(base_url, username, password) + try: + connected_nodes, subscriptions = await get_instance_connected_nodes( + session, headers, base_url, + ) + logger.info("服务实例 %s: 已连接 %d 个节点, 共 %d 个订阅源", + base_url, len(connected_nodes), len(subscriptions)) + + matched_proxies: list[ProxyInfo] = [] + matched_node_ids: set[int] = set() + + for sub_index, sub in enumerate(subscriptions): + sub_address = sub.get("address") + if not sub_address: + continue + + # 拉取订阅内容 + try: + async with session.get( + sub_address, headers=headers, + timeout=aiohttp.ClientTimeout(total=15.0), + ) as sub_resp: + content = await sub_resp.text() + share_links = parse_subscription(content) + logger.info("订阅源 %d: 解析到 %d 个节点", sub_index, len(share_links)) + except Exception as e: + logger.warning("订阅源 %d 加载失败: %s", sub_index, e) + continue + + # 该订阅源下的已连接节点 + sub_connected = [ + n for n in connected_nodes + if n["sub_index"] == sub_index and n["id"] not in matched_node_ids + ] + + # 第一次匹配：名称+地址模糊匹配 + unmatched = [] + for conn_node in sub_connected: + if conn_node["id"] in matched_node_ids: + continue + best_match = None + best_quality = 0 + for link_info in share_links: + quality = 0 + name_sim = _strings_similar(conn_node["name"], link_info.name) + addr_sim = _strings_similar(conn_node["address"], link_info.address) + if name_sim and addr_sim: + quality = 3 + elif name_sim: + quality = 2 + elif addr_sim: + quality = 1 + if quality > best_quality: + best_quality = quality + best_match = link_info + + if best_match and best_quality >= 1: + matched_node_ids.add(conn_node["id"]) + matched_proxies.append(best_match) + else: + unmatched.append(conn_node) + + # 第二次匹配：地址精确匹配 或 地址模糊+端口匹配 + for conn_node in unmatched: + if conn_node["id"] in matched_node_ids: + continue + best_match = None + best_quality = 0 + for link_info in share_links: + quality = 0 + addr_exact = (conn_node["address"] and link_info.address + and conn_node["address"] == link_info.address) + addr_fuzzy = _strings_similar(conn_node["address"], link_info.address, 0.7) + port_match = (conn_node.get("net") and link_info.port + and str(conn_node["net"]) == str(link_info.port)) + if addr_exact: + quality = 4 + elif addr_fuzzy and port_match: + quality = 3 + elif addr_fuzzy: + quality = 2 + elif port_match and _strings_similar(conn_node["name"], link_info.name, 0.3): + quality = 1 + if quality > best_quality: + best_quality = quality + best_match = link_info + + if best_match and best_quality >= 1: + matched_node_ids.add(conn_node["id"]) + matched_proxies.append(best_match) + + logger.info("服务实例 %s: 共匹配到 %d 个已连接节点配置", + base_url, len(matched_proxies)) + return matched_proxies + finally: + await session.close() diff --git a/app/routers/api.py b/app/routers/api.py index ebc9f49..a8d731d 100644 --- a/app/routers/api.py +++ b/app/routers/api.py @@ -307,7 +307,7 @@ def _subscription_response(content: str, content_type: str): headers={ "Content-Disposition": 'attachment; filename="subscription"', "Profile-Update-Interval": "24", - "Profile-Title": "ProxyPool", + "Profile-Title": "NetHub", }, ) @@ -369,3 +369,136 @@ async def delete_check_url(url_id: int): if checker: checker.check_urls = [u["url"] for u in await db.get_check_urls()] return {"message": "deleted"} + + +# ---- 服务实例源管理 ---- + +@router.get("/instance-sources") +async def get_instance_sources(): + """获取所有服务实例源""" + db = get_db() + sources = await db.get_all_instance_sources() + return { + "total": len(sources), + "sources": [_instance_source_to_dict(s) for s in sources], + } + + +class InstanceSourceCreate(BaseModel): + """添加服务实例源请求体""" + base_url: str + username: str + password: str + crontab: Optional[str] = "*/10 * * * *" + latency_threshold: Optional[float] = 1500.0 + max_concurrent: Optional[int] = 50 + + +@router.post("/instance-sources") +async def add_instance_source(req: InstanceSourceCreate): + """添加服务实例源""" + from fastapi import HTTPException + if not req.base_url: + raise HTTPException(status_code=400, detail="服务实例地址不能为空") + db = get_db() + source = await db.add_instance_source( + base_url=req.base_url, + username=req.username, + password=req.password, + crontab=req.crontab or "*/10 * * * *", + latency_threshold=req.latency_threshold or 1500.0, + max_concurrent=req.max_concurrent or 50, + enabled=True, + ) + if not source: + raise HTTPException(status_code=409, detail="服务实例地址已存在") + # 注册定时任务 + scheduler = get_scheduler() + scheduler._add_instance_source_job(source) + # 自动触发首次获取 + asyncio.create_task(scheduler._fetch_single_instance_source(source.id)) + return _instance_source_to_dict(source) + + +@router.put("/instance-sources/{source_id}") +async def update_instance_source(source_id: int, base_url: str = None, username: str = None, + password: str = None, crontab: str = None, + latency_threshold: float = None, max_concurrent: int = None, + enabled: bool = None): + """更新服务实例源""" + from fastapi import HTTPException + db = get_db() + kwargs = {} + if base_url is not None: + kwargs["base_url"] = base_url + if username is not None: + kwargs["username"] = username + if password is not None: + kwargs["password"] = password + if crontab is not None: + kwargs["crontab"] = crontab + if latency_threshold is not None: + kwargs["latency_threshold"] = latency_threshold + if max_concurrent is not None: + kwargs["max_concurrent"] = max_concurrent + if enabled is not None: + kwargs["enabled"] = enabled + + success = await db.update_instance_source(source_id, **kwargs) + if not success: + raise HTTPException(status_code=404, detail="服务实例源不存在或无更新") + # 刷新定时任务 + source = await db.get_instance_source_by_id(source_id) + scheduler = get_scheduler() + scheduler.refresh_instance_source_job(source) + + if source.enabled: + # 启用时自动获取 + asyncio.create_task(scheduler._fetch_single_instance_source(source.id)) + else: + await db.update_instance_fetch_status(source_id, "idle") + return _instance_source_to_dict(source) + + +@router.delete("/instance-sources/{source_id}") +async def delete_instance_source(source_id: int): + """删除服务实例源""" + from fastapi import HTTPException + db = get_db() + success = await db.delete_instance_source(source_id) + if not success: + raise HTTPException(status_code=404, detail="服务实例源不存在") + # 移除定时任务 + scheduler = get_scheduler() + scheduler.remove_instance_source_job(source_id) + return {"message": "deleted"} + + +@router.post("/instance-sources/{source_id}/fetch") +async def manual_fetch_instance_source(source_id: int): + """手动触发获取指定服务实例源的已连接节点""" + db = get_db() + source = await db.get_instance_source_by_id(source_id) + if not source: + from fastapi import HTTPException + raise HTTPException(status_code=404, detail="服务实例源不存在") + scheduler = get_scheduler() + asyncio.create_task(scheduler._fetch_single_instance_source(source_id)) + return {"message": f"已触发实例源 #{source_id} 的获取任务"} + + +def _instance_source_to_dict(source) -> dict: + """将 InstanceSourceRecord 转为 API 响应字典""" + return { + "id": source.id, + "base_url": source.base_url, + "username": source.username, + "password": source.password, + "crontab": source.crontab, + "latency_threshold": source.latency_threshold, + "max_concurrent": source.max_concurrent, + "enabled": source.enabled, + "created_at": source.created_at, + "total_count": source.total_count, + "fetch_status": source.fetch_status, + } diff --git a/app/routers/web.py b/app/routers/web.py index e1e6f9f..e7b579e 100644 --- a/app/routers/web.py +++ b/app/routers/web.py @@ -29,6 +29,12 @@ async def index(request: Request): grouped = await db.get_proxies_grouped_by_source(config.check.latency_threshold) # 检测目标 URL check_urls = await db.get_check_urls() + # 服务实例源 + instance_sources = await db.get_all_instance_sources() + instance_sources_json = json.dumps( + [dict(asdict(s)) for s in instance_sources], + ensure_ascii=False, + ) # 计算每个订阅源的可用代理数量 sub_available_counts = {} for sub in subscriptions: @@ -48,6 +54,8 @@ async def index(request: Request): "subscriptions_json": subscriptions_json, "sub_available_counts": sub_available_counts, "check_urls": check_urls, + "instance_sources": instance_sources, + "instance_sources_json": instance_sources_json, "grouped": grouped, "latency_threshold": config.check.latency_threshold, "last_fetch_time": scheduler.last_fetch_time, diff --git a/app/scheduler.py b/app/scheduler.py index 7ea61b4..7e4fa93 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -11,13 +11,13 @@ from app.checker import ProxyChecker from app.config import AppConfig from app.database import ProxyDatabase -from app.parser import fetch_subscription, parse_subscription +from app.parser import fetch_subscription, parse_subscription, fetch_connected_proxies logger = logging.getLogger(__name__) class TaskScheduler: - """代理池定时任务调度器""" + """节点池定时任务调度器""" def __init__(self, config: AppConfig, db: ProxyDatabase, checker: ProxyChecker): self.config = config @@ -54,6 +54,7 @@ def start(self) -> None: # 启动后注册订阅任务 asyncio.create_task(self._register_subscription_jobs()) + asyncio.create_task(self._register_instance_source_jobs()) async def _register_subscription_jobs(self) -> None: """从数据库加载订阅源并注册 crontab 定时任务""" @@ -62,6 +63,13 @@ async def _register_subscription_jobs(self) -> None: self._add_subscription_job(sub) logger.info("已注册 %d 个订阅拉取任务", len(subs)) + async def _register_instance_source_jobs(self) -> None: + """从数据库加载服务实例源并注册 crontab 定时任务""" + sources = await self.db.get_enabled_instance_sources() + for source in sources: + self._add_instance_source_job(source) + logger.info("已注册 %d 个实例源获取任务", len(sources)) + def _add_subscription_job(self, sub) -> None: """为单个订阅注册 crontab 定时任务""" job_id = f"fetch_sub_{sub.id}" @@ -203,6 +211,9 @@ async def fetch_and_check(self) -> None: for sub in subs: await self._fetch_single_subscription(sub.id) + + # 同时获取服务实例源 + await self.fetch_all_instance_sources() except Exception as e: logger.error("拉取任务异常: %s", e, exc_info=True) finally: @@ -281,6 +292,133 @@ async def verify_subscription_proxies(self, sub_id: int) -> None: self._last_verify_time = datetime.now(timezone.utc).isoformat() logger.info("订阅 #%d 验证完成: 成功 %d, 失败 %d", sub_id, success_count, fail_count) + # ---- 服务实例源管理 ---- + + def _add_instance_source_job(self, source) -> None: + """为单个服务实例源注册 crontab 定时任务""" + job_id = f"fetch_instance_{source.id}" + try: + if self.scheduler.get_job(job_id): + self.scheduler.remove_job(job_id) + + parts = source.crontab.strip().split() + if len(parts) != 5: + logger.error("服务实例源 %d crontab 格式错误: %s", source.id, source.crontab) + return + + trigger = CronTrigger( + minute=parts[0], hour=parts[1], day=parts[2], + month=parts[3], day_of_week=parts[4], + ) + self.scheduler.add_job( + self._fetch_single_instance_source, + trigger=trigger, + id=job_id, + name=f"获取实例源 #{source.id}", + args=[source.id], + ) + logger.info("注册实例源任务 #%d: crontab=%s", source.id, source.crontab) + except Exception as e: + logger.error("注册实例源任务 #%d 失败: %s", source.id, e) + + def remove_instance_source_job(self, source_id: int) -> None: + """移除服务实例源定时任务""" + job_id = f"fetch_instance_{source_id}" + if self.scheduler.get_job(job_id): + self.scheduler.remove_job(job_id) + + def refresh_instance_source_job(self, source) -> None: + """刷新服务实例源定时任务（更新 crontab 后调用）""" + self.remove_instance_source_job(source.id) + if source.enabled: + self._add_instance_source_job(source) + + async def _fetch_single_instance_source(self, source_id: int) -> None: + """从服务实例获取已连接节点配置并检测入库""" + source = await self.db.get_instance_source_by_id(source_id) + if not source or not source.enabled: + return + + # 实例源的 source 标识 + source_tag = f"instance:{source.base_url}" + + try: + logger.info("开始获取实例源 #%d: %s", source.id, source.base_url) + await self.db.update_instance_fetch_status(source_id, "updating") + + proxies = await fetch_connected_proxies( + source.base_url, source.username, source.password, + ) + + if not proxies: + logger.warning("实例源 #%d 未获取到任何已连接节点", source.id) + await self.db.update_instance_total_count(source_id, 0) + await self.db.update_instance_fetch_status(source_id, "success") + return + + logger.info("实例源 #%d: 获取到 %d 个节点，开始逐个检测...", + source.id, len(proxies)) + + sub_checker = ProxyChecker( + check_urls=self.checker.check_urls, + timeout=self.config.check.timeout, + max_concurrent=source.max_concurrent, + ) + + threshold = source.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, source_tag) + 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 + + await self.db.update_instance_total_count(source_id, len(proxies)) + await self.db.update_instance_fetch_status(source_id, "success") + + self._last_fetch_time = datetime.now(timezone.utc).isoformat() + logger.info("实例源 #%d 获取完成: 新增 %d, 更新 %d, 跳过 %d, 总获取 %d", + source.id, added, updated, skipped, len(proxies)) + except Exception as e: + await self.db.update_instance_fetch_status(source_id, "failed") + logger.error("获取实例源 #%d 异常: %s", source.id, e, exc_info=True) + + async def fetch_all_instance_sources(self) -> None: + """手动触发：获取所有启用的服务实例源的已连接节点""" + sources = await self.db.get_enabled_instance_sources() + if not sources: + logger.warning("没有启用的服务实例源") + return + for source in sources: + await self._fetch_single_instance_source(source.id) + async def cleanup_proxies(self) -> None: """清理不合格代理和空订阅源""" # 清理连续3次验证失败的代理 diff --git a/app/templates/base.html b/app/templates/base.html index 3489b57..c224014 100644 --- a/app/templates/base.html +++ b/app/templates/base.html @@ -3,7 +3,7 @@ - {% block title %}ProxyPool{% endblock %} + {% block title %}NetHub{% endblock %}