检测来自 YouTube 的重复视频

Question

考虑到我的 M.tech 项目 我想知道是否有任何算法可以检测来自 youtube 的重复视频。 例如（这里是两个视频的链接）：

random user upload

upload by official channel

其中第二个是官方视频，T系列有它的版权。

youtube 是否正式采取措施从 youtube 中删除重复视频？ 不仅视频，还存在重复的 youtube 频道。 有时原始视频的观看次数少于盗版。

所以，在搜索时发现this （见pdf第[49]页）

我从给定链接中学到了什么

使用了原创与版权侵权视频检测分类器。 给定一个查询，首先检索前k个搜索结果。然后使用三个参数对视频进行分类

1. 订阅人数
2. 用户个人资料
3. 用户名流行度

并根据这些参数，按照链接中的说明识别原始视频。

编辑 1：

基本上有两个不同的目标

• 用上述方法识别原始视频
• 消除重复视频

显然，识别原始视频比找出所有重复视频更容易。 所以我宁愿先找出原始视频。

到目前为止我能想到的方法 提高准确性：

• 我们可以先用上面的方法找到原始视频
• 然后使用该视频最流行的公开帧（可能是多个）在谷歌图像上搜索。因此，此方法会检索 google 图片搜索结果中的重复视频列表。

在得到这些重复的视频后，我们可以再次逐帧检查并达到满意的程度（是的，检索到的视频是原始视频的“完全或几乎”重复副本）

这种方法行得通吗？ 如果没有，是否有更好的算法来改进给定的方法？

如果我无法清楚地解释我的方法，请在 cmets 部分写下。 我很快会添加更多细节。

Answer 1

我最近联合了一个small tool for that purpose。它仍在进行中，但通常非常准确。这个想法是简单地比较视频中心亮度最大值之间的时间。因此，它应该适用于不同的分辨率、帧速率和视频旋转。 ffmpeg 用于解码，imageio 作为 python 的桥梁，numpy/scipy 用于最大值计算和一些 k-最近邻库（annoy、cyflann、hnsw）进行比较。

目前它根本没有完善，所以你应该知道一点 python 来运行它或简单地复制这个想法。

Answer 2

我也有同样的问题..所以自己写了一个程序..

问题是我有各种格式和分辨率的视频。所以需要对每个视频帧进行哈希并进行比较。

https://github.com/gklc811/duplicate_video_finder

您只需更改顶部的目录即可。

from os import path, walk, makedirs, rename
from time import clock
from imagehash import average_hash
from PIL import Image
from cv2 import VideoCapture, CAP_PROP_FRAME_COUNT, CAP_PROP_FRAME_WIDTH, CAP_PROP_FRAME_HEIGHT, CAP_PROP_FPS
from json import dump, load
from multiprocessing import Pool, cpu_count

input_vid_dir = r'C:\Users\gokul\Documents\data\\'
json_dir = r'C:\Users\gokul\Documents\db\\'
analyzed_dir = r'C:\Users\gokul\Documents\analyzed\\'
duplicate_dir = r'C:\Users\gokul\Documents\duplicate\\'

if not path.exists(json_dir):
    makedirs(json_dir)

if not path.exists(analyzed_dir):
    makedirs(analyzed_dir)

if not path.exists(duplicate_dir):
    makedirs(duplicate_dir)


def write_to_json(filename, data):
    file_full_path = json_dir + filename + ".json"
    with open(file_full_path, 'w') as file_pointer:
        dump(data, file_pointer)
    return


def video_to_json(filename):
    file_full_path = input_vid_dir + filename
    start = clock()
    size = round(path.getsize(file_full_path) / 1024 / 1024, 2)
    video_pointer = VideoCapture(file_full_path)
    frame_count = int(VideoCapture.get(video_pointer, int(CAP_PROP_FRAME_COUNT)))
    width = int(VideoCapture.get(video_pointer, int(CAP_PROP_FRAME_WIDTH)))
    height = int(VideoCapture.get(video_pointer, int(CAP_PROP_FRAME_HEIGHT)))
    fps = int(VideoCapture.get(video_pointer, int(CAP_PROP_FPS)))
    success, image = video_pointer.read()
    video_hash = {}
    while success:
        frame_hash = average_hash(Image.fromarray(image))
        video_hash[str(frame_hash)] = filename
        success, image = video_pointer.read()
    stop = clock()
    time_taken = stop - start
    print("Time taken for ", file_full_path, " is : ", time_taken)
    data_dict = dict()
    data_dict['size'] = size
    data_dict['time_taken'] = time_taken
    data_dict['fps'] = fps
    data_dict['height'] = height
    data_dict['width'] = width
    data_dict['frame_count'] = frame_count
    data_dict['filename'] = filename
    data_dict['video_hash'] = video_hash
    write_to_json(filename, data_dict)
    return


def multiprocess_video_to_json():
    files = next(walk(input_vid_dir))[2]
    processes = cpu_count()
    print(processes)
    pool = Pool(processes)
    start = clock()
    pool.starmap_async(video_to_json, zip(files))
    pool.close()
    pool.join()
    stop = clock()
    print("Time Taken : ", stop - start)


def key_with_max_val(d):
    max_value = 0
    required_key = ""
    for k in d:
        if d[k] > max_value:
            max_value = d[k]
            required_key = k
    return required_key


def duplicate_analyzer():
    files = next(walk(json_dir))[2]
    data_dict = {}
    for file in files:
        filename = json_dir + file
        with open(filename) as f:
            data = load(f)
        video_hash = data['video_hash']
        count = 0
        duplicate_file_dict = dict()
        for key in video_hash:
            count += 1
            if key in data_dict:
                if data_dict[key] in duplicate_file_dict:
                    duplicate_file_dict[data_dict[key]] = duplicate_file_dict[data_dict[key]] + 1
                else:
                    duplicate_file_dict[data_dict[key]] = 1
            else:
                data_dict[key] = video_hash[key]
        if duplicate_file_dict:
            duplicate_file = key_with_max_val(duplicate_file_dict)
            duplicate_percentage = ((duplicate_file_dict[duplicate_file] / count) * 100)
            if duplicate_percentage > 50:
                file = file[:-5]
                print(file, " is dup of ", duplicate_file)
                src = analyzed_dir + file
                tgt = duplicate_dir + file
                if path.exists(src):
                    rename(src, tgt)
                # else:
                #     print("File already moved")


def mv_analyzed_file():
    files = next(walk(json_dir))[2]
    for filename in files:
        filename = filename[:-5]
        src = input_vid_dir + filename
        tgt = analyzed_dir + filename
        if path.exists(src):
            rename(src, tgt)
        # else:
        #     print("File already moved")


if __name__ == '__main__':
    mv_analyzed_file()
    multiprocess_video_to_json()
    mv_analyzed_file()
    duplicate_analyzer()