diff --git a/.gitignore b/.gitignore index 213f83f..f2b7a36 100644 --- a/.gitignore +++ b/.gitignore @@ -35,6 +35,9 @@ git_cache/ backup_state.json logs.jsonl +# Downloaded mihomo kernel binary (subscription sync script) +scripts/.bin/ + # Generated web assets web/.next/ web/out/ diff --git a/Dockerfile b/Dockerfile index eafa471..1eb295f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -53,6 +53,19 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ RUN pip install --no-cache-dir uv +# mihomo 代理内核(订阅代理池用,固定 v1.19.27 与开发环境一致) +RUN case "$TARGETARCH" in \ + amd64) _ARCH="linux-amd64-compatible" ;; \ + arm64) _ARCH="linux-arm64" ;; \ + *) echo "unsupported arch: $TARGETARCH" && exit 1 ;; \ + esac && \ + _VER="v1.19.27" && \ + mkdir -p /app/scripts/.bin && \ + curl -fsSL "https://github.com/MetaCubeX/mihomo/releases/download/${_VER}/mihomo-${_ARCH}-${_VER}.gz" -o /tmp/mihomo.gz && \ + python3 -c "import gzip,shutil; shutil.copyfileobj(gzip.open('/tmp/mihomo.gz','rb'), open('/app/scripts/.bin/mihomo','wb'))" && \ + chmod +x /app/scripts/.bin/mihomo && \ + rm -f /tmp/mihomo.gz + COPY pyproject.toml uv.lock ./ RUN uv sync --frozen --no-dev --no-install-project diff --git a/api/app.py b/api/app.py index a413e74..8f0bc79 100644 --- a/api/app.py +++ b/api/app.py @@ -12,6 +12,8 @@ from api.support import resolve_web_asset, start_limited_account_watcher, web_index_asset from services.backup_service import backup_service from services.config import config +from services.mihomo_manager import mihomo_manager +from services.proxy_health_service import proxy_health_service def serve_web_asset(full_path: str): @@ -35,12 +37,20 @@ async def lifespan(_: FastAPI): thread = start_limited_account_watcher(stop_event) backup_service.start() config.cleanup_old_images() + # mihomo 常驻代理内核(启动失败不拖垮主服务,订阅代理功能降级) + try: + mihomo_manager.start() + except Exception as e: # noqa: BLE001 + print(f"[lifespan] mihomo 启动失败,订阅代理功能降级: {e}", flush=True) + proxy_health_service.start() try: yield finally: stop_event.set() thread.join(timeout=1) backup_service.stop() + proxy_health_service.stop() + mihomo_manager.stop() app = FastAPI(title="webchat2api", version=app_version, lifespan=lifespan) install_exception_handlers(app) diff --git a/api/system.py b/api/system.py index 45cf853..03c2ce7 100644 --- a/api/system.py +++ b/api/system.py @@ -14,6 +14,8 @@ from services.image_storage_service import ImageStorageError, image_storage_service from services.image_tags_service import delete_tag, get_all_tags, set_tags from services.log_service import log_service +from services.proxy_health_service import proxy_health_service +from services.proxy_pool_service import proxy_pool_service from services.proxy_service import test_proxy @@ -43,6 +45,20 @@ class LogDeleteRequest(BaseModel): class BackupDeleteRequest(BaseModel): key: str = "" +class ProxyPoolImportRequest(BaseModel): + proxies: str = "" + +class ProxyPoolDeleteRequest(BaseModel): + ids: list[str] = [] + +class ProxyPoolSubscriptionRequest(BaseModel): + name: str = "" + url: str = "" + region_keywords: str = "" + +class ProxyPoolSyncRequest(BaseModel): + subscription_id: str | None = None + def create_router(app_version: str) -> APIRouter: router = APIRouter() @@ -233,4 +249,76 @@ async def delete_image_tag(tag: str, authorization: str | None = Header(default= count = delete_tag(tag) return {"ok": True, "removed_from": count} + @router.get("/api/proxy-pool") + async def list_proxy_pool(authorization: str | None = Header(default=None)): + require_admin(authorization) + return {"items": proxy_pool_service.list_items()} + + @router.post("/api/proxy-pool") + async def import_proxy_pool(body: ProxyPoolImportRequest, authorization: str | None = Header(default=None)): + require_admin(authorization) + text = (body.proxies or "").strip() + if not text: + raise HTTPException(status_code=400, detail={"error": "proxies is required"}) + result = proxy_pool_service.import_proxies(text) + return {**result, "items": proxy_pool_service.list_items()} + + @router.delete("/api/proxy-pool") + async def delete_proxy_pool(body: ProxyPoolDeleteRequest, authorization: str | None = Header(default=None)): + require_admin(authorization) + if not body.ids: + result = proxy_pool_service.clear_all() + else: + result = proxy_pool_service.delete_proxies(body.ids) + return {**result, "items": proxy_pool_service.list_items()} + + @router.post("/api/proxy-pool/assign") + async def assign_proxy_pool(authorization: str | None = Header(default=None)): + require_admin(authorization) + return await run_in_threadpool(proxy_pool_service.assign_to_accounts) + + @router.post("/api/proxy-pool/clear") + async def clear_proxy_pool_assignments(authorization: str | None = Header(default=None)): + require_admin(authorization) + return await run_in_threadpool(proxy_pool_service.clear_assignments) + + # ── 订阅源管理 ── + @router.get("/api/proxy-pool/subscriptions") + async def list_subscriptions(authorization: str | None = Header(default=None)): + require_admin(authorization) + return {"items": proxy_health_service.list_subscriptions()} + + @router.post("/api/proxy-pool/subscriptions") + async def add_subscription(body: ProxyPoolSubscriptionRequest, authorization: str | None = Header(default=None)): + require_admin(authorization) + if not body.url.strip(): + raise HTTPException(status_code=400, detail={"error": "url is required"}) + sub = await run_in_threadpool( + proxy_health_service.add_subscription, body.name, body.url, body.region_keywords + ) + return {"item": sub, "items": proxy_health_service.list_subscriptions()} + + @router.delete("/api/proxy-pool/subscriptions/{sub_id}") + async def delete_subscription(sub_id: str, authorization: str | None = Header(default=None)): + require_admin(authorization) + result = await run_in_threadpool(proxy_health_service.delete_subscription, sub_id) + return {**result, "items": proxy_health_service.list_subscriptions()} + + # ── 同步(异步 task,前端轮询 sync/{task_id}) ── + @router.post("/api/proxy-pool/sync") + async def sync_proxy_pool(body: ProxyPoolSyncRequest, authorization: str | None = Header(default=None)): + require_admin(authorization) + return proxy_health_service.run_sync_now(body.subscription_id) + + @router.get("/api/proxy-pool/sync/{task_id}") + async def get_sync_status(task_id: str, authorization: str | None = Header(default=None)): + require_admin(authorization) + return proxy_health_service.get_task_status(task_id) + + # ── 健康视图(pool item 带 health/latency 字段) ── + @router.get("/api/proxy-pool/health") + async def proxy_pool_health(authorization: str | None = Header(default=None)): + require_admin(authorization) + return {"items": proxy_pool_service.list_items()} + return router diff --git a/docker-compose.yml b/docker-compose.yml index adf9f1e..85dbf9f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -14,7 +14,7 @@ services: # - ./config.json:/app/config.json environment: LOGIN_SECRET: admin - STORAGE_BACKEND: json + STORAGE_BACKEND: sqlite # 如需访问宿主机代理,可同时取消下一行和 extra_hosts 的注释。 # PROXY_URL: http://host.docker.internal:7890 # WEBCHAT2API_BASE_URL: https://your-domain.com diff --git a/pyproject.toml b/pyproject.toml index 0e63423..697551a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,7 @@ dependencies = [ "sqlalchemy>=2.0.0", "psycopg2-binary>=2.9.0", "gitpython>=3.1.0", + "pyyaml>=6.0.3", ] [dependency-groups] diff --git a/scripts/sync_subscription.py b/scripts/sync_subscription.py new file mode 100644 index 0000000..236119d --- /dev/null +++ b/scripts/sync_subscription.py @@ -0,0 +1,474 @@ +#!/usr/bin/env python3 +"""阶段一脚本:从 Clash 订阅拉取美国节点 → 用 mihomo 把每个节点转成独立本地 +SOCKS5 端口 → 并发用这些端口测 chatgpt.com 连通性 → 剔除超时/被挡节点,输出可用节点。 + +这是「订阅节点代理池」扩展(方案 C)的第一步:独立、可手动跑通,先验证技术链路, +跑通后再把逻辑沉淀进 webchat2api 项目(阶段二)。 + +用法(项目根目录执行): + uv run --with pyyaml python scripts/sync_subscription.py [选项] + +产物写入 --out-dir(默认 data/subscription_us): + usable_socks5.txt 一行一个 socks5://127.0.0.1:,可直接灌 proxy_pool + usable_nodes.yaml 可用节点原样 Clash 配置(name 已还原,可给其它客户端加载) + report.json 每个节点的测活详情(status/latency/error) + mihomo.yaml 生成的 mihomo 配置(调试用) +""" + +from __future__ import annotations + +import argparse +import gzip +import json +import os +import platform +import shutil +import signal +import socket +import subprocess +import sys +import time +import urllib.request +from concurrent.futures import ThreadPoolExecutor, as_completed +from pathlib import Path + +import yaml +from curl_cffi import requests as ccreq + +# ────────────────────────────────────────────────────────────── +# 默认值 +# ────────────────────────────────────────────────────────────── +DEFAULT_SUB_URL = "https://slp.19930905.xyz/c/?token=87ebe4bebccf282d558c8271a95c9e36&cdk=lkdx" +DEFAULT_REGION_KEYWORDS = ["🇺🇸", "美国", "美國", "United States", "America"] +DEFAULT_TEST_URL = "https://chatgpt.com/api/auth/csrf" +DEFAULT_PORT_BASE = 30000 +DEFAULT_TIMEOUT = 15.0 +DEFAULT_CONCURRENCY = 16 +PROJECT_ROOT = Path(__file__).resolve().parent.parent +GH_API_LATEST = "https://api.github.com/repos/MetaCubeX/mihomo/releases/latest" + + +def log(msg: str) -> None: + print(f"[sync] {msg}", flush=True) + + +def mask_url(url: str) -> str: + """订阅 URL 脱敏,避免 token 完整打印到日志。""" + if "token=" not in url: + return url + import re + + def _sub(m: "re.Match[str]") -> str: + tok = m.group(1) + if len(tok) <= 8: + return "token=***" + return f"token={tok[:4]}...{tok[-4:]}" + + return re.sub(r"token=([^&]+)", _sub, url) + + +# ────────────────────────────────────────────────────────────── +# 1. 拉取订阅 +# ────────────────────────────────────────────────────────────── +def fetch_subscription(url: str, timeout: float = 30.0) -> str: + log(f"拉取订阅:{mask_url(url)}") + s = ccreq.Session() + try: + r = s.get(url, headers={"user-agent": "clash-verge/v1.0"}, timeout=timeout) + r.raise_for_status() + return r.text + finally: + s.close() + + +# ────────────────────────────────────────────────────────────── +# 2. 解析 + 3. 地区筛选 +# ────────────────────────────────────────────────────────────── +def parse_proxies(yaml_text: str) -> list[dict]: + """解析 Clash YAML,只取真实 proxies(丢弃 proxy-groups)。""" + data = yaml.safe_load(yaml_text) or {} + proxies = data.get("proxies") or [] + return [p for p in proxies if isinstance(p, dict)] + + +def filter_region( + nodes: list[dict], keywords: list[str], prefix: str = "us" +) -> list[tuple[str, dict, str]]: + """按 name 关键词筛选节点,返回 [(node_id, 重命名后的节点副本, 原始name)]。 + + node_id = prefix + 节点在订阅中的原始下标,保证单次运行内唯一;原始 name 仅存档。 + 注意:若订阅节点顺序/数量变化,同一下标会指向不同节点,跨日不应依赖 id 稳定性。 + """ + out = [] + for i, node in enumerate(nodes): + orig = str(node.get("name") or f"node-{i}") + if any(k and k in orig for k in keywords): + node_id = f"{prefix}-{i:04d}" + renamed = dict(node) + renamed["name"] = node_id + out.append((node_id, renamed, orig)) + return out + + +# ────────────────────────────────────────────────────────────── +# 4. 生成 mihomo 配置(proxies + listeners 多端口) +# ────────────────────────────────────────────────────────────── +def build_mihomo_config( + mapped: list[tuple[str, dict, str]], port_base: int +) -> tuple[dict, dict[str, int]]: + """每节点一个 socks listener,proxy 字段钉死该节点 → 直出绕过主路由。""" + proxies: list[dict] = [] + listeners: list[dict] = [] + port_map: dict[str, int] = {} + for idx, (node_id, node, _orig) in enumerate(mapped): + proxies.append(node) + port = port_base + idx + port_map[node_id] = port + listeners.append( + { + "name": f"lis-{node_id}", + "type": "socks", + "listen": "127.0.0.1", + "port": port, + "proxy": node_id, + } + ) + config = { + "log-level": "warning", + "mode": "direct", + "allow-lan": False, + "ipv6": False, + "tcp-concurrent": True, + "proxies": proxies, + "listeners": listeners, + } + return config, port_map + + +# ────────────────────────────────────────────────────────────── +# 5. mihomo 二进制管理(自动下载) +# ────────────────────────────────────────────────────────────── +def _arch_keys() -> list[str]: + m = platform.machine().lower() + if m in ("x86_64", "amd64"): + # 优先 compatible 变体(兼容老 CPU,无 AVX 等指令集要求),其次普通版 + return ["linux-amd64-compatible", "linux-amd64"] + if m in ("aarch64", "arm64"): + return ["linux-arm64"] + raise SystemExit(f"不支持的架构:{m}") + + +def _pick_asset_url(meta: dict, keys: list[str]) -> str: + assets = {a["name"].lower(): a["browser_download_url"] for a in meta.get("assets", [])} + for k in keys: + for name, url in assets.items(): + if name.endswith(".gz") and k in name: + return url + raise SystemExit(f"未找到匹配架构 {keys} 的 mihomo release 资产") + + +def ensure_mihomo_binary(target: Path) -> Path: + if target.exists() and os.access(target, os.X_OK): + log(f"mihomo 二进制已存在:{target}") + return target + target.parent.mkdir(parents=True, exist_ok=True) + log("查询 mihomo 最新版本……") + req = urllib.request.Request(GH_API_LATEST, headers={"Accept": "application/vnd.github+json"}) + with urllib.request.urlopen(req, timeout=30) as r: + meta = json.load(r) + tag = meta.get("tag_name", "unknown") + url = _pick_asset_url(meta, _arch_keys()) + gz_path = target.with_suffix(target.suffix + ".gz") + log(f"下载 mihomo {tag}:{url}") + with urllib.request.urlopen(url, timeout=120) as r, open(gz_path, "wb") as f: + shutil.copyfileobj(r, f) + with gzip.open(gz_path, "rb") as fi, open(target, "wb") as fo: + shutil.copyfileobj(fi, fo) + os.chmod(target, 0o755) + gz_path.unlink(missing_ok=True) + log(f"mihomo 就绪:{target}") + return target + + +# ────────────────────────────────────────────────────────────── +# 6. mihomo 进程生命周期 +# ────────────────────────────────────────────────────────────── +def is_port_open(port: int, host: str = "127.0.0.1") -> bool: + try: + with socket.create_connection((host, port), timeout=1): + return True + except OSError: + return False + + +def start_mihomo( + bin_path: Path, config_path: Path, work_dir: Path, log_path: Path +) -> tuple[subprocess.Popen, object]: + work_dir.mkdir(parents=True, exist_ok=True) + logf = open(log_path, "w") + proc = subprocess.Popen( + [str(bin_path), "-f", str(config_path), "-d", str(work_dir)], + stdout=logf, + stderr=subprocess.STDOUT, + ) + return proc, logf + + +def wait_mihomo_ready( + proc: subprocess.Popen, port_map: dict[str, int], log_path: Path, deadline_s: float = 20.0 +) -> None: + if not port_map: + return + first_port = next(iter(port_map.values())) + t0 = time.perf_counter() + while time.perf_counter() - t0 < deadline_s: + if proc.poll() is not None: + tail = _tail(log_path, 40) + raise SystemExit(f"mihomo 进程已退出(code={proc.returncode})。日志末尾:\n{tail}") + if is_port_open(first_port): + time.sleep(0.5) # 让其余 listener 端口完成绑定,减少首测假阴性 + return + time.sleep(0.3) + raise SystemExit(f"mihomo 启动超时({deadline_s:.0f}s 内端口 {first_port} 未就绪)") + + +def _tail(path: Path, n: int) -> str: + try: + lines = path.read_text(errors="replace").splitlines() + return "\n".join(lines[-n:]) + except OSError: + return "(读日志失败)" + + +def cleanup_stale_mihomo(pid_path: Path) -> None: + """若存在旧 mihomo PID 文件,尝试终止残留进程并清理文件。""" + if not pid_path.exists(): + return + try: + pid = int(pid_path.read_text().strip()) + except (ValueError, OSError): + pid_path.unlink(missing_ok=True) + return + try: + os.kill(pid, 0) # 探活,不发送信号 + except ProcessLookupError: + pid_path.unlink(missing_ok=True) + return + except PermissionError: + log(f"残留 mihomo(pid={pid}) 存在但无权限管理,请手动 pkill -f mihomo") + return + log(f"发现残留 mihomo(pid={pid}),正在终止……") + try: + os.kill(pid, signal.SIGTERM) + for _ in range(10): + try: + os.kill(pid, 0) + time.sleep(0.3) + except ProcessLookupError: + break + else: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + pid_path.unlink(missing_ok=True) + + +# ────────────────────────────────────────────────────────────── +# 7. 并发测活 +# ────────────────────────────────────────────────────────────── +def probe(port: int, test_url: str, timeout: float) -> dict: + """通过指定 socks5 端口访问 chatgpt.com,收紧判据:status==200 且响应含 csrfToken。""" + s = ccreq.Session(impersonate="edge101", verify=True, proxy=f"socks5://127.0.0.1:{port}") + t0 = time.perf_counter() + try: + r = s.get(test_url, headers={"user-agent": "Mozilla/5.0"}, timeout=timeout) + latency_ms = int((time.perf_counter() - t0) * 1000) + body = r.text if r.status_code == 200 else "" + ok = r.status_code == 200 and "csrftoken" in body.lower() + if ok: + error = None + elif r.status_code != 200: + error = f"HTTP {r.status_code}" + else: + error = "200 but no csrfToken (likely CF challenge / non-JSON)" + return {"ok": ok, "status": int(r.status_code), "latency_ms": latency_ms, "error": error} + except Exception as e: + latency_ms = int((time.perf_counter() - t0) * 1000) + msg = f"{type(e).__name__}: {e}" + return {"ok": False, "status": 0, "latency_ms": latency_ms, "error": msg[:300]} + finally: + s.close() + + +def probe_all( + port_map: dict[str, int], test_url: str, timeout: float, workers: int +) -> dict[str, dict]: + results: dict[str, dict] = {} + with ThreadPoolExecutor(max_workers=workers) as ex: + fut = {ex.submit(probe, p, test_url, timeout): nid for nid, p in port_map.items()} + done = 0 + total = len(fut) + for f in as_completed(fut): + nid = fut[f] + results[nid] = f.result() + done += 1 + if done % 10 == 0 or done == total: + ok = sum(1 for v in results.values() if v["ok"]) + log(f"测活进度 {done}/{total},可用 {ok}") + return results + + +# ────────────────────────────────────────────────────────────── +# 8. 输出产物 +# ────────────────────────────────────────────────────────────── +def emit( + out_dir: Path, + mapped: list[tuple[str, dict, str]], + port_map: dict[str, int], + results: dict[str, dict], +) -> dict: + out_dir.mkdir(parents=True, exist_ok=True) + + # node_id -> 原始信息 + meta = {nid: {"orig_name": orig, "node": node} for nid, node, orig in mapped} + + usable_ids = [nid for nid, _, _ in mapped if results.get(nid, {}).get("ok")] + + # usable_socks5.txt + socks5_lines = [f"socks5://127.0.0.1:{port_map[nid]}" for nid in usable_ids] + (out_dir / "usable_socks5.txt").write_text("\n".join(socks5_lines) + ("\n" if socks5_lines else "")) + + # usable_nodes.yaml(name 还原为原始名) + usable_proxies = [] + for nid, _, _ in mapped: + if results.get(nid, {}).get("ok"): + node = dict(meta[nid]["node"]) + node["name"] = meta[nid]["orig_name"] + usable_proxies.append(node) + (out_dir / "usable_nodes.yaml").write_text( + yaml.safe_dump({"proxies": usable_proxies}, allow_unicode=True, sort_keys=False) + ) + + # report.json(全部节点,含失败原因) + report = [] + for nid, _node, orig in mapped: + r = results.get(nid, {"ok": False, "status": 0, "latency_ms": 0, "error": "not tested"}) + node = meta[nid]["node"] + report.append( + { + "node_id": nid, + "name": orig, + "type": node.get("type"), + "server": node.get("server"), + "port": node.get("port"), + "local_port": port_map.get(nid), + "usable": r["ok"], + "status": r["status"], + "latency_ms": r["latency_ms"], + "error": r["error"], + } + ) + report.sort(key=lambda x: (not x["usable"], x["latency_ms"] if x["usable"] else 0)) + (out_dir / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2)) + + # mihomo.yaml 由 main() 在 mihomo 启动前写入,此处不重复 + + return { + "total": len(mapped), + "usable": len(usable_ids), + "socks5_file": str(out_dir / "usable_socks5.txt"), + } + + +# ────────────────────────────────────────────────────────────── +# main +# ────────────────────────────────────────────────────────────── +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser(description="拉取订阅美国节点并测活(mihomo 多端口)") + p.add_argument("--url", default=DEFAULT_SUB_URL, help="订阅 URL") + p.add_argument( + "--region", + default=",".join(DEFAULT_REGION_KEYWORDS), + help="地区关键词,逗号分隔(默认美国)", + ) + p.add_argument("--port-base", type=int, default=DEFAULT_PORT_BASE, help="本地 socks5 起始端口") + p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT, help="单节点测活超时(秒)") + p.add_argument("--concurrency", type=int, default=DEFAULT_CONCURRENCY, help="并发数") + p.add_argument("--test-url", default=DEFAULT_TEST_URL, help="测活目标 URL") + p.add_argument("--out-dir", default=str(PROJECT_ROOT / "data" / "subscription_us"), help="产物目录") + p.add_argument( + "--mihomo-bin", + default=str(PROJECT_ROOT / "scripts" / ".bin" / "mihomo"), + help="mihomo 二进制路径(不存在自动下载)", + ) + p.add_argument( + "--keep-mihomo", action="store_true", help="跑完保留 mihomo 进程(默认退出即关)" + ) + return p.parse_args() + + +def main() -> int: + args = parse_args() + keywords = [k.strip() for k in args.region.split(",") if k.strip()] + out_dir = Path(args.out_dir) + bin_path = Path(args.mihomo_bin) + config_path = out_dir / "mihomo.yaml" + work_dir = out_dir / "mihomo_home" + log_path = out_dir / "mihomo.log" + pid_path = out_dir / "mihomo.pid" + + # 启动前清理上次 --keep-mihomo 残留,再检查端口占用 + cleanup_stale_mihomo(pid_path) + if is_port_open(args.port_base): + raise SystemExit( + f"端口 {args.port_base} 已被占用。请先清理旧 mihomo:pkill -f mihomo 或更换 --port-base" + ) + + yaml_text = fetch_subscription(args.url) + nodes = parse_proxies(yaml_text) + log(f"解析到真实节点 {len(nodes)} 个") + + mapped = filter_region(nodes, keywords) + log(f"命中地区({keywords})节点 {len(mapped)} 个") + if not mapped: + log("没有命中任何节点,结束。") + return 1 + + config, port_map = build_mihomo_config(mapped, args.port_base) + out_dir.mkdir(parents=True, exist_ok=True) + config_path.write_text(yaml.safe_dump(config, allow_unicode=True, sort_keys=False)) + + ensure_mihomo_binary(bin_path) + + log("启动 mihomo……") + proc, logf = start_mihomo(bin_path, config_path, work_dir, log_path) + pid_path.write_text(str(proc.pid)) + try: + wait_mihomo_ready(proc, port_map, log_path) + log(f"mihomo 就绪(pid={proc.pid}),开始测活(目标 {args.test_url},并发 {args.concurrency})") + results = probe_all(port_map, args.test_url, args.timeout, args.concurrency) + finally: + if not args.keep_mihomo: + if proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + pid_path.unlink(missing_ok=True) + logf.close() + log("已关闭 mihomo") + else: + logf.close() + log(f"保留 mihomo(pid={proc.pid}),端口 {args.port_base}..{args.port_base + len(mapped) - 1}(PID 文件 {pid_path})") + + summary = emit(out_dir, mapped, port_map, results) + log( + f"完成:{summary['usable']}/{summary['total']} 可用 → {summary['socks5_file']}" + ) + log("提示:若要看每节点失败原因,查看 report.json") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/services/account_service.py b/services/account_service.py index 9a24905..59a4a2a 100644 --- a/services/account_service.py +++ b/services/account_service.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os import re from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta, timezone @@ -156,7 +157,7 @@ def __init__(self, storage_backend: StorageBackend, now: Callable[[], float] | N self.storage = storage_backend self._lock = Lock() self._image_slot_condition = Condition(self._lock) - self._index = 0 + self._index = int.from_bytes(os.urandom(4), "big") self._now = now or (lambda: datetime.now(timezone.utc).timestamp()) self._accounts = self._load_accounts() self._image_inflight: dict[str, int] = {} @@ -384,6 +385,7 @@ def _normalize_account(self, item: dict) -> dict | None: provider_strategy = account_strategy(normalized["provider"]) if normalized["provider"] in {GROK_PROVIDER, GEMINI_PROVIDER}: normalized = provider_strategy.normalize_account(normalized) + normalized["proxy"] = str(normalized.get("proxy") or "").strip() normalized["success"] = int(normalized.get("success") or 0) normalized["fail"] = int(normalized.get("fail") or 0) normalized["last_used_at"] = normalized.get("last_used_at") @@ -549,12 +551,73 @@ def mark_text_used(self, access_token: str) -> None: next_item.update({"app_chat": True, "last_check_status": "valid_by_call", "last_success_at": now, **self._grok_clear_transient_metadata(now)}) if next_item.get("status") != "禁用": next_item["status"] = "正常" + else: + # GPT: a successful search proves the account is healthy — clear any + # lingering cooldown/abnormal state and reset the failure counter so + # it doesn't accumulate toward 禁用 across recovered transient failures. + now = self._timestamp_string() + next_item["fail"] = 0 + next_item["last_success_at"] = now + if next_item.get("status") not in ("禁用",): + next_item["status"] = "正常" + next_item.pop("cooldown_until", None) + next_item.pop("state_reason", None) account = self._normalize_account(next_item) if account is None: return self._set_account_locked(account) self._save_accounts() + def mark_search_failure(self, access_token: str, exc: Exception, provider: str = GPT_PROVIDER) -> None: + """Transition a GPT account's state based on a search failure. + + Auto state-switching across 限流 / 异常 / 禁用: + - rate-limit (429/402) or Cloudflare (403): → 限流 (cooldown 15min, auto-recovers via watcher) + - auth-invalid (401 token_invalidated): → 异常; after 3 consecutive auth failures → 禁用 + - transient network errors (timeout/TLS): no change (don't penalize healthy accounts) + + Called from the search path so degraded accounts exit rotation immediately + instead of being retried on every request (which cascades into pool-wide failure + under sustained load). Recoverable accounts are re-validated by the limited-account + watcher (list_limited_tokens includes 限流 + 异常); permanently dead ones escalate + to 禁用 so they stay out of rotation but are kept for batch re-add. + """ + if not access_token: + return + text = str(exc).lower() + is_auth = any(marker in text for marker in ( + "token_invalidated", "token_revoked", + "authentication token has been invalidated", + "invalidated oauth token", "status=401", + )) + is_rate_or_cf = any(marker in text for marker in ( + "status=403", "status=429", "status=402", + "rate_limit", "cloudflare", + )) + if not (is_auth or is_rate_or_cf): + return + account = self.get_account(access_token, provider=provider) or {} + prev_fail = int(account.get("fail") or 0) + now = self._timestamp_string() + if is_auth and prev_fail + 1 >= 3: + new_status, reason, fail_count = "禁用", "search_auth_failure_disabled", prev_fail + 1 + elif is_auth: + new_status, reason, fail_count = "异常", "search_auth_failure", prev_fail + 1 + else: + new_status, reason, fail_count = "限流", "search_rate_limited", prev_fail + updates = { + "status": new_status, + "state_reason": reason, + "last_check_error": str(exc)[:200], + "last_check_at": now, + "fail": fail_count, + } + if new_status in ("限流", "异常"): + updates["cooldown_until"] = self._timestamp_string(900) + self.update_account(access_token, updates, provider=provider) + log_service.add(LOG_TYPE_ACCOUNT, "搜索失败状态切换", + {"token": anonymize_token(access_token), "status": new_status, "reason": reason}) + def mark_grok_console_used(self, access_token: str, success: bool = True) -> None: if not access_token: return @@ -604,11 +667,14 @@ def list_accounts(self, provider: str | None = None) -> list[dict]: return self._list_account_items_locked(provider) def list_limited_tokens(self) -> list[str]: + # Includes both 限流 (rate-limited) and 异常 (auth-invalidated) accounts so the + # limited-account-watcher re-validates both: 限流 recovers when the cooldown + # passes, 异常 recovers if the access token is still valid (or stays 异常 if dead). with self._lock: return [ token for item in self._all_accounts_locked() - if item.get("status") == "限流" + if item.get("status") in ("限流", "异常") and normalize_provider(item.get("provider")) == GPT_PROVIDER and (token := item.get("access_token") or "") ] diff --git a/services/config.py b/services/config.py index 8f6a3a5..64b583a 100644 --- a/services/config.py +++ b/services/config.py @@ -66,6 +66,7 @@ "browser_bridge_url", "chat_completion_cache", "chat_completion_message_normalization", + "mihomo", } @@ -97,6 +98,29 @@ def _normalize_backup_include(value: object) -> dict[str, bool]: return normalized +DEFAULT_MIHOMO_REGION = "🇺🇸,美国,美國,United States,America" + + +def _normalize_mihomo_settings(value: object) -> dict[str, object]: + source = value if isinstance(value, dict) else {} + bin_default = "scripts/.bin/mihomo" + ctrl_default = "127.0.0.1:9090" + test_url_default = "https://chatgpt.com/api/auth/csrf" + return { + "bin_path": str(source.get("bin_path") or bin_default).strip() or bin_default, + "port_base": _normalize_positive_int(source.get("port_base"), 30000, 1024), + "port_range_size": _normalize_positive_int(source.get("port_range_size"), 500, 1), + "external_controller": str(source.get("external_controller") or ctrl_default).strip() or ctrl_default, + "external_controller_secret": str(source.get("external_controller_secret") or "").strip(), + "region_keywords": str(source.get("region_keywords") or DEFAULT_MIHOMO_REGION).strip() or DEFAULT_MIHOMO_REGION, + "sync_interval_hours": _normalize_positive_int(source.get("sync_interval_hours"), 24, 1), + "test_url": str(source.get("test_url") or test_url_default).strip() or test_url_default, + "test_timeout": _normalize_positive_int(source.get("test_timeout"), 15, 1), + "concurrency": _normalize_positive_int(source.get("concurrency"), 16, 1), + "probe_retry": _normalize_positive_int(source.get("probe_retry"), 1, 0), + } + + def _normalize_backup_settings(value: object) -> dict[str, object]: source = value if isinstance(value, dict) else {} return { @@ -525,6 +549,7 @@ def get(self) -> dict[str, object]: data["flaresolverr_url"] = self.flaresolverr_url data["flaresolverr_timeout_sec"] = self.flaresolverr_timeout_sec data["browser_bridge_url"] = self.browser_bridge_url + data["mihomo"] = self.get_mihomo_settings() data.pop("auth-key", None) return data @@ -561,6 +586,8 @@ def update(self, data: dict[str, object]) -> dict[str, object]: next_data["chat_completion_cache"] = _normalize_chat_completion_cache_settings(next_data.get("chat_completion_cache")) if "chat_completion_message_normalization" in next_data: next_data["chat_completion_message_normalization"] = _normalize_chat_completion_message_normalization_settings(next_data.get("chat_completion_message_normalization")) + if "mihomo" in next_data: + next_data["mihomo"] = _normalize_mihomo_settings(next_data.get("mihomo")) next_data.pop("backup_state", None) self.data = next_data self._save() @@ -569,6 +596,9 @@ def update(self, data: dict[str, object]) -> dict[str, object]: def get_backup_settings(self) -> dict[str, object]: return _normalize_backup_settings(self.data.get("backup")) + def get_mihomo_settings(self) -> dict[str, object]: + return _normalize_mihomo_settings(self.data.get("mihomo")) + @property def enable_turnstile_solver(self) -> bool: value = self.data.get("enable_turnstile_solver", True) diff --git a/services/mihomo_manager.py b/services/mihomo_manager.py new file mode 100644 index 0000000..a46e3d2 --- /dev/null +++ b/services/mihomo_manager.py @@ -0,0 +1,523 @@ +"""mihomo 代理内核生命周期管理 + 订阅节点配置生成。 + +把订阅里异构的加密协议节点(VLESS/VMess/SS/AnyTLS…)通过 mihomo `listeners` 多端口 +统一成本地 `socks5://127.0.0.1:` 入口,每条 listener 的 `proxy` 字段钉死一个节点 +(直出绕过主路由),供 proxy_pool 消费。消费侧(curl_cffi/account.proxy)零改动。 + +设计要点(见 plans/groovy-painting-nebula.md Review 补强): +- node_id 用节点内容指纹(server+port+type+uuid/password),同节点跨 sync 端口稳定。 +- 配置变更用 stop+start(external-controller reload 对 listeners 增删未验证,首版不依赖)。 +- start 失败不抛异常,返回 False,由 lifespan 容错降级(不拖垮主服务)。 +""" + +from __future__ import annotations + +import gzip +import hashlib +import json +import os +import platform +import shutil +import signal +import socket +import subprocess +import threading +import time +import urllib.request +from pathlib import Path +from typing import Any + +import yaml +from curl_cffi import requests as ccreq + +from services.config import config + +PROJECT_ROOT = Path(__file__).resolve().parent.parent +DATA_DIR = PROJECT_ROOT / "data" +MIHOMO_DIR = DATA_DIR / "mihomo" +GH_API_LATEST = "https://api.github.com/repos/MetaCubeX/mihomo/releases/latest" +DEFAULT_REGION_KEYWORDS = ["🇺🇸", "美国", "美國", "United States", "America"] + + +def log(msg: str) -> None: + print(f"[mihomo] {msg}", flush=True) + + +# ────────────────────────────────────────────────────────────── +# 订阅拉取 / 解析 / 筛选 / 配置生成(共享逻辑,scripts/sync_subscription.py 也复用) +# ────────────────────────────────────────────────────────────── +def fetch_subscription(url: str, timeout: float = 30.0) -> str: + """带 clash UA 拉取订阅原文。""" + s = ccreq.Session() + try: + r = s.get(url, headers={"user-agent": "clash-verge/v1.0"}, timeout=timeout) + r.raise_for_status() + return r.text + finally: + s.close() + + +def parse_proxies(yaml_text: str) -> list[dict]: + """解析 Clash YAML,只取真实 proxies(丢弃 proxy-groups)。""" + data = yaml.safe_load(yaml_text) or {} + proxies = data.get("proxies") or [] + return [p for p in proxies if isinstance(p, dict)] + + +def _node_fingerprint(node: dict) -> str: + """节点内容指纹:server+port+type+uuid/password。同节点跨 sync 稳定。""" + raw = "|".join( + str(node.get(k) or "") + for k in ("server", "port", "type", "uuid", "password", "username") + ) + return hashlib.md5(raw.encode("utf-8")).hexdigest()[:8] + + +def filter_region( + nodes: list[dict], keywords: list[str], prefix: str = "us" +) -> list[tuple[str, dict, str]]: + """按 name 关键词筛选节点,返回 [(node_id, 重命名副本, 原始name)]。 + + node_id = prefix + 内容指纹;同指纹节点去重(重复节点只留一个)。 + """ + seen: set[str] = set() + out: list[tuple[str, dict, str]] = [] + for i, node in enumerate(nodes): + orig = str(node.get("name") or f"node-{i}") + if not any(k and k in orig for k in keywords): + continue + fp = _node_fingerprint(node) + if fp in seen: + continue + seen.add(fp) + node_id = f"{prefix}-{fp}" + renamed = dict(node) + renamed["name"] = node_id + out.append((node_id, renamed, orig)) + return out + + +def allocate_ports( + new_node_ids: list[str], + old_port_map: dict[str, int], + port_base: int, + range_size: int, +) -> dict[str, int]: + """端口分配:内容稳定的节点复用旧端口(跨 sync 不变),新节点分配未用端口(不挤占)。""" + used_ports = set(old_port_map.values()) + new_map: dict[str, int] = {} + for nid in new_node_ids: + if nid in old_port_map: + new_map[nid] = old_port_map[nid] + next_port = port_base + for nid in sorted(nid for nid in new_node_ids if nid not in new_map): + while next_port in used_ports or next_port in new_map.values(): + next_port += 1 + if next_port >= port_base + range_size: + raise RuntimeError(f"端口范围 {range_size} 耗尽,无法为新节点分配端口") + new_map[nid] = next_port + used_ports.add(next_port) + next_port += 1 + return new_map + + +def build_mihomo_config( + mapped: list[tuple[str, dict, str]], + port_map: dict[str, int], + external_controller: str = "127.0.0.1:9090", + external_controller_secret: str = "", +) -> tuple[dict, dict[str, int]]: + """每节点一个 socks listener,proxy 钉死该节点。端口由 port_map 给定(已由 allocate_ports 分配)。""" + proxies: list[dict] = [] + listeners: list[dict] = [] + for node_id, node, _orig in mapped: + port = port_map.get(node_id) + if port is None: + continue + proxies.append(node) + listeners.append( + { + "name": f"lis-{node_id}", + "type": "socks", + "listen": "127.0.0.1", + "port": port, + "proxy": node_id, + } + ) + cfg: dict[str, Any] = { + "log-level": "warning", + "mode": "direct", + "allow-lan": False, + "ipv6": False, + "tcp-concurrent": True, + "external-controller": external_controller, + "proxies": proxies, + "listeners": listeners, + } + if external_controller_secret: + cfg["secret"] = external_controller_secret + return cfg, port_map + + +# ────────────────────────────────────────────────────────────── +# 测活 +# ────────────────────────────────────────────────────────────── +def probe(port: int, test_url: str, timeout: float, retry: int = 1) -> dict: + """通过指定 socks5 端口访问 test_url。chatgpt.com 用 csrf 判据,其他用 200。失败重试。""" + is_chatgpt = "chatgpt.com" in test_url + last: dict = {"ok": False, "status": 0, "latency_ms": 0, "error": "not tested"} + for attempt in range(retry + 1): + s = ccreq.Session(impersonate="edge101", verify=True, proxy=f"socks5://127.0.0.1:{port}") + t0 = time.perf_counter() + try: + r = s.get(test_url, headers={"user-agent": "Mozilla/5.0"}, timeout=timeout) + latency_ms = int((time.perf_counter() - t0) * 1000) + body = r.text if r.status_code == 200 else "" + if is_chatgpt: + ok = r.status_code == 200 and "csrftoken" in body.lower() + else: + ok = r.status_code == 200 + if ok: + return {"ok": True, "status": int(r.status_code), "latency_ms": latency_ms, "error": None} + error = "200 but no csrfToken (likely CF challenge)" if r.status_code == 200 else f"HTTP {r.status_code}" + last = {"ok": False, "status": int(r.status_code), "latency_ms": latency_ms, "error": error} + except Exception as e: + latency_ms = int((time.perf_counter() - t0) * 1000) + last = {"ok": False, "status": 0, "latency_ms": latency_ms, "error": f"{type(e).__name__}: {e}"[:300]} + finally: + s.close() + if attempt < retry: + time.sleep(0.3) + return last + + +def probe_all( + port_map: dict[str, int], test_url: str, timeout: float, workers: int, retry: int = 1, + progress_cb: Any = None, +) -> dict[str, dict]: + """并发测活所有端口。progress_cb(done, total, ok) 可选进度回调。""" + from concurrent.futures import ThreadPoolExecutor, as_completed + + results: dict[str, dict] = {} + with ThreadPoolExecutor(max_workers=workers) as ex: + fut = {ex.submit(probe, p, test_url, timeout, retry): nid for nid, p in port_map.items()} + total = len(fut) + done = 0 + for f in as_completed(fut): + nid = fut[f] + results[nid] = f.result() + done += 1 + if progress_cb: + ok = sum(1 for v in results.values() if v["ok"]) + progress_cb(done, total, ok) + return results + + +# ────────────────────────────────────────────────────────────── +# mihomo 二进制管理 +# ────────────────────────────────────────────────────────────── +def _arch_keys() -> list[str]: + m = platform.machine().lower() + if m in ("x86_64", "amd64"): + return ["linux-amd64-compatible", "linux-amd64"] + if m in ("aarch64", "arm64"): + return ["linux-arm64"] + raise RuntimeError(f"不支持的架构:{m}") + + +def _pick_asset_url(meta: dict, keys: list[str]) -> str: + assets = {a["name"].lower(): a["browser_download_url"] for a in meta.get("assets", [])} + for k in keys: + for name, url in assets.items(): + if name.endswith(".gz") and k in name: + return url + raise RuntimeError(f"未找到匹配架构 {keys} 的 mihomo release 资产") + + +def ensure_mihomo_binary(target: Path) -> Path: + """确保 mihomo 二进制存在,不存在则从 GitHub release 下载。""" + if target.exists() and os.access(target, os.X_OK): + return target + target.parent.mkdir(parents=True, exist_ok=True) + log("查询 mihomo 最新版本……") + req = urllib.request.Request(GH_API_LATEST, headers={"Accept": "application/vnd.github+json"}) + with urllib.request.urlopen(req, timeout=30) as r: + meta = json.load(r) + tag = meta.get("tag_name", "unknown") + url = _pick_asset_url(meta, _arch_keys()) + gz_path = target.with_suffix(target.suffix + ".gz") + log(f"下载 mihomo {tag}:{url}") + with urllib.request.urlopen(url, timeout=120) as r, open(gz_path, "wb") as f: + shutil.copyfileobj(r, f) + with gzip.open(gz_path, "rb") as fi, open(target, "wb") as fo: + shutil.copyfileobj(fi, fo) + os.chmod(target, 0o755) + gz_path.unlink(missing_ok=True) + log(f"mihomo 就绪:{target}") + return target + + +# ────────────────────────────────────────────────────────────── +# 进程辅助 +# ────────────────────────────────────────────────────────────── +def is_port_open(port: int, host: str = "127.0.0.1") -> bool: + try: + with socket.create_connection((host, port), timeout=1): + return True + except OSError: + return False + + +def _tail(path: Path, n: int) -> str: + try: + lines = path.read_text(errors="replace").splitlines() + return "\n".join(lines[-n:]) + except OSError: + return "(读日志失败)" + + +def cleanup_stale_mihomo(pid_path: Path) -> None: + """终止旧 mihomo 残留进程并清理 PID 文件。""" + if not pid_path.exists(): + return + try: + pid = int(pid_path.read_text().strip()) + except (ValueError, OSError): + pid_path.unlink(missing_ok=True) + return + try: + os.kill(pid, 0) + except ProcessLookupError: + pid_path.unlink(missing_ok=True) + return + except PermissionError: + log(f"残留 mihomo(pid={pid}) 无权限管理,请手动 pkill -f mihomo") + return + log(f"终止残留 mihomo(pid={pid})……") + try: + os.kill(pid, signal.SIGTERM) + for _ in range(10): + try: + os.kill(pid, 0) + time.sleep(0.3) + except ProcessLookupError: + break + else: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + pid_path.unlink(missing_ok=True) + + +# ────────────────────────────────────────────────────────────── +# MihomoManager 单例 +# ────────────────────────────────────────────────────────────── +class MihomoManager: + """mihomo 常驻进程生命周期 + 订阅配置加载。线程安全(RLock)。""" + + def __init__(self) -> None: + self._lock = threading.RLock() + self._proc: subprocess.Popen | None = None + self._logf: Any = None + self._ready = False + self._port_map: dict[str, int] = {} + self._mapped: list[tuple[str, dict, str]] = [] + self._runtime_config = MIHOMO_DIR / "runtime.yaml" + self._work_dir = MIHOMO_DIR / "home" + self._log_path = MIHOMO_DIR / "mihomo.log" + self._pid_path = MIHOMO_DIR / "mihomo.pid" + self._port_map_path = MIHOMO_DIR / "port_map.json" + + @property + def ready(self) -> bool: + with self._lock: + return self._ready and self._proc is not None and self._proc.poll() is None + + @property + def port_map(self) -> dict[str, int]: + with self._lock: + return dict(self._port_map) + + @property + def mapped(self) -> list[tuple[str, dict, str]]: + with self._lock: + return list(self._mapped) + + def _resolve_bin(self) -> Path: + settings = config.get_mihomo_settings() + bin_path = Path(str(settings.get("bin_path") or "scripts/.bin/mihomo")) + if not bin_path.is_absolute(): + bin_path = PROJECT_ROOT / bin_path + return bin_path + + def _write_empty_config(self) -> None: + """空配置(无 listener,仅 controller),供首次启动。""" + settings = config.get_mihomo_settings() + cfg = { + "log-level": "warning", + "mode": "direct", + "allow-lan": False, + "external-controller": settings.get("external_controller") or "127.0.0.1:9090", + "proxies": [], + "listeners": [], + } + if settings.get("external_controller_secret"): + cfg["secret"] = settings["external_controller_secret"] + self._runtime_config.parent.mkdir(parents=True, exist_ok=True) + self._runtime_config.write_text(yaml.safe_dump(cfg, allow_unicode=True, sort_keys=False)) + + def _start(self) -> bool: + """内部启动,不持锁。失败返回 False 不抛。""" + try: + bin_path = ensure_mihomo_binary(self._resolve_bin()) + settings = config.get_mihomo_settings() + port_base = int(settings.get("port_base") or 30000) + # 端口范围预检:若 port_base 已被占(非自己),报错 + if is_port_open(port_base) and not self._is_own_port(port_base): + log(f"端口 {port_base} 已被占用,mihomo 启动中止") + return False + cleanup_stale_mihomo(self._pid_path) + if not self._runtime_config.exists(): + self._write_empty_config() + self._work_dir.mkdir(parents=True, exist_ok=True) + self._logf = open(self._log_path, "w") + self._proc = subprocess.Popen( + [str(bin_path), "-f", str(self._runtime_config), "-d", str(self._work_dir)], + stdout=self._logf, + stderr=subprocess.STDOUT, + ) + self._pid_path.write_text(str(self._proc.pid)) + if self._wait_ready(): + self._ready = True + log(f"mihomo 启动成功(pid={self._proc.pid})") + return True + log("mihomo 启动超时") + self._kill_proc() + return False + except Exception as e: + log(f"mihomo 启动失败:{type(e).__name__}: {e}") + self._ready = False + return False + + def _is_own_port(self, port: int) -> bool: + """端口是否由当前 mihomo 占用(粗判:在 port_map 里)。""" + return port in self._port_map.values() + + def _wait_ready(self, deadline_s: float = 20.0) -> bool: + settings = config.get_mihomo_settings() + ctrl = str(settings.get("external_controller") or "127.0.0.1:9090") + host, _, port_s = ctrl.partition(":") + try: + ctrl_port = int(port_s) + except ValueError: + ctrl_port = 9090 + t0 = time.perf_counter() + while time.perf_counter() - t0 < deadline_s: + if self._proc and self._proc.poll() is not None: + log(f"mihomo 进程已退出(code={self._proc.returncode})\n{_tail(self._log_path, 30)}") + return False + if is_port_open(ctrl_port, host or "127.0.0.1"): + return True + time.sleep(0.3) + return False + + def _kill_proc(self) -> None: + if self._proc and self._proc.poll() is None: + self._proc.terminate() + try: + self._proc.wait(timeout=5) + except subprocess.TimeoutExpired: + self._proc.kill() + if self._logf: + try: + self._logf.close() + except Exception: + pass + self._logf = None + self._pid_path.unlink(missing_ok=True) + + def _stop(self) -> None: + """内部停止,不持锁。""" + self._ready = False + self._kill_proc() + log("mihomo 已停止") + + def start(self) -> bool: + """启动 mihomo(用 runtime.yaml)。失败不抛,返回 False。""" + with self._lock: + if self._proc and self._proc.poll() is None: + return True + return self._start() + + def stop(self) -> None: + with self._lock: + self._stop() + + def is_healthy(self) -> bool: + return self.ready + + def apply_new_config( + self, cfg: dict, port_map: dict[str, int], mapped: list[tuple[str, dict, str]] + ) -> bool: + """写入新配置并 stop+start 应用(首版不依赖 reload)。返回是否就绪。""" + with self._lock: + self._runtime_config.parent.mkdir(parents=True, exist_ok=True) + self._runtime_config.write_text(yaml.safe_dump(cfg, allow_unicode=True, sort_keys=False)) + self._port_map = dict(port_map) + self._mapped = list(mapped) + self._stop() + return self._start() + + def build_and_load( + self, subscription_url: str, region_keywords: list[str] + ) -> tuple[dict[str, int], list[tuple[str, dict, str]]]: + """拉订阅→筛→分配端口(复用旧映射)→生成配置→apply_new_config。返回(port_map, mapped)。""" + yaml_text = fetch_subscription(subscription_url) + nodes = parse_proxies(yaml_text) + mapped = filter_region(nodes, region_keywords) + if not mapped: + raise ValueError("订阅中没有命中地区关键词的节点") + settings = config.get_mihomo_settings() + port_base = int(settings.get("port_base") or 30000) + range_size = int(settings.get("port_range_size") or 500) + old_port_map = self._load_port_map() + new_node_ids = [nid for nid, _, _ in mapped] + port_map = allocate_ports(new_node_ids, old_port_map, port_base, range_size) + cfg, _ = build_mihomo_config( + mapped, + port_map, + str(settings.get("external_controller") or "127.0.0.1:9090"), + str(settings.get("external_controller_secret") or ""), + ) + self._save_port_map(port_map) + if not self.apply_new_config(cfg, port_map, mapped): + raise RuntimeError("mihomo 应用新配置失败") + return port_map, mapped + + def _load_port_map(self) -> dict[str, int]: + if self._port_map_path.exists(): + try: + data = json.loads(self._port_map_path.read_text()) + return {str(k): int(v) for k, v in data.items()} if isinstance(data, dict) else {} + except (json.JSONDecodeError, ValueError, OSError): + return {} + return {} + + def _save_port_map(self, port_map: dict[str, int]) -> None: + self._port_map_path.parent.mkdir(parents=True, exist_ok=True) + self._port_map_path.write_text(json.dumps(port_map)) + + def get_socks5_url(self, node_id: str) -> str: + port = self._port_map.get(node_id) + return f"socks5://127.0.0.1:{port}" if port else "" + + def get_node_meta(self, node_id: str) -> dict | None: + """返回某节点的元信息(原始 name、节点字段)。""" + with self._lock: + for nid, node, orig in self._mapped: + if nid == node_id: + return {"node_id": nid, "name": orig, "node": node} + return None + + +mihomo_manager = MihomoManager() diff --git a/services/network/client.py b/services/network/client.py index 1364453..1c83bd7 100644 --- a/services/network/client.py +++ b/services/network/client.py @@ -2,16 +2,16 @@ from typing import Any -def build_session_kwargs(*, impersonate: str | None = None, verify: bool = True, **session_kwargs: Any) -> dict[str, object]: +def build_session_kwargs(*, account_proxy: str = "", impersonate: str | None = None, verify: bool = True, **session_kwargs: Any) -> dict[str, object]: if impersonate: session_kwargs["impersonate"] = impersonate session_kwargs["verify"] = verify from services.proxy_service import proxy_settings - return proxy_settings.build_session_kwargs(**session_kwargs) + return proxy_settings.build_session_kwargs(account_proxy=account_proxy, **session_kwargs) -def create_session(*, impersonate: str | None = None, verify: bool = True, **session_kwargs: Any): +def create_session(*, account_proxy: str = "", impersonate: str | None = None, verify: bool = True, **session_kwargs: Any): from curl_cffi import requests - return requests.Session(**build_session_kwargs(impersonate=impersonate, verify=verify, **session_kwargs)) + return requests.Session(**build_session_kwargs(account_proxy=account_proxy, impersonate=impersonate, verify=verify, **session_kwargs)) diff --git a/services/openai_backend_api.py b/services/openai_backend_api.py index 72f2413..8440a2c 100644 --- a/services/openai_backend_api.py +++ b/services/openai_backend_api.py @@ -69,16 +69,18 @@ class OpenAIBackendAPI: - 协议兼容转换放在 `services.protocol` """ - def __init__(self, access_token: str = "") -> None: + def __init__(self, access_token: str = "", *, account_proxy: str = "") -> None: """初始化后端客户端。 参数: - `access_token`:可选。传入后表示使用已登录链路;不传则使用未登录链路。 + - `account_proxy`:可选。账号级代理 URL,优先于全局代理。 """ self.base_url = "https://chatgpt.com" self.client_version = DEFAULT_CLIENT_VERSION self.client_build_number = DEFAULT_CLIENT_BUILD_NUMBER self.access_token = access_token + self.account_proxy = account_proxy self.network_profile = self._build_network_profile() self.fp = self.network_profile.as_fingerprint() self.user_agent = self.fp["user-agent"] @@ -86,7 +88,7 @@ def __init__(self, access_token: str = "") -> None: self.session_id = self.fp["oai-session-id"] self.pow_script_sources: list[str] = [] self.pow_data_build = "" - self.session = create_session(impersonate=self.network_profile.impersonate, verify=self.network_profile.verify) + self.session = create_session(account_proxy=self.account_proxy, impersonate=self.network_profile.impersonate, verify=self.network_profile.verify) self.session.headers.update(build_chatgpt_web_headers( self.network_profile, base_url=self.base_url, @@ -155,7 +157,7 @@ def _refresh_session(self) -> None: self.device_id = self.fp["oai-device-id"] self.session_id = self.fp["oai-session-id"] - self.session = create_session(impersonate=fresh.impersonate, verify=fresh.verify) + self.session = create_session(account_proxy=self.account_proxy, impersonate=fresh.impersonate, verify=fresh.verify) self.session.headers.update(build_chatgpt_web_headers( fresh, base_url=self.base_url, diff --git a/services/protocol/anthropic_v1_messages.py b/services/protocol/anthropic_v1_messages.py index e1a075d..6d19078 100644 --- a/services/protocol/anthropic_v1_messages.py +++ b/services/protocol/anthropic_v1_messages.py @@ -110,8 +110,10 @@ def preprocess_payload(payload: dict[str, object], text_mapper: Callable[[str], def message_request(body: dict[str, Any]) -> MessageRequest: payload = preprocess_payload(dict(body)) + token = account_service.get_text_access_token() + account = account_service.get_account(token) or {} return MessageRequest( - backend=OpenAIBackendAPI(access_token=account_service.get_text_access_token()), + backend=OpenAIBackendAPI(access_token=token, account_proxy=str(account.get("proxy") or "")), messages=normalize_messages(payload.get("messages"), payload.get("system")), model=str(payload.get("model") or "auto").strip() or "auto", tools=payload.get("tools"), diff --git a/services/protocol/openai_search.py b/services/protocol/openai_search.py index 996554d..4f958fd 100644 --- a/services/protocol/openai_search.py +++ b/services/protocol/openai_search.py @@ -1,5 +1,7 @@ from __future__ import annotations +import logging +import time from typing import Any from fastapi import HTTPException @@ -9,7 +11,10 @@ from services.openai_backend_api import OpenAIBackendAPI, SEARCH_MODEL from services.protocol.openai_v1_chat_complete import completion_response +logger = logging.getLogger(__name__) + MODEL = SEARCH_MODEL +_MAX_RETRIES = 3 def _source_items(value: object) -> list[dict[str, str]]: @@ -36,15 +41,47 @@ def handle(body: dict[str, Any]) -> dict[str, Any]: if not prompt: raise HTTPException(status_code=400, detail={"error": "prompt is required"}) model = str(body.get("model") or MODEL).strip() or MODEL - token = account_service.get_text_access_token(provider="gpt") - if not token: + + attempted: set[str] = set() + last_error: Exception | None = None + + for attempt in range(_MAX_RETRIES): + token = account_service.get_text_access_token(excluded_tokens=attempted, provider="gpt") + if not token: + break + attempted.add(token) + account = account_service.get_account(token) or {} + proxy = str(account.get("proxy") or "") + + backend = OpenAIBackendAPI(token, account_proxy=proxy) + try: + result = backend.search(prompt, model=model, timeout_secs=90, poll_interval_secs=3) + account_service.mark_text_used(token) + break + except Exception as exc: + logger.warning( + "search failed for %s (attempt %d/%d): %s", + account.get("email", "?"), + attempt + 1, + _MAX_RETRIES, + exc, + ) + # Transition the account's state (限流/异常/禁用) so degraded accounts + # exit rotation and get auto-recovered by the limited-account-watcher + # instead of being retried on every request. + try: + account_service.mark_search_failure(token, exc) + except Exception as mark_exc: + logger.warning("failed to mark search failure for account: %s", mark_exc) + last_error = exc + continue + finally: + backend.close() + else: + if last_error: + raise HTTPException(status_code=502, detail={"error": f"search failed after {_MAX_RETRIES} attempts: {last_error}"}) raise HTTPException(status_code=429, detail={"error": "no available text account"}) - backend = OpenAIBackendAPI(token) - try: - result = backend.search(prompt, model=model) - finally: - backend.close() - account_service.mark_text_used(token) + sources = _source_items(result.get("sources")) answer = str(result.get("answer") or "") response: dict[str, Any] = { diff --git a/services/protocol/openai_v1_models.py b/services/protocol/openai_v1_models.py index 318176f..9b6e7f6 100644 --- a/services/protocol/openai_v1_models.py +++ b/services/protocol/openai_v1_models.py @@ -30,8 +30,8 @@ def _empty_model_result() -> dict[str, Any]: return {"object": "list", "data": []} -def _fetch_chatgpt_models(OpenAIBackendAPI: type, access_token: str = "") -> dict[str, Any]: - with OpenAIBackendAPI(access_token) as backend: +def _fetch_chatgpt_models(OpenAIBackendAPI: type, access_token: str = "", account_proxy: str = "") -> dict[str, Any]: + with OpenAIBackendAPI(access_token, account_proxy=account_proxy) as backend: return backend.list_models() @@ -43,6 +43,17 @@ def _get_gpt_access_token() -> str: return "" +def _get_account_proxy(token: str) -> str: + if not token: + return "" + try: + from services.account_service import account_service + account = account_service.get_account(token) + return str((account or {}).get("proxy") or "") + except Exception: + return "" + + def list_models() -> dict[str, Any]: try: from services.openai_backend_api import OpenAIBackendAPI @@ -52,7 +63,7 @@ def list_models() -> dict[str, Any]: access_token = _get_gpt_access_token() if access_token: try: - result = _fetch_chatgpt_models(OpenAIBackendAPI, access_token) + result = _fetch_chatgpt_models(OpenAIBackendAPI, access_token, _get_account_proxy(access_token)) except Exception: try: result = _fetch_chatgpt_models(OpenAIBackendAPI) diff --git a/services/providers/gemini/accounts.py b/services/providers/gemini/accounts.py index 947d756..21d8c3b 100644 --- a/services/providers/gemini/accounts.py +++ b/services/providers/gemini/accounts.py @@ -289,7 +289,7 @@ def validate_remote_info(access_token: str, account: dict[str, Any] | None = Non if access_token: source.setdefault("access_token", access_token) cookie_header_value = account_cookie_header(source) - with GeminiWebClient(cookie_header_value, source.get("user_agent")) as client: + with GeminiWebClient(cookie_header_value, source.get("user_agent"), account_proxy=str(source.get("proxy") or "")) as client: client.rotate_psidts() session_token = client.bootstrap_session_token() return gemini_session_writeback(source, client.cookie_header, session_token) diff --git a/services/providers/gemini/client.py b/services/providers/gemini/client.py index 7144d4b..a1b847d 100644 --- a/services/providers/gemini/client.py +++ b/services/providers/gemini/client.py @@ -459,11 +459,11 @@ def parse_web_response_text(raw_text: str) -> object: class GeminiWebClient: - def __init__(self, cookie_header: str, user_agent: str | None = None) -> None: + def __init__(self, cookie_header: str, user_agent: str | None = None, account_proxy: str = "") -> None: self.cookie_header = cookie_header self.user_agent = user_agent or GEMINI_BROWSER_USER_AGENT self.session_token = "" - self.session = create_session() + self.session = create_session(account_proxy=account_proxy) def __enter__(self) -> "GeminiWebClient": return self @@ -587,7 +587,7 @@ def fetch_authenticated_init_body() -> str: return "" account = account_service.get_account(access_token) or {"access_token": access_token, "provider": "gemini"} cookie_header = account_cookie_header(account) - with GeminiWebClient(cookie_header, account.get("user_agent")) as client: + with GeminiWebClient(cookie_header, account.get("user_agent"), account_proxy=str(account.get("proxy") or "")) as client: init_body = client.fetch_init_body() persist_gemini_session(account_service, access_token, account, client.cookie_header) return init_body @@ -614,7 +614,7 @@ def chat_completion(body: dict[str, Any], spec: ModelSpec, messages: list[dict[s if session_token: payload["session_token"] = session_token try: - with GeminiWebClient(cookie_header, account.get("user_agent")) as client: + with GeminiWebClient(cookie_header, account.get("user_agent"), account_proxy=str(account.get("proxy") or "")) as client: response_payload = client.generate(payload) persist_gemini_session(account_service, access_token, account, client.cookie_header, client.session_token) except GeminiWebError as exc: diff --git a/services/providers/gpt/runtime.py b/services/providers/gpt/runtime.py index f14e8d5..dc464cb 100644 --- a/services/providers/gpt/runtime.py +++ b/services/providers/gpt/runtime.py @@ -380,8 +380,16 @@ def conversation_events( yield from iter_conversation_payloads(payloads, history_text, history_messages) +def _account_proxy(token: str) -> str: + if not token: + return "" + account = account_service.get_account(token) + return str((account or {}).get("proxy") or "") + + def text_backend() -> OpenAIBackendAPI: - return OpenAIBackendAPI(access_token=account_service.get_text_access_token()) + token = account_service.get_text_access_token() + return OpenAIBackendAPI(access_token=token, account_proxy=_account_proxy(token)) def stream_text_deltas(backend: OpenAIBackendAPI, request: ConversationRequest) -> Iterator[str]: @@ -394,7 +402,7 @@ def stream_text_deltas(backend: OpenAIBackendAPI, request: ConversationRequest) if token: attempted_tokens.add(token) try: - active_backend = OpenAIBackendAPI(access_token=token) + active_backend = OpenAIBackendAPI(access_token=token, account_proxy=_account_proxy(token)) try: for event in conversation_events(active_backend, messages=request.messages, model=request.model, prompt=request.prompt): if event.get("type") != "conversation.delta": @@ -521,7 +529,7 @@ def stream_image_outputs_with_pool(request: ConversationRequest) -> Iterator[Ima emitted_for_token = False returned_message = False returned_result = False - backend = OpenAIBackendAPI(access_token=token) + backend = OpenAIBackendAPI(access_token=token, account_proxy=_account_proxy(token)) try: try: for output in stream_image_outputs(backend, request, index, request.n): diff --git a/services/providers/grok/client.py b/services/providers/grok/client.py index 07b5f9c..e2b7e70 100644 --- a/services/providers/grok/client.py +++ b/services/providers/grok/client.py @@ -665,10 +665,12 @@ def _raise_console_upstream_error(access_token: str, upstream_status: int, respo class GrokConsoleClient: - def __init__(self, access_token: str) -> None: + def __init__(self, access_token: str, account: dict[str, Any] | None = None) -> None: self.access_token = access_token + self.account = account if isinstance(account, dict) else None + self.account_proxy = str((self.account or {}).get("proxy") or "") self.network_profile = _grok_console_profile() - self.session = create_session(impersonate=self.network_profile.impersonate, verify=self.network_profile.verify) + self.session = create_session(account_proxy=self.account_proxy, impersonate=self.network_profile.impersonate, verify=self.network_profile.verify) def close(self) -> None: self.session.close() @@ -1838,9 +1840,10 @@ class GrokAppChatClient: def __init__(self, access_token: str, account: dict[str, Any] | None = None) -> None: self.access_token = access_token self.account = account if isinstance(account, dict) else None + self.account_proxy = str((self.account or {}).get("proxy") or "") self.network_profile = _grok_app_chat_profile() impersonate = _app_chat_impersonate(self.network_profile, self.account) - self.session = create_session(impersonate=impersonate, verify=self.network_profile.verify) + self.session = create_session(account_proxy=self.account_proxy, impersonate=impersonate, verify=self.network_profile.verify) def close(self) -> None: self.session.close() @@ -2421,8 +2424,9 @@ def console_chat_completion(body: dict[str, Any], spec: ModelSpec, messages: lis access_token = account_service.get_grok_console_access_token() if not access_token: raise HTTPException(status_code=503, detail={"error": "no available Grok account"}) + account = account_service.get_account(access_token, provider="grok") try: - with GrokConsoleClient(access_token) as client: + with GrokConsoleClient(access_token, account) as client: response_json = client.create_response(payload) except GrokConsoleError as exc: account_service.mark_grok_console_used(access_token, success=False) @@ -2442,8 +2446,9 @@ def console_chat_completion_events(body: dict[str, Any], spec: ModelSpec, messag access_token = account_service.get_grok_console_access_token() if not access_token: raise HTTPException(status_code=503, detail={"error": "no available Grok account"}) + account = account_service.get_account(access_token, provider="grok") try: - with GrokConsoleClient(access_token) as client: + with GrokConsoleClient(access_token, account) as client: for event in client.stream_response(payload): yield event except GrokConsoleError as exc: diff --git a/services/proxy_health_service.py b/services/proxy_health_service.py new file mode 100644 index 0000000..63d0fe5 --- /dev/null +++ b/services/proxy_health_service.py @@ -0,0 +1,242 @@ +"""订阅代理池健康服务:周期/手动同步订阅 → 测活 → 更新 proxy_pool → 重分配失效账号。 + +闭环:mihomo_manager.build_and_load(拉订阅+筛+起 mihomo 多端口)→ probe_all(curl_cffi +测 chatgpt.com)→ make_subscription_item + replace_subscription_items(只留可用) +→ reassign_invalid_accounts(仅失效账号换出口)。 + +调度仿 BackupService(threading.Thread + stop_event.wait)。sync 单飞锁防并发。 +长任务异步化:run_sync_now 返回 task_id,前端轮询 get_task_status。 + +首版限制:多订阅时每个订阅 build_and_load 会覆盖 mihomo 配置(以最后订阅为准);多订阅 +节点合并留后续。当前用户单订阅场景不受影响。 +""" + +from __future__ import annotations + +import threading +import uuid +from datetime import datetime, timezone +from typing import Any + +from services.config import config +from services.mihomo_manager import mihomo_manager, probe_all +from services.proxy_pool_service import make_subscription_item, proxy_pool_service + + +def _now() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + + +def _parse_region(raw: str) -> list[str]: + return [k.strip() for k in str(raw or "").split(",") if k.strip()] + + +class ProxyHealthService: + def __init__(self) -> None: + self._lock = threading.RLock() + self._stop_event = threading.Event() + self._thread: threading.Thread | None = None + self._sync_lock = threading.Lock() # sync 单飞锁 + self._tasks: dict[str, dict[str, Any]] = {} + self._storage = config.get_storage_backend() + + # ── 调度 ── + def start(self) -> None: + with self._lock: + if self._thread and self._thread.is_alive(): + return + self._stop_event.clear() + self._thread = threading.Thread(target=self._run, daemon=True, name="proxy-health-scheduler") + self._thread.start() + + def stop(self) -> None: + with self._lock: + self._stop_event.set() + thread = self._thread + self._thread = None + if thread and thread.is_alive(): + thread.join(timeout=2) + + def _run(self) -> None: + """周期 sync:启动后立即跑一次(若有启用订阅),之后按 sync_interval_hours 周期。""" + # 首次启动立即同步(补强5) + self._try_sync(task_id=None, subscription_id=None) + while not self._stop_event.is_set(): + try: + settings = config.get_mihomo_settings() + interval = max(1, int(settings.get("sync_interval_hours") or 24)) * 3600 + except Exception: + interval = 24 * 3600 + self._stop_event.wait(interval) + if self._stop_event.is_set(): + break + self._try_sync(task_id=None, subscription_id=None) + + # ── sync 单飞 + 异步 task ── + def run_sync_now(self, subscription_id: str | None = None) -> dict[str, Any]: + """立即触发一次 sync(异步),返回 task_id 供轮询。若已有 sync 在跑则返回 skipped。""" + task_id = str(uuid.uuid4()) + with self._lock: + self._tasks[task_id] = { + "status": "pending", + "progress": 0, + "result": None, + "error": None, + "started_at": _now(), + } + threading.Thread( + target=self._try_sync, args=(task_id, subscription_id), daemon=True, name="proxy-sync" + ).start() + return {"task_id": task_id} + + def _try_sync(self, task_id: str | None, subscription_id: str | None) -> None: + """单飞:抢不到锁则跳过(手动 sync 记 skipped)。抢到则跑 run_sync。""" + if not self._sync_lock.acquire(blocking=False): + if task_id: + self._update_task(task_id, status="skipped", error="another sync running") + return + try: + if task_id: + self._update_task(task_id, status="running", progress=10) + result = self.run_sync(subscription_id) + if task_id: + status = "error" if result.get("errors") and not result.get("usable") else "done" + self._update_task(task_id, status=status, progress=100, result=result) + except Exception as e: + if task_id: + self._update_task(task_id, status="error", error=f"{type(e).__name__}: {e}") + finally: + self._sync_lock.release() + + def get_task_status(self, task_id: str) -> dict[str, Any]: + with self._lock: + return dict(self._tasks.get(task_id, {"status": "unknown", "error": "task not found"})) + + def _update_task(self, task_id: str, **fields) -> None: + with self._lock: + task = self._tasks.get(task_id) + if task: + task.update(fields) + + # ── sync 核心闭环 ── + def run_sync(self, subscription_id: str | None = None) -> dict[str, Any]: + """同步执行 sync 闭环(阻塞,调用方负责单飞)。返回 {total, usable, reassigned, errors}。""" + settings = config.get_mihomo_settings() + test_url = str(settings.get("test_url") or "https://chatgpt.com/api/auth/csrf") + timeout = float(settings.get("test_timeout") or 15) + concurrency = int(settings.get("concurrency") or 16) + retry = int(settings.get("probe_retry") or 1) + + subs = self._storage.load_subscriptions() + if subscription_id: + subs = [s for s in subs if s.get("id") == subscription_id] + subs = [s for s in subs if s.get("enabled", True)] + if not subs: + return {"error": "无启用的订阅源", "total": 0, "usable": 0, "reassigned": 0, "errors": []} + + total_all = 0 + usable_all = 0 + errors: list[str] = [] + + for sub in subs: + region = _parse_region(sub.get("region_keywords") or settings.get("region_keywords") or "") + try: + port_map, mapped = mihomo_manager.build_and_load(sub["url"], region) + except Exception as e: + msg = f"{sub.get('name')}: {type(e).__name__}: {e}" + errors.append(msg) + self._update_sub_last_sync(sub["id"], error=msg) + continue + + results = probe_all( + port_map, test_url, timeout, concurrency, retry, + ) + items: list[dict[str, Any]] = [] + region_str = ",".join(region) + for node_id, node, _orig in mapped: + r = results.get(node_id, {}) + if r.get("ok"): + items.append( + make_subscription_item( + sub["id"], node_id, port_map[node_id], + str(node.get("type") or ""), region_str, dict(node), + int(r.get("latency_ms") or 0), + ) + ) + proxy_pool_service.replace_subscription_items(sub["id"], items) + self._update_sub_last_sync(sub["id"], total=len(mapped), usable=len(items)) + total_all += len(mapped) + usable_all += len(items) + + # sync 后重分配失效账号(仅失效的,补强决策) + try: + reassign = proxy_pool_service.reassign_invalid_accounts() + reassigned = int(reassign.get("reassigned") or 0) + except Exception as e: + reassigned = 0 + errors.append(f"reassign_failed: {e}") + + return { + "total": total_all, + "usable": usable_all, + "reassigned": reassigned, + "errors": errors, + } + + # ── 订阅源 CRUD ── + def list_subscriptions(self) -> list[dict[str, Any]]: + return self._storage.load_subscriptions() + + def add_subscription(self, name: str, url: str, region_keywords: str = "") -> dict[str, Any]: + subs = self._storage.load_subscriptions() + sub = { + "id": str(uuid.uuid4()), + "name": str(name or "").strip() or "未命名订阅", + "url": str(url or "").strip(), + "region_keywords": str(region_keywords or "").strip(), + "enabled": True, + "last_sync_at": "", + "last_total": 0, + "last_usable": 0, + "last_error": "", + } + subs.append(sub) + self._storage.save_subscriptions(subs) + return sub + + def delete_subscription(self, sub_id: str) -> dict[str, Any]: + subs = self._storage.load_subscriptions() + before = len(subs) + subs = [s for s in subs if s.get("id") != sub_id] + removed = before - len(subs) + self._storage.save_subscriptions(subs) + if removed: + proxy_pool_service.delete_subscription_items(sub_id) + return {"removed": removed} + + def update_subscription(self, sub_id: str, fields: dict[str, Any]) -> dict[str, Any]: + subs = self._storage.load_subscriptions() + updated = None + for s in subs: + if s.get("id") == sub_id: + for k in ("name", "url", "region_keywords", "enabled"): + if k in fields: + s[k] = fields[k] + updated = s + self._storage.save_subscriptions(subs) + return updated or {"error": "subscription not found"} + + def _update_sub_last_sync( + self, sub_id: str, total: int = 0, usable: int = 0, error: str = "" + ) -> None: + subs = self._storage.load_subscriptions() + for s in subs: + if s.get("id") == sub_id: + s["last_sync_at"] = _now() + s["last_total"] = total + s["last_usable"] = usable + s["last_error"] = error + self._storage.save_subscriptions(subs) + + +proxy_health_service = ProxyHealthService() diff --git a/services/proxy_pool_service.py b/services/proxy_pool_service.py new file mode 100644 index 0000000..f9f1031 --- /dev/null +++ b/services/proxy_pool_service.py @@ -0,0 +1,285 @@ +"""Proxy pool management: import, list, delete, assign to accounts.""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from threading import Lock +from typing import Any +from urllib.parse import quote as url_quote + +from services.config import config +from services.storage.base import StorageBackend + + +def _clean(value: object) -> str: + return str(value or "").strip() + + +def _parse_proxy_line(line: str) -> dict[str, Any] | None: + """Parse 'IP:port:username:password' into a proxy pool item.""" + parts = line.strip().split(":") + if len(parts) < 2: + return None + if len(parts) == 2: + host, port = parts[0], parts[1] + username, password = "", "" + elif len(parts) == 4: + host, port, username, password = parts + else: + host = parts[0] + port = parts[1] + username = parts[2] + password = ":".join(parts[3:]) + host = host.strip() + port = port.strip() + username = username.strip() + password = password.strip() + if not host or not port: + return None + try: + port_int = int(port) + except ValueError: + return None + if port_int < 1 or port_int > 65535: + return None + if username and password: + url = f"http://{url_quote(username, safe='')}:{url_quote(password, safe='')}@{host}:{port_int}" + else: + url = f"http://{host}:{port_int}" + return { + "id": str(uuid.uuid4()), + "url": url, + "host": host, + "port": port_int, + "username": username, + "created_at": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"), + } + + +def make_subscription_item( + subscription_id: str, + node_id: str, + port: int, + protocol: str, + region: str, + raw_node: dict[str, Any], + latency_ms: int, +) -> dict[str, Any]: + """构造一个订阅式代理池 item(url=socks5://127.0.0.1:port,由 mihomo listener 提供出口)。""" + now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + return { + "id": str(uuid.uuid4()), + "url": f"socks5://127.0.0.1:{port}", + "host": "127.0.0.1", + "port": port, + "username": "", + "created_at": now, + "source": "subscription", + "subscription_id": subscription_id, + "node_id": node_id, + "region": region, + "protocol": protocol, + "raw_node": raw_node, + "local_port": port, + "health": "ok", + "last_check_at": now, + "latency_ms": latency_ms, + } + + +class ProxyPoolService: + def __init__(self, storage: StorageBackend): + self._storage = storage + self._lock = Lock() + self._items: list[dict[str, Any]] = self._storage.load_proxy_pool() + + def list_items(self) -> list[dict[str, Any]]: + with self._lock: + return list(self._items) + + def import_proxies(self, text: str) -> dict[str, Any]: + """Import proxies from text (one per line, format: IP:port:user:pass).""" + lines = [line.strip() for line in text.strip().splitlines() if line.strip()] + added = 0 + skipped = 0 + existing_urls = set() + with self._lock: + existing_urls = {item["url"] for item in self._items} + for line in lines: + parsed = _parse_proxy_line(line) + if parsed is None: + skipped += 1 + continue + if parsed["url"] in existing_urls: + skipped += 1 + continue + self._items.append(parsed) + existing_urls.add(parsed["url"]) + added += 1 + self._save() + return {"added": added, "skipped": skipped, "total": len(self._items)} + + def delete_proxies(self, ids: list[str]) -> dict[str, Any]: + """Delete proxies by their IDs.""" + id_set = set(ids) + with self._lock: + before = len(self._items) + self._items = [item for item in self._items if item["id"] not in id_set] + removed = before - len(self._items) + self._save() + return {"removed": removed, "total": len(self._items)} + + def clear_all(self) -> dict[str, Any]: + """Remove all proxies from the pool.""" + with self._lock: + removed = len(self._items) + self._items = [] + self._save() + return {"removed": removed, "total": 0} + + def assign_to_accounts(self) -> dict[str, Any]: + """Assign proxies round-robin to all accounts.""" + from services.account_service import account_service + + with self._lock: + pool = list(self._items) + + if not pool: + return {"assigned": 0, "error": "代理池为空,请先导入代理"} + + accounts = account_service.list_accounts() + if not accounts: + return {"assigned": 0, "error": "没有账号可分配"} + + assigned = 0 + for i, account in enumerate(accounts): + proxy_item = pool[i % len(pool)] + access_token = _clean(account.get("access_token")) + if not access_token: + continue + account_service.update_account(access_token, {"proxy": proxy_item["url"]}, provider=account.get("provider")) + assigned += 1 + + return {"assigned": assigned, "total_proxies": len(pool), "total_accounts": len(accounts)} + + def clear_assignments(self) -> dict[str, Any]: + """Clear proxy assignments from all accounts.""" + from services.account_service import account_service + + accounts = account_service.list_accounts() + cleared = 0 + for account in accounts: + access_token = _clean(account.get("access_token")) + if not access_token: + continue + if _clean(account.get("proxy")): + account_service.update_account(access_token, {"proxy": ""}, provider=account.get("provider")) + cleared += 1 + + return {"cleared": cleared} + + def add_subscription_items(self, subscription_id: str, items: list[dict[str, Any]]) -> dict[str, Any]: + """批量追加订阅 item,去重按 (subscription_id, node_id)。""" + with self._lock: + existing_keys = { + (it.get("subscription_id"), it.get("node_id")) + for it in self._items + if it.get("source") == "subscription" + } + added = 0 + for item in items: + key = (item.get("subscription_id"), item.get("node_id")) + if key in existing_keys: + continue + self._items.append(item) + existing_keys.add(key) + added += 1 + self._save() + return {"added": added, "total": len(self._items)} + + def replace_subscription_items(self, subscription_id: str, items: list[dict[str, Any]]) -> dict[str, Any]: + """整体替换某订阅的 item:删旧加新(sync 后调用)。""" + with self._lock: + before = sum(1 for it in self._items if it.get("subscription_id") == subscription_id) + self._items = [it for it in self._items if it.get("subscription_id") != subscription_id] + existing_keys = { + (it.get("subscription_id"), it.get("node_id")) + for it in self._items + if it.get("source") == "subscription" + } + added = 0 + for item in items: + key = (item.get("subscription_id"), item.get("node_id")) + if key in existing_keys: + continue + self._items.append(item) + existing_keys.add(key) + added += 1 + self._save() + return {"removed": before, "added": added, "total": len(self._items)} + + def delete_subscription_items(self, subscription_id: str) -> dict[str, Any]: + """删除某订阅的所有 item(删订阅源时调用)。""" + with self._lock: + before = len(self._items) + self._items = [it for it in self._items if it.get("subscription_id") != subscription_id] + removed = before - len(self._items) + self._save() + return {"removed": removed, "total": len(self._items)} + + def update_subscription_health( + self, subscription_id: str, health_map: dict[str, dict[str, Any]] + ) -> dict[str, Any]: + """更新某订阅 item 的 health/latency/last_check_at。health_map: node_id -> {health, latency_ms}。""" + now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + updated = 0 + with self._lock: + for it in self._items: + if it.get("subscription_id") != subscription_id: + continue + nid = it.get("node_id") + if nid in health_map: + info = health_map[nid] + it["health"] = info.get("health", "unknown") + it["latency_ms"] = info.get("latency_ms", 0) + it["last_check_at"] = now + updated += 1 + self._save() + return {"updated": updated} + + def reassign_invalid_accounts(self) -> dict[str, Any]: + """找出 proxy 失效的账号(空或不在可用集合)并重新分配,不动仍有效的账号。""" + from services.account_service import account_service + + with self._lock: + pool = list(self._items) + if not pool: + return {"reassigned": 0, "error": "代理池为空"} + usable = [it for it in pool if it.get("health", "unknown") != "down"] + if not usable: + return {"reassigned": 0, "error": "无可用代理"} + usable_urls = {it["url"] for it in usable} + accounts = account_service.list_accounts() + if not accounts: + return {"reassigned": 0, "error": "没有账号"} + reassigned = 0 + pick = 0 + for account in accounts: + access_token = _clean(account.get("access_token")) + if not access_token: + continue + cur = _clean(account.get("proxy")) + if cur and cur in usable_urls: + continue + new_item = usable[pick % len(usable)] + pick += 1 + account_service.update_account(access_token, {"proxy": new_item["url"]}, provider=account.get("provider")) + reassigned += 1 + return {"reassigned": reassigned, "total_usable": len(usable), "total_accounts": len(accounts)} + + def _save(self) -> None: + self._storage.save_proxy_pool(self._items) + + +proxy_pool_service = ProxyPoolService(config.get_storage_backend()) diff --git a/services/proxy_service.py b/services/proxy_service.py index c3587af..59a473f 100644 --- a/services/proxy_service.py +++ b/services/proxy_service.py @@ -11,8 +11,8 @@ class ProxySettingsStore: - def build_session_kwargs(self, **session_kwargs) -> dict[str, object]: - proxy = config.get_proxy_settings() + def build_session_kwargs(self, *, account_proxy: str = "", **session_kwargs) -> dict[str, object]: + proxy = account_proxy or config.get_proxy_settings() if proxy: session_kwargs["proxy"] = proxy return session_kwargs diff --git a/services/storage/base.py b/services/storage/base.py index fe4e345..856691c 100644 --- a/services/storage/base.py +++ b/services/storage/base.py @@ -37,6 +37,26 @@ def save_settings(self, settings: dict[str, Any]) -> None: """保存全局设置""" pass + @abstractmethod + def load_proxy_pool(self) -> list[dict[str, Any]]: + """加载代理池数据""" + pass + + @abstractmethod + def save_proxy_pool(self, items: list[dict[str, Any]]) -> None: + """保存代理池数据""" + pass + + @abstractmethod + def load_subscriptions(self) -> list[dict[str, Any]]: + """加载订阅源数据""" + pass + + @abstractmethod + def save_subscriptions(self, items: list[dict[str, Any]]) -> None: + """保存订阅源数据""" + pass + @abstractmethod def health_check(self) -> dict[str, Any]: """健康检查,返回存储后端状态""" diff --git a/services/storage/database_storage.py b/services/storage/database_storage.py index 60bd781..941428b 100644 --- a/services/storage/database_storage.py +++ b/services/storage/database_storage.py @@ -79,6 +79,70 @@ def save_auth_keys(self, auth_keys: list[dict[str, Any]]) -> None: """保存鉴权密钥数据到数据库""" self._save_rows(AuthKeyModel, auth_keys, "id", "key_id") + def load_proxy_pool(self) -> list[dict[str, Any]]: + """从数据库加载代理池数据""" + session = self.Session() + try: + row = session.query(AppSettingModel).filter_by(key="proxy_pool").one_or_none() + if row is None: + return [] + try: + data = json.loads(row.data) + except json.JSONDecodeError: + return [] + return data if isinstance(data, list) else [] + finally: + session.close() + + def save_proxy_pool(self, items: list[dict[str, Any]]) -> None: + """保存代理池数据到数据库""" + session = self.Session() + try: + row = session.query(AppSettingModel).filter_by(key="proxy_pool").one_or_none() + data = json.dumps(items, ensure_ascii=False) + if row is None: + session.add(AppSettingModel(key="proxy_pool", data=data)) + else: + row.data = data + session.commit() + except Exception as e: + session.rollback() + raise e + finally: + session.close() + + def load_subscriptions(self) -> list[dict[str, Any]]: + """从数据库加载订阅源数据""" + session = self.Session() + try: + row = session.query(AppSettingModel).filter_by(key="subscriptions").one_or_none() + if row is None: + return [] + try: + data = json.loads(row.data) + except json.JSONDecodeError: + return [] + return data if isinstance(data, list) else [] + finally: + session.close() + + def save_subscriptions(self, items: list[dict[str, Any]]) -> None: + """保存订阅源数据到数据库""" + session = self.Session() + try: + row = session.query(AppSettingModel).filter_by(key="subscriptions").one_or_none() + data = json.dumps(items, ensure_ascii=False) + if row is None: + session.add(AppSettingModel(key="subscriptions", data=data)) + else: + row.data = data + session.commit() + except Exception as e: + session.rollback() + raise e + finally: + session.close() + def load_settings(self) -> dict[str, Any]: """从数据库加载全局设置""" session = self.Session() diff --git a/services/storage/git_storage.py b/services/storage/git_storage.py index f3fd54b..af05afa 100644 --- a/services/storage/git_storage.py +++ b/services/storage/git_storage.py @@ -119,6 +119,40 @@ def save_auth_keys(self, auth_keys: list[dict[str, Any]]) -> None: print(f"[git-storage] save failed: {e}") raise e + def load_proxy_pool(self) -> list[dict[str, Any]]: + """从 Git 仓库加载代理池数据""" + try: + return self._load_json_file("proxy_pool.json") + except (FileNotFoundError, json.JSONDecodeError): + return [] + except Exception: + return [] + + def save_proxy_pool(self, items: list[dict[str, Any]]) -> None: + """保存代理池数据到 Git 仓库""" + try: + self._save_json_file("proxy_pool.json", items, "Update proxy pool data") + except Exception as e: + print(f"[git-storage] save proxy pool failed: {e}") + raise e + + def load_subscriptions(self) -> list[dict[str, Any]]: + """从 Git 仓库加载订阅源数据""" + try: + return self._load_json_file("subscriptions.json") + except (FileNotFoundError, json.JSONDecodeError): + return [] + except Exception: + return [] + + def save_subscriptions(self, items: list[dict[str, Any]]) -> None: + """保存订阅源数据到 Git 仓库""" + try: + self._save_json_file("subscriptions.json", items, "Update subscriptions data") + except Exception as e: + print(f"[git-storage] save subscriptions failed: {e}") + raise e + def load_settings(self) -> dict[str, Any]: """从 Git 仓库加载全局设置""" try: diff --git a/services/storage/json_storage.py b/services/storage/json_storage.py index c28b4f7..d01af54 100644 --- a/services/storage/json_storage.py +++ b/services/storage/json_storage.py @@ -82,6 +82,22 @@ def save_auth_keys(self, auth_keys: list[dict[str, Any]]) -> None: encoding="utf-8", ) + def load_proxy_pool(self) -> list[dict[str, Any]]: + """从 JSON 文件加载代理池数据""" + return self._load_json_list(self.file_path.with_name("proxy_pool.json")) + + def save_proxy_pool(self, items: list[dict[str, Any]]) -> None: + """保存代理池数据到 JSON 文件""" + self._save_json_list(self.file_path.with_name("proxy_pool.json"), items) + + def load_subscriptions(self) -> list[dict[str, Any]]: + """从 JSON 文件加载订阅源数据""" + return self._load_json_list(self.file_path.with_name("subscriptions.json")) + + def save_subscriptions(self, items: list[dict[str, Any]]) -> None: + """保存订阅源数据到 JSON 文件""" + self._save_json_list(self.file_path.with_name("subscriptions.json"), items) + def load_settings(self) -> dict[str, Any]: """从 JSON 文件加载全局设置""" return self._load_json_object(self.settings_path) diff --git a/test/test_account_api_sanitization.py b/test/test_account_api_sanitization.py index 8ed76c5..ccddd57 100644 --- a/test/test_account_api_sanitization.py +++ b/test/test_account_api_sanitization.py @@ -315,6 +315,12 @@ def load_settings(self) -> dict[str, Any]: def save_settings(self, settings: dict[str, Any]) -> None: pass + def load_proxy_pool(self) -> list[dict[str, Any]]: + return [] + + def save_proxy_pool(self, items: list[dict[str, Any]]) -> None: + pass + def health_check(self) -> dict[str, Any]: return {"ok": True} diff --git a/test/test_account_provider.py b/test/test_account_provider.py index 521fb24..e54256e 100644 --- a/test/test_account_provider.py +++ b/test/test_account_provider.py @@ -350,7 +350,7 @@ def test_gemini_refresh_by_sanitized_row_id_uses_client_helpers(self) -> None: self.assertTrue(strategy.supports_refresh(accounts[0])) self.assertEqual(result["refreshed"], 1) self.assertEqual(result["errors"], []) - client_class.assert_called_once_with("__Secure-1PSID=psid; __Secure-1PSIDTS=old-psidts", None) + client_class.assert_called_once_with("__Secure-1PSID=psid; __Secure-1PSIDTS=old-psidts", None, account_proxy="") client.rotate_psidts.assert_called_once_with() client.bootstrap_session_token.assert_called_once_with() refreshed_accounts = service.list_accounts(provider=GEMINI_PROVIDER) diff --git a/test/test_grok_client_parity.py b/test/test_grok_client_parity.py index a317537..73b7e3f 100644 --- a/test/test_grok_client_parity.py +++ b/test/test_grok_client_parity.py @@ -410,12 +410,13 @@ def test_app_chat_completion_refreshes_once_before_no_account_failure(self) -> N def test_console_chat_completion_marks_console_used_on_success(self) -> None: account_service = types.SimpleNamespace( get_grok_console_access_token=mock.Mock(return_value="console-token"), + get_account=mock.Mock(return_value={}), mark_grok_console_used=mock.Mock(), ) spec = resolve_model("grok-4.20-non-reasoning") class FakeConsoleClient: - def __init__(self, access_token: str) -> None: + def __init__(self, access_token: str, account: Any = None) -> None: self.access_token = access_token def __enter__(self) -> "FakeConsoleClient": @@ -439,12 +440,13 @@ def create_response(self, payload: dict[str, Any]) -> dict[str, Any]: def test_console_chat_completion_events_marks_console_used_on_success(self) -> None: account_service = types.SimpleNamespace( get_grok_console_access_token=mock.Mock(return_value="console-token"), + get_account=mock.Mock(return_value={}), mark_grok_console_used=mock.Mock(), ) spec = resolve_model("grok-4.20-non-reasoning") class FakeConsoleClient: - def __init__(self, access_token: str) -> None: + def __init__(self, access_token: str, account: Any = None) -> None: self.access_token = access_token def __enter__(self) -> "FakeConsoleClient": diff --git a/test/test_grok_provider.py b/test/test_grok_provider.py index 640dcd3..dd169a6 100644 --- a/test/test_grok_provider.py +++ b/test/test_grok_provider.py @@ -1003,6 +1003,7 @@ def test_non_streaming_grok_app_chat_completion_includes_reasoning_content(self) def test_console_chat_completion_uses_reserved_console_quota(self) -> None: account_service = types.SimpleNamespace( get_grok_console_access_token=mock.Mock(return_value="grok-token"), + get_account=mock.Mock(return_value={}), mark_grok_console_used=mock.Mock(), ) response_json = { @@ -1026,6 +1027,7 @@ def test_console_chat_completion_uses_reserved_console_quota(self) -> None: def test_console_chat_completion_marks_failed_request_without_extra_quota_decrement(self) -> None: account_service = types.SimpleNamespace( get_grok_console_access_token=mock.Mock(return_value="grok-token"), + get_account=mock.Mock(return_value={}), mark_grok_console_used=mock.Mock(), ) client = mock.Mock() @@ -1046,6 +1048,7 @@ def test_console_chat_completion_marks_failed_request_without_extra_quota_decrem def test_console_chat_completion_marks_empty_response_failed_without_extra_quota_decrement(self) -> None: account_service = types.SimpleNamespace( get_grok_console_access_token=mock.Mock(return_value="grok-token"), + get_account=mock.Mock(return_value={}), mark_grok_console_used=mock.Mock(), ) client = mock.Mock() @@ -1068,6 +1071,7 @@ def test_console_chat_completion_marks_empty_response_failed_without_extra_quota def test_console_chat_completion_validates_payload_before_reserving_quota(self) -> None: account_service = types.SimpleNamespace( get_grok_console_access_token=mock.Mock(return_value="grok-token"), + get_account=mock.Mock(return_value={}), mark_grok_console_used=mock.Mock(), ) @@ -1081,6 +1085,7 @@ def test_console_chat_completion_validates_payload_before_reserving_quota(self) def test_console_stream_uses_reserved_console_quota(self) -> None: account_service = types.SimpleNamespace( get_grok_console_access_token=mock.Mock(return_value="grok-token"), + get_account=mock.Mock(return_value={}), mark_grok_console_used=mock.Mock(), ) client = mock.Mock() @@ -2167,12 +2172,13 @@ def close(self) -> None: def test_grok_console_stream_does_not_mark_reserved_quota_when_generator_is_closed(self) -> None: account_service = types.SimpleNamespace( get_grok_console_access_token=mock.Mock(return_value="selected-token"), + get_account=mock.Mock(return_value={}), mark_grok_console_used=mock.Mock(), ) spec = resolve_model("grok-4.3") class FakeClient: - def __init__(self, access_token: str) -> None: + def __init__(self, access_token: str, account=None) -> None: self.access_token = access_token def __enter__(self) -> "FakeClient": @@ -2202,12 +2208,13 @@ def stream_response(self, payload): def test_grok_console_stream_does_not_mark_reserved_quota_when_stream_completes_without_events(self) -> None: account_service = types.SimpleNamespace( get_grok_console_access_token=mock.Mock(return_value="selected-token"), + get_account=mock.Mock(return_value={}), mark_grok_console_used=mock.Mock(), ) spec = resolve_model("grok-4.3") class FakeClient: - def __init__(self, access_token: str) -> None: + def __init__(self, access_token: str, account=None) -> None: self.access_token = access_token def __enter__(self) -> "FakeClient": @@ -2235,12 +2242,13 @@ def stream_response(self, payload): def test_grok_console_stream_marks_account_used_after_partial_stream_error(self) -> None: account_service = types.SimpleNamespace( get_grok_console_access_token=mock.Mock(return_value="selected-token"), + get_account=mock.Mock(return_value={}), mark_grok_console_used=mock.Mock(), ) spec = resolve_model("grok-4.3") class FakeClient: - def __init__(self, access_token: str) -> None: + def __init__(self, access_token: str, account=None) -> None: self.access_token = access_token def __enter__(self) -> "FakeClient": diff --git a/test/test_provider_models.py b/test/test_provider_models.py index 104b85f..043f49b 100644 --- a/test/test_provider_models.py +++ b/test/test_provider_models.py @@ -37,7 +37,7 @@ class FakeBackend: fail_authenticated = False fail_anonymous = False - def __init__(self, access_token: str = "") -> None: + def __init__(self, access_token: str = "", **kwargs) -> None: self.access_token = access_token self.__class__.calls.append(access_token) @@ -160,6 +160,7 @@ def test_list_models_tries_gpt_account_token_before_anonymous(self) -> None: account_service = FakeAccountService("stored-token") with mock.patch.object(openai_v1_models, "_get_gpt_access_token", account_service.get_text_access_token), \ + mock.patch.object(openai_v1_models, "_get_account_proxy", return_value=""), \ mock.patch.dict(sys.modules, {"services.openai_backend_api": types.SimpleNamespace( OpenAIBackendAPI=FakeBackend, )}): @@ -180,6 +181,7 @@ def test_list_models_falls_back_to_anonymous_when_account_fetch_fails(self) -> N FakeBackend.fail_authenticated = True with mock.patch.object(openai_v1_models, "_get_gpt_access_token", account_service.get_text_access_token), \ + mock.patch.object(openai_v1_models, "_get_account_proxy", return_value=""), \ mock.patch.dict(sys.modules, {"services.openai_backend_api": types.SimpleNamespace( OpenAIBackendAPI=FakeBackend, )}): @@ -199,6 +201,7 @@ def test_list_models_uses_fallbacks_when_authenticated_and_anonymous_fetch_fail( FakeBackend.fail_anonymous = True with mock.patch.object(openai_v1_models, "_get_gpt_access_token", account_service.get_text_access_token), \ + mock.patch.object(openai_v1_models, "_get_account_proxy", return_value=""), \ mock.patch.dict(sys.modules, {"services.openai_backend_api": types.SimpleNamespace( OpenAIBackendAPI=FakeBackend, )}): diff --git a/test/test_proxy_subscription.py b/test/test_proxy_subscription.py new file mode 100644 index 0000000..bb3f1ba --- /dev/null +++ b/test/test_proxy_subscription.py @@ -0,0 +1,143 @@ +"""订阅代理池扩展单元测试(阶段二 M3/M4)。 + +覆盖 mihomo_manager 配置生成(filter_region 去重/稳定、allocate_ports 复用/新分配、 +build_mihomo_config listener-proxy 一致性)+ proxy_pool_service 订阅方法 +(make_subscription_item、replace/delete_subscription_items、reassign_invalid_accounts)。 +""" +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + + +# ── M3: mihomo_manager 纯函数 ── +def test_filter_region_dedup_and_stable(): + from services.mihomo_manager import filter_region + + nodes = [ + {"name": "🇺🇸 美国 | No.1", "type": "vless", "server": "1.1.1.1", "port": 443, "uuid": "abc"}, + {"name": "🇺🇸 美国 | 重复", "type": "vless", "server": "1.1.1.1", "port": 443, "uuid": "abc"}, + {"name": "🇯🇵 日本", "type": "vless", "server": "2.2.2.2", "port": 443, "uuid": "def"}, + ] + m1 = filter_region(nodes, ["🇺🇸", "美国"]) + m2 = filter_region(nodes, ["🇺🇸", "美国"]) + assert len(m1) == 1, "重复节点应去重" + assert m1[0][0] == m2[0][0], "node_id 跨调用稳定" + assert m1[0][0].startswith("us-") + assert m1[0][2] == "🇺🇸 美国 | No.1", "保留原始 name" + + +def test_filter_region_empty_when_no_match(): + from services.mihomo_manager import filter_region + + nodes = [{"name": "🇯🇵 日本", "type": "vless", "server": "1.1.1.1", "port": 443}] + assert filter_region(nodes, ["🇺🇸"]) == [] + + +def test_allocate_ports_reuse_and_new(): + from services.mihomo_manager import allocate_ports + + old = {"us-aaa": 30000, "us-bbb": 30001} + new = allocate_ports(["us-aaa", "us-ccc"], old, 30000, 500) + assert new["us-aaa"] == 30000, "旧节点复用端口" + assert new["us-ccc"] == 30002, "新节点分配未用端口(避开 30000/30001)" + assert "us-bbb" not in new, "删除的节点不残留" + + +def test_allocate_ports_returns_deleted_node_port_after_save(): + """已删除节点的端口在 port_map 清理后可被新节点复用。""" + from services.mihomo_manager import allocate_ports + + # 模拟 port_map 已清理(只含现有节点) + old = {"us-aaa": 30000} + new = allocate_ports(["us-aaa", "us-bbb"], old, 30000, 500) + assert new["us-bbb"] == 30001 + + +def test_build_mihomo_config_listener_proxy_match(): + from services.mihomo_manager import build_mihomo_config, filter_region + + nodes = [ + {"name": "🇺🇸 美国 | A", "type": "vless", "server": "1.1.1.1", "port": 443, "uuid": "a"}, + {"name": "🇺🇸 美国 | B", "type": "vless", "server": "2.2.2.2", "port": 8443, "uuid": "b"}, + ] + mapped = filter_region(nodes, ["🇺🇸"]) + port_map = {nid: 30000 + i for i, (nid, _, _) in enumerate(mapped)} + cfg, _ = build_mihomo_config(mapped, port_map) + proxy_names = {p["name"] for p in cfg["proxies"]} + listener_proxies = {l["proxy"] for l in cfg["listeners"]} + assert listener_proxies == proxy_names, "每个 listener.proxy 必须对应一个 proxy" + assert all(l["type"] == "socks" for l in cfg["listeners"]) + assert all(l["listen"] == "127.0.0.1" for l in cfg["listeners"]) + assert cfg["external-controller"] + + +# ── M4: proxy_pool_service 订阅方法 ── +def _make_svc(): + from services.storage.factory import create_storage_backend + from services.proxy_pool_service import ProxyPoolService + + tmp = Path(tempfile.mkdtemp()) + return ProxyPoolService(create_storage_backend(tmp / "accounts.json")) + + +def test_make_subscription_item(): + from services.proxy_pool_service import make_subscription_item + + it = make_subscription_item("sub1", "us-abc", 30000, "vless", "US", {"server": "1.1.1.1"}, 100) + assert it["url"] == "socks5://127.0.0.1:30000" + assert it["source"] == "subscription" + assert it["subscription_id"] == "sub1" + assert it["node_id"] == "us-abc" + assert it["health"] == "ok" + assert it["local_port"] == 30000 + + +def test_replace_and_delete_subscription_items(): + from services.proxy_pool_service import make_subscription_item + + svc = _make_svc() + it1 = make_subscription_item("sub1", "us-a", 30000, "vless", "US", {}, 100) + r = svc.add_subscription_items("sub1", [it1]) + assert r["added"] == 1 and len(svc.list_items()) == 1 + + it2 = make_subscription_item("sub1", "us-b", 30001, "vless", "US", {}, 200) + r = svc.replace_subscription_items("sub1", [it2]) + assert r["removed"] == 1 and r["added"] == 1 + assert len(svc.list_items()) == 1 + assert svc.list_items()[0]["node_id"] == "us-b" + + r = svc.delete_subscription_items("sub1") + assert r["removed"] == 1 and len(svc.list_items()) == 0 + + +def test_reassign_invalid_accounts_only_invalid(): + import services.account_service as am + from services.proxy_pool_service import make_subscription_item + + class FakeAS: + def __init__(self): + self.accounts = [] + self.updates = [] + + def list_accounts(self): + return self.accounts + + def update_account(self, t, d, provider=None): + self.updates.append(t) + + am.account_service = FakeAS() + svc = _make_svc() + # usable item at 30002 + svc.add_subscription_items("sub1", [make_subscription_item("sub1", "us-ok", 30002, "vless", "US", {}, 50)]) + am.account_service.accounts = [ + {"access_token": "tok1", "proxy": "socks5://127.0.0.1:30002", "provider": "gpt"}, # 有效 + {"access_token": "tok2", "proxy": "", "provider": "gpt"}, # 失效(空) + {"access_token": "tok3", "proxy": "socks5://127.0.0.1:99999", "provider": "gpt"}, # 失效(不在pool) + ] + r = svc.reassign_invalid_accounts() + assert r["reassigned"] == 2, r + assert "tok1" not in am.account_service.updates, "有效账号不该被重分配" + assert "tok2" in am.account_service.updates + assert "tok3" in am.account_service.updates diff --git a/test/test_remote_account_api.py b/test/test_remote_account_api.py index 18c89e5..66f4397 100644 --- a/test/test_remote_account_api.py +++ b/test/test_remote_account_api.py @@ -51,6 +51,12 @@ def load_settings(self) -> dict[str, Any]: def save_settings(self, settings: dict[str, Any]) -> None: pass + def load_proxy_pool(self) -> list[dict[str, Any]]: + return [] + + def save_proxy_pool(self, items: list[dict[str, Any]]) -> None: + pass + def health_check(self) -> dict[str, Any]: return {"ok": True} diff --git a/test/test_turnstile_retry.py b/test/test_turnstile_retry.py index 12bf818..d8ebf8a 100644 --- a/test/test_turnstile_retry.py +++ b/test/test_turnstile_retry.py @@ -23,7 +23,7 @@ def test_retryable_turnstile_error_rotates_to_next_account_without_removing_old_ created_tokens: list[str] = [] class FakeBackend: - def __init__(self, access_token: str = "") -> None: + def __init__(self, access_token: str = "", **kwargs) -> None: self.access_token = access_token created_tokens.append(access_token) @@ -53,6 +53,7 @@ def get_text_access_token(attempted_tokens: set[str]) -> str: mock.patch.object(gpt_runtime, "OpenAIBackendAPI", FakeBackend), mock.patch.object(gpt_runtime, "conversation_events", fake_conversation_events), mock.patch.object(gpt_runtime, "account_service", account_service), + mock.patch.object(gpt_runtime, "_account_proxy", return_value=""), ): deltas = list(conversation.stream_text_deltas(initial_backend, request)) diff --git a/uv.lock b/uv.lock index 29ac550..310db49 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = ">=3.13" [[package]] @@ -143,45 +143,6 @@ wheels = [ { url = "https://mirrors.aliyun.com/pypi/packages/db/8f/61959034484a4a7c527811f4721e75d02d653a35afb0b6054474d8185d4c/charset_normalizer-3.4.7-py3-none-any.whl", hash = "sha256:3dce51d0f5e7951f8bb4900c257dad282f49190fdbebecd4ba99bcc41fef404d" }, ] -[[package]] -name = "webchat2api" -version = "0.0.7" -source = { virtual = "." } -dependencies = [ - { name = "curl-cffi" }, - { name = "fastapi" }, - { name = "gitpython" }, - { name = "pillow" }, - { name = "psycopg2-binary" }, - { name = "pybase64" }, - { name = "python-multipart" }, - { name = "sqlalchemy" }, - { name = "tiktoken" }, - { name = "uvicorn" }, -] - -[package.dev-dependencies] -dev = [ - { name = "httpx" }, -] - -[package.metadata] -requires-dist = [ - { name = "curl-cffi", specifier = ">=0.15.0" }, - { name = "fastapi", specifier = ">=0.136.0" }, - { name = "gitpython", specifier = ">=3.1.0" }, - { name = "pillow", specifier = ">=12.2.0" }, - { name = "psycopg2-binary", specifier = ">=2.9.0" }, - { name = "pybase64", specifier = ">=1.4.3" }, - { name = "python-multipart", specifier = ">=0.0.26" }, - { name = "sqlalchemy", specifier = ">=2.0.0" }, - { name = "tiktoken", specifier = ">=0.12.0" }, - { name = "uvicorn", specifier = ">=0.44.0" }, -] - -[package.metadata.requires-dev] -dev = [{ name = "httpx", specifier = ">=0.28.1" }] - [[package]] name = "click" version = "8.3.2" @@ -285,9 +246,7 @@ wheels = [ { url = "https://mirrors.aliyun.com/pypi/packages/7a/75/7e9cd1126a1e1f0cd67b0eda02e5221b28488d352684704a78ed505bd719/greenlet-3.4.0-cp313-cp313-macosx_11_0_universal2.whl", hash = "sha256:43748988b097f9c6f09364f260741aa73c80747f63389824435c7a50bfdfd5c1" }, { url = "https://mirrors.aliyun.com/pypi/packages/9d/c4/3e2df392e5cb199527c4d9dbcaa75c14edcc394b45040f0189f649631e3c/greenlet-3.4.0-cp313-cp313-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:5566e4e2cd7a880e8c27618e3eab20f3494452d12fd5129edef7b2f7aa9a36d1" }, { url = "https://mirrors.aliyun.com/pypi/packages/da/af/750cdfda1d1bd30a6c28080245be8d0346e669a98fdbae7f4102aa95fff3/greenlet-3.4.0-cp313-cp313-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:1054c5a3c78e2ab599d452f23f7adafef55062a783a8e241d24f3b633ba6ff82" }, - { url = "https://mirrors.aliyun.com/pypi/packages/e0/93/c8c508d68ba93232784bbc1b5474d92371f2897dfc6bc281b419f2e0d492/greenlet-3.4.0-cp313-cp313-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:98eedd1803353daf1cd9ef23eef23eda5a4d22f99b1f998d273a8b78b70dd47f" }, { url = "https://mirrors.aliyun.com/pypi/packages/54/78/0cbc693622cd54ebe25207efbb3a0eb07c2639cb8594f6e3aaaa0bb077a8/greenlet-3.4.0-cp313-cp313-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:f82cb6cddc27dd81c96b1506f4aa7def15070c3b2a67d4e46fd19016aacce6cf" }, - { url = "https://mirrors.aliyun.com/pypi/packages/7f/46/cfaaa0ade435a60550fd83d07dfd5c41f873a01da17ede5c4cade0b9bab8/greenlet-3.4.0-cp313-cp313-manylinux_2_39_riscv64.whl", hash = "sha256:b7857e2202aae67bc5725e0c1f6403c20a8ff46094ece015e7d474f5f7020b55" }, { url = "https://mirrors.aliyun.com/pypi/packages/ba/c0/8966767de01343c1ff47e8b855dc78e7d1a8ed2b7b9c83576a57e289f81d/greenlet-3.4.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:227a46251ecba4ff46ae742bc5ce95c91d5aceb4b02f885487aff269c127a729" }, { url = "https://mirrors.aliyun.com/pypi/packages/b8/38/bcdc71ba05e9a5fda87f63ffc2abcd1f15693b659346df994a48c968003d/greenlet-3.4.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:5b99e87be7eba788dd5b75ba1cde5639edffdec5f91fe0d734a249535ec3408c" }, { url = "https://mirrors.aliyun.com/pypi/packages/a1/c2/19b664b7173b9e4ef5f77e8cef9f14c20ec7fce7920dc1ccd7afd955d093/greenlet-3.4.0-cp313-cp313-win_amd64.whl", hash = "sha256:849f8bc17acd6295fcb5de8e46d55cc0e52381c56eaf50a2afd258e97bc65940" }, @@ -295,9 +254,7 @@ wheels = [ { url = "https://mirrors.aliyun.com/pypi/packages/78/02/bde66806e8f169cf90b14d02c500c44cdbe02c8e224c9c67bafd1b8cadd1/greenlet-3.4.0-cp314-cp314-macosx_11_0_universal2.whl", hash = "sha256:10a07aca6babdd18c16a3f4f8880acfffc2b88dfe431ad6aa5f5740759d7d75e" }, { url = "https://mirrors.aliyun.com/pypi/packages/05/1f/39da1c336a87d47c58352fb8a78541ce63d63ae57c5b9dae1fe02801bbc2/greenlet-3.4.0-cp314-cp314-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:076e21040b3a917d3ce4ad68fb5c3c6b32f1405616c4a57aa83120979649bd3d" }, { url = "https://mirrors.aliyun.com/pypi/packages/d3/6c/90ee29a4ee27af7aa2e2ec408799eeb69ee3fcc5abcecac6ddd07a5cd0f2/greenlet-3.4.0-cp314-cp314-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:e82689eea4a237e530bb5cb41b180ef81fa2160e1f89422a67be7d90da67f615" }, - { url = "https://mirrors.aliyun.com/pypi/packages/d2/4a/74078d3936712cff6d3c91a930016f476ce4198d84e224fe6d81d3e02880/greenlet-3.4.0-cp314-cp314-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:06c2d3b89e0c62ba50bd7adf491b14f39da9e7e701647cb7b9ff4c99bee04b19" }, { url = "https://mirrors.aliyun.com/pypi/packages/07/49/d4cad6e5381a50947bb973d2f6cf6592621451b09368b8c20d9b8af49c5b/greenlet-3.4.0-cp314-cp314-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:4df3b0b2289ec686d3c821a5fee44259c05cfe824dd5e6e12c8e5f5df23085cf" }, - { url = "https://mirrors.aliyun.com/pypi/packages/79/3e/df8a83ab894751bc31e1106fdfaa80ca9753222f106b04de93faaa55feb7/greenlet-3.4.0-cp314-cp314-manylinux_2_39_riscv64.whl", hash = "sha256:070b8bac2ff3b4d9e0ff36a0d19e42103331d9737e8504747cd1e659f76297bd" }, { url = "https://mirrors.aliyun.com/pypi/packages/37/31/d1edd54f424761b5d47718822f506b435b6aab2f3f93b465441143ea5119/greenlet-3.4.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:8bff29d586ea415688f4cec96a591fcc3bf762d046a796cdadc1fdb6e7f2d5bf" }, { url = "https://mirrors.aliyun.com/pypi/packages/b0/c6/6d3f9cdcb21c4e12a79cb332579f1c6aa1af78eb68059c5a957c7812d95e/greenlet-3.4.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:8a569c2fb840c53c13a2b8967c63621fafbd1a0e015b9c82f408c33d626a2fda" }, { url = "https://mirrors.aliyun.com/pypi/packages/63/45/c1ca4a1ad975de4727e52d3ffe641ae23e1d7a8ffaa8ff7a0477e1827b92/greenlet-3.4.0-cp314-cp314-win_amd64.whl", hash = "sha256:207ba5b97ea8b0b60eb43ffcacf26969dd83726095161d676aac03ff913ee50d" }, @@ -305,9 +262,7 @@ wheels = [ { url = "https://mirrors.aliyun.com/pypi/packages/d4/8f/18d72b629783f5e8d045a76f5325c1e938e659a9e4da79c7dcd10169a48d/greenlet-3.4.0-cp314-cp314t-macosx_11_0_universal2.whl", hash = "sha256:d70012e51df2dbbccfaf63a40aaf9b40c8bed37c3e3a38751c926301ce538ece" }, { url = "https://mirrors.aliyun.com/pypi/packages/9e/ad/5fa86ec46769c4153820d58a04062285b3b9e10ba3d461ee257b68dcbf53/greenlet-3.4.0-cp314-cp314t-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:a58bec0751f43068cd40cff31bb3ca02ad6000b3a51ca81367af4eb5abc480c8" }, { url = "https://mirrors.aliyun.com/pypi/packages/43/f0/4e8174ca0e87ae748c409f055a1ba161038c43cc0a5a6f1433a26ac2e5bf/greenlet-3.4.0-cp314-cp314t-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:05fa0803561028f4b2e3b490ee41216a842eaee11aed004cc343a996d9523aa2" }, - { url = "https://mirrors.aliyun.com/pypi/packages/ef/92/466b0d9afd44b8af623139a3599d651c7564fa4152f25f117e1ee5949ffb/greenlet-3.4.0-cp314-cp314t-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:c4cd56a9eb7a6444edbc19062f7b6fbc8f287c663b946e3171d899693b1c19fa" }, { url = "https://mirrors.aliyun.com/pypi/packages/19/da/991cf7cd33662e2df92a1274b7eb4d61769294d38a1bba8a45f31364845e/greenlet-3.4.0-cp314-cp314t-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:e60d38719cb80b3ab5e85f9f1aed4960acfde09868af6762ccb27b260d68f4ed" }, - { url = "https://mirrors.aliyun.com/pypi/packages/0d/14/3395a7ef3e260de0325152ddfe19dffb3e49fe10873b94654352b53ad48e/greenlet-3.4.0-cp314-cp314t-manylinux_2_39_riscv64.whl", hash = "sha256:1f85f204c4d54134ae850d401fa435c89cd667d5ce9dc567571776b45941af72" }, { url = "https://mirrors.aliyun.com/pypi/packages/36/c5/6c2c708e14db3d9caea4b459d8464f58c32047451142fe2cfd90e7458f41/greenlet-3.4.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:7f50c804733b43eded05ae694691c9aa68bca7d0a867d67d4a3f514742a2d53f" }, { url = "https://mirrors.aliyun.com/pypi/packages/7a/4c/50c5fed19378e11a29fabab1f6be39ea95358f4a0a07e115a51ca93385d8/greenlet-3.4.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:2d4f0635dc4aa638cda4b2f5a07ae9a2cff9280327b581a3fcb6f317b4fbc38a" }, { url = "https://mirrors.aliyun.com/pypi/packages/db/72/85ae954d734703ab48e622c59d4ce35d77ce840c265814af9c078cacc7aa/greenlet-3.4.0-cp314-cp314t-win_amd64.whl", hash = "sha256:1a4a48f24681300c640f143ba7c404270e1ebbbcf34331d7104a4ff40f8ea705" }, @@ -660,6 +615,42 @@ wheels = [ { url = "https://mirrors.aliyun.com/pypi/packages/9a/22/f1925cdda983ab66fc8ec6ec8014b959262747e58bdca26a4e3d1da29d56/python_multipart-0.0.26-py3-none-any.whl", hash = "sha256:c0b169f8c4484c13b0dcf2ef0ec3a4adb255c4b7d18d8e420477d2b1dd03f185" }, ] +[[package]] +name = "pyyaml" +version = "6.0.3" +source = { registry = "https://mirrors.aliyun.com/pypi/simple" } +sdist = { url = "https://mirrors.aliyun.com/pypi/packages/05/8e/961c0007c59b8dd7729d542c61a4d537767a59645b82a0b521206e1e25c2/pyyaml-6.0.3.tar.gz", hash = "sha256:d76623373421df22fb4cf8817020cbb7ef15c725b9d5e45f17e189bfc384190f" } +wheels = [ + { url = "https://mirrors.aliyun.com/pypi/packages/d1/11/0fd08f8192109f7169db964b5707a2f1e8b745d4e239b784a5a1dd80d1db/pyyaml-6.0.3-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:8da9669d359f02c0b91ccc01cac4a67f16afec0dac22c2ad09f46bee0697eba8" }, + { url = "https://mirrors.aliyun.com/pypi/packages/b1/16/95309993f1d3748cd644e02e38b75d50cbc0d9561d21f390a76242ce073f/pyyaml-6.0.3-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:2283a07e2c21a2aa78d9c4442724ec1eb15f5e42a723b99cb3d822d48f5f7ad1" }, + { url = "https://mirrors.aliyun.com/pypi/packages/50/31/b20f376d3f810b9b2371e72ef5adb33879b25edb7a6d072cb7ca0c486398/pyyaml-6.0.3-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ee2922902c45ae8ccada2c5b501ab86c36525b883eff4255313a253a3160861c" }, + { url = "https://mirrors.aliyun.com/pypi/packages/49/1e/a55ca81e949270d5d4432fbbd19dfea5321eda7c41a849d443dc92fd1ff7/pyyaml-6.0.3-cp313-cp313-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:a33284e20b78bd4a18c8c2282d549d10bc8408a2a7ff57653c0cf0b9be0afce5" }, + { url = "https://mirrors.aliyun.com/pypi/packages/74/27/e5b8f34d02d9995b80abcef563ea1f8b56d20134d8f4e5e81733b1feceb2/pyyaml-6.0.3-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0f29edc409a6392443abf94b9cf89ce99889a1dd5376d94316ae5145dfedd5d6" }, + { url = "https://mirrors.aliyun.com/pypi/packages/f9/11/ba845c23988798f40e52ba45f34849aa8a1f2d4af4b798588010792ebad6/pyyaml-6.0.3-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:f7057c9a337546edc7973c0d3ba84ddcdf0daa14533c2065749c9075001090e6" }, + { url = "https://mirrors.aliyun.com/pypi/packages/3d/e0/7966e1a7bfc0a45bf0a7fb6b98ea03fc9b8d84fa7f2229e9659680b69ee3/pyyaml-6.0.3-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:eda16858a3cab07b80edaf74336ece1f986ba330fdb8ee0d6c0d68fe82bc96be" }, + { url = "https://mirrors.aliyun.com/pypi/packages/de/94/980b50a6531b3019e45ddeada0626d45fa85cbe22300844a7983285bed3b/pyyaml-6.0.3-cp313-cp313-win32.whl", hash = "sha256:d0eae10f8159e8fdad514efdc92d74fd8d682c933a6dd088030f3834bc8e6b26" }, + { url = "https://mirrors.aliyun.com/pypi/packages/97/c9/39d5b874e8b28845e4ec2202b5da735d0199dbe5b8fb85f91398814a9a46/pyyaml-6.0.3-cp313-cp313-win_amd64.whl", hash = "sha256:79005a0d97d5ddabfeeea4cf676af11e647e41d81c9a7722a193022accdb6b7c" }, + { url = "https://mirrors.aliyun.com/pypi/packages/73/e8/2bdf3ca2090f68bb3d75b44da7bbc71843b19c9f2b9cb9b0f4ab7a5a4329/pyyaml-6.0.3-cp313-cp313-win_arm64.whl", hash = "sha256:5498cd1645aa724a7c71c8f378eb29ebe23da2fc0d7a08071d89469bf1d2defb" }, + { url = "https://mirrors.aliyun.com/pypi/packages/9d/8c/f4bd7f6465179953d3ac9bc44ac1a8a3e6122cf8ada906b4f96c60172d43/pyyaml-6.0.3-cp314-cp314-macosx_10_13_x86_64.whl", hash = "sha256:8d1fab6bb153a416f9aeb4b8763bc0f22a5586065f86f7664fc23339fc1c1fac" }, + { url = "https://mirrors.aliyun.com/pypi/packages/bd/9c/4d95bb87eb2063d20db7b60faa3840c1b18025517ae857371c4dd55a6b3a/pyyaml-6.0.3-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:34d5fcd24b8445fadc33f9cf348c1047101756fd760b4dacb5c3e99755703310" }, + { url = "https://mirrors.aliyun.com/pypi/packages/92/b5/47e807c2623074914e29dabd16cbbdd4bf5e9b2db9f8090fa64411fc5382/pyyaml-6.0.3-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:501a031947e3a9025ed4405a168e6ef5ae3126c59f90ce0cd6f2bfc477be31b7" }, + { url = "https://mirrors.aliyun.com/pypi/packages/02/9e/e5e9b168be58564121efb3de6859c452fccde0ab093d8438905899a3a483/pyyaml-6.0.3-cp314-cp314-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:b3bc83488de33889877a0f2543ade9f70c67d66d9ebb4ac959502e12de895788" }, + { url = "https://mirrors.aliyun.com/pypi/packages/88/f9/16491d7ed2a919954993e48aa941b200f38040928474c9e85ea9e64222c3/pyyaml-6.0.3-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:c458b6d084f9b935061bc36216e8a69a7e293a2f1e68bf956dcd9e6cbcd143f5" }, + { url = "https://mirrors.aliyun.com/pypi/packages/dd/3f/5989debef34dc6397317802b527dbbafb2b4760878a53d4166579111411e/pyyaml-6.0.3-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:7c6610def4f163542a622a73fb39f534f8c101d690126992300bf3207eab9764" }, + { url = "https://mirrors.aliyun.com/pypi/packages/d7/ce/af88a49043cd2e265be63d083fc75b27b6ed062f5f9fd6cdc223ad62f03e/pyyaml-6.0.3-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:5190d403f121660ce8d1d2c1bb2ef1bd05b5f68533fc5c2ea899bd15f4399b35" }, + { url = "https://mirrors.aliyun.com/pypi/packages/23/20/bb6982b26a40bb43951265ba29d4c246ef0ff59c9fdcdf0ed04e0687de4d/pyyaml-6.0.3-cp314-cp314-win_amd64.whl", hash = "sha256:4a2e8cebe2ff6ab7d1050ecd59c25d4c8bd7e6f400f5f82b96557ac0abafd0ac" }, + { url = "https://mirrors.aliyun.com/pypi/packages/f4/f4/a4541072bb9422c8a883ab55255f918fa378ecf083f5b85e87fc2b4eda1b/pyyaml-6.0.3-cp314-cp314-win_arm64.whl", hash = "sha256:93dda82c9c22deb0a405ea4dc5f2d0cda384168e466364dec6255b293923b2f3" }, + { url = "https://mirrors.aliyun.com/pypi/packages/7c/f9/07dd09ae774e4616edf6cda684ee78f97777bdd15847253637a6f052a62f/pyyaml-6.0.3-cp314-cp314t-macosx_10_13_x86_64.whl", hash = "sha256:02893d100e99e03eda1c8fd5c441d8c60103fd175728e23e431db1b589cf5ab3" }, + { url = "https://mirrors.aliyun.com/pypi/packages/4e/78/8d08c9fb7ce09ad8c38ad533c1191cf27f7ae1effe5bb9400a46d9437fcf/pyyaml-6.0.3-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:c1ff362665ae507275af2853520967820d9124984e0f7466736aea23d8611fba" }, + { url = "https://mirrors.aliyun.com/pypi/packages/7b/5b/3babb19104a46945cf816d047db2788bcaf8c94527a805610b0289a01c6b/pyyaml-6.0.3-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:6adc77889b628398debc7b65c073bcb99c4a0237b248cacaf3fe8a557563ef6c" }, + { url = "https://mirrors.aliyun.com/pypi/packages/8b/cc/dff0684d8dc44da4d22a13f35f073d558c268780ce3c6ba1b87055bb0b87/pyyaml-6.0.3-cp314-cp314t-manylinux2014_s390x.manylinux_2_17_s390x.manylinux_2_28_s390x.whl", hash = "sha256:a80cb027f6b349846a3bf6d73b5e95e782175e52f22108cfa17876aaeff93702" }, + { url = "https://mirrors.aliyun.com/pypi/packages/b1/5e/f77dc6b9036943e285ba76b49e118d9ea929885becb0a29ba8a7c75e29fe/pyyaml-6.0.3-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:00c4bdeba853cc34e7dd471f16b4114f4162dc03e6b7afcc2128711f0eca823c" }, + { url = "https://mirrors.aliyun.com/pypi/packages/ce/88/a9db1376aa2a228197c58b37302f284b5617f56a5d959fd1763fb1675ce6/pyyaml-6.0.3-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:66e1674c3ef6f541c35191caae2d429b967b99e02040f5ba928632d9a7f0f065" }, + { url = "https://mirrors.aliyun.com/pypi/packages/da/92/1446574745d74df0c92e6aa4a7b0b3130706a4142b2d1a5869f2eaa423c6/pyyaml-6.0.3-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:16249ee61e95f858e83976573de0f5b2893b3677ba71c9dd36b9cf8be9ac6d65" }, + { url = "https://mirrors.aliyun.com/pypi/packages/f0/7a/1c7270340330e575b92f397352af856a8c06f230aa3e76f86b39d01b416a/pyyaml-6.0.3-cp314-cp314t-win_amd64.whl", hash = "sha256:4ad1906908f2f5ae4e5a8ddfce73c320c2a1429ec52eafd27138b7f1cbe341c9" }, + { url = "https://mirrors.aliyun.com/pypi/packages/f1/12/de94a39c2ef588c7e6455cfbe7343d3b2dc9d6b6b2f40c4c6565744c873d/pyyaml-6.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:ebc55a14a21cb14062aa4162f906cd962b28e2e9ea38f9b4391244cd8de4ae0b" }, +] + [[package]] name = "regex" version = "2026.4.4" @@ -902,3 +893,44 @@ sdist = { url = "https://mirrors.aliyun.com/pypi/packages/5e/da/6eee1ff8b6cbeed4 wheels = [ { url = "https://mirrors.aliyun.com/pypi/packages/b7/23/a5bbd9600dd607411fa644c06ff4951bec3a4d82c4b852374024359c19c0/uvicorn-0.44.0-py3-none-any.whl", hash = "sha256:ce937c99a2cc70279556967274414c087888e8cec9f9c94644dfca11bd3ced89" }, ] + +[[package]] +name = "webchat2api" +version = "0.0.11" +source = { virtual = "." } +dependencies = [ + { name = "curl-cffi" }, + { name = "fastapi" }, + { name = "gitpython" }, + { name = "pillow" }, + { name = "psycopg2-binary" }, + { name = "pybase64" }, + { name = "python-multipart" }, + { name = "pyyaml" }, + { name = "sqlalchemy" }, + { name = "tiktoken" }, + { name = "uvicorn" }, +] + +[package.dev-dependencies] +dev = [ + { name = "httpx" }, +] + +[package.metadata] +requires-dist = [ + { name = "curl-cffi", specifier = ">=0.15.0" }, + { name = "fastapi", specifier = ">=0.136.0" }, + { name = "gitpython", specifier = ">=3.1.0" }, + { name = "pillow", specifier = ">=12.2.0" }, + { name = "psycopg2-binary", specifier = ">=2.9.0" }, + { name = "pybase64", specifier = ">=1.4.3" }, + { name = "python-multipart", specifier = ">=0.0.26" }, + { name = "pyyaml", specifier = ">=6.0.3" }, + { name = "sqlalchemy", specifier = ">=2.0.0" }, + { name = "tiktoken", specifier = ">=0.12.0" }, + { name = "uvicorn", specifier = ">=0.44.0" }, +] + +[package.metadata.requires-dev] +dev = [{ name = "httpx", specifier = ">=0.28.1" }] diff --git a/web/src/app/settings/components/proxy-pool-card.tsx b/web/src/app/settings/components/proxy-pool-card.tsx new file mode 100644 index 0000000..ef36c8d --- /dev/null +++ b/web/src/app/settings/components/proxy-pool-card.tsx @@ -0,0 +1,407 @@ +"use client"; + +import { useEffect, useRef, useState } from "react"; +import { + LoaderCircle, + Network, + Plus, + RefreshCw, + Shuffle, + Trash2, + XCircle, +} from "lucide-react"; +import { toast } from "sonner"; + +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { Card, CardContent } from "@/components/ui/card"; +import { Textarea } from "@/components/ui/textarea"; +import { + addSubscription, + assignProxyPool, + clearProxyPoolAssignments, + deleteProxyPool, + deleteSubscription, + fetchProxyPool, + fetchSubscriptions, + getSyncStatus, + importProxyPool, + syncProxyPool, + type ProxyPoolItem, + type SubscriptionSource, +} from "@/lib/api"; + +export function ProxyPoolCard() { + const didLoadRef = useRef(false); + const [items, setItems] = useState([]); + const [subs, setSubs] = useState([]); + const [isLoading, setIsLoading] = useState(true); + const [isImporting, setIsImporting] = useState(false); + const [isAssigning, setIsAssigning] = useState(false); + const [isClearing, setIsClearing] = useState(false); + const [showImport, setShowImport] = useState(false); + const [importText, setImportText] = useState(""); + + // 订阅源 + const [showAddSub, setShowAddSub] = useState(false); + const [subForm, setSubForm] = useState({ name: "", url: "", region_keywords: "🇺🇸,美国,美國,United States,America" }); + const [isAddingSub, setIsAddingSub] = useState(false); + const [syncTaskId, setSyncTaskId] = useState(null); + const [syncStatus, setSyncStatus] = useState(""); + + const load = async () => { + setIsLoading(true); + try { + const data = await fetchProxyPool(); + setItems(data.items); + } catch (error) { + toast.error(error instanceof Error ? error.message : "加载代理池失败"); + } finally { + setIsLoading(false); + } + }; + + const loadSubs = async () => { + try { + const data = await fetchSubscriptions(); + setSubs(data.items); + } catch (error) { + toast.error(error instanceof Error ? error.message : "加载订阅源失败"); + } + }; + + useEffect(() => { + if (didLoadRef.current) return; + didLoadRef.current = true; + void load(); + void loadSubs(); + }, []); + + // 同步状态轮询 + useEffect(() => { + if (!syncTaskId) return; + const timer = window.setInterval(async () => { + try { + const s = await getSyncStatus(syncTaskId); + setSyncStatus(s.status); + if (s.status === "done" || s.status === "error" || s.status === "skipped") { + window.clearInterval(timer); + setSyncTaskId(null); + if (s.status === "done") { + const r = s.result; + toast.success(`同步完成:可用 ${r?.usable ?? 0}/${r?.total ?? 0},重分配 ${r?.reassigned ?? 0} 账号`); + void load(); + void loadSubs(); + } else { + toast.error(`同步${s.status === "skipped" ? "跳过(另一任务在跑)" : "失败"}:${s.error ?? ""}`); + } + } + } catch (error) { + window.clearInterval(timer); + setSyncTaskId(null); + toast.error(error instanceof Error ? error.message : "查询同步状态失败"); + } + }, 2000); + return () => window.clearInterval(timer); + }, [syncTaskId]); + + const handleImport = async () => { + const text = importText.trim(); + if (!text) { + toast.error("请输入代理列表"); + return; + } + setIsImporting(true); + try { + const data = await importProxyPool(text); + setItems(data.items); + setImportText(""); + setShowImport(false); + toast.success(`导入完成:新增 ${data.added ?? 0} 个,跳过 ${data.skipped ?? 0} 个`); + } catch (error) { + toast.error(error instanceof Error ? error.message : "导入失败"); + } finally { + setIsImporting(false); + } + }; + + const handleDelete = async (id: string) => { + try { + const data = await deleteProxyPool([id]); + setItems(data.items); + toast.success("已删除"); + } catch (error) { + toast.error(error instanceof Error ? error.message : "删除失败"); + } + }; + + const handleClearAll = async () => { + try { + const data = await deleteProxyPool([]); + setItems(data.items); + toast.success(`已清空代理池(删除 ${data.removed ?? 0} 个)`); + } catch (error) { + toast.error(error instanceof Error ? error.message : "清空失败"); + } + }; + + const handleAssign = async () => { + setIsAssigning(true); + try { + const data = await assignProxyPool(); + if (data.error) { + toast.error(data.error); + } else { + toast.success(`已分配 ${data.assigned} 个账号(${data.total_proxies} 个代理轮转分配给 ${data.total_accounts} 个账号)`); + } + } catch (error) { + toast.error(error instanceof Error ? error.message : "分配失败"); + } finally { + setIsAssigning(false); + } + }; + + const handleClearAssignments = async () => { + setIsClearing(true); + try { + const data = await clearProxyPoolAssignments(); + toast.success(`已清除 ${data.cleared} 个账号的代理分配`); + } catch (error) { + toast.error(error instanceof Error ? error.message : "清除失败"); + } finally { + setIsClearing(false); + } + }; + + const handleAddSub = async () => { + if (!subForm.url.trim()) { + toast.error("请输入订阅 URL"); + return; + } + setIsAddingSub(true); + try { + const data = await addSubscription(subForm.name, subForm.url, subForm.region_keywords); + setSubs(data.items); + setSubForm({ ...subForm, name: "", url: "" }); + setShowAddSub(false); + toast.success("订阅源已添加"); + } catch (error) { + toast.error(error instanceof Error ? error.message : "添加失败"); + } finally { + setIsAddingSub(false); + } + }; + + const handleDeleteSub = async (id: string) => { + try { + const data = await deleteSubscription(id); + setSubs(data.items); + void load(); + toast.success("订阅源已删除(含其代理项)"); + } catch (error) { + toast.error(error instanceof Error ? error.message : "删除失败"); + } + }; + + const handleSync = async () => { + try { + const r = await syncProxyPool(); + setSyncTaskId(r.task_id); + setSyncStatus("pending"); + toast.info("同步已启动(拉订阅 + 测活 + 更新池,约 1-2 分钟)"); + } catch (error) { + toast.error(error instanceof Error ? error.message : "启动同步失败"); + } + }; + + const syncing = !!syncTaskId; + + return ( + + +
+
+
+ +
+
+

代理池

+

+ 批量导入代理或订阅节点,自动分配给账号,每个账号使用独立出口。 +

+
+
+ 0 ? "success" : "secondary"} className="w-fit rounded-md px-2.5 py-1"> + {items.length > 0 ? `${items.length} 个代理` : "未配置"} + +
+ + {isLoading ? ( +
+ +
+ ) : ( + <> + {/* Action buttons */} +
+ + + + + {items.length > 0 && ( + + )} +
+ + {/* Import area */} + {showImport && ( +
+ +