一、背景与意义

校园霸凌问题长期困扰着教育工作者和家长。传统的事后处理方式往往无法及时发现霸凌行为，导致受害者持续受到伤害。随着人工智能技术的发展，利用Python构建智能检测系统，实现对校园环境的实时监控与预警，成为防范霸凌行为的有效技术手段。

本方案旨在通过计算机视觉、自然语言处理和音频分析等技术，构建一套多模态的防霸凌检测系统，能够在霸凌行为发生初期进行识别并触发预警机制。

二、系统整体架构

text

┌─────────────────────────────────────────────────────┐
│                    数据采集层                         │
│  (摄像头、麦克风阵列、社交媒体API、校园论坛爬虫)        │
└─────────────────────────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────┐
│                    数据处理层                         │
│  (视频流解码、音频降噪、文本清洗、特征提取)            │
└─────────────────────────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────┐
│                    模型推理层                         │
│  (行为识别、情绪检测、关键词匹配、社交网络分析)        │
└─────────────────────────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────┐
│                    预警响应层                         │
│  (实时告警、证据保存、事件溯源、通知推送)              │
└─────────────────────────────────────────────────────┘

三、核心技术模块与Python实现

3.1 视频行为识别模块

利用姿态估计和时序动作检测，识别推搡、踢打、围堵等霸凌典型动作。

python

import cv2
import mediapipe as mp
import numpy as np
from collections import deque
import warnings
warnings.filterwarnings('ignore')

