python --视频去重工具
同级目录下新建dup_video
- 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()
-