Files
workphone-sdk/sdk/app/services/content_guard.py

150 lines
4.6 KiB
Python

"""
防封模块 — 内容防封守卫
1. 敏感词过滤(黑名单 + 正则模式匹配)
2. 内容差异化(变量替换 + 零宽字符 + 同义词 + 表情随机)
3. 消息唯一性保障
"""
import random
import re
import time
import logging
from typing import List, Optional
logger = logging.getLogger(__name__)
SENSITIVE_WORDS: List[str] = [
"加我微信", "微信号", "扫码", "二维码", "转账", "红包",
"免费领", "赚钱", "兼职", "日赚", "月入", "暴利",
"刷单", "代理", "优惠券", "点击链接", "限时", "秒杀",
"抢购", "加群", "进群", "私聊我", "代购", "佣金",
"保证赚", "零风险", "稳赚不赔", "日入过万", "躺赚",
"加Telegram", "加TG", "加WhatsApp", "加WX",
"V信", "威信", "薇信", "WX号",
"银行卡", "信用卡套现", "贷款", "网贷",
"色情", "赌博", "博彩", "棋牌",
]
SENSITIVE_PATTERNS: List[re.Pattern] = [
re.compile(r"https?://\S+"),
re.compile(r"www\.\S+"),
re.compile(r"[a-zA-Z0-9]{6,}\.(?:com|cn|net|org|xyz|top|cc|vip)\b"),
re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)"),
re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"),
re.compile(r"(?:微信|wx|WX|vx|VX)[号:\s]*[a-zA-Z0-9_]{5,20}"),
re.compile(r"(?:QQ|qq)[号:\s]*\d{5,12}"),
]
SAFE_REPLACEMENTS = {
"加我微信": "联系我",
"微信号": "联系方式",
"扫码": "查看",
"二维码": "图片",
"免费领": "获取",
"赚钱": "收益",
"兼职": "合作",
"优惠券": "福利",
"点击链接": "查看详情",
"限时": "近期",
"秒杀": "特价",
"抢购": "选购",
"V信": "联系方式",
"威信": "联系方式",
"薇信": "联系方式",
}
GREETINGS_BY_HOUR = {
range(6, 9): ["早上好", "早安", "上午好"],
range(9, 12): ["上午好", "您好"],
range(12, 14): ["中午好", "午安"],
range(14, 18): ["下午好", "您好"],
range(18, 22): ["晚上好", "您好"],
range(22, 24): ["晚安", "您好"],
range(0, 6): ["您好"],
}
EMOJI_POOL = [
"😊", "👍", "🙏", "", "💯", "🎉", "😄", "🤝",
"💪", "🌟", "👏", "😁", "🙂", "❤️", "🔥", "💐",
]
ZWS = "\u200b" # 零宽空格
ZWNJ = "\u200c" # 零宽非连接符
ZWJ = "\u200d" # 零宽连接符
INVISIBLE_CHARS = [ZWS, ZWNJ, ZWJ]
def filter_sensitive(text: str) -> str:
result = text
for word, replacement in SAFE_REPLACEMENTS.items():
result = result.replace(word, replacement)
for word in SENSITIVE_WORDS:
if word in result and word not in SAFE_REPLACEMENTS:
result = result.replace(word, "*" * len(word))
for pat in SENSITIVE_PATTERNS:
result = pat.sub("[已过滤]", result)
return result
def has_sensitive_words(text: str) -> List[str]:
found = []
for word in SENSITIVE_WORDS:
if word in text:
found.append(word)
for pat in SENSITIVE_PATTERNS:
matches = pat.findall(text)
if matches:
found.extend(matches[:3])
return found
def _get_greeting() -> str:
hour = time.localtime().tm_hour
for hr_range, greetings in GREETINGS_BY_HOUR.items():
if hour in hr_range:
return random.choice(greetings)
return "您好"
def diversify_content(
text: str,
variables: Optional[dict] = None,
add_invisible: bool = True,
add_emoji: bool = False,
) -> str:
"""
内容差异化处理,确保每条消息唯一。
Args:
text: 原始消息文本
variables: 自定义变量 {"{昵称}": "小明"} 等
add_invisible: 是否插入零宽字符增加唯一性
add_emoji: 是否在末尾随机添加表情
"""
result = text
result = result.replace("{时间}", _get_greeting())
result = result.replace("{日期}", time.strftime("%m月%d"))
if variables:
for k, v in variables.items():
result = result.replace(k, str(v))
result = filter_sensitive(result)
if add_invisible and len(result) > 2 and not result.startswith("CMD:") and not result.startswith("{") and not result.startswith("[") and not result.startswith("http"):
num_inserts = random.randint(1, min(3, len(result) // 4))
for _ in range(num_inserts):
pos = random.randint(1, len(result) - 1)
char = random.choice(INVISIBLE_CHARS)
result = result[:pos] + char + result[pos:]
if add_emoji and random.random() < 0.4:
result = result.rstrip() + random.choice(EMOJI_POOL)
return result
def batch_diversify(texts: List[str], **kwargs) -> List[str]:
return [diversify_content(t, **kwargs) for t in texts]