class BullyingActionDetector:
    """基于姿态的霸凌行为检测器"""
    
    def __init__(self, sequence_length=30):
        self.mp_pose = mp.solutions.pose
        self.pose = self.mp_pose.Pose(
            static_image_mode=False,
            model_complexity=1,
            min_detection_confidence=0.5,
            min_tracking_confidence=0.5
        )
        self.mp_drawing = mp.solutions.drawing_utils
        
        # 动作序列缓存
        self.sequence_buffer = deque(maxlen=sequence_length)
        self.suspicious_actions = []
        
        # 霸凌行为阈值
        self.thresholds = {
            'push': 0.7,      # 推搡动作阈值
            'kick': 0.65,     # 踢打动作阈值
            'punch': 0.75,    # 击打动作阈值
            'grab': 0.6       # 抓扯动作阈值
        }
        
    def calculate_keypoint_angles(self, landmarks):
        """计算关键点角度特征"""
        angles = {}
        
        # 定义关键点索引
        keypoints = {
            'shoulder': [11, 12],  # 肩膀
            'elbow': [13, 14],      # 手肘
            'wrist': [15, 16],      # 手腕
            'hip': [23, 24],        # 髋部
            'knee': [25, 26],       # 膝盖
            'ankle': [27, 28]       # 脚踝
        }
        
        def get_angle(a, b, c):
            """计算三点夹角"""
            a = np.array([a.x, a.y])
            b = np.array([b.x, b.y])
            c = np.array([c.x, c.y])
            
            ba = a - b
            bc = c - b
            
            cosine_angle = np.dot(ba, bc) / (np.linalg.norm(ba) * np.linalg.norm(bc) + 1e-8)
            angle = np.arccos(np.clip(cosine_angle, -1.0, 1.0))
            
            return np.degrees(angle)
        
        # 计算各关节角度
        # 左臂
        angles['left_elbow'] = get_angle(
            landmarks[mp.solutions.pose.PoseLandmark.LEFT_SHOULDER.value],
            landmarks[mp.solutions.pose.PoseLandmark.LEFT_ELBOW.value],
            landmarks[mp.solutions.pose.PoseLandmark.LEFT_WRIST.value]
        )
        
        # 右臂
        angles['right_elbow'] = get_angle(
            landmarks[mp.solutions.pose.PoseLandmark.RIGHT_SHOULDER.value],
            landmarks[mp.solutions.pose.PoseLandmark.RIGHT_ELBOW.value],
            landmarks[mp.solutions.pose.PoseLandmark.RIGHT_WRIST.value]
        )
        
        # 躯干倾斜角度
        shoulders_center = np.array([
            (landmarks[11].x + landmarks[12].x) / 2,
            (landmarks[11].y + landmarks[12].y) / 2
        ])
        hips_center = np.array([
            (landmarks[23].x + landmarks[24].x) / 2,
            (landmarks[23].y + landmarks[24].y) / 2
        ])
        
        torso_vector = shoulders_center - hips_center
        vertical_vector = np.array([0, 1])
        
        cos_torso = np.dot(torso_vector, vertical_vector) / (np.linalg.norm(torso_vector) + 1e-8)
        angles['torso_tilt'] = np.degrees(np.arccos(np.clip(cos_torso, -1.0, 1.0)))
        
        return angles
    
    def detect_aggressive_motion(self, prev_landmarks, curr_landmarks):
        """检测攻击性运动特征"""
        if prev_landmarks is None or curr_landmarks is None:
            return 0.0
        
        # 计算手腕运动速度
        left_wrist_prev = np.array([
            prev_landmarks[15].x, prev_landmarks[15].y
        ])
        left_wrist_curr = np.array([
            curr_landmarks[15].x, curr_landmarks[15].y
        ])
        
        right_wrist_prev = np.array([
            prev_landmarks[16].x, prev_landmarks[16].y
        ])
        right_wrist_curr = np.array([
            curr_landmarks[16].x, curr_landmarks[16].y
        ])
        
        left_speed = np.linalg.norm(left_wrist_curr - left_wrist_prev)
        right_speed = np.linalg.norm(right_wrist_curr - right_wrist_prev)
        
        # 快速运动可能表示攻击行为
        aggression_score = min(1.0, (left_speed + right_speed) / 0.5)
        
        return aggression_score
    
    def process_frame(self, frame):
        """处理单帧图像"""
        rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        results = self.pose.process(rgb_frame)
        
        detection_result = {
            'has_person': False,
            'aggression_score': 0.0,
            'action_type': None,
            'landmarks': None
        }
        
        if results.pose_landmarks:
            detection_result['has_person'] = True
            detection_result['landmarks'] = results.pose_landmarks.landmark
            
            # 计算姿态角度
            angles = self.calculate_keypoint_angles(results.pose_landmarks.landmark)
            
            # 检测攻击性动作
            if len(self.sequence_buffer) > 0:
                prev_landmarks = self.sequence_buffer[-1].get('landmarks')
                if prev_landmarks:
                    aggression = self.detect_aggressive_motion(
                        prev_landmarks, 
                        results.pose_landmarks.landmark
                    )
                    detection_result['aggression_score'] = aggression
                    
                    # 基于角度阈值判定动作类型
                    if aggression > self.thresholds['push']:
                        if angles['torso_tilt'] > 30:
                            detection_result['action_type'] = 'push'
                        elif angles['left_elbow'] < 45 or angles['right_elbow'] < 45:
                            detection_result['action_type'] = 'punch'
                        else:
                            detection_result['action_type'] = 'aggressive_gesture'
            
            # 缓存当前帧数据
            self.sequence_buffer.append(detection_result)
        
        return detection_result
    
    def draw_detection(self, frame, detection_result):
        """在图像上绘制检测结果"""
        if detection_result['has_person']:
            # 根据攻击性分数改变边框颜色
            if detection_result['aggression_score'] > 0.6:
                color = (0, 0, 255)  # 红色 - 高危
            elif detection_result['aggression_score'] > 0.3:
                color = (0, 165, 255)  # 橙色 - 警告
            else:
                color = (0, 255, 0)  # 绿色 - 正常
            
            cv2.putText(frame, 
                       f"Aggression: {detection_result['aggression_score']:.2f}",
                       (10, 30),
                       cv2.FONT_HERSHEY_SIMPLEX,
                       1, color, 2)
            
            if detection_result['action_type']:
                cv2.putText(frame,
                           f"Action: {detection_result['action_type']}",
                           (10, 70),
                           cv2.FONT_HERSHEY_SIMPLEX,
                           1, color, 2)
        
        return frame

3.2 音频情绪检测模块

通过分析语音中的情绪特征（音量、语速、基频等），识别恐惧、愤怒等与霸凌相关的情绪状态。

python

import librosa
import numpy as np
from scipy.signal import find_peaks
import speech_recognition as sr
from collections import deque
import threading

