diff --git a/README.md b/README.md index 0ad2caf..d00fdd2 100644 --- a/README.md +++ b/README.md @@ -11,3 +11,55 @@ File operations must be conservative. Any functionality that renames, moves, or `scripts/match_filenames.py` matches local video files against the external reference database. `scripts/diagnose_duration_match.py` is a manual diagnostic tool that compares one local file with current and historical durations for a specified `movie.id`; it is not an automated test. Both scripts require an explicit `--database` argument and open the reference database strictly read-only with SQLite `mode=ro`. + +## WCX title detection + +`scripts/detect_wcx_title.py` identifies local video files that probably +belong to WCX by analyzing the layout of the typical WCX title screen near the +beginning of each video. It samples frames at 8, 10, and 12 seconds, normalizes +them to 640x360 without changing their proportions, and measures black +background areas, red and bright text areas, and the edge layout. The +best-matching frame receives a score from 0 to 8: + +- `8/8`: `wcx` +- `6-7/8`: `uncertain` +- `0-5/8`: `not_wcx` + +The detector is completely non-destructive: it does not rename, move, or +delete video files. Batch mode also creates no PNG files. + +Run it on MP4 files directly in a directory: + +```bash +scripts/detect_wcx_title.py \ + /storage/disk1/X \ + --batch \ + --ending mp4 +``` + +Search recursively for multiple video extensions: + +```bash +scripts/detect_wcx_title.py \ + /storage/disk1/X \ + --batch \ + --recursive \ + --ending mp4,avi +``` + +Batch output contains one line per file with its classification, score, best +timestamp, and full file path. A summary after the run reports the number +processed and the totals for `wcx`, `uncertain`, `not_wcx`, and errors. + +In one verified test run, 62 files were analyzed: 41 were classified as +`wcx`, 21 as `not_wcx`, none as `uncertain`, and no files produced +errors. All 62 classifications were checked manually and were correct for +that test material. + +This is a heuristic detector, not a guaranteed source of truth. Heavily +edited or unusual files may require manual handling. Its goal is useful +accuracy with low complexity and reasonable runtime. + +For detailed inspection of one video, use `--diagnose-layout` together with +the required `--output-dir`; this mode saves normalized diagnostic images +and prints the measurements for all three timestamps. diff --git a/scripts/detect_wcx_title.py b/scripts/detect_wcx_title.py new file mode 100755 index 0000000..490c03c --- /dev/null +++ b/scripts/detect_wcx_title.py @@ -0,0 +1,769 @@ +#!/usr/bin/env python3 + +""" +Extract diagnostic frames from a video for future WCX title detection. + +This script never renames, moves, deletes, or modifies the input video. + +Example: + detect_wcx_title.py /path/to/video.mp4 \ + --output-dir /tmp/wcx-title-frames + + detect_wcx_title.py /path/to/video.mp4 \ + --output-dir /tmp/wcx-diagnostic \ + --diagnose-layout +""" + +import argparse +import shlex +import shutil +import subprocess +import sys +from pathlib import Path +from typing import Callable, Sequence + + +FRAME_TIMESTAMPS = tuple(range(6, 16)) +DIAGNOSTIC_TIMESTAMPS = (8, 10, 12) +NORMALIZED_WIDTH = 640 +NORMALIZED_HEIGHT = 360 +DEFAULT_ENDINGS = ("mp4", "avi", "mkv", "mov", "m4v", "webm") +IGNORED_FILENAMES = {".ds_store", "thumbs.db", "desktop.ini"} + + +def parse_arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Extract diagnostic PNG frames at 6 through 15 seconds " + "without modifying the input video." + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog="""Examples: + scripts/detect_wcx_title.py /path/to/video.mp4 \ + --output-dir /tmp/wcx-diagnostic --diagnose-layout + + scripts/detect_wcx_title.py /storage/disk1/X --batch + + scripts/detect_wcx_title.py /storage/disk1/X \ + --batch --recursive --ending mp4,avi""", + ) + + parser.add_argument( + "input_path", + type=Path, + help="Video file, or a directory when --batch is used.", + ) + + parser.add_argument( + "--output-dir", + type=Path, + help="Directory in which the extracted PNG frames will be written.", + ) + + parser.add_argument( + "--diagnose-layout", + action="store_true", + help=( + "Extract normalized frames at 8.0, 10.0, and 12.0 seconds " + "and report layout metrics." + ), + ) + + parser.add_argument( + "--batch", + action="store_true", + help="Classify video files in a directory without writing PNG files.", + ) + + parser.add_argument( + "--recursive", + action="store_true", + help="Search subdirectories in batch mode.", + ) + + parser.add_argument( + "--ending", + action="append", + help=( + "Video extension for batch mode; may be comma-separated or " + "repeated." + ), + ) + + return parser.parse_args() + + +def prepare_output_directory(output_dir: Path) -> None: + try: + output_dir.mkdir(parents=True, exist_ok=True) + except OSError as error: + raise RuntimeError( + f"Could not create output directory {output_dir}: {error}" + ) from error + + if not output_dir.is_dir(): + raise RuntimeError( + f"Output path is not a directory: {output_dir}" + ) + + +def parse_endings(values: Sequence[str] | None) -> set[str]: + if values is None: + return set(DEFAULT_ENDINGS) + + endings = { + ending.strip().lower().removeprefix(".") + for value in values + for ending in value.split(",") + if ending.strip() + } + + if not endings: + raise ValueError("--ending must contain at least one extension") + + return endings + + +def is_ignored_file(path: Path) -> bool: + return ( + path.name.startswith("._") + or path.name.lower() in IGNORED_FILENAMES + ) + + +def is_video_file(path: Path, endings: set[str]) -> bool: + return ( + path.is_file() + and not is_ignored_file(path) + and path.suffix.lower().removeprefix(".") in endings + ) + + +def find_video_files( + directory: Path, + endings: set[str], + recursive: bool, +) -> list[Path]: + directory = directory.resolve() + candidates = directory.rglob("*") if recursive else directory.iterdir() + return sorted( + ( + path + for path in candidates + if is_video_file(path, endings) + ), + key=lambda path: str(path), + ) + + +def extract_frame( + ffmpeg: str, + video_file: Path, + output_file: Path, + timestamp: int, +) -> None: + command = [ + ffmpeg, + "-hide_banner", + "-loglevel", + "error", + "-ss", + f"{timestamp:.1f}", + "-i", + str(video_file), + "-frames:v", + "1", + "-n", + str(output_file), + ] + + print(f"Running: {shlex.join(command)}") + + try: + result = subprocess.run( + command, + capture_output=True, + text=True, + ) + except OSError as error: + raise RuntimeError( + f"Could not run ffmpeg for {timestamp:.1f} seconds: {error}" + ) from error + + if result.returncode != 0: + detail = result.stderr.strip() or "ffmpeg returned no error message" + raise RuntimeError( + f"ffmpeg could not extract the frame at " + f"{timestamp:.1f} seconds: {detail}" + ) + + if not output_file.is_file(): + raise RuntimeError( + f"ffmpeg reported success but did not create the frame at " + f"{timestamp:.1f} seconds: {output_file}" + ) + + print(f"Created: {output_file}") + + +def is_nearly_black(red: int, green: int, blue: int) -> bool: + return red <= 30 and green <= 30 and blue <= 30 + + +def is_red(red: int, green: int, blue: int) -> bool: + return ( + red >= 140 + and red >= green * 1.5 + and red >= blue * 1.5 + ) + + +def is_bright(red: int, green: int, blue: int) -> bool: + return red >= 180 and green >= 180 and blue >= 180 + + +def calculate_region_ratio( + rgb_data: bytes, + width: int, + x_start: int, + x_end: int, + y_start: int, + y_end: int, + predicate, +) -> float: + matching_pixels = 0 + total_pixels = (x_end - x_start) * (y_end - y_start) + + for y_position in range(y_start, y_end): + row_offset = y_position * width * 3 + for x_position in range(x_start, x_end): + offset = row_offset + x_position * 3 + if predicate( + rgb_data[offset], + rgb_data[offset + 1], + rgb_data[offset + 2], + ): + matching_pixels += 1 + + return matching_pixels / total_pixels + + +def calculate_layout_metrics( + rgb_data: bytes, + width: int = NORMALIZED_WIDTH, + height: int = NORMALIZED_HEIGHT, +) -> dict[str, float]: + expected_bytes = width * height * 3 + if len(rgb_data) != expected_bytes: + raise ValueError( + f"Expected {expected_bytes} RGB bytes for {width}x{height}, " + f"received {len(rgb_data)}" + ) + + edge_width = max(1, int(width * 0.10)) + edge_height = max(1, int(height * 0.10)) + band_height = max(1, int(height * 0.15)) + + return { + "black_ratio_total": calculate_region_ratio( + rgb_data, width, 0, width, 0, height, is_nearly_black + ), + "black_ratio_top": calculate_region_ratio( + rgb_data, width, 0, width, 0, edge_height, is_nearly_black + ), + "black_ratio_bottom": calculate_region_ratio( + rgb_data, + width, + 0, + width, + height - edge_height, + height, + is_nearly_black, + ), + "black_ratio_left": calculate_region_ratio( + rgb_data, width, 0, edge_width, 0, height, is_nearly_black + ), + "black_ratio_right": calculate_region_ratio( + rgb_data, + width, + width - edge_width, + width, + 0, + height, + is_nearly_black, + ), + "red_ratio_top": calculate_region_ratio( + rgb_data, width, 0, width, 0, band_height, is_red + ), + "bright_ratio_top": calculate_region_ratio( + rgb_data, width, 0, width, 0, band_height, is_bright + ), + "bright_ratio_bottom": calculate_region_ratio( + rgb_data, + width, + 0, + width, + height - band_height, + height, + is_bright, + ), + } + + +def calculate_wcx_layout_score(metrics: dict[str, float]) -> int: + conditions = ( + 0.20 <= metrics["black_ratio_total"] <= 0.60, + metrics["black_ratio_top"] >= 0.45, + metrics["black_ratio_bottom"] >= 0.50, + metrics["black_ratio_left"] >= 0.75, + metrics["black_ratio_right"] >= 0.75, + metrics["red_ratio_top"] >= 0.05, + metrics["bright_ratio_top"] >= 0.02, + metrics["bright_ratio_bottom"] >= 0.03, + ) + return sum(conditions) + + +def select_best_timestamp( + timestamp_scores: Sequence[tuple[float, int]], +) -> tuple[float, int]: + if not timestamp_scores: + raise ValueError("At least one timestamp and score is required") + + return min( + timestamp_scores, + key=lambda item: (-item[1], item[0]), + ) + + +def classify_wcx_score(score: int) -> str: + if not 0 <= score <= 8: + raise ValueError(f"WCX layout score must be between 0 and 8: {score}") + + if score == 8: + return "wcx" + + if score >= 6: + return "uncertain" + + return "not_wcx" + + +def split_rgb24_frames( + rgb_data: bytes, + width: int = NORMALIZED_WIDTH, + height: int = NORMALIZED_HEIGHT, + frame_count: int = len(DIAGNOSTIC_TIMESTAMPS), +) -> tuple[bytes, ...]: + frame_size = width * height * 3 + expected_bytes = frame_size * frame_count + actual_bytes = len(rgb_data) + + if actual_bytes < expected_bytes: + raise ValueError( + f"RGB24 buffer is too short: expected {expected_bytes} bytes, " + f"received {actual_bytes}" + ) + + if actual_bytes > expected_bytes: + raise ValueError( + f"RGB24 buffer is too long: expected {expected_bytes} bytes, " + f"received {actual_bytes}" + ) + + return tuple( + rgb_data[offset:offset + frame_size] + for offset in range(0, expected_bytes, frame_size) + ) + + +def analyze_rgb24_frames( + rgb_data: bytes, +) -> tuple[ + tuple[tuple[float, dict[str, float], int], ...], + float, + int, + str, +]: + frames = split_rgb24_frames(rgb_data) + analyses = tuple( + ( + float(timestamp), + metrics, + calculate_wcx_layout_score(metrics), + ) + for timestamp, frame in zip( + DIAGNOSTIC_TIMESTAMPS, + frames, + ) + for metrics in (calculate_layout_metrics(frame),) + ) + best_timestamp, best_score = select_best_timestamp( + tuple( + (timestamp, score) + for timestamp, _metrics, score in analyses + ) + ) + return ( + analyses, + best_timestamp, + best_score, + classify_wcx_score(best_score), + ) + + +def extract_batch_rgb24( + ffmpeg: str, + video_file: Path, +) -> bytes: + filter_graph = ( + "[0:v]setpts=PTS-STARTPTS," + "fps=1/2:start_time=0," + "scale=640:360:force_original_aspect_ratio=decrease," + "pad=640:360:(ow-iw)/2:(oh-ih)/2:black" + ) + command = [ + ffmpeg, + "-hide_banner", + "-loglevel", + "error", + "-ss", + f"{DIAGNOSTIC_TIMESTAMPS[0]:.1f}", + "-t", + "5.0", + "-i", + str(video_file), + "-filter_complex", + filter_graph, + "-frames:v", + str(len(DIAGNOSTIC_TIMESTAMPS)), + "-c:v", + "rawvideo", + "-pix_fmt", + "rgb24", + "-f", + "rawvideo", + "pipe:1", + ] + + try: + result = subprocess.run( + command, + capture_output=True, + ) + except OSError as error: + raise RuntimeError(f"Could not run ffmpeg: {error}") from error + + if result.returncode != 0: + detail = ( + result.stderr.decode("utf-8", errors="replace").strip() + or "ffmpeg returned no error message" + ) + raise RuntimeError(f"ffmpeg could not analyze the video: {detail}") + + try: + split_rgb24_frames(result.stdout) + except ValueError as error: + raise RuntimeError(f"Invalid RGB24 output from ffmpeg: {error}") from error + + return result.stdout + + +def analyze_video_for_batch( + ffmpeg: str, + video_file: Path, +) -> tuple[float, int, str]: + _analyses, best_timestamp, best_score, classification = ( + analyze_rgb24_frames( + extract_batch_rgb24(ffmpeg, video_file) + ) + ) + return best_timestamp, best_score, classification + + +def process_batch( + video_files: Sequence[Path], + analyzer: Callable[[Path], tuple[float, int, str]], +) -> int: + counts = { + "processed": 0, + "wcx": 0, + "uncertain": 0, + "not_wcx": 0, + "errors": 0, + } + + for video_file in video_files: + counts["processed"] += 1 + try: + timestamp, score, classification = analyzer(video_file) + except (OSError, RuntimeError, ValueError) as error: + counts["errors"] += 1 + print(f"error - - {video_file}: {error}") + continue + + counts[classification] += 1 + print( + f"{classification:<10} {score}/8 " + f"{timestamp:>4.1f} {video_file}" + ) + + print() + print(f"processed: {counts['processed']}") + print(f"wcx: {counts['wcx']}") + print(f"uncertain: {counts['uncertain']}") + print(f"not_wcx: {counts['not_wcx']}") + print(f"errors: {counts['errors']}") + + return 1 if counts["errors"] else 0 + + +def diagnose_layout( + ffmpeg: str, + video_file: Path, + output_dir: Path, +) -> None: + normalized_files = tuple( + output_dir / f"normalized-{timestamp:04.1f}.png" + for timestamp in DIAGNOSTIC_TIMESTAMPS + ) + + for output_file in normalized_files: + if output_file.exists(): + raise RuntimeError( + f"Output file already exists and will not be overwritten: " + f"{output_file}" + ) + + filter_graph = ( + "[0:v]setpts=PTS-STARTPTS," + "fps=1/2:start_time=0," + "scale=640:360:force_original_aspect_ratio=decrease," + "pad=640:360:(ow-iw)/2:(oh-ih)/2:black," + "split=4[normalized_raw][frame_8_input]" + "[frame_10_input][frame_12_input];" + "[frame_8_input]select=eq(n\\,0)[frame_8];" + "[frame_10_input]select=eq(n\\,1)[frame_10];" + "[frame_12_input]select=eq(n\\,2)[frame_12]" + ) + + command = [ + ffmpeg, + "-hide_banner", + "-loglevel", + "error", + "-n", + "-ss", + f"{DIAGNOSTIC_TIMESTAMPS[0]:.1f}", + "-t", + "5.0", + "-i", + str(video_file), + "-filter_complex", + filter_graph, + "-map", + "[frame_8]", + "-frames:v", + "1", + "-c:v", + "png", + str(normalized_files[0]), + "-map", + "[frame_10]", + "-frames:v", + "1", + "-c:v", + "png", + str(normalized_files[1]), + "-map", + "[frame_12]", + "-frames:v", + "1", + "-c:v", + "png", + str(normalized_files[2]), + "-map", + "[normalized_raw]", + "-frames:v", + str(len(DIAGNOSTIC_TIMESTAMPS)), + "-c:v", + "rawvideo", + "-pix_fmt", + "rgb24", + "-f", + "rawvideo", + "pipe:1", + ] + + print(f"Running: {shlex.join(command)}") + + try: + result = subprocess.run( + command, + capture_output=True, + ) + except OSError as error: + raise RuntimeError( + f"Could not run ffmpeg for diagnostic frames: {error}" + ) from error + + if result.returncode != 0: + detail = ( + result.stderr.decode("utf-8", errors="replace").strip() + or "ffmpeg returned no error message" + ) + raise RuntimeError( + f"ffmpeg could not extract the diagnostic frames: {detail}" + ) + + for output_file in normalized_files: + if not output_file.is_file(): + raise RuntimeError( + f"ffmpeg reported success but did not create: {output_file}" + ) + print(f"Created: {output_file}") + + try: + analyses, best_timestamp, best_score, classification = ( + analyze_rgb24_frames(result.stdout) + ) + except ValueError as error: + raise RuntimeError(f"Invalid RGB24 output from ffmpeg: {error}") from error + + for index, (timestamp, metrics, score) in enumerate(analyses): + if index: + print() + print(f"timestamp: {timestamp:.1f}") + print(f"wcx_layout_score: {score}/8") + print(f"normalized_size: {NORMALIZED_WIDTH}x{NORMALIZED_HEIGHT}") + for name, ratio in metrics.items(): + print(f"{name}: {ratio:.6f}") + + print() + print(f"best_timestamp: {best_timestamp:.1f}") + print(f"best_wcx_layout_score: {best_score}/8") + print(f"classification: {classification}") + + +def main() -> int: + args = parse_arguments() + + if args.batch: + if args.diagnose_layout: + print( + "Error: --batch cannot be combined with --diagnose-layout.", + file=sys.stderr, + ) + return 2 + + if args.output_dir is not None: + print( + "Error: --batch cannot be combined with --output-dir.", + file=sys.stderr, + ) + return 2 + + if not args.input_path.is_dir(): + print( + f"Error: Batch directory does not exist: {args.input_path}", + file=sys.stderr, + ) + return 2 + + try: + endings = parse_endings(args.ending) + video_files = find_video_files( + args.input_path, + endings=endings, + recursive=args.recursive, + ) + except (OSError, ValueError) as error: + print(f"Error: {error}", file=sys.stderr) + return 2 + + if not video_files: + print( + f"No matching video files found in: " + f"{args.input_path.resolve()}" + ) + return process_batch((), lambda _path: (0.0, 0, "not_wcx")) + + ffmpeg = shutil.which("ffmpeg") + if ffmpeg is None: + print("Error: ffmpeg was not found in PATH.", file=sys.stderr) + return 1 + + return process_batch( + video_files, + lambda video_file: analyze_video_for_batch( + ffmpeg, + video_file, + ), + ) + + if args.recursive: + print( + "Error: --recursive requires --batch.", + file=sys.stderr, + ) + return 2 + + if args.ending is not None: + print( + "Error: --ending requires --batch.", + file=sys.stderr, + ) + return 2 + + if args.output_dir is None: + print( + "Error: --output-dir is required unless --batch is used.", + file=sys.stderr, + ) + return 2 + + if not args.input_path.is_file(): + print( + f"Error: Video file does not exist: {args.input_path}", + file=sys.stderr, + ) + return 1 + + ffmpeg = shutil.which("ffmpeg") + if ffmpeg is None: + print( + "Error: ffmpeg was not found in PATH.", + file=sys.stderr, + ) + return 1 + + try: + prepare_output_directory(args.output_dir) + + if args.diagnose_layout: + diagnose_layout( + ffmpeg=ffmpeg, + video_file=args.input_path, + output_dir=args.output_dir, + ) + else: + for timestamp in FRAME_TIMESTAMPS: + output_file = ( + args.output_dir + / f"frame-{timestamp:04.1f}.png" + ) + extract_frame( + ffmpeg=ffmpeg, + video_file=args.input_path, + output_file=output_file, + timestamp=timestamp, + ) + except RuntimeError as error: + print(f"Error: {error}", file=sys.stderr) + return 1 + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_detect_wcx_title.py b/tests/test_detect_wcx_title.py new file mode 100644 index 0000000..fdb3c8a --- /dev/null +++ b/tests/test_detect_wcx_title.py @@ -0,0 +1,453 @@ +import contextlib +import importlib.util +import io +import tempfile +import unittest +from pathlib import Path + + +SCRIPT_PATH = ( + Path(__file__).resolve().parents[1] + / "scripts" + / "detect_wcx_title.py" +) +SPEC = importlib.util.spec_from_file_location( + "detect_wcx_title", + SCRIPT_PATH, +) +if SPEC is None or SPEC.loader is None: + raise RuntimeError(f"Could not load script: {SCRIPT_PATH}") + +detect_wcx_title = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(detect_wcx_title) + + +class PixelClassificationTests(unittest.TestCase): + def test_nearly_black_thresholds(self) -> None: + self.assertTrue( + detect_wcx_title.is_nearly_black(30, 30, 30) + ) + self.assertFalse( + detect_wcx_title.is_nearly_black(31, 30, 30) + ) + + def test_red_thresholds(self) -> None: + self.assertTrue( + detect_wcx_title.is_red(150, 100, 100) + ) + self.assertFalse( + detect_wcx_title.is_red(139, 0, 0) + ) + self.assertFalse( + detect_wcx_title.is_red(150, 101, 100) + ) + + def test_bright_thresholds(self) -> None: + self.assertTrue( + detect_wcx_title.is_bright(180, 180, 180) + ) + self.assertFalse( + detect_wcx_title.is_bright(180, 179, 180) + ) + + def test_layout_metrics_for_constructed_buffer(self) -> None: + width = 10 + height = 20 + pixels = [(100, 100, 100)] * (width * height) + + for x_position in range(width): + pixels[x_position] = (0, 0, 0) + pixels[width + x_position] = (0, 0, 0) + + for y_position in range(3): + pixels[y_position * width] = (150, 0, 0) + + for y_position in range(height - 3, height): + pixels[y_position * width + 1] = (255, 255, 255) + + rgb_data = bytes( + channel + for pixel in pixels + for channel in pixel + ) + metrics = detect_wcx_title.calculate_layout_metrics( + rgb_data, + width=width, + height=height, + ) + + self.assertAlmostEqual( + metrics["black_ratio_top"], + 18 / 20, + ) + self.assertAlmostEqual( + metrics["red_ratio_top"], + 3 / 30, + ) + self.assertAlmostEqual( + metrics["bright_ratio_bottom"], + 3 / 30, + ) + + def test_layout_metrics_reject_wrong_buffer_size(self) -> None: + with self.assertRaisesRegex(ValueError, "Expected 12 RGB bytes"): + detect_wcx_title.calculate_layout_metrics( + bytes(11), + width=2, + height=2, + ) + + def test_split_rgb24_frames(self) -> None: + frame_size = 2 * 2 * 3 + frames = ( + bytes([1]) * frame_size, + bytes([2]) * frame_size, + bytes([3]) * frame_size, + ) + + result = detect_wcx_title.split_rgb24_frames( + b"".join(frames), + width=2, + height=2, + ) + + self.assertEqual(result, frames) + + def test_split_rgb24_frames_rejects_short_buffer(self) -> None: + with self.assertRaisesRegex(ValueError, "too short"): + detect_wcx_title.split_rgb24_frames( + bytes(35), + width=2, + height=2, + ) + + def test_split_rgb24_frames_rejects_long_buffer(self) -> None: + with self.assertRaisesRegex(ValueError, "too long"): + detect_wcx_title.split_rgb24_frames( + bytes(37), + width=2, + height=2, + ) + + def test_wcx_layout_score_all_conditions(self) -> None: + metrics = { + "black_ratio_total": 0.40, + "black_ratio_top": 0.60, + "black_ratio_bottom": 0.70, + "black_ratio_left": 0.90, + "black_ratio_right": 0.90, + "red_ratio_top": 0.10, + "bright_ratio_top": 0.05, + "bright_ratio_bottom": 0.08, + } + + self.assertEqual( + detect_wcx_title.calculate_wcx_layout_score(metrics), + 8, + ) + + def test_wcx_layout_score_no_conditions(self) -> None: + metrics = { + "black_ratio_total": 0.10, + "black_ratio_top": 0.10, + "black_ratio_bottom": 0.10, + "black_ratio_left": 0.10, + "black_ratio_right": 0.10, + "red_ratio_top": 0.01, + "bright_ratio_top": 0.01, + "bright_ratio_bottom": 0.01, + } + + self.assertEqual( + detect_wcx_title.calculate_wcx_layout_score(metrics), + 0, + ) + + def test_wcx_layout_score_includes_boundaries(self) -> None: + metrics = { + "black_ratio_total": 0.20, + "black_ratio_top": 0.45, + "black_ratio_bottom": 0.50, + "black_ratio_left": 0.75, + "black_ratio_right": 0.75, + "red_ratio_top": 0.05, + "bright_ratio_top": 0.02, + "bright_ratio_bottom": 0.03, + } + + self.assertEqual( + detect_wcx_title.calculate_wcx_layout_score(metrics), + 8, + ) + + metrics["black_ratio_total"] = 0.60 + self.assertEqual( + detect_wcx_title.calculate_wcx_layout_score(metrics), + 8, + ) + + def test_wcs_like_metrics_score_low(self) -> None: + metrics = { + "black_ratio_total": 0.88, + "black_ratio_top": 1.00, + "black_ratio_bottom": 1.00, + "black_ratio_left": 0.96, + "black_ratio_right": 0.96, + "red_ratio_top": 0.00, + "bright_ratio_top": 0.00, + "bright_ratio_bottom": 0.00, + } + + self.assertEqual( + detect_wcx_title.calculate_wcx_layout_score(metrics), + 4, + ) + + def test_wunf_like_metrics_score_low(self) -> None: + metrics = { + "black_ratio_total": 0.15, + "black_ratio_top": 0.20, + "black_ratio_bottom": 0.25, + "black_ratio_left": 0.10, + "black_ratio_right": 0.10, + "red_ratio_top": 0.20, + "bright_ratio_top": 0.01, + "bright_ratio_bottom": 0.01, + } + + self.assertLessEqual( + detect_wcx_title.calculate_wcx_layout_score(metrics), + 2, + ) + + def test_select_best_timestamp_uses_highest_score(self) -> None: + self.assertEqual( + detect_wcx_title.select_best_timestamp( + ((8.0, 3), (10.0, 8), (12.0, 5)) + ), + (10.0, 8), + ) + + def test_select_best_timestamp_uses_earliest_tie(self) -> None: + self.assertEqual( + detect_wcx_title.select_best_timestamp( + ((12.0, 8), (10.0, 8), (8.0, 7)) + ), + (10.0, 8), + ) + + def test_classify_wcx_score_eight(self) -> None: + self.assertEqual( + detect_wcx_title.classify_wcx_score(8), + "wcx", + ) + + def test_classify_wcx_score_seven(self) -> None: + self.assertEqual( + detect_wcx_title.classify_wcx_score(7), + "uncertain", + ) + + def test_classify_wcx_score_six(self) -> None: + self.assertEqual( + detect_wcx_title.classify_wcx_score(6), + "uncertain", + ) + + def test_classify_wcx_score_five(self) -> None: + self.assertEqual( + detect_wcx_title.classify_wcx_score(5), + "not_wcx", + ) + + def test_classify_wcx_score_zero(self) -> None: + self.assertEqual( + detect_wcx_title.classify_wcx_score(0), + "not_wcx", + ) + + def test_classify_wcx_score_rejects_negative(self) -> None: + with self.assertRaises(ValueError): + detect_wcx_title.classify_wcx_score(-1) + + def test_classify_wcx_score_rejects_above_maximum(self) -> None: + with self.assertRaises(ValueError): + detect_wcx_title.classify_wcx_score(9) + + +class BatchModeTests(unittest.TestCase): + def test_parse_comma_separated_endings(self) -> None: + self.assertEqual( + detect_wcx_title.parse_endings(["mp4,avi"]), + {"mp4", "avi"}, + ) + + def test_parse_repeated_endings(self) -> None: + self.assertEqual( + detect_wcx_title.parse_endings(["mp4", "avi"]), + {"mp4", "avi"}, + ) + + def test_parse_endings_normalizes_dot_and_case(self) -> None: + self.assertEqual( + detect_wcx_title.parse_endings([".MP4"]), + {"mp4"}, + ) + + def test_video_file_filtering_and_ignored_files(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + video = directory / "movie.MP4" + wrong_ending = directory / "movie.txt" + apple_double = directory / "._movie.mp4" + ds_store = directory / ".DS_Store" + thumbs = directory / "Thumbs.db" + desktop = directory / "desktop.ini" + + for path in ( + video, + wrong_ending, + apple_double, + ds_store, + thumbs, + desktop, + ): + path.touch() + + endings = {"mp4"} + self.assertTrue( + detect_wcx_title.is_video_file(video, endings) + ) + for path in (apple_double, ds_store, thumbs, desktop): + self.assertTrue( + detect_wcx_title.is_ignored_file(path) + ) + for path in ( + wrong_ending, + apple_double, + ds_store, + thumbs, + desktop, + ): + self.assertFalse( + detect_wcx_title.is_video_file(path, endings) + ) + + def test_non_recursive_file_search(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + direct = directory / "direct.mp4" + nested_directory = directory / "nested" + nested = nested_directory / "nested.mp4" + nested_directory.mkdir() + direct.touch() + nested.touch() + + self.assertEqual( + detect_wcx_title.find_video_files( + directory, + endings={"mp4"}, + recursive=False, + ), + [direct.resolve()], + ) + + def test_recursive_file_search(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + direct = directory / "direct.mp4" + nested_directory = directory / "nested" + nested = nested_directory / "nested.mp4" + nested_directory.mkdir() + direct.touch() + nested.touch() + + self.assertEqual( + detect_wcx_title.find_video_files( + directory, + endings={"mp4"}, + recursive=True, + ), + sorted( + [direct.resolve(), nested.resolve()], + key=lambda path: str(path), + ), + ) + + def test_file_search_is_sorted_by_full_path(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + paths = [ + directory / "z.mp4", + directory / "A.mp4", + directory / "m.mp4", + ] + for path in paths: + path.touch() + + result = detect_wcx_title.find_video_files( + directory, + endings={"mp4"}, + recursive=False, + ) + + self.assertEqual( + result, + sorted( + (path.resolve() for path in paths), + key=lambda path: str(path), + ), + ) + + def test_batch_counts_classifications(self) -> None: + video_files = ( + Path("/videos/wcx.mp4"), + Path("/videos/uncertain.mp4"), + Path("/videos/not-wcx.mp4"), + ) + results = { + video_files[0]: (12.0, 8, "wcx"), + video_files[1]: (10.0, 7, "uncertain"), + video_files[2]: (8.0, 4, "not_wcx"), + } + + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = detect_wcx_title.process_batch( + video_files, + lambda path: results[path], + ) + + self.assertEqual(exit_code, 0) + self.assertIn("processed: 3", output.getvalue()) + self.assertIn("wcx: 1", output.getvalue()) + self.assertIn("uncertain: 1", output.getvalue()) + self.assertIn("not_wcx: 1", output.getvalue()) + self.assertIn("errors: 0", output.getvalue()) + + def test_batch_returns_one_when_file_errors(self) -> None: + video_files = ( + Path("/videos/good.mp4"), + Path("/videos/broken.mp4"), + ) + + def analyze(path: Path) -> tuple[float, int, str]: + if path.name == "broken.mp4": + raise RuntimeError("mock analysis failure") + return 8.0, 8, "wcx" + + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = detect_wcx_title.process_batch( + video_files, + analyze, + ) + + self.assertEqual(exit_code, 1) + self.assertIn("error", output.getvalue()) + self.assertIn("processed: 2", output.getvalue()) + self.assertIn("errors: 1", output.getvalue()) + + +if __name__ == "__main__": + unittest.main()