feat: add spam rule tester page

web/api.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
+import re
 import hmac
 import hashlib
 import ipaddress
@@ -22,6 +23,7 @@ from config import WEB_AUTH_ENABLED, WEB_AUTH_PASSWORD, WEB_AUTH_USERNAME, WEB_S
 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
 
@@ -504,6 +506,64 @@ async def api_restart_bot():
     return JSONResponse({"ok": True, "message": "TG Bot 正在重启"})
 
 
+@app.post("/api/spam-test")
+async def api_spam_test(request: Request):
+    data = await request.json()
+    text = str(data.get("text") or "").strip()
+    use_ai = bool(data.get("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)},
+    }
+    if use_ai:
+        ai_is_spam, ai_reason = await spam_detector.detect(text)
+        result.update({"ai_is_spam": ai_is_spam, "ai_reason": ai_reason, "final_is_spam": ai_is_spam})
+    else:
+        result["final_is_spam"] = bool(decision) if decision is not None else None
+    return JSONResponse(result)
+
+
+@app.post("/api/spam-rules")
+async def api_add_spam_rule(request: Request):
+    data = await request.json()
+    rule_type = str(data.get("type") or "keyword").strip().lower()
+    pattern = str(data.get("pattern") or "").strip()
+    try:
+        score = int(data.get("score") or 60)
+    except (TypeError, ValueError):
+        score = 60
+    if rule_type not in {"keyword", "regex"}:
+        return JSONResponse({"ok": False, "error": "规则类型只能是 keyword 或 regex"}, status_code=400)
+    if not pattern or len(pattern) > 500:
+        return JSONResponse({"ok": False, "error": "规则内容不能为空，且不能超过 500 字"}, status_code=400)
+    score = max(1, min(score, 200))
+    if rule_type == "regex":
+        try:
+            re.compile(pattern)
+        except re.error as e:
+            return JSONResponse({"ok": False, "error": f"正则无效：{e}"}, status_code=400)
+    path = spam_detector.RULES_DIR / ("spam_keywords.txt" if rule_type == "keyword" else "spam_regex.txt")
+    line = f"{pattern}|{score}"
+    existing = path.read_text(encoding="utf-8") if path.exists() else ""
+    existing_patterns = {raw.rsplit("|", 1)[0].strip() for raw in existing.splitlines() if raw.strip() and not raw.strip().startswith("#")}
+    if pattern in existing_patterns:
+        return JSONResponse({"ok": False, "error": "规则已存在"}, status_code=409)
+    with path.open("a", encoding="utf-8") as f:
+        if existing and not existing.endswith("\n"):
+            f.write("\n")
+        f.write(f"\n# Web 手动添加规则 {format_dt(now_tz())}\n{line}\n")
+    counts = spam_detector.reload_external_rules()
+    return JSONResponse({"ok": True, "type": rule_type, "pattern": pattern, "score": score, "rule_counts": counts})
+
+
 @app.post("/api/restart")
 async def api_restart():
     """重启服务：返回响应后退出当前进程，由 Docker restart 策略拉起。"""