class AudioBullyingDetector:
    """基于音频的霸凌检测器"""
    
    def __init__(self, sample_rate=16000, frame_duration=3.0):
        self.sample_rate = sample_rate
        self.frame_duration = frame_duration
        self.recognizer = sr.Recognizer()
        
        # 情绪特征缓存
        self.emotion_buffer = deque(maxlen=10)
        
        # 霸凌相关关键词（中文）
        self.bullying_keywords = [
            "去死", "废物", "垃圾", "蠢货", "白痴", "傻逼",
            "打你", "揍你", "等着瞧", "叫你做人", "跪下",
            "废物", "弱智", "滚开", "闭嘴", "欠揍"
        ]
        
        # 痛苦表达关键词
        self.distress_keywords = [
            "救命", "放开", "不要", "疼", "停下", "求求你",
            "别打了", "害怕", "呜呜", "help", "stop"
        ]
        
    def extract_acoustic_features(self, audio_data):
        """提取声学特征"""
        features = {}
        
        # 短时能量（音量）
        energy = np.sum(audio_data**2) / len(audio_data)
        features['energy'] = energy
        
        # 过零率
        zcr = librosa.feature.zero_crossing_rate(audio_data).mean()
        features['zcr'] = zcr
        
        # 基频提取（F0）
        pitches, magnitudes = librosa.piptrack(
            y=audio_data, 
            sr=self.sample_rate
        )
        f0 = pitches[magnitudes.argmax()]
        f0 = f0[f0 > 0]
        features['f0_mean'] = np.mean(f0) if len(f0) > 0 else 0
        features['f0_std'] = np.std(f0) if len(f0) > 0 else 0
        
        # 梅尔频率倒谱系数（MFCC）
        mfcc = librosa.feature.mfcc(
            y=audio_data, 
            sr=self.sample_rate, 
            n_mfcc=13
        )
        features['mfcc_mean'] = mfcc.mean(axis=1)
        
        # 谐波噪声比
        harmonic, percussive = librosa.effects.hpss(audio_data)
        features['hnr'] = 10 * np.log10(
            (np.sum(harmonic**2) + 1e-8) / 
            (np.sum(percussive**2) + 1e-8)
        )
        
        return features
    
    def detect_emotion_from_features(self, features):
        """基于声学特征进行情绪分类"""
        scores = {
            'anger': 0.0,
            'fear': 0.0,
            'neutral': 0.0,
            'distress': 0.0
        }
        
        # 愤怒特征：高能量、高ZCR、高F0
        if features['energy'] > 0.01:
            scores['anger'] += min(1.0, features['energy'] * 50)
        if features['zcr'] > 0.1:
            scores['anger'] += min(1.0, features['zcr'] * 5)
        if features['f0_mean'] > 300:
            scores['anger'] += min(1.0, (features['f0_mean'] - 300) / 100)
        
        # 恐惧特征：中高能量、高F0、高HNR（颤抖）
        if 0.005 < features['energy'] < 0.03:
            scores['fear'] += 0.4
        if features['f0_mean'] > 280:
            scores['fear'] += min(1.0, (features['f0_mean'] - 280) / 150)
        if features['hnr'] < 20:  # 较低的谐波噪声比表示声音颤抖
            scores['fear'] += 0.3
        
        # 痛苦特征：高能量、极高F0变化
        if features['energy'] > 0.02:
            scores['distress'] += min(1.0, features['energy'] * 30)
        if features['f0_std'] > 50:
            scores['distress'] += min(1.0, features['f0_std'] / 100)
            
        return scores
    
    def keyword_matching(self, text):
        """关键词匹配"""
        if not text:
            return {'bullying_score': 0, 'distress_score': 0, 'matched_keywords': []}
        
        text_lower = text.lower()
        matched_bullying = []
        matched_distress = []
        
        for kw in self.bullying_keywords:
            if kw.lower() in text_lower:
                matched_bullying.append(kw)
        
        for kw in self.distress_keywords:
            if kw.lower() in text_lower:
                matched_distress.append(kw)
        
        return {
            'bullying_score': min(1.0, len(matched_bullying) / 3),
            'distress_score': min(1.0, len(matched_distress) / 2),
            'matched_keywords': matched_bullying + matched_distress
        }
    
    def process_audio(self, audio_file_path):
        """处理音频文件"""
        # 加载音频
        audio_data, sr = librosa.load(audio_file_path, sr=self.sample_rate)
        
        # 提取声学特征
        features = self.extract_acoustic_features(audio_data)
        
        # 情绪检测
        emotion_scores = self.detect_emotion_from_features(features)
        
        # 语音识别（可选）
        transcript = ""
        try:
            with sr.AudioFile(audio_file_path) as source:
                audio = self.recognizer.record(source)
                transcript = self.recognizer.recognize_google(audio, language='zh-CN')
        except Exception as e:
            print(f"Speech recognition failed: {e}")
        
        # 关键词匹配
        keyword_result = self.keyword_matching(transcript)
        
        # 综合评分
        total_score = (
            emotion_scores['anger'] * 0.3 +
            emotion_scores['fear'] * 0.3 +
            emotion_scores['distress'] * 0.25 +
            keyword_result['bullying_score'] * 0.15
        )
        
        result = {
            'total_bullying_score': total_score,
            'emotion_scores': emotion_scores,
            'keyword_analysis': keyword_result,
            'transcript': transcript,
            'is_bullying_suspected': total_score > 0.5
        }
        
        return result

