"""Web 管理面板 API""" from __future__ import annotations import asyncio import json import logging import hmac import hashlib import ipaddress import socket import time from datetime import datetime from pathlib import Path from typing import Any from fastapi import FastAPI, Query, Request, Response from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from db.models import Action, Ban, GameScore, Message, LotteryEvent, LotteryParticipant, ShopItem, ShopTransaction from config import WEB_AUTH_ENABLED, WEB_AUTH_PASSWORD, WEB_AUTH_USERNAME, WEB_SESSION_SECRET, WEB_TRUST_SAME_SUBNET from services.stats_service import get_daily_stats from services.time_utils import format_dt, from_timestamp, now_tz from services.game_service import list_shop_items, upsert_shop_item, delete_shop_item, list_shop_transactions, draw_lottery_event BASE_DIR = Path(__file__).parent app = FastAPI(title="TG Spam Guard Manager") app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static") # SSE _sse_queues: list[asyncio.Queue] = [] _log_buffer: list[dict] = [] MAX_LOG_BUFFER = 300 MAX_LOG_MESSAGE_CHARS = 1200 _restart_bot_callback = None def set_restart_bot_callback(callback): global _restart_bot_callback _restart_bot_callback = callback def compact_log_message(record: logging.LogRecord) -> str: """Web 日志只展示单行摘要，避免大段 traceback/长文本刷屏。""" msg = record.getMessage().replace("\r", " ").replace("\n", " ") msg = " ".join(msg.split()) if len(msg) > MAX_LOG_MESSAGE_CHARS: msg = msg[:MAX_LOG_MESSAGE_CHARS] + "..." return msg class SSELogHandler(logging.Handler): """把应用日志推送到 Web 实时日志窗口。""" def emit(self, record: logging.LogRecord): try: item = { "time": format_dt(from_timestamp(record.created)), "level": record.levelname, "message": compact_log_message(record), } _log_buffer.append(item) del _log_buffer[:-MAX_LOG_BUFFER] try: loop = asyncio.get_running_loop() loop.create_task(broadcast_sse("log", item)) except RuntimeError: pass except Exception: pass def install_log_handler(): logger = logging.getLogger("spam_guard") if any(isinstance(h, SSELogHandler) for h in logger.handlers): return handler = SSELogHandler() handler.setLevel(logging.INFO) handler.setFormatter(logging.Formatter("%(message)s")) logger.addHandler(handler) install_log_handler() AUTH_COOKIE_NAME = "tg_spam_guard_session" AUTH_COOKIE_MAX_AGE = 7 * 24 * 3600 LOCAL_NETWORKS: list[ipaddress._BaseNetwork] = [] def _local_ipv4_addresses() -> set[str]: ips = {"127.0.0.1"} try: hostname = socket.gethostname() for item in socket.getaddrinfo(hostname, None, family=socket.AF_INET): ips.add(item[4][0]) except Exception: pass try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.connect(("8.8.8.8", 80)) ips.add(sock.getsockname()[0]) sock.close() except Exception: pass return ips def _build_local_networks() -> list[ipaddress._BaseNetwork]: nets: list[ipaddress._BaseNetwork] = [ipaddress.ip_network("127.0.0.0/8")] for ip in _local_ipv4_addresses(): try: addr = ipaddress.ip_address(ip) if addr.is_loopback: continue nets.append(ipaddress.ip_network(f"{ip}/24", strict=False)) except Exception: pass return nets LOCAL_NETWORKS = _build_local_networks() def _sign_session(username: str, ts: int | None = None) -> str: ts = ts or int(time.time()) payload = f"{username}:{ts}" sig = hmac.new(WEB_SESSION_SECRET.encode(), payload.encode(), hashlib.sha256).hexdigest() return f"{payload}:{sig}" def _verify_session(token: str | None) -> bool: if not token: return False try: username, ts_text, sig = token.split(":", 2) ts = int(ts_text) if username != WEB_AUTH_USERNAME or time.time() - ts > AUTH_COOKIE_MAX_AGE: return False expected = hmac.new(WEB_SESSION_SECRET.encode(), f"{username}:{ts}".encode(), hashlib.sha256).hexdigest() return hmac.compare_digest(sig, expected) except Exception: return False def _client_ip(request: Request) -> str: forwarded = request.headers.get("x-forwarded-for", "").split(",")[0].strip() return forwarded or (request.client.host if request.client else "") def _same_subnet_allowed(request: Request) -> bool: if not WEB_TRUST_SAME_SUBNET: return False try: addr = ipaddress.ip_address(_client_ip(request)) return any(addr in net for net in LOCAL_NETWORKS) except Exception: return False def _is_auth_allowed(request: Request) -> tuple[bool, str]: if not WEB_AUTH_ENABLED: return True, "disabled" if _same_subnet_allowed(request): return True, "same_subnet" if _verify_session(request.cookies.get(AUTH_COOKIE_NAME)): return True, "session" return False, "login_required" @app.middleware("http") async def web_auth_middleware(request: Request, call_next): path = request.url.path public = path == "/" or path == "/health" or path.startswith("/static/") or path in {"/api/auth/status", "/api/auth/login"} if public: return await call_next(request) ok, reason = _is_auth_allowed(request) if not ok: return JSONResponse({"ok": False, "error": "未登录", "reason": reason}, status_code=401) return await call_next(request) @app.get("/api/auth/status") async def api_auth_status(request: Request): ok, reason = _is_auth_allowed(request) return JSONResponse({"ok": True, "authenticated": ok, "reason": reason, "username": WEB_AUTH_USERNAME if ok else None}) @app.post("/api/auth/login") async def api_auth_login(request: Request): data = await request.json() username = str(data.get("username") or "") password = str(data.get("password") or "") if not (hmac.compare_digest(username, WEB_AUTH_USERNAME) and hmac.compare_digest(password, WEB_AUTH_PASSWORD)): return JSONResponse({"ok": False, "error": "账号或密码错误"}, status_code=401) resp = JSONResponse({"ok": True, "username": username}) resp.set_cookie(AUTH_COOKIE_NAME, _sign_session(username), max_age=AUTH_COOKIE_MAX_AGE, httponly=True, samesite="lax") return resp @app.post("/api/auth/logout") async def api_auth_logout(): resp = JSONResponse({"ok": True}) resp.delete_cookie(AUTH_COOKIE_NAME) return resp def json_safe(value: Any) -> Any: """将 ORM values() 结果转成 JSONResponse 可序列化对象。""" if isinstance(value, datetime): return format_dt(value) if isinstance(value, list): return [json_safe(v) for v in value] if isinstance(value, dict): return {k: json_safe(v) for k, v in value.items()} return value async def broadcast_sse(event: str, data: dict): dead = [] payload = {"event": event, "data": json.dumps(data, ensure_ascii=False)} for q in _sse_queues: try: await q.put(payload) except Exception: dead.append(q) for q in dead: if q in _sse_queues: _sse_queues.remove(q) @app.get("/", response_class=HTMLResponse) async def index(): return HTMLResponse((BASE_DIR / "static" / "index.html").read_text(encoding="utf-8")) @app.get("/health") async def health(): return {"ok": True, "service": "tg-spam-guard", "time": format_dt(now_tz())} @app.get("/api/stats") async def api_stats(day: str = Query(default=None, pattern=r"^\d{4}-\d{2}-\d{2}$"), chat_id: int | None = None): return JSONResponse(await get_daily_stats(day, chat_id=chat_id)) def parse_draw_at(value): if not value: return None try: dt = datetime.fromisoformat(str(value).replace("Z", "+00:00")) if dt.tzinfo is None: dt = dt.replace(tzinfo=now_tz().tzinfo) return dt except Exception: return None def normalize_message(row: dict) -> dict: row = dict(row) row["time"] = row.get("created_at") row["spam"] = row.get("is_spam", False) row["deleted"] = row.get("manually_deleted", False) return row def normalize_ban(row: dict) -> dict: row = dict(row) row["time"] = row.get("created_at") return row @app.get("/api/messages") async def api_messages(limit: int = Query(default=50, ge=1, le=500), chat_id: int | None = None): qs = Message.all() if chat_id is not None: qs = qs.filter(chat_id=chat_id) msgs = await qs.order_by("-created_at").limit(limit).values() return JSONResponse(json_safe([normalize_message(m) for m in msgs])) @app.get("/api/chat") async def api_chat(limit: int = Query(default=100, ge=1, le=500), chat_id: int | None = None): """兼容旧前端：返回聊天记录。""" qs = Message.all() if chat_id is not None: qs = qs.filter(chat_id=chat_id) msgs = await qs.order_by("-created_at").limit(limit).values() return JSONResponse(json_safe([normalize_message(m) for m in msgs])) @app.get("/api/bans") async def api_bans(limit: int = Query(default=100, ge=1, le=500), chat_id: int | None = None): qs = Ban.all() if chat_id is not None: qs = qs.filter(chat_id=chat_id) bans = await qs.order_by("-created_at").limit(limit).values() return JSONResponse(json_safe([normalize_ban(b) for b in bans])) @app.get("/api/actions") async def api_actions(limit: int = Query(default=100, ge=1, le=500)): actions = await Action.all().order_by("-created_at").limit(limit).values() return JSONResponse(json_safe(list(actions))) @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: sources.extend(await model.all().distinct().values("chat_id")) except Exception: pass ids = sorted({int(r["chat_id"]) for r in sources if r.get("chat_id")}) 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) name = chat.title or chat.full_name or chat.username or str(chat_id) except Exception: pass result.append({"chat_id": chat_id, "name": name, "label": f"{name} ({chat_id})"}) return JSONResponse(result) @app.get("/api/shop-items") async def api_shop_items(chat_id: int = 0): return JSONResponse(json_safe(await list_shop_items(chat_id, include_disabled=True))) @app.post("/api/shop-items") async def api_save_shop_item(request: Request): data = await request.json() item = await upsert_shop_item( chat_id=int(data.get("chat_id") or 0), item_key=str(data.get("item_key") or "").strip(), name=str(data.get("name") or "").strip(), cost=int(data.get("cost") or 0), desc=str(data.get("desc") or ""), enabled=bool(data.get("enabled", True)), ) return JSONResponse(json_safe({"ok": True, "item": {"id": item.id, "chat_id": item.chat_id, "item_key": item.item_key, "name": item.name, "cost": item.cost, "desc": item.desc, "enabled": item.enabled}})) @app.delete("/api/shop-items/{item_key}") async def api_delete_shop_item(item_key: str, chat_id: int = 0): ok = await delete_shop_item(chat_id, item_key) return JSONResponse({"ok": ok}) @app.get("/api/shop-transactions") async def api_shop_transactions(chat_id: int = 0, limit: int = Query(default=100, ge=1, le=500)): return JSONResponse(json_safe(await list_shop_transactions(chat_id, limit))) @app.get("/api/lotteries") async def api_lotteries(chat_id: int | None = None, limit: int = Query(default=100, ge=1, le=300)): qs = LotteryEvent.all() if chat_id is not None: qs = qs.filter(chat_id=chat_id) rows = await qs.order_by("-created_at").limit(limit).values() return JSONResponse(json_safe(list(rows))) @app.post("/api/lotteries") async def api_create_lottery(request: Request): data = await request.json() event = await LotteryEvent.create( chat_id=int(data.get("chat_id") or 0), creator_id=int(data.get("creator_id") or 0), creator_name=str(data.get("creator_name") or "web-admin"), title=str(data.get("title") or data.get("prize") or "抽奖活动"), description=str(data.get("description") or ""), prize=str(data.get("prize") or data.get("title") or "奖品"), condition_type=str(data.get("condition_type") or "all"), condition_value=int(data.get("condition_value") or 0), draw_type=str(data.get("draw_type") or data.get("end_type") or "manual"), draw_at=parse_draw_at(data.get("draw_at")), end_type=str(data.get("end_type") or data.get("draw_type") or "manual"), end_value=int(data.get("end_value") or data.get("max_participants") or 0), max_participants=int(data.get("max_participants") or 0), status=str(data.get("status") or "draft"), ) return JSONResponse(json_safe({"ok": True, "event": { "id": event.id, "chat_id": event.chat_id, "title": event.title, "description": event.description, "prize": event.prize, "condition_type": event.condition_type, "condition_value": event.condition_value, "draw_type": event.draw_type, "end_type": event.end_type, "end_value": event.end_value, "max_participants": event.max_participants, "draw_at": event.draw_at, "status": event.status, "created_at": event.created_at, }})) @app.put("/api/lotteries/{event_id}") async def api_update_lottery(event_id: int, request: Request): event = await LotteryEvent.filter(id=event_id).first() if not event: return JSONResponse({"ok": False, "error": "抽奖不存在"}, status_code=404) data = await request.json() for key in ["title", "description", "prize", "condition_type", "draw_type", "end_type", "status"]: if key in data: setattr(event, key, str(data.get(key) or "")) if "draw_at" in data: event.draw_at = parse_draw_at(data.get("draw_at")) for key in ["condition_value", "end_value", "max_participants"]: if key in data: setattr(event, key, int(data.get(key) or 0)) await event.save() return JSONResponse(json_safe({"ok": True, "event": { "id": event.id, "chat_id": event.chat_id, "title": event.title, "description": event.description, "prize": event.prize, "condition_type": event.condition_type, "condition_value": event.condition_value, "draw_type": event.draw_type, "end_type": event.end_type, "end_value": event.end_value, "max_participants": event.max_participants, "draw_at": event.draw_at, "status": event.status, "created_at": event.created_at, }})) @app.delete("/api/lotteries/{event_id}") async def api_delete_lottery(event_id: int): await LotteryParticipant.filter(event_id=event_id).delete() ok = await LotteryEvent.filter(id=event_id).delete() return JSONResponse({"ok": bool(ok)}) @app.get("/api/lotteries/{event_id}/participants") async def api_lottery_participants(event_id: int): rows = await LotteryParticipant.filter(event_id=event_id).order_by("created_at").values() return JSONResponse(json_safe(list(rows))) @app.post("/api/lotteries/{event_id}/cancel") async def api_cancel_lottery(event_id: int): event = await LotteryEvent.filter(id=event_id).first() if not event: return JSONResponse({"ok": False, "error": "抽奖不存在"}, status_code=404) event.status = "cancelled" event.ended_at = now_tz() await event.save() return JSONResponse({"ok": True}) @app.post("/api/lotteries/{event_id}/draw") async def api_draw_lottery(event_id: int): result = await draw_lottery_event(event_id) return JSONResponse(json_safe(result)) @app.get("/api/logs") async def api_logs(limit: int = Query(default=100, ge=1, le=300)): return JSONResponse(_log_buffer[-limit:]) @app.post("/api/restart-bot") async def api_restart_bot(): """只重启 Telegram Bot，不重启 Web 服务。""" if not _restart_bot_callback: return JSONResponse({"ok": False, "error": "Bot 重启回调未就绪"}, status_code=503) logging.getLogger("spam_guard").warning("🔄 Web 管理端请求重启 TG Bot") asyncio.create_task(_restart_bot_callback()) return JSONResponse({"ok": True, "message": "TG Bot 正在重启"}) @app.post("/api/restart") async def api_restart(): """重启服务：返回响应后退出当前进程，由 Docker restart 策略拉起。""" import os import signal logging.getLogger("spam_guard").warning("🔄 Web 管理端请求重启服务") async def delayed_exit(): await asyncio.sleep(1) os.kill(os.getpid(), signal.SIGTERM) asyncio.create_task(delayed_exit()) return JSONResponse({"ok": True, "message": "服务正在重启"}) @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: return JSONResponse({"ok": False, "error": "消息不能为空"}, status_code=400) if len(text) > 3500: 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) rec = await Message.create( msg_id=msg.message_id, chat_id=chat_id, user_id=0, username="web-admin", first_name="Web 管理端", text=text[:500], ) logging.getLogger("spam_guard").info(f"🌐 Web 发送消息: {text[:120]}") await broadcast_sse("chat", { "id": rec.id, "msg_id": msg.message_id, "chat_id": chat_id, "user_id": 0, "username": "web-admin", "first_name": "Web 管理端", "text": text, "time": format_dt(now_tz()), "spam": False, }) return JSONResponse({"ok": True, "msg_id": msg.message_id}) @app.post("/api/chat/{message_id}/delete") @app.post("/api/messages/{message_id}/delete") 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")) rec = await Message.get_or_none(id=message_id) if not rec: return JSONResponse({"ok": False, "error": "消息不存在"}, status_code=404) if rec.manually_deleted: return JSONResponse({"ok": True, "already_deleted": True}) 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) except BadRequest as e: # 消息已不存在时，允许继续标记数据库；其他 TG 删除失败不标记。 if "message to delete not found" in str(e).lower(): tg_error = str(e) else: return JSONResponse({"ok": False, "error": f"TG 删除失败: {e}"}, status_code=400) except Exception as e: return JSONResponse({"ok": False, "error": f"TG 删除失败: {type(e).__name__}: {e}"}, status_code=502) rec.manually_deleted = True rec.deleted_at = now_tz() rec.delete_reason = "web_manual_delete" if not tg_error else f"web_manual_delete_after_tg_error: {tg_error[:160]}" await rec.save() ban = None if ban_user and rec.user_id: 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())) ban = await Ban.create( user_id=rec.user_id, username=rec.username, first_name=rec.first_name, reason="Web 删除消息时手动封禁", method="web_delete_ban", duration_minutes=config.BAN_DURATION_MINUTES, unban_at=unban_at, ) except Exception as e: return JSONResponse({"ok": False, "error": f"消息已删除，但封禁失败: {type(e).__name__}: {e}"}, status_code=502) await Action.create( action="MANUAL_DELETE_MESSAGE", user_id=rec.user_id, username=rec.username, details={"message_id": rec.id, "chat_id": rec.chat_id, "msg_id": rec.msg_id, "tg_error": tg_error}, ) logging.getLogger("spam_guard").info(f"🗑️ Web 手动删除聊天消息: db_id={rec.id}, tg_msg_id={rec.msg_id}, user=@{rec.username}") await broadcast_sse("chat_delete", json_safe({ "id": rec.id, "msg_id": rec.msg_id, "chat_id": rec.chat_id, "deleted_at": rec.deleted_at, })) if ban: await Action.create(action="BAN", user_id=rec.user_id, username=rec.username, details={"method": "web_delete_ban", "message_id": rec.id}) await broadcast_sse("ban", json_safe({ "id": ban.id, "user_id": rec.user_id, "username": rec.username, "first_name": rec.first_name, "reason": ban.reason, "method": ban.method, "duration_minutes": config.BAN_DURATION_MINUTES, "auto_unban": True, "unban_at": unban_at, "time": ban.created_at, })) return JSONResponse(json_safe({"ok": True, "tg_deleted": tg_deleted, "tg_error": tg_error, "deleted_at": rec.deleted_at, "banned": bool(ban)})) @app.get("/api/scores") async def api_scores(limit: int = Query(default=50, ge=1, le=200)): scores = await GameScore.all().order_by("-score").limit(limit).values() return JSONResponse(json_safe(list(scores))) @app.post("/api/unban/{user_id}") 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 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}) return JSONResponse({"ok": True, "user_id": user_id}) except Exception as e: return JSONResponse({"ok": False, "error": str(e)}, status_code=500) @app.get("/api/stream") async def api_stream(request: Request): from sse_starlette.sse import EventSourceResponse queue: asyncio.Queue = asyncio.Queue() _sse_queues.append(queue) async def event_generator(): try: yield {"event": "connected", "data": "ok"} while True: if await request.is_disconnected(): break try: data = await asyncio.wait_for(queue.get(), timeout=30) yield {"event": data["event"], "data": data["data"]} except asyncio.TimeoutError: yield {"event": "ping", "data": "ok"} finally: if queue in _sse_queues: _sse_queues.remove(queue) return EventSourceResponse(event_generator())