diff --git a/web/api.py b/web/api.py index 8a35f11..ec77ca8 100644 --- a/web/api.py +++ b/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")