diff --git a/scripts/video_analysis/download_video.py b/scripts/video_analysis/download_video.py new file mode 100644 index 00000000..b6acef78 --- /dev/null +++ b/scripts/video_analysis/download_video.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from scripts.video_analysis.error_types import ErrorInfo, make_error + + +@dataclass +class _Ok: + value: Any + + def is_ok(self) -> bool: + return True + + def is_err(self) -> bool: + return False + + +@dataclass +class _Err: + err: ErrorInfo + + def is_ok(self) -> bool: + return False + + def is_err(self) -> bool: + return True + + +def ok(value: Any) -> _Ok: + return _Ok(value) + + +def err(error: ErrorInfo) -> _Err: + return _Err(error) + + +def validate_output_path(path: Path) -> _Ok | _Err: + if path.exists() and path.is_dir(): + return err(make_error("OutputIsDirectory", "validate_output_path", str(path))) + path.parent.mkdir(parents=True, exist_ok=True) + return ok(path) + + +def build_ydl_args(url: str, output: Path) -> list[str]: + return [ + "yt-dlp", + "--format", "bestvideo[ext=mp4]/best", + "--output", str(output), + "--no-warnings", + "--quiet", + url, + ] + + +def download_video(url: str, output: Path) -> _Ok | _Err: + validated = validate_output_path(output) + if validated.is_err(): + return validated + completed = subprocess.run( + build_ydl_args(url, output), + capture_output=True, + text=True, + ) + log_path = output.with_suffix(".log") + log_path.write_text( + f"# yt-dlp log\n# url: {url}\n# output: {output}\n# returncode: {completed.returncode}\n\nstdout:\n{completed.stdout}\n\nstderr:\n{completed.stderr}\n", + encoding="utf-8", + ) + if completed.returncode != 0: + return err(make_error("YtdlpError", "download_video", completed.stderr[:500])) + return ok({"output": str(output), "log": str(log_path), "returncode": completed.returncode}) diff --git a/tests/test_video_analysis_download_video.py b/tests/test_video_analysis_download_video.py new file mode 100644 index 00000000..70007f3a --- /dev/null +++ b/tests/test_video_analysis_download_video.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock, patch + +from scripts.video_analysis.download_video import ( + build_ydl_args, + download_video, + validate_output_path, +) + + +def test_validate_output_path_creates_parent(tmp_path: Path) -> None: + out = tmp_path / "subdir" / "video.mp4" + result = validate_output_path(out) + assert result.is_ok() + assert out.parent.exists() + + +def test_validate_output_path_rejects_existing_dir(tmp_path: Path) -> None: + out = tmp_path / "existing_dir" + out.mkdir() + result = validate_output_path(out) + assert result.is_err() + + +def test_build_ydl_args_basic(tmp_path: Path) -> None: + out = tmp_path / "v.mp4" + args = build_ydl_args("https://youtu.be/VID", out) + assert "--output" in args + assert str(out) in args + assert "https://youtu.be/VID" in args + + +def test_download_video_success(tmp_path: Path) -> None: + out = tmp_path / "video.mp4" + out.write_bytes(b"fake-mp4-content") + with patch("subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="") + result = download_video("https://youtu.be/VID", out) + assert result.is_ok() + assert (tmp_path / "video.log").exists() + + +def test_download_video_failure(tmp_path: Path) -> None: + out = tmp_path / "video.mp4" + with patch("subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="ERROR: video unavailable") + result = download_video("https://youtu.be/VID", out) + assert result.is_err()