urban/WCX-collection
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add WCX title detection

Browse Source
This commit is contained in:
Urban
2026-07-21 19:32:49 +02:00
parent 8fe68d684e
commit a60ef87ddc
3 changed files with 1274 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
README.md
View File

@ -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.

769
scripts/detect_wcx_title.py Executable file
View File

@ -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())

453
tests/test_detect_wcx_title.py Normal file
View File

@ -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()