基于Python实现视频去重

基本原理

一款基于Python语言的视频去重复程序，它可以根据视频的特征参数，将重复的视频剔除，以减少视频的存储空间。它的基本原理是：首先利用Python语言对视频文件进行解析，提取视频的特征参数，如帧率、码率等；然后根据特征参数，生成视频的哈希值；最后，将每个视频的哈希值进行对比，如果哈希值相同，则表示视频内容相同，可以将其中一个视频剔除，以节省存储空间。

实现方法

基于Python实现视频去重小工具，需要使用Python语言的一些第三方库，如OpenCV、PIL等。实现的基本步骤如下：

1、首先，使用OpenCV库对视频文件进行解析，提取视频的特征参数；

2、然后，使用PIL库将视频的每一帧转换为灰度图片，计算每一帧图片的哈希值；

3、最后，将每个视频的哈希值进行对比，如果哈希值相同，则表示视频内容相同，可以将其中一个视频剔除。

• 示例代码

import cv2
from PIL import Image
 
# 读取视频文件
video_file = "video.mp4"
cap = cv2.VideoCapture(video_file)
 
# 提取视频的特征参数
frame_rate = cap.get(cv2.CAP_PROP_FPS)
frame_width = cap.get(cv2.CAP_PROP_FRAME_WIDTH)
frame_height = cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
 
# 计算每一帧图片的哈希值
while True:
    # 读取每一帧图片
    success, frame = cap.read()
    if not success:
        break
    # 将图片转换为灰度图
    gray_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    # 计算图片的哈希值
    image = Image.fromarray(gray_frame)
    hash_value = image.hash()
    # 将哈希值进行对比，如果哈希值相同，则表示视频内容相同
    if hash_value == other_hash_value:
        # 剔除重复视频
        pass
 
# 释放视频文件
cap.release()

其它视频去重code

同级目录下新建dup_video

# -*- coding:utf-8 -*-
import json
import os
import shutil

import cv2
import imagehash
from PIL import Image
from loguru import logger
from PySimpleGUI import popup_get_folder


class VideoDuplicate(object):
    '''
    返回整个视频的图片指纹列表
    从1秒开始，每3秒抽帧，计算一张图像指纹
    '''

    def __init__(self):
        self._over_length_video: list = []
        self._no_video: list = []

    def _video_hash(self, video_path) -> list:
        '''
        @param video_path -> 视频绝对路径;
        '''
        hash_arr = []
        cap = cv2.VideoCapture(video_path)  ##打开视频文件
        logger.info(f'开始抽帧【{video_path}】')

        n_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))  # 视频的帧数
        logger.warning(f'视频帧数:{n_frames}')

        fps = cap.get(cv2.CAP_PROP_FPS)  # 视频的帧率
        logger.warning(f'视频帧率:{fps}')

        dur = n_frames / fps * 1000  # 视频大致总长度
        cap_set = 1000
        logger.warning(f'视频大约总长:{dur / 1000}')
        if dur // 1000 > 11:
            logger.error(f'视频时长超出规定范围【6~10】;当前时长:【{dur // 1000}】;跳过该视频;')
            self._over_length_video.append(video_path)
            return []

        while cap_set < dur:  # 从3秒开始，每60秒抽帧，计算图像指纹。总长度-3s，是因为有的时候计算出来的长度不准。
            cap.set(cv2.CAP_PROP_POS_MSEC, cap_set)
            logger.debug(f'开始提取:【{cap_set // 1000}】/s的图片;')
            # 返回该时间点的，图像(numpy数组)，及读取是否成功
            success, image_np = cap.read()
            if success:
                img = Image.fromarray(cv2.cvtColor(image_np, cv2.COLOR_BGR2RGB))  # 转成cv图像格式
                h = str(imagehash.dhash(img))
                logger.success(f'【{cap_set}/s图像指纹:【{h}】')
                hash_arr.append(h)  # 图像指纹
            else:
                logger.error(str(cap_set / 1000))
            cap_set += 1000 * 2
        cap.release()  # 释放视频
        return hash_arr

    def start(self, base_dir):
        '''
        @param base_dir -> 主文件路径;
        '''
        data: list = []
        for video in os.listdir(base_dir):
            logger.debug(f'-' * 80)
            name, ext = os.path.splitext(video)
            if ext not in ('.mp4', '.MP4'):
                logger.error(f'视频文件格式不符;【{video}】;执行跳过;')
                continue

            abs_video_path = os.path.join(base_dir, video)
            video_hash_list = self._video_hash(abs_video_path)
            if video_hash_list:
                data.append({'video_abs_path': abs_video_path, 'hash': video_hash_list})

        self._write_log(data)
        return data

    @staticmethod
    def _write_log(data: list) -> None:
        '''视频哈希后的值写入日志文件'''
        with open(f'log.txt', 'w+', encoding='utf-8') as f:
            f.write(json.dumps(data))

    def __call__(self, base_dir, *args, **kwargs):
        self.start(base_dir)
        logger.debug(f'-----------------------------------开始比对关键帧差值感知余弦算法-----------------------------')

        with open('log.txt') as f:
            data = json.loads(f.read())
            for i in range(0, len(data) - 1):
                for j in range(i + 1, len(data)):
                    if data[i]['hash'] == data[j]['hash']:
                        _, filename = os.path.split(data[i]['video_abs_path'])
                        logger.error(f'移动文件:【{filename}】')
                        shutil.move(
                            os.path.join(base_dir, filename),
                            os.path.join(os.path.join(os.getcwd(), 'dup_video'), filename)
                        )
        logger.warning('---------------------超长视频----------------------')
        for i in self._over_length_video:
            _, name = os.path.split(i)
            logger.error(name)


def main():
    path = popup_get_folder('请选择[视频去重]文件夹')
    v = VideoDuplicate()
    v(path)


if __name__ == '__main__':
    main()