feat: generate spam rules from text with ai
This commit is contained in:
ngfchl
+149
-25
@@ -10,6 +10,7 @@ import hashlib
|
||||
import ipaddress
|
||||
import socket
|
||||
import time
|
||||
import httpx
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
@@ -626,6 +627,107 @@ def sanitize_manual_rule(rule_type: str, pattern: str) -> tuple[str, str | None]
|
||||
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):
|
||||
@@ -676,7 +778,7 @@ async def api_spam_test(data: SpamTestPayload):
|
||||
"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)},
|
||||
"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:
|
||||
ai_is_spam, ai_reason = await spam_detector.detect(text)
|
||||
@@ -686,38 +788,60 @@ async def api_spam_test(data: SpamTestPayload):
|
||||
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(data: SpamRulePayload):
|
||||
rule_type = data.type.strip().lower()
|
||||
pattern = data.pattern.strip()
|
||||
score = data.score
|
||||
if rule_type not in {"keyword", "regex"}:
|
||||
return JSONResponse({"ok": False, "error": "规则类型只能是 keyword 或 regex"}, status_code=400)
|
||||
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)
|
||||
score = max(1, min(score, 200))
|
||||
try:
|
||||
stored_pattern, original_pattern = sanitize_manual_rule(rule_type, pattern)
|
||||
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)
|
||||
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.rsplit("|", 1)[0].strip() for raw in existing.splitlines() if raw.strip() and not raw.strip().startswith("#")}
|
||||
if stored_pattern in existing_patterns:
|
||||
return JSONResponse({"ok": False, "error": "规则已存在"}, status_code=409)
|
||||
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# Web 手动添加规则 {format_dt(now_tz())}{note}\n{line}\n")
|
||||
counts = spam_detector.reload_external_rules()
|
||||
await Action.create(action="SPAM_RULE_ADD", details={"type": rule_type, "pattern": stored_pattern, "score": score, "original": original_pattern})
|
||||
return JSONResponse({"ok": True, "type": rule_type, "pattern": stored_pattern, "original_pattern": original_pattern, "score": score, "rule_counts": counts})
|
||||
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")
|
||||
|
||||
Reference in New Issue
Block a user