ngfchl
683 lines
25 KiB
Python
683 lines
25 KiB
Python
"""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 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
|
|
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
|
|
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)
|
|
|
|
result = []
|
|
for chat_id in ids:
|
|
name = str(chat_id)
|
|
try:
|
|
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
|
|
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
|
|
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)
|
|
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,
|
|
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.error import BadRequest
|
|
|
|
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)
|
|
|
|
tg_deleted = True
|
|
tg_error = None
|
|
try:
|
|
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():
|
|
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 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,
|
|
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:
|
|
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})
|
|
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())
|