fix: sanitize manual spam rules before saving

web/api.py

@@ -512,6 +512,28 @@ async def api_restart_bot():
     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)
+
+
 @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):
@@ -576,23 +598,24 @@ async def api_add_spam_rule(request: Request):
     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)
+    try:
+        stored_pattern, original_pattern = sanitize_manual_rule(rule_type, pattern)
+    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)
     path = spam_detector.RULES_DIR / ("spam_keywords.txt" if rule_type == "keyword" else "spam_regex.txt")
-    line = f"{pattern}|{score}"
+    line = f"{stored_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:
+    if stored_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})
+    return JSONResponse({"ok": True, "type": rule_type, "pattern": stored_pattern, "original_pattern": original_pattern, "score": score, "rule_counts": counts})
 
 
 @app.post("/api/restart")