3.3 文本霸凌检测模块

检测校园论坛、聊天记录、社交平台中的网络霸凌内容。

python

import jieba
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.svm import LinearSVC
import pickle
import re

class TextBullyingDetector:
    """文本霸凌检测器（网络霸凌）"""
    
    def __init__(self, model_path=None):
        # 霸凌类别定义
        self.bullying_types = [
            'insult',      # 辱骂
            'threat',      # 威胁
            'exclusion',   # 排挤
            'gossip',      # 谣言/八卦
            'harassment'   # 骚扰
        ]
        
        # 加载预训练模型或使用规则引擎
        if model_path:
            with open(model_path, 'rb') as f:
                self.model = pickle.load(f)
            self.use_ml = True
        else:
            self.use_ml = False
            self.init_rule_engine()
        
        # 敏感词词典
        self.sensitive_words = {
            'insult': ['傻逼', '弱智', '废物', '垃圾', '蠢货', '白痴', '脑残'],
            'threat': ['打死', '揍你', '等着', '收拾', '教训', '砍', '杀'],
            'exclusion': ['滚', '别来', '孤立', '不理', '不带', '排挤'],
            'sexual': ['操', '干', '草', '强奸', '性'],  # 性骚扰相关
        }
        
    def init_rule_engine(self):
        """初始化规则引擎"""
        self.insult_patterns = [
            r'(你|他|她|它)\s*[真好超]?\s*(傻|蠢|笨|废)\s*[逼货物]?',
            r'(白痴|智障|脑残|弱智)',
            r'(垃圾|废物|人渣|败类)'
        ]
        
        self.threat_patterns = [
            r'(打|揍|砍|杀|弄)\s*死',
            r'(等着|小心)\s*点',
            r'(收拾|教训|修理)\s*你',
        ]
        
    def preprocess_text(self, text):
        """文本预处理"""
        # 移除多余空白
        text = re.sub(r'\s+', ' ', text)
        
        # 移除特殊字符但保留中文、英文、数字
        text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9\s]', '', text)
        
        # 分词
        words = jieba.lcut(text)
        
        return words, ' '.join(words)
    
    def extract_features_rule(self, text):
        """基于规则的特征提取"""
        features = {
            'insult_score': 0.0,
            'threat_score': 0.0,
            'exclusion_score': 0.0,
            'harassment_score': 0.0,
            'total_score': 0.0
        }
        
        # 辱骂检测
        insult_count = 0
        for word in self.sensitive_words['insult']:
            if word in text:
                insult_count += text.count(word)
        features['insult_score'] = min(1.0, insult_count / 5)
        
        # 威胁检测
        threat_count = 0
        for word in self.sensitive_words['threat']:
            if word in text:
                threat_count += text.count(word)
        features['threat_score'] = min(1.0, threat_count / 3)
        
        # 排挤检测
        exclusion_count = 0
        for word in self.sensitive_words['exclusion']:
            if word in text:
                exclusion_count += text.count(word)
        features['exclusion_score'] = min(1.0, exclusion_count / 2)
        
        # 正则匹配增强
        for pattern in self.insult_patterns:
            if re.search(pattern, text):
                features['insult_score'] = min(1.0, features['insult_score'] + 0.3)
        
        for pattern in self.threat_patterns:
            if re.search(pattern, text):
                features['threat_score'] = min(1.0, features['threat_score'] + 0.4)
        
        # 综合评分（加权）
        features['total_score'] = (
            features['insult_score'] * 0.4 +
            features['threat_score'] * 0.35 +
            features['exclusion_score'] * 0.25
        )
        
        return features
    
    def detect_bullying(self, text, user_id=None, timestamp=None):
        """检测文本是否包含霸凌内容"""
        words, cleaned_text = self.preprocess_text(text)
        
        if self.use_ml:
            # 使用机器学习模型
            features = self.vectorizer.transform([cleaned_text])
            prediction = self.model.predict(features)[0]
            probability = self.model.predict_proba(features)[0]
            
            result = {
                'is_bullying': prediction == 1,
                'confidence': float(max(probability)),
                'bullying_type': self.bullying_types[np.argmax(probability)] if prediction == 1 else None,
                'details': {
                    'text': text,
                    'cleaned_text': cleaned_text,
                    'user_id': user_id,
                    'timestamp': timestamp
                }
            }
        else:
            # 使用规则引擎
            features = self.extract_features_rule(cleaned_text)
            result = {
                'is_bullying': features['total_score'] > 0.4,
                'confidence': features['total_score'],
                'bullying_type': self.get_dominant_type(features),
                'details': {
                    'text': text,
                    'cleaned_text': cleaned_text,
                    'user_id': user_id,
                    'timestamp': timestamp,
                    'scores': features
                }
            }
        
        return result
    
    def get_dominant_type(self, features):
        """获取主要的霸凌类型"""
        scores = {
            'insult': features['insult_score'],
            'threat': features['threat_score'],
            'exclusion': features['exclusion_score']
        }
        return max(scores, key=scores.get) if max(scores.values()) > 0 else None
    
    def batch_detect(self, texts):
        """批量检测"""
        results = []
        for text in texts:
            results.append(self.detect_bullying(text))
        return results

