"""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())