similar_video.py (8343B)
1 import os 2 import json 3 from pathlib import Path 4 import cv2 5 import numpy as np 6 from PIL import Image 7 import imagehash 8 import ctypes 9 from ctypes import wintypes 10 11 def get_video_properties(video_path, root_path): 12 cap = cv2.VideoCapture(str(video_path)) 13 if not cap.isOpened(): 14 return {} 15 16 width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) 17 height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) 18 fps = cap.get(cv2.CAP_PROP_FPS) 19 total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) 20 21 duration = total_frames / fps if fps > 0 else 0 22 file_size_mb = os.path.getsize(video_path) / (1024 * 1024) 23 24 cap.release() 25 26 try: 27 rel_path = str(video_path.relative_to(root_path)) 28 except ValueError: 29 rel_path = str(video_path) 30 31 return { 32 'rel_path': rel_path, 33 'resolution': f"{width}x{height}", 34 'duration': f"{duration:.2f}s", 35 'fps': f"{fps:.2f}", 36 'size': f"{file_size_mb:.2f} MB" 37 } 38 39 def calculate_video_phash(video_path, sample_frames=10): 40 cap = cv2.VideoCapture(str(video_path)) 41 if not cap.isOpened(): 42 return None 43 44 total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) 45 if total_frames <= 0: 46 cap.release() 47 return None 48 49 frame_indices = np.linspace(0, total_frames - 1, sample_frames, dtype=int) 50 hashes = [] 51 52 for idx in frame_indices: 53 cap.set(cv2.CAP_PROP_POS_FRAMES, idx) 54 ret, frame = cap.read() 55 if not ret: 56 continue 57 58 frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) 59 pil_img = Image.fromarray(frame_rgb) 60 61 try: 62 hash_val = imagehash.phash(pil_img) 63 hashes.append(hash_val) 64 except Exception: 65 continue 66 67 cap.release() 68 return hashes 69 70 def load_or_calculate_hashes(root_dir, video_paths, sample_frames=10): 71 cache_path = Path(root_dir) / ".video_phash_cache.json" 72 cache_data = {} 73 74 if cache_path.exists(): 75 try: 76 with open(cache_path, "r", encoding="utf-8") as f: 77 cache_data = json.load(f) 78 except Exception: 79 pass 80 81 video_hashes = {} 82 updated_cache = {} 83 cache_changed = False 84 85 for path in video_paths: 86 try: 87 stat = path.stat() 88 mtime = stat.st_mtime 89 size = stat.st_size 90 except Exception: 91 continue 92 93 str_path = str(path) 94 95 if str_path in cache_data: 96 cached = cache_data[str_path] 97 if cached.get('mtime') == mtime and cached.get('size') == size: 98 if cached['hashes']: 99 video_hashes[path] = [imagehash.hex_to_hash(h) for h in cached['hashes']] 100 updated_cache[str_path] = cached 101 continue 102 103 hashes = calculate_video_phash(path, sample_frames) 104 105 hash_strings = [str(h) for h in hashes] if hashes else [] 106 if hashes: 107 video_hashes[path] = hashes 108 109 updated_cache[str_path] = { 110 'mtime': mtime, 111 'size': size, 112 'hashes': hash_strings 113 } 114 cache_changed = True 115 116 if len(cache_data) != len(updated_cache): 117 cache_changed = True 118 119 if cache_changed: 120 try: 121 with open(cache_path, "w", encoding="utf-8") as f: 122 json.dump(updated_cache, f, indent=2, ensure_ascii=False) 123 except Exception: 124 pass 125 126 return video_hashes 127 128 def open_in_explorer(file_paths): 129 if not file_paths: 130 return 131 132 shell32 = ctypes.windll.shell32 133 ole32 = ctypes.windll.ole32 134 135 ole32.CoInitialize(None) 136 137 dir_path = str(file_paths[0].parent) 138 139 # 関数の引数と戻り値の型を厳密に定義 140 ILCreateFromPathW = shell32.ILCreateFromPathW 141 ILCreateFromPathW.restype = ctypes.c_void_p 142 ILCreateFromPathW.argtypes = [wintypes.LPCWSTR] 143 144 ILFree = shell32.ILFree 145 ILFree.argtypes = [ctypes.c_void_p] 146 147 SHOpenFolderAndSelectItems = shell32.SHOpenFolderAndSelectItems 148 SHOpenFolderAndSelectItems.restype = ctypes.HRESULT 149 SHOpenFolderAndSelectItems.argtypes = [ctypes.c_void_p, wintypes.UINT, ctypes.c_void_p, wintypes.DWORD] 150 151 dir_pidl = ILCreateFromPathW(dir_path) 152 if not dir_pidl: 153 ole32.CoUninitialize() 154 return 155 156 file_pidls = [] 157 for p in file_paths: 158 pidl = ILCreateFromPathW(str(p)) 159 if pidl: 160 file_pidls.append(pidl) 161 162 if file_pidls: 163 # c_void_p の配列として正確に確保 164 pidl_array = (ctypes.c_void_p * len(file_pidls))(*file_pidls) 165 SHOpenFolderAndSelectItems(dir_pidl, len(file_pidls), ctypes.byref(pidl_array), 0) 166 167 for pidl in file_pidls: 168 ILFree(pidl) 169 170 ILFree(dir_pidl) 171 ole32.CoUninitialize() 172 173 def group_similar_videos(video_hashes, threshold=15): 174 paths = list(video_hashes.keys()) 175 num_videos = len(paths) 176 177 parent_map = {p: p for p in paths} 178 179 def find(p): 180 if parent_map[p] == p: 181 return p 182 parent_map[p] = find(parent_map[p]) 183 return parent_map[p] 184 185 def union(p1, p2): 186 root1 = find(p1) 187 root2 = find(p2) 188 if root1 != root2: 189 parent_map[root2] = root1 190 191 for i in range(num_videos): 192 for j in range(i + 1, num_videos): 193 path_a = paths[i] 194 path_b = paths[j] 195 196 hashes_a = video_hashes[path_a] 197 hashes_b = video_hashes[path_b] 198 199 min_len = min(len(hashes_a), len(hashes_b)) 200 if min_len == 0: 201 continue 202 203 distances = [hashes_a[k] - hashes_b[k] for k in range(min_len)] 204 avg_distance = sum(distances) / min_len 205 206 if avg_distance <= threshold: 207 union(path_a, path_b) 208 209 groups = {} 210 for p in paths: 211 root = find(p) 212 if root not in groups: 213 groups[root] = [] 214 groups[root].append(p) 215 216 return [g for g in groups.values() if len(g) > 1] 217 218 def display_table(properties_list): 219 headers = ["Relative Path", "Resolution", "Duration", "FPS", "File Size"] 220 col_widths = [max(len(str(p.get(h, ''))) for p in properties_list) for h in ['rel_path', 'resolution', 'duration', 'fps', 'size']] 221 col_widths = [max(w, len(h)) for w, h in zip(col_widths, headers)] 222 223 header_str = " | ".join(f"{h:<{w}}" for h, w in zip(headers, col_widths)) 224 print("-" * len(header_str)) 225 print(header_str) 226 print("-" * len(header_str)) 227 228 for props in properties_list: 229 row_str = " | ".join(f"{str(props.get(k, '')):<{w}}" for k, w in zip(['rel_path', 'resolution', 'duration', 'fps', 'size'], col_widths)) 230 print(row_str) 231 print("-" * len(header_str)) 232 233 if __name__ == '__main__': 234 target_folder = r"E:\Pictures\他人作品\ホロ" 235 root_path = Path(target_folder) 236 237 video_extensions = {'.mp4', '.avi', '.mkv', '.mov', '.flv', '.wmv'} 238 video_paths = [] 239 240 for p in root_path.rglob('*'): 241 if p.suffix.lower() in video_extensions: 242 video_paths.append(p) 243 244 video_hashes = load_or_calculate_hashes(root_path, video_paths, sample_frames=10) 245 groups = group_similar_videos(video_hashes, threshold=12) 246 247 if not groups: 248 print("類似動画は検出されませんでした。") 249 else: 250 for idx, group in enumerate(groups, 1): 251 print(f"\n[Match Group {idx}/{len(groups)}] 一致数: {len(group)}") 252 253 props_list = [get_video_properties(p, root_path) for p in group] 254 display_table(props_list) 255 256 open_in_explorer(group) 257 258 if idx < len(groups): 259 input("\n次のグループを表示するには Enter キーを押してください...") 260 else: 261 print("\nすべてのグループの処理が完了しました。")