ngfchl
1181 lines
49 KiB
Python
1181 lines
49 KiB
Python
"""Web 管理面板 API"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import re
|
||
import hmac
|
||
import hashlib
|
||
import ipaddress
|
||
import socket
|
||
import time
|
||
import httpx
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from fastapi import FastAPI, Query, Request, Response
|
||
from fastapi.exceptions import RequestValidationError
|
||
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
from pydantic import BaseModel, Field
|
||
from tortoise.expressions import Q
|
||
|
||
from db.models import Action, Ban, GameScore, Message, LotteryEvent, LotteryParticipant, ShopItem, ShopTransaction, Whitelist
|
||
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
|
||
from services import spam_detector
|
||
|
||
BASE_DIR = Path(__file__).parent
|
||
|
||
app = FastAPI(title="TG Spam Guard Manager")
|
||
|
||
|
||
@app.exception_handler(RequestValidationError)
|
||
async def validation_exception_handler(request: Request, exc: RequestValidationError):
|
||
return JSONResponse({"ok": False, "error": "请求参数无效", "details": exc.errors()}, status_code=422)
|
||
|
||
|
||
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
|
||
_rule_write_lock = asyncio.Lock()
|
||
|
||
|
||
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):
|
||
try:
|
||
data = await request.json()
|
||
if not isinstance(data, dict):
|
||
data = {}
|
||
except Exception:
|
||
return JSONResponse({"ok": False, "error": "请求体必须是合法 JSON"}, status_code=400)
|
||
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
|
||
|
||
|
||
|
||
|
||
|
||
class LoginPayload(BaseModel):
|
||
username: str = Field(default='', max_length=80)
|
||
password: str = Field(default='', max_length=200)
|
||
|
||
|
||
class SendPayload(BaseModel):
|
||
text: str = Field(default='', max_length=4000)
|
||
chat_id: int | None = None
|
||
reply_to_message_id: int | None = None
|
||
|
||
|
||
class BanPayload(BaseModel):
|
||
duration_minutes: int = Field(default=0, ge=0, le=525600)
|
||
|
||
|
||
class ShopItemPayload(BaseModel):
|
||
chat_id: int = 0
|
||
item_key: str = Field(default='', max_length=64)
|
||
name: str = Field(default='', max_length=80)
|
||
cost: int = Field(default=0, ge=0, le=1_000_000)
|
||
desc: str = Field(default='', max_length=500)
|
||
enabled: bool = True
|
||
stock: int = Field(default=-1, ge=-1, le=1_000_000)
|
||
|
||
|
||
class LotteryPayload(BaseModel):
|
||
chat_id: int = 0
|
||
creator_id: int = 0
|
||
creator_name: str = Field(default='web-admin', max_length=80)
|
||
title: str = Field(default='', max_length=120)
|
||
description: str = Field(default='', max_length=500)
|
||
prize: str = Field(default='', max_length=120)
|
||
condition_type: str = Field(default='all', max_length=32)
|
||
condition_value: int = Field(default=0, ge=0, le=1_000_000)
|
||
draw_type: str = Field(default='people', max_length=32)
|
||
draw_at: str | None = None
|
||
end_type: str = Field(default='people', max_length=32)
|
||
end_value: int = Field(default=0, ge=0, le=1_000_000)
|
||
max_participants: int = Field(default=0, ge=0, le=1_000_000)
|
||
prize_count: int = Field(default=1, ge=1, le=1_000_000)
|
||
remaining_count: int | None = Field(default=None, ge=0, le=1_000_000)
|
||
status: str = Field(default='active', max_length=32)
|
||
|
||
|
||
class SpamTestPayload(BaseModel):
|
||
text: str = Field(default='', max_length=5000)
|
||
use_ai: bool = False
|
||
|
||
|
||
class SpamRulePayload(BaseModel):
|
||
type: str = Field(default='keyword', max_length=16)
|
||
pattern: str = Field(default='', max_length=500)
|
||
score: int = Field(default=60, ge=1, le=200)
|
||
async def safe_json(request: Request) -> dict:
|
||
if not request.headers.get("content-type", "").startswith("application/json"):
|
||
return {}
|
||
try:
|
||
data = await request.json()
|
||
return data if isinstance(data, dict) else {}
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def normalize_condition_type(value: str) -> str:
|
||
value = (value or "all").strip()
|
||
return "min_points" if value == "points" else value
|
||
|
||
|
||
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.exclude(text="__service__:new_chat_member")
|
||
if chat_id is not None:
|
||
qs = qs.filter(Q(chat_id=chat_id) | Q(chat_id=0))
|
||
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.exclude(text="__service__:new_chat_member")
|
||
if chat_id is not None:
|
||
qs = qs.filter(Q(chat_id=chat_id) | Q(chat_id=0))
|
||
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(Q(chat_id=chat_id) | Q(chat_id=0))
|
||
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(data: ShopItemPayload):
|
||
item_key = data.item_key.strip()
|
||
name = data.name.strip()
|
||
if not item_key or not name:
|
||
return JSONResponse({"ok": False, "error": "商品编号和名称不能为空"}, status_code=400)
|
||
item = await upsert_shop_item(
|
||
chat_id=data.chat_id,
|
||
item_key=item_key,
|
||
name=name,
|
||
cost=data.cost,
|
||
desc=data.desc,
|
||
enabled=data.enabled,
|
||
stock=data.stock,
|
||
)
|
||
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, "stock": item.stock, "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(Q(chat_id=chat_id) | Q(chat_id=0))
|
||
rows = list(await qs.order_by("-created_at").limit(limit).values())
|
||
for row in rows:
|
||
row["participants_count"] = await LotteryParticipant.filter(event_id=row["id"]).count()
|
||
return JSONResponse(json_safe(rows))
|
||
|
||
|
||
@app.post("/api/lotteries")
|
||
async def api_create_lottery(data: LotteryPayload):
|
||
title = (data.title or data.prize or "抽奖活动").strip()
|
||
prize = (data.prize or data.title or "奖品").strip()
|
||
if not title or not prize:
|
||
return JSONResponse({"ok": False, "error": "抽奖标题和奖品不能为空"}, status_code=400)
|
||
draw_type = data.draw_type or data.end_type or "manual"
|
||
event = await LotteryEvent.create(
|
||
chat_id=data.chat_id,
|
||
creator_id=data.creator_id,
|
||
creator_name=data.creator_name or "web-admin",
|
||
title=title,
|
||
description=data.description,
|
||
prize=prize,
|
||
condition_type=normalize_condition_type(data.condition_type),
|
||
condition_value=data.condition_value,
|
||
draw_type=draw_type,
|
||
draw_at=parse_draw_at(data.draw_at),
|
||
end_type=data.end_type or draw_type,
|
||
end_value=data.end_value or data.max_participants,
|
||
max_participants=data.max_participants,
|
||
prize_count=max(1, data.prize_count),
|
||
remaining_count=max(0, data.remaining_count if data.remaining_count is not None else data.prize_count),
|
||
status=data.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, "prize_count": event.prize_count, "remaining_count": event.remaining_count,
|
||
"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, data: LotteryPayload):
|
||
event = await LotteryEvent.filter(id=event_id).first()
|
||
if not event:
|
||
return JSONResponse({"ok": False, "error": "抽奖不存在"}, status_code=404)
|
||
participant_count = await LotteryParticipant.filter(event_id=event_id).count()
|
||
if participant_count > 0:
|
||
return JSONResponse({"ok": False, "error": "已有用户参与，不能编辑抽奖；仅可强制开奖或关闭"}, status_code=409)
|
||
for key in ["title", "description", "prize", "draw_type", "end_type", "status"]:
|
||
value = getattr(data, key)
|
||
if value is not None:
|
||
setattr(event, key, str(value or ""))
|
||
event.condition_type = normalize_condition_type(data.condition_type)
|
||
event.draw_at = parse_draw_at(data.draw_at)
|
||
event.condition_value = data.condition_value
|
||
event.end_value = data.end_value
|
||
event.max_participants = data.max_participants
|
||
event.prize_count = max(1, data.prize_count)
|
||
event.remaining_count = max(0, data.remaining_count if data.remaining_count is not None else data.prize_count)
|
||
await event.save()
|
||
return JSONResponse(json_safe({"ok": True, "event": await LotteryEvent.filter(id=event_id).values().first()}))
|
||
|
||
|
||
@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 正在重启"})
|
||
|
||
|
||
def sanitize_manual_rule(rule_type: str, pattern: str) -> tuple[str, str | None]:
|
||
"""清洗后台手动规则。关键词会去零宽/压缩空白，正则只做首尾清理并校验。"""
|
||
original = str(pattern or "")
|
||
cleaned = spam_detector.normalize(original)
|
||
cleaned = cleaned.replace("｜", "|").strip()
|
||
if rule_type == "keyword":
|
||
# 关键词最终写紧凑版本：外部关键词会同时匹配 normalize/compact 两种目标，
|
||
# 紧凑写入能覆盖广告常用的零宽字符、断词、插空格逃逸。
|
||
cleaned = spam_detector.compact(cleaned)
|
||
cleaned = cleaned.strip("｜|")
|
||
if len(cleaned) < 2:
|
||
raise ValueError("关键词太短，至少 2 个字符")
|
||
if cleaned.lower() in {"qq", "tg", "ok", "yes", "http", "https", "招聘", "联系", "项目"}:
|
||
raise ValueError("关键词过于泛化，容易误杀，请改用更具体词组或正则")
|
||
else:
|
||
if not cleaned:
|
||
raise ValueError("正则不能为空")
|
||
re.compile(cleaned)
|
||
changed = cleaned != original.strip()
|
||
return cleaned, (original if changed else None)
|
||
|
||
|
||
def parse_json_object(text: str) -> dict[str, Any] | None:
|
||
body = (text or "").strip()
|
||
if not body:
|
||
return None
|
||
try:
|
||
data = json.loads(body)
|
||
return data if isinstance(data, dict) else None
|
||
except Exception:
|
||
pass
|
||
m = re.search(r"\{[\s\S]*\}", body)
|
||
if not m:
|
||
return None
|
||
try:
|
||
data = json.loads(m.group(0))
|
||
return data if isinstance(data, dict) else None
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def fallback_rule_from_text(text: str) -> dict[str, Any]:
|
||
norm = spam_detector.normalize(text)
|
||
domains = sorted(spam_detector.extract_domains_from_text(norm))
|
||
if domains:
|
||
return {"type": "keyword", "pattern": domains[0], "score": 80, "reason": "AI不可用，按域名生成关键词规则"}
|
||
compacted = spam_detector.compact(norm)
|
||
compacted = re.sub(r"[，。！？、,.!?;；：:\s]+", "", compacted)
|
||
if len(compacted) > 80:
|
||
compacted = compacted[:80]
|
||
return {"type": "keyword", "pattern": compacted, "score": 90, "reason": "AI不可用，按紧凑文本生成关键词规则"}
|
||
|
||
|
||
async def generate_rule_from_text(text: str, use_ai: bool = True) -> dict[str, Any]:
|
||
import config
|
||
if not use_ai:
|
||
return fallback_rule_from_text(text)
|
||
prompt = f"""你是中文群聊垃圾广告规则提取器。请从下面的聊天内容中提取一条最适合写入拦截库的规则。
|
||
|
||
要求：
|
||
1. 只输出 JSON，不要解释，不要 Markdown。
|
||
2. JSON 格式：{{"type":"keyword|regex","pattern":"规则内容","score":1-200,"reason":"一句话说明"}}
|
||
3. 优先生成能覆盖变体但不容易误杀的规则。
|
||
4. 如果内容包含明确广告域名/链接，优先提取域名作为 keyword。
|
||
5. 如果有零宽字符、断词、空格插入、黑U/USDT/跑分/贷款/博彩等变体，优先生成 regex。
|
||
6. keyword 不要太泛，禁止只输出：联系、项目、赚钱、福利、微信、QQ、TG、http、https。
|
||
7. regex 必须是 Python re 可编译的正则，不要带 /.../ 包裹。
|
||
8. 分数建议：明确灰产/博彩/色情/诈骗 90-120；普通营销 60-80；域名 80。
|
||
|
||
聊天内容：
|
||
{text[:2000]}
|
||
"""
|
||
try:
|
||
async with httpx.AsyncClient(timeout=60, proxy=None) as client:
|
||
resp = await client.post(
|
||
f"{config.OLLAMA_URL}/api/generate",
|
||
json={"model": config.OLLAMA_MODEL, "prompt": prompt, "stream": False, "think": False, "options": {"temperature": 0.1, "num_predict": 220}},
|
||
)
|
||
resp.raise_for_status()
|
||
reply = resp.json().get("response", "")
|
||
data = parse_json_object(reply)
|
||
if not data:
|
||
raise ValueError("AI 未返回合法 JSON")
|
||
rule_type = str(data.get("type") or "keyword").strip().lower()
|
||
pattern = str(data.get("pattern") or "").strip()
|
||
score = int(data.get("score") or 90)
|
||
reason = str(data.get("reason") or "AI 生成规则").strip()
|
||
if rule_type not in {"keyword", "regex"}:
|
||
rule_type = "keyword"
|
||
if not pattern:
|
||
raise ValueError("AI 生成规则为空")
|
||
return {"type": rule_type, "pattern": pattern, "score": max(1, min(score, 200)), "reason": reason, "ai_used": True}
|
||
except Exception as e:
|
||
logging.getLogger("spam_guard").warning(f"⚠️ AI 生成规则失败，使用兜底规则: {e}")
|
||
data = fallback_rule_from_text(text)
|
||
data["ai_used"] = False
|
||
data["reason"] = f"{data.get('reason', '兜底生成')}；AI异常: {type(e).__name__}"
|
||
return data
|
||
|
||
|
||
async def append_spam_rule(rule_type: str, pattern: str, score: int, source_note: str = "Web 添加规则") -> dict[str, Any]:
|
||
rule_type = rule_type.strip().lower()
|
||
if rule_type not in {"keyword", "regex"}:
|
||
raise ValueError("规则类型只能是 keyword 或 regex")
|
||
score = max(1, min(int(score), 200))
|
||
stored_pattern, original_pattern = sanitize_manual_rule(rule_type, pattern)
|
||
path = spam_detector.RULES_DIR / ("spam_keywords.txt" if rule_type == "keyword" else "spam_regex.txt")
|
||
line = f"{stored_pattern}|{score}"
|
||
async with _rule_write_lock:
|
||
existing = path.read_text(encoding="utf-8") if path.exists() else ""
|
||
existing_patterns = {raw.lstrip("\ufeff").rsplit("|", 1)[0].strip() for raw in existing.splitlines() if raw.strip() and not raw.strip().lstrip("\ufeff").startswith("#")}
|
||
if stored_pattern in existing_patterns:
|
||
raise FileExistsError("规则已存在")
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with path.open("a", encoding="utf-8") as f:
|
||
if existing and not existing.endswith("\n"):
|
||
f.write("\n")
|
||
note = f" 原始: {original_pattern}" if original_pattern else ""
|
||
f.write(f"\n# {source_note} {format_dt(now_tz())}{note}\n{line}\n")
|
||
counts = spam_detector.reload_external_rules()
|
||
return {"type": rule_type, "pattern": stored_pattern, "original_pattern": original_pattern, "score": score, "rule_counts": counts}
|
||
|
||
|
||
@app.get("/api/spam-rules")
|
||
async def api_list_spam_rules(limit: int = Query(default=500, ge=1, le=2000)):
|
||
def read_rules(path: Path, rule_type: str):
|
||
items = []
|
||
if not path.exists():
|
||
return items
|
||
for i, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
||
line = raw.strip().lstrip("\ufeff")
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
if "|" in line:
|
||
pattern, score_text = line.rsplit("|", 1)
|
||
try:
|
||
score = int(score_text.strip())
|
||
except ValueError:
|
||
pattern, score = line, None
|
||
else:
|
||
pattern, score = line, None
|
||
items.append({"type": rule_type, "line": i, "pattern": pattern.strip(), "score": score})
|
||
return items
|
||
keywords = read_rules(spam_detector.RULES_DIR / "spam_keywords.txt", "keyword")
|
||
regex = read_rules(spam_detector.RULES_DIR / "spam_regex.txt", "regex")
|
||
adblock_path = spam_detector.RULES_DIR / "adblock_domains.txt"
|
||
adblock_preview = []
|
||
adblock_count = 0
|
||
if adblock_path.exists():
|
||
for line in adblock_path.read_text(encoding="utf-8", errors="ignore").splitlines():
|
||
raw = line.strip()
|
||
if not raw or raw.startswith("#"):
|
||
continue
|
||
adblock_count += 1
|
||
if len(adblock_preview) < min(limit, 200):
|
||
adblock_preview.append({"type": "adblock_domain", "line": adblock_count, "pattern": raw, "score": 80})
|
||
return JSONResponse({"ok": True, "keywords": keywords[-limit:], "regex": regex[-limit:], "adblock_domains": adblock_preview, "counts": {"keywords": len(keywords), "regex": len(regex), "adblock_domains": adblock_count}})
|
||
|
||
|
||
@app.post("/api/spam-test")
|
||
async def api_spam_test(data: SpamTestPayload):
|
||
text = data.text.strip()
|
||
use_ai = data.use_ai
|
||
if not text:
|
||
return JSONResponse({"ok": False, "error": "请输入要检测的文本"}, status_code=400)
|
||
decision, reason, score = spam_detector.local_classify(text)
|
||
result = {
|
||
"ok": True,
|
||
"text": text,
|
||
"local_decision": decision,
|
||
"local_is_spam": bool(decision) if decision is not None else None,
|
||
"reason": reason,
|
||
"score": score,
|
||
"rule_counts": {"keywords": len(spam_detector.EXTERNAL_KEYWORD_RULES), "regex": len(spam_detector.EXTERNAL_REGEX_RULES), "adblock_domains": len(spam_detector.ADBLOCK_DOMAINS)},
|
||
}
|
||
if use_ai:
|
||
if decision is True:
|
||
result.update({"ai_is_spam": None, "ai_reason": "本地强规则已判定为垃圾，未额外调用 AI", "final_is_spam": True})
|
||
else:
|
||
ai_is_spam, ai_reason = await spam_detector.ai_classify(text)
|
||
final_is_spam = True if ai_is_spam else (bool(decision) if decision is not None else False)
|
||
if ai_is_spam and score < 75:
|
||
result["score"] = 80
|
||
result["reason"] = f"{reason}; AI加权判定"
|
||
result.update({"ai_is_spam": ai_is_spam, "ai_reason": ai_reason, "final_is_spam": final_is_spam})
|
||
else:
|
||
result["final_is_spam"] = bool(decision) if decision is not None else None
|
||
return JSONResponse(result)
|
||
|
||
|
||
@app.post("/api/spam-rules/from-text")
|
||
async def api_add_spam_rule_from_text(request: Request):
|
||
try:
|
||
raw = await request.json()
|
||
if not isinstance(raw, dict):
|
||
raw = {}
|
||
except Exception:
|
||
return JSONResponse({"ok": False, "error": "请求体必须是合法 JSON"}, status_code=400)
|
||
text = str(raw.get("text") or "").strip()
|
||
use_ai = bool(raw.get("use_ai", True))
|
||
if not text:
|
||
return JSONResponse({"ok": False, "error": "请输入要入库的聊天内容"}, status_code=400)
|
||
if len(text) > 4000:
|
||
return JSONResponse({"ok": False, "error": "聊天内容不能超过 4000 字"}, status_code=400)
|
||
generated = None
|
||
try:
|
||
generated = await generate_rule_from_text(text, use_ai)
|
||
saved = await append_spam_rule(generated["type"], generated["pattern"], generated["score"], "Web AI生成规则")
|
||
except FileExistsError as e:
|
||
return JSONResponse({"ok": False, "error": str(e), "generated_rule": generated}, status_code=409)
|
||
except re.error as e:
|
||
return JSONResponse({"ok": False, "error": f"AI 生成的正则无效：{e}", "generated_rule": generated}, status_code=400)
|
||
except ValueError as e:
|
||
return JSONResponse({"ok": False, "error": str(e), "generated_rule": generated}, status_code=400)
|
||
await Action.create(action="SPAM_RULE_AI_ADD", details={"text": text[:500], "generated": generated, "saved": saved})
|
||
return JSONResponse({"ok": True, **saved, "generated_rule": generated})
|
||
|
||
|
||
@app.post("/api/spam-rules")
|
||
async def api_add_spam_rule(request: Request):
|
||
try:
|
||
raw = await request.json()
|
||
if not isinstance(raw, dict):
|
||
raw = {}
|
||
except Exception:
|
||
return JSONResponse({"ok": False, "error": "请求体必须是合法 JSON"}, status_code=400)
|
||
rule_type = str(raw.get("type") or "keyword")
|
||
pattern = str(raw.get("pattern") or "").strip()
|
||
try:
|
||
score = int(raw.get("score") or 75)
|
||
except Exception:
|
||
score = 75
|
||
if not pattern or len(pattern) > 500:
|
||
return JSONResponse({"ok": False, "error": "规则内容不能为空，且不能超过 500 字"}, status_code=400)
|
||
try:
|
||
saved = await append_spam_rule(rule_type, pattern, score, "Web 手动添加规则")
|
||
except FileExistsError as e:
|
||
return JSONResponse({"ok": False, "error": str(e)}, status_code=409)
|
||
except re.error as e:
|
||
return JSONResponse({"ok": False, "error": f"正则无效：{e}"}, status_code=400)
|
||
except ValueError as e:
|
||
return JSONResponse({"ok": False, "error": str(e)}, status_code=400)
|
||
await Action.create(action="SPAM_RULE_ADD", details={"type": saved["type"], "pattern": saved["pattern"], "score": saved["score"], "original": saved.get("original_pattern")})
|
||
return JSONResponse({"ok": True, **saved})
|
||
|
||
|
||
@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": "服务正在重启"})
|
||
|
||
|
||
def build_ban_options(duration_minutes: int):
|
||
"""duration_minutes=0 表示永久封禁；否则按分钟自动解封。"""
|
||
duration = int(duration_minutes or 0)
|
||
if duration <= 0:
|
||
return {"duration_minutes": 0, "auto_unban": False, "unban_at": None, "until_date": None}
|
||
unban_at = now_tz() + timedelta(minutes=duration)
|
||
return {"duration_minutes": duration, "auto_unban": True, "unban_at": unban_at, "until_date": int(unban_at.timestamp())}
|
||
|
||
def ban_duration_label(duration_minutes: int) -> str:
|
||
duration = int(duration_minutes or 0)
|
||
if duration <= 0:
|
||
return "永久"
|
||
if duration % 1440 == 0:
|
||
return f"{duration // 1440} 天"
|
||
if duration % 60 == 0:
|
||
return f"{duration // 60} 小时"
|
||
return f"{duration} 分钟"
|
||
|
||
@app.post("/api/send")
|
||
async def api_send(data: SendPayload):
|
||
import config
|
||
text = data.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.chat_id or config.TG_CHAT_ID)
|
||
reply_to_message_id = data.reply_to_message_id
|
||
msg = await run_tg_request(
|
||
lambda bot: bot.send_message(chat_id=chat_id, text=text, reply_to_message_id=reply_to_message_id),
|
||
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/messages/{message_id}/ban")
|
||
async def api_ban_message_user(message_id: int, data: BanPayload | None = None):
|
||
"""根据聊天记录封禁用户，不删除消息。duration_minutes=0 为永久封禁。"""
|
||
rec = await Message.get_or_none(id=message_id)
|
||
if not rec:
|
||
return JSONResponse({"ok": False, "error": "消息不存在"}, status_code=404)
|
||
if not rec.chat_id or not rec.user_id:
|
||
return JSONResponse({"ok": False, "error": "缺少 chat_id 或 user_id，无法封禁"}, status_code=400)
|
||
if await Whitelist.get_or_none(chat_id=rec.chat_id, user_id=rec.user_id, enabled=True):
|
||
return JSONResponse({"ok": False, "error": "该用户在白名单中，不允许封禁；如需处理请先移出白名单"}, status_code=409)
|
||
opts = build_ban_options((data.duration_minutes if data else 0))
|
||
try:
|
||
if opts["until_date"]:
|
||
await run_tg_request(lambda bot: bot.ban_chat_member(chat_id=rec.chat_id, user_id=rec.user_id, until_date=opts["until_date"]), connect_timeout=20.0, read_timeout=20.0, label="ban_chat_member")
|
||
else:
|
||
await run_tg_request(lambda bot: bot.ban_chat_member(chat_id=rec.chat_id, user_id=rec.user_id), 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=f"Web 手动封禁（{ban_duration_label(opts['duration_minutes'])}）",
|
||
method="web_manual_ban",
|
||
chat_id=rec.chat_id,
|
||
duration_minutes=opts["duration_minutes"],
|
||
auto_unban=opts["auto_unban"],
|
||
unban_at=opts["unban_at"],
|
||
)
|
||
await Action.create(action="BAN", user_id=rec.user_id, username=rec.username, chat_id=rec.chat_id, details={"method": "web_manual_ban", "message_id": rec.id, "duration_minutes": opts["duration_minutes"]})
|
||
await broadcast_sse("ban", json_safe({
|
||
"id": ban.id,
|
||
"chat_id": rec.chat_id,
|
||
"user_id": rec.user_id,
|
||
"username": rec.username,
|
||
"first_name": rec.first_name,
|
||
"reason": ban.reason,
|
||
"method": ban.method,
|
||
"duration_minutes": opts["duration_minutes"],
|
||
"auto_unban": opts["auto_unban"],
|
||
"unban_at": opts["unban_at"],
|
||
"time": ban.created_at,
|
||
}))
|
||
return JSONResponse(json_safe({"ok": True, "banned": True, "ban_id": ban.id, "duration_minutes": opts["duration_minutes"], "auto_unban": opts["auto_unban"]}))
|
||
except Exception as e:
|
||
return JSONResponse({"ok": False, "error": f"封禁失败: {type(e).__name__}: {e}"}, status_code=502)
|
||
|
||
|
||
@app.post("/api/messages/{message_id}/whitelist")
|
||
async def api_add_message_user_to_whitelist(message_id: int, request: Request):
|
||
"""把聊天记录对应用户加入白名单。白名单用户自动检测只删消息不封号。"""
|
||
rec = await Message.get_or_none(id=message_id)
|
||
if not rec:
|
||
return JSONResponse({"ok": False, "error": "消息不存在"}, status_code=404)
|
||
if not rec.chat_id or not rec.user_id:
|
||
return JSONResponse({"ok": False, "error": "缺少 chat_id 或 user_id，无法加入白名单"}, status_code=400)
|
||
data = await safe_json(request)
|
||
reason = str(data.get("reason") or "Web 右键加入白名单")[:500]
|
||
item = await Whitelist.get_or_none(chat_id=rec.chat_id, user_id=rec.user_id)
|
||
if item:
|
||
item.enabled = True
|
||
item.username = rec.username
|
||
item.first_name = rec.first_name
|
||
item.reason = reason
|
||
item.updated_at = now_tz()
|
||
await item.save()
|
||
else:
|
||
item = await Whitelist.create(chat_id=rec.chat_id, user_id=rec.user_id, username=rec.username, first_name=rec.first_name, reason=reason)
|
||
await Action.create(action="WHITELIST_ADD", chat_id=rec.chat_id, user_id=rec.user_id, username=rec.username, details={"message_id": rec.id, "reason": reason})
|
||
return JSONResponse(json_safe({"ok": True, "id": item.id, "chat_id": item.chat_id, "user_id": item.user_id, "username": item.username, "first_name": item.first_name, "reason": item.reason, "enabled": item.enabled}))
|
||
|
||
|
||
@app.get("/api/whitelist")
|
||
async def api_list_whitelist(chat_id: int = Query(...), limit: int = Query(default=200, ge=1, le=1000)):
|
||
qs = Whitelist.filter(enabled=True, chat_id=chat_id)
|
||
rows = await qs.order_by("-updated_at").limit(limit).values()
|
||
return JSONResponse(json_safe(list(rows)))
|
||
|
||
|
||
@app.post("/api/whitelist")
|
||
async def api_create_whitelist(request: Request):
|
||
"""手动添加当前群组白名单用户。白名单严格按群组隔离。"""
|
||
data = await safe_json(request)
|
||
try:
|
||
user_id = int(data.get("user_id") or 0)
|
||
chat_id = int(data.get("chat_id") or 0)
|
||
except (TypeError, ValueError):
|
||
return JSONResponse({"ok": False, "error": "chat_id/user_id 必须是数字"}, status_code=400)
|
||
if not chat_id:
|
||
return JSONResponse({"ok": False, "error": "chat_id 不能为空，请先选择群组"}, status_code=400)
|
||
if not user_id:
|
||
return JSONResponse({"ok": False, "error": "user_id 不能为空"}, status_code=400)
|
||
username = str(data.get("username") or "").strip().lstrip("@")[:255] or None
|
||
first_name = str(data.get("first_name") or "").strip()[:255] or None
|
||
reason = str(data.get("reason") or "Web 手动添加白名单").strip()[:500]
|
||
item = await Whitelist.get_or_none(chat_id=chat_id, user_id=user_id)
|
||
if item:
|
||
item.enabled = True
|
||
item.username = username or item.username
|
||
item.first_name = first_name or item.first_name
|
||
item.reason = reason
|
||
item.updated_at = now_tz()
|
||
await item.save()
|
||
else:
|
||
item = await Whitelist.create(chat_id=chat_id, user_id=user_id, username=username, first_name=first_name, reason=reason)
|
||
await Action.create(action="WHITELIST_ADD", chat_id=chat_id, user_id=user_id, username=username, details={"source": "web_manual", "reason": reason})
|
||
return JSONResponse(json_safe({"ok": True, "id": item.id, "chat_id": item.chat_id, "user_id": item.user_id, "username": item.username, "first_name": item.first_name, "reason": item.reason, "enabled": item.enabled}))
|
||
|
||
|
||
@app.delete("/api/whitelist/{item_id}")
|
||
async def api_delete_whitelist(item_id: int):
|
||
item = await Whitelist.get_or_none(id=item_id, enabled=True)
|
||
if not item:
|
||
return JSONResponse({"ok": False, "error": "白名单记录不存在"}, status_code=404)
|
||
item.enabled = False
|
||
item.updated_at = now_tz()
|
||
await item.save()
|
||
await Action.create(action="WHITELIST_DELETE", chat_id=item.chat_id, user_id=item.user_id, username=item.username, details={"whitelist_id": item.id})
|
||
return JSONResponse({"ok": True})
|
||
|
||
|
||
@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 safe_json(request)
|
||
ban_user = bool(data.get("ban_user") or data.get("ban"))
|
||
ban_opts = build_ban_options(int(data.get("duration_minutes") or 0))
|
||
|
||
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:
|
||
if await Whitelist.get_or_none(chat_id=rec.chat_id, user_id=rec.user_id, enabled=True):
|
||
await Action.create(action="DELETE_BAN_SKIPPED_WHITELIST", chat_id=rec.chat_id, user_id=rec.user_id, username=rec.username, details={"message_id": rec.id})
|
||
ban_user = False
|
||
if ban_user and rec.user_id:
|
||
try:
|
||
if ban_opts["until_date"]:
|
||
await run_tg_request(lambda bot: bot.ban_chat_member(chat_id=rec.chat_id, user_id=rec.user_id, until_date=ban_opts["until_date"]), connect_timeout=20.0, read_timeout=20.0, label="ban_chat_member")
|
||
else:
|
||
await run_tg_request(lambda bot: bot.ban_chat_member(chat_id=rec.chat_id, user_id=rec.user_id), 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=f"Web 删除消息时手动封禁（{ban_duration_label(ban_opts['duration_minutes'])}）",
|
||
method="web_delete_ban",
|
||
chat_id=rec.chat_id,
|
||
duration_minutes=ban_opts["duration_minutes"],
|
||
auto_unban=ban_opts["auto_unban"],
|
||
unban_at=ban_opts["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, "duration_minutes": ban_opts["duration_minutes"]})
|
||
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": ban_opts["duration_minutes"],
|
||
"auto_unban": ban_opts["auto_unban"],
|
||
"unban_at": ban_opts["unban_at"],
|
||
"chat_id": rec.chat_id,
|
||
"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, chat_id: int | None = None):
|
||
import config
|
||
target_chat_id = int(chat_id or config.TG_CHAT_ID)
|
||
try:
|
||
await run_tg_request(lambda bot: bot.unban_chat_member(chat_id=target_chat_id, user_id=user_id), label="unban_chat_member")
|
||
await Ban.filter(user_id=user_id, chat_id=target_chat_id).update(auto_unban=False, unban_at=now_tz().replace(tzinfo=None))
|
||
await Action.create(action="UNBAN", chat_id=target_chat_id, user_id=user_id, details={"method": "web"})
|
||
await broadcast_sse("unban", {"chat_id": target_chat_id, "user_id": user_id})
|
||
return JSONResponse({"ok": True, "chat_id": target_chat_id, "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())
|