EAI-deepface/video_handler.py

import cv2
from deepface import DeepFace
import time
from moviepy.editor import VideoFileClip, concatenate_videoclips


def process_video(video_url):
    # 打开视频流
    cap = cv2.VideoCapture(video_url)
    # 检查视频是否成功打开
    if not cap.isOpened():
        print("Error opening video stream from URL")
        return {}, None

    # 获取视频的帧率、宽度和高度
    fps = cap.get(cv2.CAP_PROP_FPS)
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

    # 获取当前时间戳
    timestamp = int(time.time())

    # 定义视频编码器和创建视频写入对象，使用MP4格式
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    # 在输出文件名中加入时间戳
    output_path = f'annotated_video_{timestamp}.mp4'
    out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))

    # 初始化情绪字典
    emotion_dict = {}

    frame_count = 0
    current_emotion = None

    # 循环读取视频帧
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break

        if frame_count % 20 == 0:
            try:
                # 使用DeepFace进行情绪识别
                result = DeepFace.analyze(frame, actions=['emotion'], enforce_detection=False)
                current_emotion = result[0]['dominant_emotion']

                # 更新情绪字典
                if current_emotion in emotion_dict:
                    emotion_dict[current_emotion] += 1
                else:
                    emotion_dict[current_emotion] = 1

            except Exception as e:
                print(f"Error processing frame: {e}")

        if current_emotion is not None:
            # 在帧上显示情绪信息
            cv2.putText(frame, f"Emotion: {current_emotion}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)

        # 写入标注后的帧到输出视频
        out.write(frame)
        frame_count += 1

    # 释放视频捕获对象和写入对象
    cap.release()
    out.release()

    return emotion_dict, output_path


def merge_videos(video_paths):
    clips = []
    try:
        clips = [VideoFileClip(p) for p in video_paths]
        final = concatenate_videoclips(clips)
        output = f'merged_video_{int(time.time())}.mp4'
        final.write_videofile(output, threads=4, codec='libx264')
        return output
    finally:
        for clip in clips:
            clip.close()
        time.sleep(0.5)  # 额外等待