"""Web 管理面板 API""" from __future__ import annotations import asyncio import json import logging from datetime import datetime from pathlib import Path from typing import Any from fastapi import FastAPI, Query, Request from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from db.models import Action, Ban, GameScore, Message from services.stats_service import get_daily_stats 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 class SSELogHandler(logging.Handler): """把应用日志推送到 Web 实时日志窗口。""" def emit(self, record: logging.LogRecord): try: item = { "time": datetime.fromtimestamp(record.created).isoformat(sep=" ", timespec="seconds"), "level": record.levelname, "message": self.format(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("%(asctime)s [%(levelname)s] %(message)s", "%Y-%m-%d %H:%M:%S")) logger.addHandler(handler) install_log_handler() def json_safe(value: Any) -> Any: """将 ORM values() 结果转成 JSONResponse 可序列化对象。""" if isinstance(value, datetime): return value.isoformat(sep=" ", timespec="seconds") 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": datetime.now().isoformat(timespec="seconds")} @app.get("/api/stats") async def api_stats(day: str = Query(default=None, pattern=r"^\d{4}-\d{2}-\d{2}$")): return JSONResponse(await get_daily_stats(day)) def normalize_message(row: dict) -> dict: row = dict(row) row["time"] = row.get("created_at") row["spam"] = row.get("is_spam", 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)): msgs = await Message.all().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)): """兼容旧前端：返回聊天记录。""" msgs = await Message.all().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)): bans = await Ban.all().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/logs") async def api_logs(limit: int = Query(default=100, ge=1, le=300)): return JSONResponse(_log_buffer[-limit:]) @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) bot = Bot(token=config.TG_BOT_TOKEN, request=HTTPXRequest(proxy=config.PROXY_URL)) msg = await bot.send_message(chat_id=config.TG_CHAT_ID, text=text) rec = await Message.create( msg_id=msg.message_id, chat_id=config.TG_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": config.TG_CHAT_ID, "user_id": 0, "username": "web-admin", "first_name": "Web 管理端", "text": text, "time": datetime.now().isoformat(sep=" ", timespec="seconds"), "spam": False, }) return JSONResponse({"ok": True, "msg_id": msg.message_id}) @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=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=datetime.now()) 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())