3.4 集成检测与预警系统

将各模块整合，实现综合判断和实时预警。

python

import asyncio
import json
import time
from datetime import datetime
from typing import Dict, List
import smtplib
from email.mime.text import MIMEText
import requests

class BullyingDetectionSystem:
    """霸凌检测集成系统"""
    
    def __init__(self, config: Dict):
        self.config = config
        
        # 初始化各检测模块
        self.video_detector = BullyingActionDetector()
        self.audio_detector = AudioBullyingDetector()
        self.text_detector = TextBullyingDetector()
        
        # 预警配置
        self.alert_threshold = config.get('alert_threshold', 0.6)
        self.alert_cooldown = config.get('alert_cooldown', 60)  # 冷却时间（秒）
        
        # 事件存储
        self.events = []
        self.last_alert_time = {}
        
        # 通知方式
        self.enable_email = config.get('enable_email', False)
        self.enable_sms = config.get('enable_sms', False)
        self.enable_webhook = config.get('enable_webhook', False)
        
    def fuse_results(self, video_result=None, audio_result=None, text_result=None):
        """多模态结果融合"""
        scores = []
        weights = {
            'video': 0.4,
            'audio': 0.35,
            'text': 0.25
        }
        
        fused_result = {
            'total_score': 0.0,
            'risk_level': 'low',
            'evidence': {},
            'recommendation': ''
        }
        
        if video_result and video_result.get('aggression_score'):
            video_score = video_result['aggression_score']
            scores.append(video_score * weights['video'])
            fused_result['evidence']['video'] = video_result
            
        if audio_result and audio_result.get('total_bullying_score'):
            audio_score = audio_result['total_bullying_score']
            scores.append(audio_score * weights['audio'])
            fused_result['evidence']['audio'] = audio_result
            
        if text_result and text_result.get('confidence'):
            text_score = text_result['confidence']
            scores.append(text_score * weights['text'])
            fused_result['evidence']['text'] = text_result
            
        if scores:
            fused_result['total_score'] = sum(scores)
            
            # 风险等级判定
            if fused_result['total_score'] >= 0.7:
                fused_result['risk_level'] = 'high'
                fused_result['recommendation'] = '立即干预，通知安保人员和辅导员'
            elif fused_result['total_score'] >= 0.4:
                fused_result['risk_level'] = 'medium'
                fused_result['recommendation'] = '密切关注，记录事件并通知班主任'
            else:
                fused_result['risk_level'] = 'low'
                fused_result['recommendation'] = '常规监控'
        
        return fused_result
    
    def create_event_record(self, fused_result, location, timestamp):
        """创建事件记录"""
        event = {
            'event_id': f"BLY_{int(timestamp)}_{hash(str(fused_result)) % 10000:04d}",
            'timestamp': datetime.fromtimestamp(timestamp).isoformat(),
            'location': location,
            'risk_level': fused_result['risk_level'],
            'total_score': fused_result['total_score'],
            'evidence_summary': {
                'has_video': 'video' in fused_result['evidence'],
                'has_audio': 'audio' in fused_result['evidence'],
                'has_text': 'text' in fused_result['evidence']
            },
            'recommendation': fused_result['recommendation'],
            'status': 'pending'
        }
        return event
    
    async def send_alert(self, event):
        """发送预警通知"""
        alert_key = f"{event['location']}_{event['risk_level']}"
        current_time = time.time()
        
        # 检查冷却时间
        if alert_key in self.last_alert_time:
            if current_time - self.last_alert_time[alert_key] < self.alert_cooldown:
                return
        
        self.last_alert_time[alert_key] = current_time
        
        # 邮件通知
        if self.enable_email:
            await self.send_email_alert(event)
        
        # Webhook回调
        if self.enable_webhook:
            await self.send_webhook(event)
        
        # 短信通知（使用第三方API）
        if self.enable_sms:
            await self.send_sms_alert(event)
        
        # 控制台输出
        print(f"\n⚠️ 霸凌预警 [{event['risk_level'].upper()}]")
        print(f"   位置: {event['location']}")
        print(f"   时间: {event['timestamp']}")
        print(f"   置信度: {event['total_score']:.2f}")
        print(f"   建议: {event['recommendation']}")
    
    async def send_email_alert(self, event):
        """发送邮件预警"""
        try:
            msg = MIMEText(
                f"""
                <html>
                <body>
                <h2>校园霸凌检测预警</h2>
                <p><strong>风险等级:</strong> {event['risk_level']}</p>
                <p><strong>发生位置:</strong> {event['location']}</p>
                <p><strong>发生时间:</strong> {event['timestamp']}</p>
                <p><strong>置信度:</strong> {event['total_score']:.2f}</p>
                <p><strong>处理建议:</strong> {event['recommendation']}</p>
                <p><strong>事件ID:</strong> {event['event_id']}</p>
                </body>
                </html>
                """,
                "html", "utf-8"
            )
            
            msg['Subject'] = f"[霸凌预警] {event['risk_level']} - {event['location']}"
            msg['From'] = self.config['email_sender']
            msg['To'] = self.config['email_receiver']
            
            # 异步发送邮件（示例使用同步方式）
            with smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) as server:
                server.starttls()
                server.login(self.config['email_sender'], self.config['email_password'])
                server.send_message(msg)
                
        except Exception as e:
            print(f"邮件发送失败: {e}")
    
    async def send_webhook(self, event):
        """发送Webhook回调"""
        try:
            payload = {
                'event_type': 'bullying_alert',
                'data': event
            }
            
            response = requests.post(
                self.config['webhook_url'],
                json=payload,
                timeout=5
            )
            
            if response.status_code != 200:
                print(f"Webhook发送失败: {response.status_code}")
                
        except Exception as e:
            print(f"Webhook异常: {e}")
    
    async def send_sms_alert(self, event):
        """发送短信预警（示例使用阿里云短信）"""
        # 实际使用时需接入具体的短信服务商API
        # 这里仅作示例
        print(f"[模拟短信] 发送至 {self.config['sms_receiver']}: "
              f"霸凌预警 {event['risk_level']} @ {event['location']}")
    
    async def process_detection(self, video_frame=None, audio_path=None, 
                                text_content=None, location="unknown"):
        """处理单次检测"""
        timestamp = time.time()
        
        # 执行各模块检测
        video_result = None
        audio_result = None
        text_result = None
        
        if video_frame is not None:
            video_result = self.video_detector.process_frame(video_frame)
            
        if audio_path is not None:
            audio_result = self.audio_detector.process_audio(audio_path)
            
        if text_content is not None:
            text_result = self.text_detector.detect_bullying(text_content)
        
        # 融合结果
        fused_result = self.fuse_results(video_result, audio_result, text_result)
        
        # 创建事件记录
        if fused_result['total_score'] >= self.alert_threshold:
            event = self.create_event_record(fused_result, location, timestamp)
            self.events.append(event)
            
            # 发送预警
            await self.send_alert(event)
            
            return event
        
        return None
    
    def get_statistics(self):
        """获取检测统计信息"""
        if not self.events:
            return {"total_events": 0}
        
        high_risk = sum(1 for e in self.events if e['risk_level'] == 'high')
        medium_risk = sum(1 for e in self.events if e['risk_level'] == 'medium')
        
        return {
            'total_events': len(self.events),
            'high_risk_count': high_risk,
            'medium_risk_count': medium_risk,
            'high_risk_rate': high_risk / len(self.events) if self.events else 0,
            'latest_event': self.events[-1] if self.events else None
        }

