""" VideoDetection - Main Entrypoint. Phần mềm nhận diện trùng lặp video bản quyền với V-JEPA 2 & NeuralFP. Chạy mặc định giao diện Desktop GUI Windows 10-11, hỗ trợ CLI batching. """ from __future__ import annotations import argparse import json import sys from pathlib import Path # Đảm bảo UTF-8 an toàn trên Windows console (tránh lỗi cp1252) if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") if hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8") # Đảm bảo import được các module nội bộ theo đường dẫn tương đối sys.path.insert(0, str(Path(__file__).resolve().parent)) from src.core.paths import TARGET_DIR, clean_temp_dir from src.gui.desktop_app import launch_gui from src.pipeline import VideoDetectionPipeline def run_cli() -> None: parser = argparse.ArgumentParser( description="VideoDetection CLI - Nhận diện trùng lặp video với V-JEPA 2 & NeuralFP" ) subparsers = parser.add_subparsers(dest="command", help="Lệnh thực thi") # Lệnh sync subparsers.add_parser("sync", help="Tự động đồng bộ và nạp đặc trưng cho video mới trong Data/ và Target/") # Lệnh compare p_compare = subparsers.add_parser("compare", help="Đối soát 1 video target với cả kho hoặc các video tham chiếu cụ thể") p_compare.add_argument("target_video", type=str, help="Tên file hoặc đường dẫn video target") p_compare.add_argument("--refs", nargs="*", default=None, help="Danh sách video tham chiếu cụ thể (bỏ trống để so sánh cả kho)") p_compare.add_argument("--json", action="store_true", help="Xuất kết quả dạng JSON") # Lệnh index p_index = subparsers.add_parser("index", help="Nạp video gốc của chủ sở hữu vào thư mục Data/") p_index.add_argument("video_path", type=str, help="Đường dẫn file video gốc") # Lệnh scan p_scan = subparsers.add_parser("scan", help="Quét video mục tiêu để tìm đoạn trùng lặp") p_scan.add_argument("--target", type=str, default=None, help="Đường dẫn video cụ thể (nếu bỏ trống sẽ quét toàn bộ thư mục Target/)") p_scan.add_argument("--json", action="store_true", help="Xuất kết quả dạng JSON") # Lệnh clean subparsers.add_parser("clean", help="Dọn dẹp thư mục file tạm Temp/") # Lệnh download (YouTube) p_dl = subparsers.add_parser("download", help="Tải video hoặc 1 đoạn chỉ định từ YouTube vào Target/ hoặc Data/") p_dl.add_argument("url_or_id", type=str, help="ID video (11 ký tự) hoặc Link YouTube đầy đủ/rút gọn") p_dl.add_argument( "--section", type=str, default="00:00:00-01:00:00", help="Khoảng thời gian cần tải (mặc định: '00:00:00-01:00:00' - 1h đầu; nhập 'all' để tải toàn bộ)", ) p_dl.add_argument( "--dest", type=str, choices=["target", "data"], default="target", help="Thư mục lưu video: 'target' (mặc định) hoặc 'data'", ) p_dl.add_argument( "--sync", action="store_true", help="Tự động đồng bộ và trích xuất đặc trưng V-JEPA 2 & NeuralFP ngay sau khi tải", ) # Lệnh gui subparsers.add_parser("gui", help="Khởi động giao diện Desktop GUI Windows 10-11") args = parser.parse_args() if args.command == "gui" or args.command is None: launch_gui() return if args.command == "clean": removed = clean_temp_dir() print(f"[OK] Đã dọn dẹp {removed} file tạm trong thư mục Temp/.") return pipeline = VideoDetectionPipeline() if args.command == "download": print(f"[*] Đang chuẩn bị tải từ YouTube: {args.url_or_id}") sec_desc = "1 Giờ đầu tiên" if args.section in ("00:00:00-01:00:00", "default") else (args.section if args.section != "all" else "Toàn bộ video") print(f" Phân đoạn : {args.section} ({sec_desc})") print(f" Thư mục lưu: {args.dest.upper()}/") result = pipeline.download_youtube_video( url_or_id=args.url_or_id, section=args.section, dest=args.dest, auto_sync=args.sync, progress_callback=lambda p, m: print(f" [{p}%] {m}"), ) if result.success and result.file_path: print("\n" + "=" * 80) print("[THÀNH CÔNG] ĐÃ TẢI VIDEO YOUTUBE THÀNH CÔNG!") print(f" Tiêu đề : {result.title}") print(f" YouTube ID : {result.video_id}") print(f" File video : {result.file_path.name}") print(f" Đường dẫn : {result.file_path}") print(f" Thư mục : {result.dest_dir}") print("=" * 80) else: print(f"\n[THẤT BẠI] Lỗi khi tải video: {result.error_message}") sys.exit(1) return if args.command == "sync": print("[*] Đang đồng bộ kho video gốc Data/...") n_d, c_d, _ = pipeline.sync_data_repository(lambda p, m: print(f" [{p}%] {m}")) print("[*] Đang đồng bộ thư mục video mục tiêu Target/...") n_t, c_t, _ = pipeline.sync_target_repository(lambda p, m: print(f" [{p}%] {m}")) print(f"[HOÀN TẤT ĐỒNG BỘ] Data: {c_d} sẵn có, {n_d} mới | Target: {c_t} sẵn có, {n_t} mới.") return if args.command == "compare": t_path = Path(args.target_video) if not t_path.exists(): # Thử tìm trong TARGET_DIR t_path = TARGET_DIR / args.target_video if not t_path.exists(): print(f"[LỖI] Không tìm thấy video target: {args.target_video}") sys.exit(1) ref_mode = f"với {len(args.refs)} video tham chiếu được chọn" if args.refs else "với TOÀN BỘ kho Data/" print(f"[*] Đang đối soát '{t_path.name}' {ref_mode}...") matches = pipeline.compare_target_with_references( target_video_path=t_path, selected_reference_ids=args.refs, progress_callback=lambda p, m: print(f" [{p}%] {m}"), ) if args.json: print(json.dumps([m.to_dict() for m in matches], indent=2, ensure_ascii=False)) else: print("\n" + "=" * 80) print(f"KẾT QUẢ ĐỐI SOÁT: {t_path.name}") print("=" * 80) if not matches: print(">> AN TOÀN: Không phát hiện đoạn nào trùng lặp với các video được so sánh.") else: print(f">> PHÁT HIỆN: {len(matches)} đoạn trùng lặp!") for idx, m in enumerate(matches, start=1): print(f" {idx}. Video gốc: {m.reference_video_id}") print(f" Target: {m._format_time(m.target_start_sec)} -> {m._format_time(m.target_end_sec)}") print(f" Gốc : {m._format_time(m.ref_start_sec)} -> {m._format_time(m.ref_end_sec)}") print(f" Độ dài: {m.duration_sec:.1f}s | Tỷ lệ khớp: {m.confidence_score*100:.1f}% ({m.match_type})") print("=" * 80) return if args.command == "index": v_path = Path(args.video_path).resolve() if not v_path.exists(): print(f"[LỖI] Không tìm thấy file: {v_path}") sys.exit(1) print(f"[*] Đang nạp video '{v_path.name}' vào kho bản quyền Data/...") record = pipeline.index_reference_video(v_path, progress_callback=lambda p, msg: print(f" [{p}%] {msg}")) print(f"[THÀNH CÔNG] Đã đăng ký video ID '{record.video_id}' ({record.duration_sec:.1f}s) vào Data/.") elif args.command == "scan": if args.target: t_path = Path(args.target).resolve() if not t_path.exists(): print(f"[LỖI] Không tìm thấy file target: {t_path}") sys.exit(1) print(f"[*] Đang quét video target '{t_path.name}'...") matches = pipeline.scan_target_video(t_path, progress_callback=lambda p, msg: print(f" [{p}%] {msg}")) results = {t_path.name: matches} else: print(f"[*] Đang quét tất cả video trong thư mục: {TARGET_DIR}...") results = pipeline.scan_all_targets(progress_callback=lambda p, msg: print(f" [{p}%] {msg}")) if args.json: out = {k: [m.to_dict() for m in v] for k, v in results.items()} print(json.dumps(out, indent=2, ensure_ascii=False)) else: print("\n" + "=" * 80) print("KẾT QUẢ ĐỐI SOÁT TRÙNG LẶP") print("=" * 80) total = sum(len(v) for v in results.values()) if total == 0: print(">> AN TOÀN: Không phát hiện đoạn video nào bị trùng lặp với kho dữ liệu chủ sở hữu.") else: for target_name, m_list in results.items(): print(f"\n[TARGET]: {target_name} ({len(m_list)} đoạn trùng lặp)") for idx, m in enumerate(m_list, start=1): print(f" {idx}. Gốc: {m.reference_video_id}") print(f" Target range : {m._format_time(m.target_start_sec)} -> {m._format_time(m.target_end_sec)}") print(f" Gốc range : {m._format_time(m.ref_start_sec)} -> {m._format_time(m.ref_end_sec)}") print(f" Thời lượng : {m.duration_sec:.1f} giây | Độ trùng khớp: {m.confidence_score*100:.1f}% ({m.match_type})") print("=" * 80) if __name__ == "__main__": if len(sys.argv) > 1 and sys.argv[1] != "gui": run_cli() else: launch_gui()