基于多模态大模型的智能视频审核员

一、项目结构

video_processing/
├── config.py        # 配置信息（如API密钥、路径等）
├── database/        # 数据库相关文件
│   ├── models.py    # ORM模型定义
│   └── utils.py     # 数据库连接与初始化工具
├── processors/      # 各个处理模块
│   ├── video_processor.py  # 视频处理（提取关键帧、转换音频）
│   ├── text_analyzer.py    # 文本分析模块
│   ├── image_analyzer.py   # 图像分析模块
│   └── status_updater.py   # 状态汇总与清理
├── utils/           # 辅助工具函数
│   ├── file_utils.py  # 文件操作相关
│   └── log_utils.py   # 日志记录
└── main.py          # 主程序入口，启动所有线程

---

二、数据库表设计（SQLite示例）

# database/models.py
from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class VideoTask(Base):
    __tablename__ = 'video_tasks'

    id = Column(Integer, primary_key=True)
    video_path = Column(String(255), nullable=False)  # 原始视频路径
    created_at = Column(DateTime, nullable=False)
    processed_text = Column(Boolean, default=False)
    text_result = Column(String(255))                 # 文本分析结果
    analyzed_images = Column(Boolean, default=False)
    image_violations = Column(Integer, default=0)     # 违规图片数量
    final_status = Column(String(50), nullable=True)  # 最终状态（如"安全"/"违规"）

---

三、核心代码实现

1. 配置文件 (config.py)

VIDEO_DIR = 'videos/'
TEXT_OUTPUT_DIR = 'text_files/'
IMAGE_OUTPUT_DIR = 'image_frames/'
DB_PATH = 'sqlite:///video_tasks.db'

# API配置（假设使用OpenAI API）
API_KEY = "your_openai_api_key"
MODEL_NAME = "gpt-3.5-turbo"

---

2. 数据库初始化 (database/utils.py)

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from .models import Base

def init_db():
    engine = create_engine(config.DB_PATH)
    Base.metadata.create_all(engine)
    return sessionmaker(bind=engine)()

---

3. 视频处理模块 (processors/video_processor.py)

import cv2
import os
import time
from config import *

class VideoProcessor:
    def __init__(self):
        self.db_session = init_db()

    def process_video(self, video_id):
        task = self.db_session.query(VideoTask).get(video_id)
        if not task:
            return

        # 提取关键帧（每秒1张，假设24fps）
        vidcap = cv2.VideoCapture(task.video_path)
        success,img = vidcap.read()
        count = 0
        while success:
            if count % 24 == 0:  # 每隔24帧保存一张
                img_path = os.path.join(IMAGE_OUTPUT_DIR, f"{video_id}_{count}.jpg")
                cv2.imwrite(img_path, img)
                task.analyzed_images +=1
            success,img = vidcap.read()
            count +=1

        # 转换音频为文本（伪代码）
        text_content = self._audio_to_text(task.video_path)

        with open(os.path.join(TEXT_OUTPUT_DIR, f"{video_id}.txt"), 'w') as f:
            f.write(text_content)

        task.processed_text = True
        self.db_session.commit()

    def _audio_to_text(self, video_path):
        # 这里需要实现音频转文本的逻辑，比如使用SpeechRecognition库
        return "Sample text content from audio"

---

4. 文本分析模块 (processors/text_analyzer.py)

import openai
from config import *

class TextAnalyzer:
    def analyze_text(self, task_id):
        with open(os.path.join(TEXT_OUTPUT_DIR, f"{task_id}.txt"), 'r') as f:
            text = f.read()

        # 调用API进行内容分析（示例逻辑）
        response = openai.Completion.create(
            engine=MODEL_NAME,
            prompt=f"检测以下文本是否包含违规内容：\n\n{text}",
            max_tokens=10
        )

        result = "安全"
        if "违规" in response.choices[0].text:
            result = "存在违规内容"

        # 更新数据库状态
        task = session.query(VideoTask).get(task_id)
        task.text_result = result
        task.processed_text = True
        session.commit()

---

5. 图像分析模块 (processors/image_analyzer.py)

import torch
from torchvision import transforms
from PIL import Image

class ImageAnalyzer:
    def __init__(self):
        self.model = torch.hub.load('ultralytics/yolov5', 'yolov5s')  # 示例模型

    def analyze_images(self, task_id):
        image_dir = os.path.join(IMAGE_OUTPUT_DIR, str(task_id))

        violations = 0
        for img_name in os.listdir(image_dir):
            img_path = os.path.join(image_dir, img_name)

            results = self.model(img_path)  # 运行检测模型
            if any('违规类别' in pred for pred in results.pandas().xyxy[0].name):
                violations +=1

        task = session.query(VideoTask).get(task_id)
        task.image_violations = violations
        task.analyzed_images = True
        session.commit()

---

6. 主程序 (main.py)

from processors import VideoProcessor, TextAnalyzer, ImageAnalyzer
import threading

def main():
    vp = VideoProcessor()
    ta = TextAnalyzer()
    ia = ImageAnalyzer()

    # 启动监听线程（简化版）
    while True:
        tasks = session.query(VideoTask).filter_by(processed=False).all()  # 假设未处理任务
        for task in tasks:
            vp_thread = threading.Thread(target=vp.process_video, args=(task.id,))
            ta_thread = threading.Thread(target=ta.analyze_text, args=(task.id,))
            ia_thread = threading.Thread(target=ia.analyze_images, args=(task.id,))

            vp_thread.start()
            ta_thread.start()
            ia_thread.start()

if __name__ == "__main__":
    main()

Donate