# 使用示例
async def main():
    """主函数示例"""
    config = {
        'alert_threshold': 0.5,
        'alert_cooldown': 30,
        'enable_email': True,
        'enable_sms': False,
        'enable_webhook': True,
        'email_sender': 'alert@school.com',
        'email_receiver': 'teacher@school.com',
        'smtp_server': 'smtp.gmail.com',
        'smtp_port': 587,
        'email_password': 'your_password',
        'webhook_url': 'https://your-server.com/api/alerts',
        'sms_receiver': '+861234567890'
    }
    
    system = BullyingDetectionSystem(config)
    
    # 模拟文本检测
    text_result = await system.process_detection(
        text_content="你这个废物，再让我看到你就打死你！",
        location="campus_forum"
    )
    
    # 获取统计信息
    stats = system.get_statistics()
    print(f"\n统计信息: {json.dumps(stats, indent=2, ensure_ascii=False)}")

if __name__ == "__main__":
    asyncio.run(main())

四、部署方案

4.1 硬件部署架构

部署位置	设备类型	配置要求	数量
教室/走廊	边缘计算节点	NVIDIA Jetson Orin / 树莓派4B+Google Coral	每楼层2-3台
服务器机房	中央服务器	GPU服务器(RTX 3090) + 大容量存储	1-2台
安保中心	监控终端	普通PC + 显示器	1-2台

