feat: 支持自定义时长和大小分桶粒度

This commit is contained in:
kim
2026-04-02 22:07:17 +08:00
parent f243664acf
commit ff0b9228f8
6 changed files with 356 additions and 46 deletions
+64 -32
View File
@@ -1,29 +1,29 @@
# 视频元数据分析器 # 视频元数据分析器
基于 Python + `uv` 的命令行工具,用于递归扫描一个或多个目录中的视频文件,调用 `ffprobe` 提取结构化元数据,并调用 `ffmpeg` 记录运行环境能力,最终输出: 一个基于 Python + `uv` 的命令行工具,用于递归扫描一个或多个目录中的视频文件,调用 `ffprobe` 提取结构化元数据,并生成三类报告:
- `report.json`:完整元数据与汇总统计 - `report.json`:完整元数据、失败列表与统计摘要
- `files.csv`:逐文件明细 - `files.csv`:适合后续处理的明细表
- `report.html`:可直接打开的离线可视化报告 - `report.html`:可直接打开的人类可读报告
## 当前能力 ## 当前能力
- 输入一个或多个目录,递归发现常见视频文件 - 递归扫描一个或多个目录,自动发现常见视频文件
- 提取视频时长、大小、编码、分辨率、帧率、音频编码等字段 - 提取视频时长、大小、码率、分辨率、帧率、流数量等字段
- 单个视频探测失败时不中断整次扫描,会单独记录错误清单 - 单个视频探测失败时不中断整体扫描,会单独记录到失败列表
- 汇总时长分布、大小分布、视频编码分布、分辨率分布 - 输出时长分布、大小分布、视频编码分布、分辨率分布
- 时长分布、大小分布会根据当前目录样本范围自动调整分桶粒度,避免结果全部挤在默认大桶里 - 默认根据当前样本范围自动调整时长/大小分桶粒度
- 可以基于已有 `report.json` 重新计算 `summary`,并重建 `report.html` / `files.csv`,无需再次扫描视频文件 - 可用参数显式指定时长和大小分布的粒度,覆盖自动分桶
- 生成适合人工查看的 HTML 报告,以及适合后续二次处理的 JSON / CSV - 可以基于已有 `report.json` 重新计算 `summary`,并重建 `report.html` / `files.csv`,无需重新扫描视频文件
## 依赖要求 ## 环境要求
- Python 3.11+ - Python 3.11+
- `uv` - `uv`
- `ffprobe` - `ffprobe`
- `ffmpeg` - `ffmpeg`
要求 `ffprobe` 与 `ffmpeg` 已经在系统 `PATH` 中,或者在运行时通过参数显式指定。 要求 `ffprobe` 和 `ffmpeg` 已在系统 `PATH` 中,或在运行时通过参数显式指定路径。
## 初始化 ## 初始化
@@ -33,13 +33,13 @@ uv sync --dev
## 运行方式 ## 运行方式
最直接的方式: 直接扫描目录:
```powershell ```powershell
uv run video-metadata-analyzer D:\Videos E:\Archive --output-dir .\reports\latest uv run video-metadata-analyzer D:\Videos E:\Archive --output-dir .\reports\latest
``` ```
如果你更习惯模块方式: 如果你习惯模块方式:
```powershell ```powershell
uv run python -m video_metadata_analyzer D:\Videos --output-dir .\reports\latest uv run python -m video_metadata_analyzer D:\Videos --output-dir .\reports\latest
@@ -54,19 +54,41 @@ uv run video-metadata-analyzer D:\Videos `
--output-dir .\reports\latest --output-dir .\reports\latest
``` ```
如果你已经有一份旧的 `report.json`,只想在新版本逻辑下重算汇总并重建报告,不想重新扫描视频文件: ## 粒度参数
默认情况下,时长分布和大小分布会根据样本范围自动选择合适的分桶粒度。
如果你希望固定粒度,可以使用:
- `--duration-bucket-seconds`:显式指定时长分布的分桶粒度,单位为秒
- `--size-bucket-megabytes`:显式指定大小分布的分桶粒度,单位为 MB
例如:
```powershell
uv run video-metadata-analyzer D:\Videos `
--duration-bucket-seconds 60 `
--size-bucket-megabytes 100 `
--output-dir .\reports\latest
```
上面的命令会把时长分布按 60 秒一桶、大小分布按 100 MB 一桶来统计。
## 基于旧报告重建
如果你已经有一份旧的 `report.json`,只想用新版本逻辑重新计算摘要并重建报告,而不重新扫描视频文件:
```powershell ```powershell
uv run video-metadata-analyzer --report-json .\reports\latest\report.json uv run video-metadata-analyzer --report-json .\reports\latest\report.json
``` ```
默认会直接在该 `report.json` 所在目录重写: 默认会直接在该 `report.json` 所在目录中重写:
- `report.json` - `report.json`
- `files.csv` - `files.csv`
- `report.html` - `report.html`
如果你希望把重建后的结果写到新目录: 如果你希望把重建结果写到其他目录:
```powershell ```powershell
uv run video-metadata-analyzer ` uv run video-metadata-analyzer `
@@ -74,9 +96,19 @@ uv run video-metadata-analyzer `
--output-dir .\reports\rebuilt --output-dir .\reports\rebuilt
``` ```
## 输出内容 重建时同样可以指定新的粒度参数:
生成目录内默认包含以下文件: ```powershell
uv run video-metadata-analyzer `
--report-json .\reports\latest\report.json `
--duration-bucket-seconds 300
```
扫描阶段如果使用了显式粒度,这些设置会写入 `report.json` 的 `summary_options`。之后再次用 `--report-json` 重建时,如果不额外传参,会继承原先保存的粒度;如果只传其中一个参数,则只覆盖对应那一项。
## 输出结果说明
输出目录默认包含三个文件:
- `report.json` - `report.json`
- `files.csv` - `files.csv`
@@ -84,29 +116,29 @@ uv run video-metadata-analyzer `
其中: 其中:
- `report.json` 包含 `summary`、成功文件列表 `files`,以及失败文件列表 `failures` - `report.json` 包含 `summary`、成功文件列表 `files`、失败文件列表 `failures`,以及用于重建的 `summary_options`
- `files.csv` 同时输出成功和失败结果,带 `status`、`error_type`、`error_message` 字段 - `files.csv` 同时包含成功和失败结果,并带有 `status`、`error_type`、`error_message` 字段
- `report.html` 是离线静态页面,可以直接双击打开查看,并显示错误清单面板 - `report.html` 是离线静态页面,可以直接双击打开查看摘要、图表和失败清单
- 当使用 `--report-json` 时,会从 `files` / `failures` 重新计算 `summary`,再生成新的 `report.json`、`files.csv`、`report.html` - 当使用 `--report-json` 时,会根据 `files` / `failures` 重新计算 `summary`,并生成新的 `report.json`、`files.csv`、`report.html`
## 开发与测试 ## 运行测试
运行测试: 运行全部测试:
```powershell ```powershell
uv run pytest -q uv run pytest -q
``` ```
查看命令帮助: 查看命令行帮助:
```powershell ```powershell
uv run video-metadata-analyzer --help uv run video-metadata-analyzer --help
``` ```
## 第一版边界 ## 当前边界
当前版本优先保证“扫描、提取、汇总、可视化”这条主链路可用,暂时没有做这些增强项: 当前版本优先保证扫描、提取、统计、可视化和报告产出,暂时没有实现这些增强项:
- 损坏视频的深度解码验证 - 损坏视频内容级校验
- 更复杂的前端交互图表 - 交互式前端筛选界面
- 增量扫描与数据库缓存 - 增量扫描缓存或数据库
+32 -1
View File
@@ -46,6 +46,28 @@ class CategoryCount:
count: int count: int
@dataclass(slots=True, frozen=True)
class SummaryOptions:
duration_bucket_seconds: float | None = None
size_bucket_bytes: int | None = None
def merged_with(self, override: SummaryOptions | None) -> SummaryOptions:
if override is None:
return self
return SummaryOptions(
duration_bucket_seconds=(
override.duration_bucket_seconds
if override.duration_bucket_seconds is not None
else self.duration_bucket_seconds
),
size_bucket_bytes=(
override.size_bucket_bytes
if override.size_bucket_bytes is not None
else self.size_bucket_bytes
),
)
@dataclass(slots=True) @dataclass(slots=True)
class AnalysisSummary: class AnalysisSummary:
total_files: int total_files: int
@@ -64,11 +86,16 @@ def _bucketize_numeric(
*, *,
step_candidates: list[float], step_candidates: list[float],
label_formatter: Callable[[float, float, float], str], label_formatter: Callable[[float, float, float], str],
explicit_step: float | None = None,
) -> list[HistogramBin]: ) -> list[HistogramBin]:
if not values: if not values:
return [] return []
step = _select_histogram_step(max_value=max(values), step_candidates=step_candidates, value_count=len(values)) step = explicit_step or _select_histogram_step(
max_value=max(values),
step_candidates=step_candidates,
value_count=len(values),
)
bin_count = max(1, math.ceil(max(values) / step)) bin_count = max(1, math.ceil(max(values) / step))
results: list[HistogramBin] = [] results: list[HistogramBin] = []
@@ -155,8 +182,10 @@ def build_summary(
records: list[VideoMetadata], records: list[VideoMetadata],
*, *,
failures: list[ProbeFailure] | None = None, failures: list[ProbeFailure] | None = None,
summary_options: SummaryOptions | None = None,
) -> AnalysisSummary: ) -> AnalysisSummary:
failures = failures or [] failures = failures or []
summary_options = summary_options or SummaryOptions()
durations = [record.duration_seconds for record in records if record.duration_seconds is not None] durations = [record.duration_seconds for record in records if record.duration_seconds is not None]
sizes = [record.size_bytes for record in records] sizes = [record.size_bytes for record in records]
codecs = [record.video_codec for record in records if record.video_codec] codecs = [record.video_codec for record in records if record.video_codec]
@@ -176,11 +205,13 @@ def build_summary(
durations, durations,
step_candidates=DURATION_STEP_CANDIDATES, step_candidates=DURATION_STEP_CANDIDATES,
label_formatter=_format_duration_bucket_label, label_formatter=_format_duration_bucket_label,
explicit_step=summary_options.duration_bucket_seconds,
), ),
size_histogram=_bucketize_numeric( size_histogram=_bucketize_numeric(
sizes, sizes,
step_candidates=SIZE_STEP_CANDIDATES, step_candidates=SIZE_STEP_CANDIDATES,
label_formatter=_format_size_bucket_label, label_formatter=_format_size_bucket_label,
explicit_step=float(summary_options.size_bucket_bytes) if summary_options.size_bucket_bytes is not None else None,
), ),
video_codecs=_build_ranked_counts(codecs), video_codecs=_build_ranked_counts(codecs),
resolutions=_build_ranked_counts(resolutions), resolutions=_build_ranked_counts(resolutions),
+50 -10
View File
@@ -1,31 +1,65 @@
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import math
import sys import sys
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import Sequence from typing import Sequence
from video_metadata_analyzer.analysis import build_summary from video_metadata_analyzer.analysis import MEGABYTE, SummaryOptions, build_summary
from video_metadata_analyzer.discovery import discover_video_files from video_metadata_analyzer.discovery import discover_video_files
from video_metadata_analyzer.models import ProbeFailure, VideoMetadata from video_metadata_analyzer.models import ProbeFailure, VideoMetadata
from video_metadata_analyzer.probe import CommandRunner, probe_tool_versions, probe_video_metadata from video_metadata_analyzer.probe import CommandRunner, probe_tool_versions, probe_video_metadata
from video_metadata_analyzer.reporting import rebuild_report_bundle_from_json, write_report_bundle from video_metadata_analyzer.reporting import rebuild_report_bundle_from_json, write_report_bundle
def _parse_positive_float(value: str, *, label: str) -> float:
try:
parsed = float(value)
except ValueError as error:
raise argparse.ArgumentTypeError(f"{label} 必须是大于 0 的数字。") from error
if not math.isfinite(parsed) or parsed <= 0:
raise argparse.ArgumentTypeError(f"{label} 必须是大于 0 的数字。")
return parsed
def _parse_duration_bucket_seconds(value: str) -> float:
return _parse_positive_float(value, label="时长粒度")
def _parse_size_bucket_megabytes(value: str) -> int:
parsed = _parse_positive_float(value, label="大小粒度")
size_bytes = round(parsed * MEGABYTE)
if size_bytes <= 0:
raise argparse.ArgumentTypeError("大小粒度换算后的字节数必须大于 0。")
return size_bytes
def build_parser() -> argparse.ArgumentParser: def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="video-metadata-analyzer", prog="video-metadata-analyzer",
description="递归扫描目录中的视频文件并生成元数据分析报告。", description="递归扫描目录中的视频文件,并输出元数据分析报告。",
) )
parser.add_argument("inputs", nargs="*", help="一个或多个待扫描的目录或视频文件路径") parser.add_argument("inputs", nargs="*", help="一个或多个需要扫描的目录或视频文件路径")
parser.add_argument( parser.add_argument(
"--output-dir", "--output-dir",
help="报告输出目录,默认写入 reports/<timestamp>", help="报告输出目录,默认写入 reports/<timestamp>",
) )
parser.add_argument( parser.add_argument(
"--report-json", "--report-json",
help="基于已有 report.json 重建 summary 和 HTML,不再重新扫描视频文件", help="基于已有 report.json 重建 summary、CSV 和 HTML,而不是重新扫描视频文件",
)
parser.add_argument(
"--duration-bucket-seconds",
type=_parse_duration_bucket_seconds,
help="显式指定时长分布直方图的分桶粒度,单位为秒。",
)
parser.add_argument(
"--size-bucket-megabytes",
type=_parse_size_bucket_megabytes,
help="显式指定大小分布直方图的分桶粒度,单位为 MB。",
) )
parser.add_argument("--ffprobe-bin", default="ffprobe", help="ffprobe 可执行文件路径") parser.add_argument("--ffprobe-bin", default="ffprobe", help="ffprobe 可执行文件路径")
parser.add_argument("--ffmpeg-bin", default="ffmpeg", help="ffmpeg 可执行文件路径") parser.add_argument("--ffmpeg-bin", default="ffmpeg", help="ffmpeg 可执行文件路径")
@@ -63,14 +97,19 @@ def main(
) -> int: ) -> int:
parser = build_parser() parser = build_parser()
args = parser.parse_args(list(argv) if argv is not None else None) args = parser.parse_args(list(argv) if argv is not None else None)
summary_options = SummaryOptions(
duration_bucket_seconds=args.duration_bucket_seconds,
size_bucket_bytes=args.size_bucket_megabytes,
)
if args.report_json: if args.report_json:
if args.inputs: if args.inputs:
parser.error("使用 --report-json 时不要再传入扫描输入路径。") parser.error("使用 --report-json 时,不要再传入扫描输入路径。")
paths = rebuild_report_bundle_from_json( paths = rebuild_report_bundle_from_json(
Path(args.report_json), Path(args.report_json),
output_dir=Path(args.output_dir) if args.output_dir else None, output_dir=Path(args.output_dir) if args.output_dir else None,
summary_options=summary_options,
) )
print("已基于现有 report.json 重建 summary 与报告。") print("已基于现有 report.json 重建 summary 与报告。")
print(f"JSON 报告:{paths['json']}") print(f"JSON 报告:{paths['json']}")
@@ -93,9 +132,9 @@ def main(
ffmpeg_bin=args.ffmpeg_bin, ffmpeg_bin=args.ffmpeg_bin,
runner=command_runner, runner=command_runner,
) )
print("正在递归扫描输入路径,统计待分析的视频文件...", file=sys.stderr) print("正在递归扫描输入路径,并统计待处理的视频文件...", file=sys.stderr)
video_files = discover_video_files(input_paths) video_files = discover_video_files(input_paths)
print(f"已发现 {len(video_files)} 个视频文件,开始提取元数据...", file=sys.stderr) print(f"已发现 {len(video_files)} 个视频文件,开始读取元数据...", file=sys.stderr)
records: list[VideoMetadata] = [] records: list[VideoMetadata] = []
failures: list[ProbeFailure] = [] failures: list[ProbeFailure] = []
total_files = len(video_files) total_files = len(video_files)
@@ -121,19 +160,20 @@ def main(
sys.stderr.write("\n") sys.stderr.write("\n")
sys.stderr.flush() sys.stderr.flush()
summary = build_summary(records, failures=failures) summary = build_summary(records, failures=failures, summary_options=summary_options)
paths = write_report_bundle( paths = write_report_bundle(
output_dir=output_dir, output_dir=output_dir,
scanned_inputs=input_paths, scanned_inputs=input_paths,
records=records, records=records,
failures=failures, failures=failures,
summary=summary, summary=summary,
summary_options=summary_options,
tool_versions=tool_versions, tool_versions=tool_versions,
) )
print( print(
"扫描完成:" "扫描完成,"
f"共发现 {summary.total_files} 个视频文件," f"共处理 {summary.total_files} 个视频文件,"
f"成功 {summary.successful_files} 个," f"成功 {summary.successful_files} 个,"
f"失败 {summary.failed_files} 个。" f"失败 {summary.failed_files} 个。"
) )
+35 -2
View File
@@ -8,7 +8,7 @@ from html import escape
from pathlib import Path from pathlib import Path
from typing import Iterable from typing import Iterable
from video_metadata_analyzer.analysis import AnalysisSummary, CategoryCount, HistogramBin, build_summary from video_metadata_analyzer.analysis import AnalysisSummary, CategoryCount, HistogramBin, SummaryOptions, build_summary
from video_metadata_analyzer.models import ProbeFailure, VideoMetadata from video_metadata_analyzer.models import ProbeFailure, VideoMetadata
@@ -52,6 +52,26 @@ def _summary_to_dict(summary: AnalysisSummary) -> dict[str, object]:
} }
def _summary_options_to_dict(summary_options: SummaryOptions) -> dict[str, object]:
return {
"duration_bucket_seconds": summary_options.duration_bucket_seconds,
"size_bucket_bytes": summary_options.size_bucket_bytes,
}
def _summary_options_from_dict(payload: dict[str, object]) -> SummaryOptions:
duration_bucket_seconds = payload.get("duration_bucket_seconds")
size_bucket_bytes = payload.get("size_bucket_bytes")
return SummaryOptions(
duration_bucket_seconds=(
float(duration_bucket_seconds)
if duration_bucket_seconds is not None
else None
),
size_bucket_bytes=int(size_bucket_bytes) if size_bucket_bytes is not None else None,
)
def _record_from_dict(payload: dict[str, object]) -> VideoMetadata: def _record_from_dict(payload: dict[str, object]) -> VideoMetadata:
return VideoMetadata( return VideoMetadata(
path=Path(str(payload["path"])), path=Path(str(payload["path"])),
@@ -85,12 +105,14 @@ def _write_json(
records: list[VideoMetadata], records: list[VideoMetadata],
failures: list[ProbeFailure], failures: list[ProbeFailure],
summary: AnalysisSummary, summary: AnalysisSummary,
summary_options: SummaryOptions,
tool_versions: dict[str, str], tool_versions: dict[str, str],
) -> None: ) -> None:
payload = { payload = {
"generated_at": datetime.now().isoformat(timespec="seconds"), "generated_at": datetime.now().isoformat(timespec="seconds"),
"scanned_inputs": [path.as_posix() for path in scanned_inputs], "scanned_inputs": [path.as_posix() for path in scanned_inputs],
"tool_versions": tool_versions, "tool_versions": tool_versions,
"summary_options": _summary_options_to_dict(summary_options),
"summary": _summary_to_dict(summary), "summary": _summary_to_dict(summary),
"files": [_record_to_dict(record) for record in records], "files": [_record_to_dict(record) for record in records],
"failures": [_failure_to_dict(failure) for failure in failures], "failures": [_failure_to_dict(failure) for failure in failures],
@@ -404,7 +426,9 @@ def write_report_bundle(
failures: list[ProbeFailure], failures: list[ProbeFailure],
summary: AnalysisSummary, summary: AnalysisSummary,
tool_versions: dict[str, str], tool_versions: dict[str, str],
summary_options: SummaryOptions | None = None,
) -> dict[str, Path]: ) -> dict[str, Path]:
summary_options = summary_options or SummaryOptions()
output_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True)
json_path = output_dir / "report.json" json_path = output_dir / "report.json"
csv_path = output_dir / "files.csv" csv_path = output_dir / "files.csv"
@@ -416,6 +440,7 @@ def write_report_bundle(
records=records, records=records,
failures=failures, failures=failures,
summary=summary, summary=summary,
summary_options=summary_options,
tool_versions=tool_versions, tool_versions=tool_versions,
) )
_write_csv(csv_path, records, failures) _write_csv(csv_path, records, failures)
@@ -434,6 +459,7 @@ def rebuild_report_bundle_from_json(
report_json_path: Path, report_json_path: Path,
*, *,
output_dir: Path | None = None, output_dir: Path | None = None,
summary_options: SummaryOptions | None = None,
) -> dict[str, Path]: ) -> dict[str, Path]:
payload = json.loads(report_json_path.read_text(encoding="utf-8")) payload = json.loads(report_json_path.read_text(encoding="utf-8"))
records = [_record_from_dict(item) for item in payload.get("files", [])] records = [_record_from_dict(item) for item in payload.get("files", [])]
@@ -443,7 +469,13 @@ def rebuild_report_bundle_from_json(
str(key): str(value) str(key): str(value)
for key, value in dict(payload.get("tool_versions", {})).items() for key, value in dict(payload.get("tool_versions", {})).items()
} }
summary = build_summary(records, failures=failures) stored_summary_options = _summary_options_from_dict(dict(payload.get("summary_options", {})))
resolved_summary_options = stored_summary_options.merged_with(summary_options)
summary = build_summary(
records,
failures=failures,
summary_options=resolved_summary_options,
)
return write_report_bundle( return write_report_bundle(
output_dir=output_dir or report_json_path.parent, output_dir=output_dir or report_json_path.parent,
@@ -451,5 +483,6 @@ def rebuild_report_bundle_from_json(
records=records, records=records,
failures=failures, failures=failures,
summary=summary, summary=summary,
summary_options=resolved_summary_options,
tool_versions=tool_versions, tool_versions=tool_versions,
) )
+29 -1
View File
@@ -1,6 +1,6 @@
from pathlib import Path from pathlib import Path
from video_metadata_analyzer.analysis import build_summary from video_metadata_analyzer.analysis import MEGABYTE, SummaryOptions, build_summary
from video_metadata_analyzer.models import ProbeFailure, VideoMetadata from video_metadata_analyzer.models import ProbeFailure, VideoMetadata
@@ -145,6 +145,34 @@ def test_build_summary_uses_finer_buckets_for_large_sample_sets_with_outliers()
assert summary.size_histogram[0].label == "0-500 MB" assert summary.size_histogram[0].label == "0-500 MB"
def test_build_summary_respects_explicit_histogram_steps() -> None:
records = [
_build_record("a.mp4", size_mb=10, duration_seconds=25),
_build_record("b.mp4", size_mb=40, duration_seconds=65),
_build_record("c.mp4", size_mb=60, duration_seconds=115),
]
summary = build_summary(
records,
summary_options=SummaryOptions(
duration_bucket_seconds=30,
size_bucket_bytes=25 * MEGABYTE,
),
)
assert [(item.label, item.count) for item in summary.duration_histogram] == [
("0-30 s", 1),
("30-60 s", 0),
("60-90 s", 1),
("90-120 s", 1),
]
assert [(item.label, item.count) for item in summary.size_histogram] == [
("0-25 MB", 1),
("25-50 MB", 1),
("50-75 MB", 1),
]
def test_build_summary_tracks_successful_and_failed_files() -> None: def test_build_summary_tracks_successful_and_failed_files() -> None:
records = [ records = [
VideoMetadata( VideoMetadata(
+146
View File
@@ -53,6 +53,72 @@ def test_main_scans_input_directory_and_writes_reports(tmp_path: Path) -> None:
assert (output_dir / "report.html").exists() assert (output_dir / "report.html").exists()
def test_main_writes_reports_with_explicit_histogram_granularity(tmp_path: Path) -> None:
input_dir = tmp_path / "videos"
input_dir.mkdir()
sample = input_dir / "sample.mp4"
sample.write_bytes(b"x" * 2048)
output_dir = tmp_path / "output"
def fake_runner(command: list[str]) -> str:
if command[:2] == ["ffprobe", "-version"]:
return "ffprobe version 8.0.1\nbuilt with ...\n"
if command[:2] == ["ffmpeg", "-version"]:
return "ffmpeg version 8.0.1\nbuilt with ...\n"
if command[0] == "ffprobe":
return json.dumps(
{
"format": {
"filename": str(sample),
"format_name": "mov,mp4,m4a,3gp,3g2,mj2",
"duration": "65.000000",
"bit_rate": "1024000",
"size": str(60 * 1024 * 1024),
},
"streams": [
{
"codec_type": "video",
"codec_name": "h264",
"width": 1280,
"height": 720,
"avg_frame_rate": "25/1",
}
],
}
)
raise AssertionError(f"unexpected command: {command}")
exit_code = main(
[
str(input_dir),
"--output-dir",
str(output_dir),
"--duration-bucket-seconds",
"30",
"--size-bucket-megabytes",
"25",
],
command_runner=fake_runner,
)
assert exit_code == 0
payload = json.loads((output_dir / "report.json").read_text(encoding="utf-8"))
assert payload["summary_options"] == {
"duration_bucket_seconds": 30.0,
"size_bucket_bytes": 25 * 1024 * 1024,
}
assert payload["summary"]["duration_histogram"] == [
{"label": "0-30 s", "count": 0},
{"label": "30-60 s", "count": 0},
{"label": "60-90 s", "count": 1},
]
assert payload["summary"]["size_histogram"] == [
{"label": "0-25 MB", "count": 0},
{"label": "25-50 MB", "count": 0},
{"label": "50-75 MB", "count": 1},
]
def test_main_continues_when_single_file_probe_fails(tmp_path: Path) -> None: def test_main_continues_when_single_file_probe_fails(tmp_path: Path) -> None:
input_dir = tmp_path / "videos" input_dir = tmp_path / "videos"
input_dir.mkdir() input_dir.mkdir()
@@ -228,3 +294,83 @@ def test_main_rebuilds_html_from_existing_report_json_without_running_probe(tmp_
assert payload["summary"]["duration_histogram"] == [{"label": "0-2 min", "count": 1}] assert payload["summary"]["duration_histogram"] == [{"label": "0-2 min", "count": 1}]
assert payload["summary"]["size_histogram"] == [{"label": "0-100 MB", "count": 1}] assert payload["summary"]["size_histogram"] == [{"label": "0-100 MB", "count": 1}]
assert "0-2 min" in html_text assert "0-2 min" in html_text
def test_main_rebuilds_report_json_with_saved_histogram_granularity_and_cli_override(tmp_path: Path) -> None:
output_dir = tmp_path / "report"
output_dir.mkdir()
report_json = output_dir / "report.json"
report_json.write_text(
json.dumps(
{
"generated_at": "2026-04-02T12:00:00",
"scanned_inputs": ["videos"],
"tool_versions": {"ffprobe": "cached ffprobe", "ffmpeg": "cached ffmpeg"},
"summary_options": {
"duration_bucket_seconds": 30.0,
"size_bucket_bytes": 25 * 1024 * 1024,
},
"summary": {
"total_files": 1,
"successful_files": 1,
"failed_files": 0,
"total_size_bytes": 0,
"total_duration_seconds": 0,
"duration_histogram": [{"label": "0-30 s", "count": 1}],
"size_histogram": [{"label": "0-25 MB", "count": 1}],
"video_codecs": [],
"resolutions": [],
},
"files": [
{
"path": "videos/sample.mp4",
"size_bytes": 60 * 1024 * 1024,
"duration_seconds": 65.0,
"format_name": "mov,mp4,m4a,3gp,3g2,mj2",
"bit_rate": 1024000,
"video_codec": "h264",
"audio_codec": "aac",
"width": 1280,
"height": 720,
"frame_rate": 25.0,
"video_stream_count": 1,
"audio_stream_count": 1,
"subtitle_stream_count": 0,
}
],
"failures": [],
},
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
def forbidden_runner(command: list[str]) -> str:
raise AssertionError(f"should not invoke external command: {command}")
exit_code = main(
[
"--report-json",
str(report_json),
"--duration-bucket-seconds",
"60",
],
command_runner=forbidden_runner,
)
assert exit_code == 0
payload = json.loads(report_json.read_text(encoding="utf-8"))
assert payload["summary_options"] == {
"duration_bucket_seconds": 60.0,
"size_bucket_bytes": 25 * 1024 * 1024,
}
assert payload["summary"]["duration_histogram"] == [
{"label": "0-1 min", "count": 0},
{"label": "1-2 min", "count": 1},
]
assert payload["summary"]["size_histogram"] == [
{"label": "0-25 MB", "count": 0},
{"label": "25-50 MB", "count": 0},
{"label": "50-75 MB", "count": 1},
]