ngfchl
326 lines
12 KiB
Python
326 lines
12 KiB
Python
"""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
|
|
from services.time_utils import format_dt, from_timestamp, now_tz
|
|
|
|
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
|
|
|
|
|
|
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 f"{format_dt(from_timestamp(record.created))} [{record.levelname}] {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()
|
|
|
|
|
|
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}$")):
|
|
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)
|
|
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)):
|
|
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": format_dt(now_tz()),
|
|
"spam": False,
|
|
})
|
|
return JSONResponse({"ok": True, "msg_id": msg.message_id})
|
|
|
|
|
|
@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"))
|
|
|
|
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=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=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())
|