diff --git a/scripts/batch_media_manager.py b/scripts/batch_media_manager.py index babea18..d6d9769 100644 --- a/scripts/batch_media_manager.py +++ b/scripts/batch_media_manager.py @@ -10,7 +10,7 @@ import queue import shutil import subprocess import threading -import datetime # 导入 datetime +import datetime from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass from pathlib import Path @@ -65,7 +65,7 @@ TRANSLATIONS: Dict[str, Dict[str, str]] = { "option_live_pair": "检测 Live Photo 配对并移动缺失伴侣的视频", "option_convert_motion": "将三星 Motion Photo 转换为 Live Photo", "option_cleanup_mp4": "清除带有 MotionPhoto_Data 标记的多余 MP4", - "option_archive_originals": "归档原始三星照片 (转换为zip)", # 新增 + "option_archive_originals": "归档原始三星照片 (转换为zip)", "label_workers": "并发线程数:", "button_start": "开始处理", "progress_status": "进度: {done} / {total}", @@ -84,20 +84,21 @@ TRANSLATIONS: Dict[str, Dict[str, str]] = { "log_task_entry": "[{task}] {result}", "log_finished": "全部任务已完成。", "log_processing_done": "处理完毕。", - "log_archiving": "正在归档原始照片...", # 新增 - "log_archive_complete": "归档完成: {zip_path}", # 新增 - "log_archive_failed": "归档失败: {error}", # 新增 - "task_orphan_video": "孤立视频: {path}", + "log_archiving": "正在归档原始照片...", + "log_archive_complete": "归档完成: {zip_path}", + "log_archive_failed": "归档失败: {error}", + "task_orphan_video": "孤立视频: {path}", # (此翻译保留，但逻辑已更改) "task_motion_convert": "Motion Photo 转 Live Photo: {path}", - "task_cleanup_mp4": "清理 MP4: {path}", - "result_moved": "已移动到 {dest}", + "task_cleanup_mp4": "清理 MP4: {path}", # (此翻译保留，但逻辑已更改) + "task_process_video": "处理视频: {path}", # 新增 + "result_moved": "已移动到 {dest} (孤立LivePhoto)", # 修改 "result_skipped_exists": "已存在转换结果，跳过", "result_converted": "已转换为 {heic} / {mov}", "result_cleanup_removed": "已删除 (MotionPhoto_Data)", - "result_cleanup_kept": "保留", + "result_cleanup_kept": "保留 (非目标视频)", # 修改 "result_cleanup_skipped": "跳过 (缺少 exiftool)", "result_failed": "失败: {error}", - "result_archived": "已归档原始文件", # 新增 + "result_archived": "已归档原始文件", }, "en": { "app_title": "Live / Motion Photo Batch Manager", @@ -110,7 +111,7 @@ TRANSLATIONS: Dict[str, Dict[str, str]] = { "option_live_pair": "Check Live Photo pairs and move orphaned videos", "option_convert_motion": "Convert Samsung Motion Photos to Live Photo", "option_cleanup_mp4": "Remove redundant MP4 files tagged MotionPhoto_Data", - "option_archive_originals": "Archive original Samsung photos (as zip)", # New + "option_archive_originals": "Archive original Samsung photos (as zip)", "label_workers": "Worker threads:", "button_start": "Start", "progress_status": "Progress: {done} / {total}", @@ -129,20 +130,21 @@ TRANSLATIONS: Dict[str, Dict[str, str]] = { "log_task_entry": "[{task}] {result}", "log_finished": "All tasks completed.", "log_processing_done": "Processing finished.", - "log_archiving": "Archiving original photos...", # New - "log_archive_complete": "Archive complete: {zip_path}", # New - "log_archive_failed": "Archive failed: {error}", # New - "task_orphan_video": "Orphan video: {path}", + "log_archiving": "Archiving original photos...", + "log_archive_complete": "Archive complete: {zip_path}", + "log_archive_failed": "Archive failed: {error}", + "task_orphan_video": "Orphan video: {path}", # (Retained, logic changed) "task_motion_convert": "Motion Photo → Live Photo: {path}", - "task_cleanup_mp4": "Cleanup MP4: {path}", - "result_moved": "Moved to {dest}", + "task_cleanup_mp4": "Cleanup MP4: {path}", # (Retained, logic changed) + "task_process_video": "Processing video: {path}", # New + "result_moved": "Moved to {dest} (Orphaned LivePhoto)", # Modified "result_skipped_exists": "Skipped (outputs already exist)", "result_converted": "Converted to {heic} / {mov}", "result_cleanup_removed": "Removed (MotionPhoto_Data)", - "result_cleanup_kept": "Kept", + "result_cleanup_kept": "Kept (Not a target video)", # Modified "result_cleanup_skipped": "Skipped (exiftool unavailable)", "result_failed": "Failed: {error}", - "result_archived": "Archived original file", # New + "result_archived": "Archived original file", }, } @@ -153,13 +155,13 @@ class ProcessingOptions: handle_live_pairs: bool convert_motion: bool remove_motion_mp4: bool - archive_originals: bool # 新增 + archive_originals: bool max_workers: int @dataclass class Task: - func: Callable[[], dict] # 修改：返回 dict 而不是 str + func: Callable[[], dict] description: str @@ -172,6 +174,7 @@ def _is_motion_photo(heic_path: Path) -> bool: def _move_orphan_video(mov_path: Path, root_dir: Path, tr: Callable[[str], str]) -> dict: + """辅助函数：移动孤立视频并返回结果字典""" error_dir = root_dir / "错误的LivePhoto视频" error_dir.mkdir(parents=True, exist_ok=True) @@ -203,29 +206,68 @@ def _convert_motion_photo(heic_path: Path, inject_mov_tag: bool, tr: Callable[[s return { "status": "CONVERTED", "message": tr("result_converted", heic=heic_out.name, mov=mov_out.name), - "original_path": heic_path, # 返回原始路径以供归档 + "original_path": heic_path, } - -def _cleanup_motion_mp4(mp4_path: Path, has_exiftool: bool, tr: Callable[[str], str]) -> dict: +# --- 新增：统一的视频处理任务 --- +def _process_video( + video_path: Path, + has_still_pair: bool, + options: ProcessingOptions, + has_exiftool: bool, + tr: Callable[[str], str], +) -> dict: + """ + 根据元数据和用户选项处理单个视频文件（MP4, MOV 等）。 + 实现新的优先级逻辑。 + """ + + # 如果两个选项都没选，或者没有 exiftool，我们只能执行旧的“孤立”逻辑 if not has_exiftool: - return {"status": "SKIPPED_EXIF", "message": tr("result_cleanup_skipped")} + if options.handle_live_pairs and not has_still_pair: + # 这是没有 exiftool 时能做的唯一判断 + return _move_orphan_video(video_path, options.root_dir, tr) + elif options.remove_motion_mp4: + return {"status": "SKIPPED_EXIF", "message": tr("result_cleanup_skipped")} + else: + return {"status": "SKIPPED", "message": tr("result_cleanup_kept")} - cmd = [ - "exiftool", - "-s", - "-s", - "-s", - "-EmbeddedVideoType", - str(mp4_path), - ] - result = subprocess.run(cmd, capture_output=True, text=True, check=False) - output = result.stdout.strip() + # --- 有 Exiftool，执行新逻辑 --- + try: + cmd = [ + "exiftool", + "-s", + "-s", + "-s", + "-EmbeddedVideoType", + "-ContentIdentifier", + str(video_path), + ] + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + output = result.stdout.strip() + + has_motion_data = "MotionPhoto_Data" in output + has_content_id = "ContentIdentifier" in output # 注意：exiftool 找到标签就会返回 True - if "MotionPhoto_Data" in output: - mp4_path.unlink(missing_ok=True) - return {"status": "REMOVED", "message": tr("result_cleanup_removed")} - return {"status": "KEPT", "message": tr("result_cleanup_kept")} + # 优先级 1：检查是否为多余的 MotionPhoto_Data (如果选项开启) + if options.remove_motion_mp4 and has_motion_data: + video_path.unlink(missing_ok=True) + return {"status": "REMOVED", "message": tr("result_cleanup_removed")} + + # 优先级 2：检查是否为孤立的 Live Photo (如果选项开启) + if ( + options.handle_live_pairs and + not has_still_pair and + has_content_id and + not has_motion_data # 关键：确保它不是 MotionPhoto + ): + return _move_orphan_video(video_path, options.root_dir, tr) + + # 其他所有情况（例如：有配对的视频、无元数据的视频等） + return {"status": "KEPT", "message": tr("result_cleanup_kept")} + + except Exception as e: + return {"status": "FAILED", "message": tr("result_failed", error=e)} def _gather_tasks( @@ -237,39 +279,26 @@ def _gather_tasks( tasks: List[Task] = [] base_map: Dict[Tuple[Path, str], Dict[str, object]] = {} heic_files: List[Path] = [] - mp4_files: List[Path] = [] + all_video_files: List[Path] = [] # 修改：收集所有视频 log(tr("log_scanning")) for path in options.root_dir.rglob("*"): if not path.is_file(): continue suffix = path.suffix.lower() + key = (path.parent, path.stem.lower()) + entry = base_map.setdefault(key, {}) if suffix in STILL_SUFFIXES: heic_files.append(path) - key = (path.parent, path.stem.lower()) - entry = base_map.setdefault(key, {}) entry["still"] = path elif suffix in VIDEO_SUFFIXES: - key = (path.parent, path.stem.lower()) - entry = base_map.setdefault(key, {}) + all_video_files.append(path) # 修改 entry.setdefault("videos", []).append(path) # type: ignore - if suffix == ".mp4": - mp4_files.append(path) - if options.handle_live_pairs: - for entry in base_map.values(): - has_still = "still" in entry - videos = entry.get("videos") or [] - if not has_still and videos: - for video_path in videos: # type: ignore - tasks.append( - Task( - func=lambda p=video_path, tr=tr: _move_orphan_video(p, options.root_dir, tr), - description=tr("task_orphan_video", path=str(video_path)), - ) - ) + # --- 任务分派逻辑修改 --- + # 1. (不变) 三星 Motion Photo 转换任务 if options.convert_motion: inject_mov_tag = has_exiftool for heic_path in heic_files: @@ -288,12 +317,18 @@ def _gather_tasks( ) ) - if options.remove_motion_mp4: - for mp4_path in mp4_files: + # 2. (新增) 统一的视频处理任务 (替代旧的 MP4 清理和孤立视频移动) + if options.handle_live_pairs or options.remove_motion_mp4: + for video_path in all_video_files: + key = (video_path.parent, video_path.stem.lower()) + has_still = "still" in base_map.get(key, {}) + tasks.append( Task( - func=lambda p=mp4_path, tr=tr: _cleanup_motion_mp4(p, has_exiftool, tr), - description=tr("task_cleanup_mp4", path=str(mp4_path)), + func=lambda p=video_path, hs=has_still: _process_video( + p, hs, options, has_exiftool, tr + ), + description=tr("task_process_video", path=str(video_path)), ) ) @@ -316,7 +351,7 @@ def _worker(options: ProcessingOptions, message_queue: queue.Queue, language: st status_text = tr("general_yes") if has_exiftool else tr("general_no") emit_log(tr("log_exiftool", status=status_text)) - files_to_archive: List[Path] = [] # 新增：待归档文件列表 + files_to_archive: List[Path] = [] try: tasks = _gather_tasks(options, has_exiftool, emit_log, tr) @@ -337,10 +372,9 @@ def _worker(options: ProcessingOptions, message_queue: queue.Queue, language: st for future in as_completed(future_map): task = future_map[future] try: - outcome = future.result() # 现在是 dict + outcome = future.result() result_text = outcome.get("message", "done") - # 新增：检查是否需要归档 if ( options.archive_originals and outcome.get("status") == "CONVERTED" and @@ -355,7 +389,6 @@ def _worker(options: ProcessingOptions, message_queue: queue.Queue, language: st completed += 1 message_queue.put(("progress", completed)) - # --- 新增：处理归档 --- if options.archive_originals and files_to_archive: emit_log(tr("log_archiving")) archive_dir_name = "三星原始照片" @@ -363,22 +396,14 @@ def _worker(options: ProcessingOptions, message_queue: queue.Queue, language: st try: for original_path in files_to_archive: - # 计算相对路径 relative_path = original_path.relative_to(options.root_dir) - # 创建目标路径 archive_dest = archive_base_dir / relative_path - - # 确保父目录存在 archive_dest.parent.mkdir(parents=True, exist_ok=True) - - # 移动文件 (使用 str() 确保兼容性) shutil.move(str(original_path), str(archive_dest)) - # 打包 Zip datestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") zip_name_base = f"{archive_dir_name}_{datestamp}" - # shutil.make_archive 会自动添加 .zip 扩展名 zip_path_str = shutil.make_archive( base_name=str(options.root_dir / zip_name_base), format="zip", @@ -386,14 +411,12 @@ def _worker(options: ProcessingOptions, message_queue: queue.Queue, language: st base_dir=archive_dir_name, ) - # 可选：删除原始归档文件夹 (暂时保留) - # shutil.rmtree(archive_base_dir) + # shutil.rmtree(archive_base_dir) # 归档后删除原文件夹 emit_log(tr("log_archive_complete", zip_path=Path(zip_path_str).name)) except Exception as e: emit_log(tr("log_archive_failed", error=e)) - # --- 归档结束 --- emit_log(tr("log_finished")) except Exception as e: @@ -418,7 +441,7 @@ class BatchManagerGUI: self.handle_live_var = tk.BooleanVar(value=True) self.convert_motion_var = tk.BooleanVar(value=True) self.remove_mp4_var = tk.BooleanVar(value=True) - self.archive_originals_var = tk.BooleanVar(value=True) # 新增 + self.archive_originals_var = tk.BooleanVar(value=True) default_workers = max(1, min(4, (os.cpu_count() or 4))) self.worker_count_var = tk.IntVar(value=default_workers) @@ -431,7 +454,6 @@ class BatchManagerGUI: def run(self) -> None: self.root.mainloop() - # 辅助函数，用于获取翻译后的文本 def _t(self, key: str) -> str: lang_map = TRANSLATIONS.get(self.lang.get(), TRANSLATIONS["zh"]) return lang_map.get(key, TRANSLATIONS["zh"].get(key, key)) @@ -466,13 +488,11 @@ class BatchManagerGUI: text=self._t("option_cleanup_mp4"), variable=self.remove_mp4_var, ).pack(anchor="w", padx=8, pady=2) - # --- 新增归档复选框 --- ttk.Checkbutton( options_frame, text=self._t("option_archive_originals"), variable=self.archive_originals_var, ).pack(anchor="w", padx=8, pady=2) - # --- 结束 --- worker_frame = ttk.Frame(self.root) worker_frame.pack(fill="x", **padding) @@ -508,11 +528,9 @@ class BatchManagerGUI: self._update_ui_text() def _update_ui_text(self) -> None: - """更新界面上所有可翻译的文本 (此脚本中未实现语言切换，但为未来做准备)""" self.root.title(self._t("app_title")) pass - def _choose_directory(self) -> None: selected = filedialog.askdirectory() if selected: @@ -533,7 +551,7 @@ class BatchManagerGUI: handle_live_pairs=self.handle_live_var.get(), convert_motion=self.convert_motion_var.get(), remove_motion_mp4=self.remove_mp4_var.get(), - archive_originals=self.archive_originals_var.get(), # 新增 + archive_originals=self.archive_originals_var.get(), max_workers=self.worker_count_var.get(), )