kim/video-metadata-analyzer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add report rebuild mode and improve histogram buckets

Browse Source
This commit is contained in:
kim
2026-04-02 21:37:32 +08:00
parent ee002eb1a9
commit f243664acf
7 changed files with 256 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+22
View File
@@ -13,6 +13,7 @@
- 单个视频探测失败时不中断整次扫描，会单独记录错误清单
- 汇总时长分布、大小分布、视频编码分布、分辨率分布
- 时长分布、大小分布会根据当前目录样本范围自动调整分桶粒度，避免结果全部挤在默认大桶里
- 可以基于已有 `report.json` 重新计算 `summary`，并重建 `report.html` / `files.csv`，无需再次扫描视频文件
- 生成适合人工查看的 HTML 报告，以及适合后续二次处理的 JSON / CSV
## 依赖要求
@@ -53,6 +54,26 @@ uv run video-metadata-analyzer D:\Videos `
--output-dir .\reports\latest
```
如果你已经有一份旧的 `report.json`，只想在新版本逻辑下重算汇总并重建报告，不想重新扫描视频文件：
```powershell
uv run video-metadata-analyzer --report-json .\reports\latest\report.json
```
默认会直接在该 `report.json` 所在目录重写：
- `report.json`
- `files.csv`
- `report.html`
如果你希望把重建后的结果写到新目录：
```powershell
uv run video-metadata-analyzer `
--report-json .\reports\latest\report.json `
--output-dir .\reports\rebuilt
```
## 输出内容
生成目录内默认包含以下文件：
@@ -66,6 +87,7 @@ uv run video-metadata-analyzer D:\Videos `
- `report.json` 包含 `summary`、成功文件列表 `files`，以及失败文件列表 `failures`
- `files.csv` 同时输出成功和失败结果，带 `status`、`error_type`、`error_message` 字段
- `report.html` 是离线静态页面，可以直接双击打开查看，并显示错误清单面板
- 当使用 `--report-json` 时，会从 `files` / `failures` 重新计算 `summary`，再生成新的 `report.json`、`files.csv`、`report.html`
## 开发与测试
src/video_metadata_analyzer/analysis.py
+11 -2
View File
@@ -28,7 +28,10 @@ SIZE_STEP_CANDIDATES = [
4 * GIGABYTE,
8 * GIGABYTE,
]
MAX_DYNAMIC_HISTOGRAM_BINS = 8
DEFAULT_DYNAMIC_HISTOGRAM_BINS = 8
MEDIUM_DYNAMIC_HISTOGRAM_BINS = 12
LARGE_DYNAMIC_HISTOGRAM_BINS = 16
XLARGE_DYNAMIC_HISTOGRAM_BINS = 24
@dataclass(slots=True)
@@ -89,7 +92,13 @@ def _bucketize_numeric(
def _resolve_histogram_max_bins(value_count: int) -> int:
if value_count <= 1:
return 1
return min(MAX_DYNAMIC_HISTOGRAM_BINS, max(2, value_count * 2))
if value_count <= 10:
return min(DEFAULT_DYNAMIC_HISTOGRAM_BINS, max(2, value_count * 2))
if value_count <= 100:
return MEDIUM_DYNAMIC_HISTOGRAM_BINS
if value_count <= 1_000:
return LARGE_DYNAMIC_HISTOGRAM_BINS
return XLARGE_DYNAMIC_HISTOGRAM_BINS
def _select_histogram_step(*, max_value: float, step_candidates: list[float], value_count: int) -> float:
src/video_metadata_analyzer/cli.py
+23 -2
View File
@@ -10,7 +10,7 @@ from video_metadata_analyzer.analysis import build_summary
from video_metadata_analyzer.discovery import discover_video_files
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.reporting import write_report_bundle
from video_metadata_analyzer.reporting import rebuild_report_bundle_from_json, write_report_bundle
def build_parser() -> argparse.ArgumentParser:
@@ -18,11 +18,15 @@ def build_parser() -> argparse.ArgumentParser:
prog="video-metadata-analyzer",
description="递归扫描目录中的视频文件并生成元数据分析报告。",
)
parser.add_argument("inputs", nargs="+", help="一个或多个待扫描的目录或视频文件路径")
parser.add_argument("inputs", nargs="*", help="一个或多个待扫描的目录或视频文件路径")
parser.add_argument(
"--output-dir",
help="报告输出目录，默认写入 reports/<timestamp>",
)
parser.add_argument(
"--report-json",
help="基于已有 report.json 重建 summary 和 HTML，不再重新扫描视频文件",
)
parser.add_argument("--ffprobe-bin", default="ffprobe", help="ffprobe 可执行文件路径")
parser.add_argument("--ffmpeg-bin", default="ffmpeg", help="ffmpeg 可执行文件路径")
return parser
@@ -60,6 +64,23 @@ def main(
parser = build_parser()
args = parser.parse_args(list(argv) if argv is not None else None)
if args.report_json:
if args.inputs:
parser.error("使用 --report-json 时不要再传入扫描输入路径。")
paths = rebuild_report_bundle_from_json(
Path(args.report_json),
output_dir=Path(args.output_dir) if args.output_dir else None,
)
print("已基于现有 report.json 重建 summary 与报告。")
print(f"JSON 报告：{paths['json']}")
print(f"CSV 明细：{paths['csv']}")
print(f"HTML 报告：{paths['html']}")
return 0
if not args.inputs:
parser.error("请提供至少一个扫描输入路径，或使用 --report-json 指定已有报告。")
input_paths = [Path(item) for item in args.inputs]
output_dir = (
Path(args.output_dir)
src/video_metadata_analyzer/reporting.py
+52 -1
View File
@@ -8,7 +8,7 @@ from html import escape
from pathlib import Path
from typing import Iterable
from video_metadata_analyzer.analysis import AnalysisSummary, CategoryCount, HistogramBin
from video_metadata_analyzer.analysis import AnalysisSummary, CategoryCount, HistogramBin, build_summary
from video_metadata_analyzer.models import ProbeFailure, VideoMetadata
@@ -52,6 +52,32 @@ def _summary_to_dict(summary: AnalysisSummary) -> dict[str, object]:
}
def _record_from_dict(payload: dict[str, object]) -> VideoMetadata:
return VideoMetadata(
path=Path(str(payload["path"])),
size_bytes=int(payload["size_bytes"]),
duration_seconds=float(payload["duration_seconds"]) if payload["duration_seconds"] is not None else None,
format_name=str(payload["format_name"]) if payload["format_name"] is not None else None,
bit_rate=int(payload["bit_rate"]) if payload["bit_rate"] is not None else None,
video_codec=str(payload["video_codec"]) if payload["video_codec"] is not None else None,
audio_codec=str(payload["audio_codec"]) if payload["audio_codec"] is not None else None,
width=int(payload["width"]) if payload["width"] is not None else None,
height=int(payload["height"]) if payload["height"] is not None else None,
frame_rate=float(payload["frame_rate"]) if payload["frame_rate"] is not None else None,
video_stream_count=int(payload["video_stream_count"]),
audio_stream_count=int(payload["audio_stream_count"]),
subtitle_stream_count=int(payload["subtitle_stream_count"]),
)
def _failure_from_dict(payload: dict[str, object]) -> ProbeFailure:
return ProbeFailure(
path=Path(str(payload["path"])),
error_type=str(payload["error_type"]),
error_message=str(payload["error_message"]),
)
def _write_json(
output_path: Path,
*,
@@ -402,3 +428,28 @@ def write_report_bundle(
)
return {"json": json_path, "csv": csv_path, "html": html_path}
def rebuild_report_bundle_from_json(
report_json_path: Path,
*,
output_dir: Path | None = None,
) -> dict[str, Path]:
payload = json.loads(report_json_path.read_text(encoding="utf-8"))
records = [_record_from_dict(item) for item in payload.get("files", [])]
failures = [_failure_from_dict(item) for item in payload.get("failures", [])]
scanned_inputs = [Path(str(item)) for item in payload.get("scanned_inputs", [])]
tool_versions = {
str(key): str(value)
for key, value in dict(payload.get("tool_versions", {})).items()
}
summary = build_summary(records, failures=failures)
return write_report_bundle(
output_dir=output_dir or report_json_path.parent,
scanned_inputs=scanned_inputs,
records=records,
failures=failures,
summary=summary,
tool_versions=tool_versions,
)
tests/test_analysis.py
+24
View File
@@ -121,6 +121,30 @@ def test_build_summary_uses_adaptive_buckets_for_compact_ranges() -> None:
]
def test_build_summary_uses_finer_buckets_for_large_sample_sets_with_outliers() -> None:
records = [
_build_record(
f"clip-{index}.mp4",
size_mb=80 + (index % 8) * 60,
duration_seconds=300 + (index % 9) * 300,
)
for index in range(1_499)
]
records.append(
_build_record(
"feature-film.mkv",
size_mb=6 * 1024,
duration_seconds=6 * 3600,
codec="hevc",
)
)
summary = build_summary(records)
assert summary.duration_histogram[0].label == "0-15 min"
assert summary.size_histogram[0].label == "0-500 MB"
def test_build_summary_tracks_successful_and_failed_files() -> None:
records = [
VideoMetadata(
tests/test_cli.py
+62
View File
@@ -166,3 +166,65 @@ def test_main_reports_scan_progress_to_stderr(tmp_path: Path, capsys) -> None:
assert "已发现 2 个视频文件" in captured.err
assert "2/2" in captured.err
assert "02-second.mp4" in captured.err
def test_main_rebuilds_html_from_existing_report_json_without_running_probe(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": {
"total_files": 500,
"successful_files": 500,
"failed_files": 0,
"total_size_bytes": 0,
"total_duration_seconds": 0,
"duration_histogram": [{"label": "0-1 h", "count": 500}],
"size_histogram": [{"label": "0-1 GB", "count": 500}],
"video_codecs": [],
"resolutions": [],
},
"files": [
{
"path": "videos/sample.mp4",
"size_bytes": 60 * 1024 * 1024,
"duration_seconds": 120.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)],
command_runner=forbidden_runner,
)
assert exit_code == 0
payload = json.loads(report_json.read_text(encoding="utf-8"))
html_text = (output_dir / "report.html").read_text(encoding="utf-8")
assert payload["summary"]["duration_histogram"] == [{"label": "0-2 min", "count": 1}]
assert payload["summary"]["size_histogram"] == [{"label": "0-100 MB", "count": 1}]
assert "0-2 min" in html_text
tests/test_report.py
+62 -1
View File
@@ -5,7 +5,7 @@ from pathlib import Path
from video_metadata_analyzer.analysis import build_summary
from video_metadata_analyzer.models import ProbeFailure, VideoMetadata
from video_metadata_analyzer.reporting import write_report_bundle
from video_metadata_analyzer.reporting import rebuild_report_bundle_from_json, write_report_bundle
def test_write_report_bundle_creates_json_csv_and_html(tmp_path: Path) -> None:
@@ -70,3 +70,64 @@ def test_write_report_bundle_creates_json_csv_and_html(tmp_path: Path) -> None:
assert "grid-template-columns: repeat(auto-fit, minmax(520px, 1fr));" in html_text
assert "font-size: 16px;" in html_text
assert "font-size: 18px;" in html_text
def test_rebuild_report_bundle_from_json_recomputes_summary_and_html(tmp_path: Path) -> None:
report_dir = tmp_path / "report"
report_dir.mkdir()
report_json = report_dir / "report.json"
payload = {
"generated_at": "2026-04-02T12:00:00",
"scanned_inputs": ["videos"],
"tool_versions": {"ffprobe": "ffprobe version x", "ffmpeg": "ffmpeg version y"},
"summary": {
"total_files": 999,
"successful_files": 999,
"failed_files": 0,
"total_size_bytes": 0,
"total_duration_seconds": 0,
"duration_histogram": [{"label": "0-1 h", "count": 999}],
"size_histogram": [{"label": "0-1 GB", "count": 999}],
"video_codecs": [],
"resolutions": [],
},
"files": [
{
"path": "videos/a.mp4",
"size_bytes": 60 * 1024 * 1024,
"duration_seconds": 120.0,
"format_name": "mov,mp4,m4a,3gp,3g2,mj2",
"bit_rate": 1500000,
"video_codec": "h264",
"audio_codec": "aac",
"width": 1920,
"height": 1080,
"frame_rate": 30.0,
"video_stream_count": 1,
"audio_stream_count": 1,
"subtitle_stream_count": 0,
}
],
"failures": [
{
"path": "videos/broken.mp4",
"error_type": "CalledProcessError",
"error_message": "moov atom not found",
}
],
}
report_json.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
paths = rebuild_report_bundle_from_json(report_json)
rebuilt_payload = json.loads(paths["json"].read_text(encoding="utf-8"))
rebuilt_html = paths["html"].read_text(encoding="utf-8")
assert rebuilt_payload["summary"]["total_files"] == 2
assert rebuilt_payload["summary"]["successful_files"] == 1
assert rebuilt_payload["summary"]["failed_files"] == 1
assert rebuilt_payload["summary"]["duration_histogram"] == [{"label": "0-2 min", "count": 1}]
assert rebuilt_payload["summary"]["size_histogram"] == [{"label": "0-100 MB", "count": 1}]
assert "视频元数据分析报告" in rebuilt_html
assert "0-2 min" in rebuilt_html
assert "0-1 h" not in rebuilt_html