4.2 软件依赖

bash

# 创建虚拟环境
python -m venv bullying_env
source bullying_env/bin/activate  # Linux/Mac
# bullying_env\Scripts\activate  # Windows

# 安装依赖
pip install opencv-python==4.8.1
pip install mediapipe==0.10.7
pip install numpy==1.24.3
pip install librosa==0.10.1
pip install scikit-learn==1.3.0
pip install jieba==0.42.1
pip install speechrecognition==3.10.0
pip install aiohttp==3.9.0
pip install pandas==2.1.0

4.3 Docker部署

dockerfile

FROM python:3.10-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    libgl1-mesa-glx \
    libglib2.0-0 \
    libsm6 \
    libxext6 \
    libxrender-dev \
    libgomp1 \
    ffmpeg \
    && rm -rf /var/lib/apt/lists/*

# 复制项目文件
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# 启动服务
CMD ["python", "main.py"]

4.4 API服务（FastAPI）

python

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
import uvicorn

app = FastAPI(title="校园霸凌检测API")

# 初始化检测系统
detection_system = BullyingDetectionSystem(config)

@app.post("/detect/video")
async def detect_video(video: UploadFile = File(...)):
    """视频检测接口"""
    # 处理逻辑
    pass

@app.post("/detect/audio")
async def detect_audio(audio: UploadFile = File(...)):
    """音频检测接口"""
    # 处理逻辑
    pass

@app.post("/detect/text")
async def detect_text(text: str):
    """文本检测接口"""
    result = await detection_system.process_detection(text_content=text)
    return JSONResponse(content=result)

@app.get("/