From 290f5f0dbe82d82a10fcda80c6150a5830241185 Mon Sep 17 00:00:00 2001 From: ngfchl Date: Tue, 7 Jul 2026 14:24:52 +0800 Subject: [PATCH] fix: use telegram proxy pool in web api --- config.py | 1 + web/api.py | 66 +++++++++++++++++++++++++++++++++++++----------------- 2 files changed, 47 insertions(+), 20 deletions(-) diff --git a/config.py b/config.py index 975079c..0531a3e 100644 --- a/config.py +++ b/config.py @@ -18,6 +18,7 @@ OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen3:8b") # Proxy PROXY_URL = os.getenv("PROXY_URL", "http://192.168.123.10:11055") TG_PROXY_URL = os.getenv("TG_PROXY_URL", PROXY_URL) +TG_PROXY_URLS = [x.strip() for x in os.getenv("TG_PROXY_URLS", TG_PROXY_URL).split(",") if x.strip()] # Web WEB_PORT = int(os.getenv("WEB_PORT", "8766")) diff --git a/web/api.py b/web/api.py index 8144015..cc14b9f 100644 --- a/web/api.py +++ b/web/api.py @@ -245,6 +245,47 @@ async def api_stats(day: str = Query(default=None, pattern=r"^\d{4}-\d{2}-\d{2}$ return JSONResponse(await get_daily_stats(day, chat_id=chat_id)) + +def tg_proxy_urls(config) -> list[str | None]: + urls: list[str | None] = [] + for value in getattr(config, "TG_PROXY_URLS", []) or []: + if value and value not in urls: + urls.append(value) + for value in [getattr(config, "TG_PROXY_URL", None), getattr(config, "PROXY_URL", None)]: + if value and value not in urls: + urls.append(value) + return urls or [None] + + +async def run_tg_request(operation, *, connect_timeout: float = 20.0, read_timeout: float = 20.0, label: str = "tg"): + import config + from telegram import Bot + from telegram.error import BadRequest, Forbidden, NetworkError, TimedOut + from telegram.request import HTTPXRequest + + last_error: Exception | None = None + proxies = tg_proxy_urls(config) + for idx, proxy in enumerate(proxies, 1): + try: + bot = Bot( + token=config.TG_BOT_TOKEN, + request=HTTPXRequest(proxy=proxy, connect_timeout=connect_timeout, read_timeout=read_timeout), + ) + return await operation(bot) + except (BadRequest, Forbidden): + raise + except (NetworkError, TimedOut, Exception) as e: + last_error = e + if idx < len(proxies): + logging.getLogger("spam_guard").warning( + f"🌐 Web TG 请求失败，切换代理重试: {label}: {type(e).__name__}: {e}; next={proxies[idx]}" + ) + continue + raise + if last_error: + raise last_error + raise RuntimeError("TG 请求失败") + def parse_draw_at(value): if not value: return None @@ -307,9 +348,6 @@ async def api_actions(limit: int = Query(default=100, ge=1, le=500)): @app.get("/api/groups") async def api_groups(): import config - from telegram import Bot - from telegram.request import HTTPXRequest - sources = [] for model in (Message, Ban, Action, GameScore, LotteryEvent, ShopItem): try: @@ -320,12 +358,11 @@ async def api_groups(): if config.TG_CHAT_ID and config.TG_CHAT_ID not in ids: ids.insert(0, config.TG_CHAT_ID) - bot = Bot(token=config.TG_BOT_TOKEN, request=HTTPXRequest(proxy=getattr(config, "TG_PROXY_URL", config.PROXY_URL), connect_timeout=10.0, read_timeout=10.0)) result = [] for chat_id in ids: name = str(chat_id) try: - chat = await bot.get_chat(chat_id=chat_id) + chat = await run_tg_request(lambda bot, cid=chat_id: bot.get_chat(chat_id=cid), connect_timeout=10.0, read_timeout=10.0, label=f"get_chat:{chat_id}") name = chat.title or chat.full_name or chat.username or str(chat_id) except Exception: pass @@ -486,9 +523,6 @@ async def api_restart(): @app.post("/api/send") async def api_send(request: Request): import config - from telegram import Bot - from telegram.request import HTTPXRequest - data = await request.json() text = str(data.get("text", "")).strip() if not text: @@ -497,8 +531,7 @@ async def api_send(request: Request): return JSONResponse({"ok": False, "error": "消息太长"}, status_code=400) chat_id = int(data.get("chat_id") or config.TG_CHAT_ID) - bot = Bot(token=config.TG_BOT_TOKEN, request=HTTPXRequest(proxy=getattr(config, "TG_PROXY_URL", config.PROXY_URL))) - msg = await bot.send_message(chat_id=chat_id, text=text) + msg = await run_tg_request(lambda bot: bot.send_message(chat_id=chat_id, text=text), label="send_message") rec = await Message.create( msg_id=msg.message_id, chat_id=chat_id, @@ -527,9 +560,7 @@ async def api_send(request: Request): async def api_delete_message(message_id: int, request: Request): """手动删除聊天记录：先删 Telegram 消息，再把数据库记录标记为已删除。""" import config - from telegram import Bot from telegram.error import BadRequest - from telegram.request import HTTPXRequest data = await request.json() if request.headers.get("content-type", "").startswith("application/json") else {} ban_user = bool(data.get("ban_user") or data.get("ban")) @@ -542,11 +573,10 @@ async def api_delete_message(message_id: int, request: Request): if not rec.chat_id or not rec.msg_id: return JSONResponse({"ok": False, "error": "缺少 TG chat_id/msg_id，无法删除 TG 消息"}, status_code=400) - bot = Bot(token=config.TG_BOT_TOKEN, request=HTTPXRequest(proxy=getattr(config, "TG_PROXY_URL", config.PROXY_URL), connect_timeout=20.0, read_timeout=20.0)) tg_deleted = True tg_error = None try: - await bot.delete_message(chat_id=rec.chat_id, message_id=rec.msg_id) + await run_tg_request(lambda bot: bot.delete_message(chat_id=rec.chat_id, message_id=rec.msg_id), connect_timeout=20.0, read_timeout=20.0, label="delete_message") except BadRequest as e: # 消息已不存在时，允许继续标记数据库；其他 TG 删除失败不标记。 if "message to delete not found" in str(e).lower(): @@ -566,7 +596,7 @@ async def api_delete_message(message_id: int, request: Request): try: from datetime import timedelta unban_at = now_tz() + timedelta(minutes=config.BAN_DURATION_MINUTES) - await bot.ban_chat_member(chat_id=rec.chat_id, user_id=rec.user_id, until_date=int(unban_at.timestamp())) + await run_tg_request(lambda bot: bot.ban_chat_member(chat_id=rec.chat_id, user_id=rec.user_id, until_date=int(unban_at.timestamp())), connect_timeout=20.0, read_timeout=20.0, label="ban_chat_member") ban = await Ban.create( user_id=rec.user_id, username=rec.username, @@ -618,11 +648,7 @@ async def api_scores(limit: int = Query(default=50, ge=1, le=200)): async def api_unban(user_id: int): import config try: - from telegram import Bot - from telegram.request import HTTPXRequest - - bot = Bot(token=config.TG_BOT_TOKEN, request=HTTPXRequest(proxy=getattr(config, "TG_PROXY_URL", config.PROXY_URL))) - await bot.unban_chat_member(chat_id=config.TG_CHAT_ID, user_id=user_id) + await run_tg_request(lambda bot: bot.unban_chat_member(chat_id=config.TG_CHAT_ID, user_id=user_id), label="unban_chat_member") await Ban.filter(user_id=user_id).update(auto_unban=False, unban_at=now_tz().replace(tzinfo=None)) await Action.create(action="UNBAN", user_id=user_id, details={"method": "web"}) await broadcast_sse("unban", {"user_id": user_id})