脚本怎样批量拦截违规内容

wen 实用脚本 2026-06-25 31

本文目录导读：

脚本怎样批量拦截违规内容

基础方案：基于关键词的黑名单拦截（适合简单敏感词）
加强版：正则表达式与模糊匹配（支持变体/谐音/拆字）
批量处理文件/数据库/API流（高性能方案）
专业级方案：调用云端AI审核API（支持图片/视频/文本）
防绕过技巧与增强建议
完整脚本模板（命令行工具）
注意事项

的需求，通常需要结合关键词匹配、正则表达式、AI模型或第三方API来实现，以下提供几种不同场景下的脚本方案（以Python为例）：

基础方案：基于关键词的黑名单拦截（适合简单敏感词）

import re
class ContentFilter:
    def __init__(self, keyword_file="blacklist.txt"):
        # 从文件加载黑名单关键词（每行一个）
        with open(keyword_file, "r", encoding="utf-8") as f:
            self.blacklist = [line.strip() for line in f if line.strip()]
    def contains_illegal(self, text):
        """检查文本是否包含任何黑名单关键词"""
        for keyword in self.blacklist:
            if keyword in text:
                return True, keyword
        return False, None
    def replace_with_star(self, text, placeholder="***"):
        """将违禁词替换为占位符"""
        for keyword in self.blacklist:
            text = text.replace(keyword, placeholder)
        return text
# 使用示例
filter = ContentFilter()
text = "这是包含敏感词A的文本"
if filter.contains_illegal(text):
    print("已拦截：包含违禁词")
else:
    print("内容安全")

加强版：正则表达式与模糊匹配（支持变体/谐音/拆字）

import re
class AdvancedFilter:
    def __init__(self):
        # 定义高级规则：支持正则、通配符、前缀后缀
        self.patterns = [
            # 1. 常规敏感词
            "违禁词A|违禁词B",
            # 2. 带通配符（如"赌.*博"匹配"赌博"、"赌大博"）
            "赌.*博|毒.*品",
            # 3. 谐音替换（如"fa轮功"变体"fa论功"）
            "法[轮伦]功|fa[轮伦]功",
            # 4. 拆分词（如"自 杀"、"自.S"）
            "自\\s*[杀刹]|自\\.*[杀刹]"
        ]
        self.combined_pattern = re.compile("|".join(self.patterns), re.IGNORECASE)
    def is_illegal(self, text):
        return bool(self.combined_pattern.search(text))
    def find_all_matches(self, text):
        return self.combined_pattern.findall(text)
# 使用示例
filter = AdvancedFilter()
texts = ["这是赌大博的内容", "法轮功是XX", "自 杀方法"]
for t in texts:
    if filter.is_illegal(t):
        print(f"拦截: {t}")

批量处理文件/数据库/API流（高性能方案）

import multiprocessing
import pandas as pd  # 用于CSV/Excel处理
def batch_check_text(text):
    """单条检测函数（可被多进程调用）"""
    # 这里调用你的过滤逻辑（如上面的类）
    filter = ContentFilter()  # 注意：多进程下需在函数内初始化
    return filter.contains_illegal(text)
# 方案1：读取CSV批量检查
def batch_check_csv(input_file, output_file, text_column="content"):
    df = pd.read_csv(input_file)
    # 多进程加速（建议使用进程池）
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(batch_check_text, df[text_column].tolist())
    df["is_illegal"] = results
    df.to_csv(output_file, index=False)
    print(f"处理完成，违规条数: {sum(results)}")
# 方案2：读取数据库（以MySQL为例）
def batch_check_mysql(host, user, password, db, table, field="content"):
    import pymysql
    conn = pymysql.connect(host=host, user=user, password=password, db=db)
    cursor = conn.cursor()
    cursor.execute(f"SELECT id, {field} FROM {table}")
    rows = cursor.fetchall()
    # 批量判断
    illegal_ids = []
    for id, text in rows:
        if filter.contains_illegal(text)[0]:
            illegal_ids.append(id)
    # 标记或删除（谨慎操作）
    if illegal_ids:
        cursor.execute(f"UPDATE {table} SET status='illegal' WHERE id IN ({','.join(map(str, illegal_ids))})")
        conn.commit()
    conn.close()
# 方案3：实时API流（如Webhook）
def api_middleware(request):
    content = request.json.get("content")
    if filter.contains_illegal(content):
        return {"code": 403, "msg": "内容违规"}, 403
    return {"code": 200, "msg": "通过"}

专业级方案：调用云端AI审核API（支持图片/视频/文本）

import requests
import json
class CloudCheckService:
    def __init__(self, api_key, secret_key):
        self.api_url = "https://api.example.com/v1/content/check"  # 替换为实际API地址
        self.api_key = api_key
    def check_text(self, text):
        payload = {
            "api_key": self.api_key,
            "content": text,
            "type": "text"  # 或 "image"、"video"
        }
        response = requests.post(self.api_url, json=payload)
        result = response.json()
        # 假设接口返回 {"code":0, "label":"normal/ad/porn/violence"}
        if result.get("label") != "normal":
            return False, result.get("label")
        return True, None
# 使用示例（腾讯云、阿里云、百度AI、网易易盾等）
# cloud = CloudCheckService("your_api_key", "your_secret")
# safe, label = cloud.check_text("测试文本")

防绕过技巧与增强建议

分词对比：使用jieba分词，将文本切词后再匹配，防止“赌..博”类变体。
拼音映射：将文本转为拼音后匹配（如“falu”映射为“法轮”）。
Unicode归一化：统一全半角、大小写（如“ｆａｌｕｎ”还原）。
图片OCR：如需拦截图片中的文字,使用Tesseract或云OCR服务。
白名单机制：对部分合法内容（如医学文章中的“毒品”一词）放行。
性能优化：使用AC自动机（Aho-Corasick）算法实现多模式高效匹配（适合百万级关键词）。

完整脚本模板（命令行工具）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-过滤工具"""
import argparse
import sys
def main():
    parser = argparse.ArgumentParser(description="批量拦截违规内容")
    parser.add_argument("-i", "--input", required=True, help="输入文件（CSV/TXT）")
    parser.add_argument("-o", "--output", help="输出文件（默认输出到控制台）")
    parser.add_argument("-k", "--keywords", required=True, help="关键词文件（每行一个）")
    parser.add_argument("-c", "--column", default="content", help="CSV中的文本列名")
    args = parser.parse_args()
    filter = ContentFilter(args.keywords)
    if args.input.endswith(".csv"):
        import pandas as pd
        df = pd.read_csv(args.input)
        results = df[args.column].apply(lambda x: filter.contains_illegal(x)[0])
        df["illegal"] = results
        if args.output:
            df.to_csv(args.output, index=False)
        else:
            print(df[df["illegal"] == True])
    else:
        with open(args.input, "r", encoding="utf-8") as f:
            for line in f:
                text = line.strip()
                if text:
                    illegal, word = filter.contains_illegal(text)
                    if illegal:
                        print(f"[拦截] {text} -> 违规词: {word}")
if __name__ == "__main__":
    main()

注意事项

误杀风险：建议对拦截结果进行人工复核，或设置分级拦截（如“禁止”和“警告”）。
数据安全：敏感词库和API密钥不要硬编码在脚本中,应使用环境变量或加密存储。
法律合规过滤应遵守当地法律法规,注意个人隐私保护和言论自由的平衡。
性能监控：批量处理大数据时，建议添加进度条（tqdm库）和错误日志。

根据你的实际需求（如处理的是文本、图片还是音视频？数据量级？实时性要求？），可以选择上述方案组合或定制开发，如果需要特定场景（如微信公众号、论坛评论、文件服务器）的详细实现,请补充说明。