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

feat: isolate probe failures in reports

Browse Source
This commit is contained in:
kim
2026-04-02 18:38:09 +08:00
parent bc242f8a31
commit 7dd47a51ff
8 changed files with 293 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+6 -4
View File
@@ -10,6 +10,7 @@
- 输入一个或多个目录，递归发现常见视频文件
- 提取视频时长、大小、编码、分辨率、帧率、音频编码等字段
- 单个视频探测失败时不中断整次扫描，会单独记录错误清单
- 汇总时长分布、大小分布、视频编码分布、分辨率分布
- 生成适合人工查看的 HTML 报告，以及适合后续二次处理的 JSON / CSV
@@ -59,7 +60,11 @@ uv run video-metadata-analyzer D:\Videos `
- `files.csv`
- `report.html`
其中 `report.html` 是离线静态页面，可以直接双击打开查看。
其中：
- `report.json` 包含 `summary`、成功文件列表 `files`，以及失败文件列表 `failures`
- `files.csv` 同时输出成功和失败结果，带 `status`、`error_type`、`error_message` 字段
- `report.html` 是离线静态页面，可以直接双击打开查看，并显示错误清单面板
## 开发与测试
@@ -80,8 +85,5 @@ uv run video-metadata-analyzer --help
当前版本优先保证“扫描、提取、汇总、可视化”这条主链路可用，暂时没有做这些增强项：
- 损坏视频的深度解码验证
- 单文件失败后的错误报告面板
- 更复杂的前端交互图表
- 增量扫描与数据库缓存
如果后续要继续做，最值得优先加的是“探测失败文件单独记录，不让整次扫描因为一个坏文件直接中断”。
src/video_metadata_analyzer/analysis.py
+12 -3
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
from collections import Counter
from dataclasses import dataclass
from video_metadata_analyzer.models import VideoMetadata
from video_metadata_analyzer.models import ProbeFailure, VideoMetadata
MEGABYTE = 1024 * 1024
@@ -41,6 +41,8 @@ class CategoryCount:
@dataclass(slots=True)
class AnalysisSummary:
total_files: int
successful_files: int
failed_files: int
total_size_bytes: int
total_duration_seconds: float
duration_histogram: list[HistogramBin]
@@ -77,7 +79,12 @@ def _build_ranked_counts(labels: list[str]) -> list[CategoryCount]:
return [CategoryCount(label=label, count=count) for label, count in ordered]
def build_summary(records: list[VideoMetadata]) -> AnalysisSummary:
def build_summary(
records: list[VideoMetadata],
*,
failures: list[ProbeFailure] | None = None,
) -> AnalysisSummary:
failures = failures or []
durations = [record.duration_seconds for record in records if record.duration_seconds is not None]
sizes = [record.size_bytes for record in records]
codecs = [record.video_codec for record in records if record.video_codec]
@@ -88,7 +95,9 @@ def build_summary(records: list[VideoMetadata]) -> AnalysisSummary:
]
return AnalysisSummary(
total_files=len(records),
total_files=len(records) + len(failures),
successful_files=len(records),
failed_files=len(failures),
total_size_bytes=sum(sizes),
total_duration_seconds=sum(durations),
duration_histogram=_bucketize_numeric(durations, DURATION_BUCKETS),
src/video_metadata_analyzer/cli.py
+33 -5
View File
@@ -7,6 +7,7 @@ from typing import Sequence
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
@@ -26,6 +27,15 @@ def build_parser() -> argparse.ArgumentParser:
return parser
def _format_probe_error(error: Exception) -> str:
for field_name in ("stderr", "stdout"):
value = getattr(error, field_name, None)
if isinstance(value, str) and value.strip():
return value.strip()
message = str(error).strip()
return message or error.__class__.__name__
def main(
argv: Sequence[str] | None = None,
*,
@@ -47,24 +57,42 @@ def main(
runner=command_runner,
)
video_files = discover_video_files(input_paths)
records = [
records: list[VideoMetadata] = []
failures: list[ProbeFailure] = []
for path in video_files:
try:
records.append(
probe_video_metadata(
path,
ffprobe_bin=args.ffprobe_bin,
runner=command_runner,
)
for path in video_files
]
summary = build_summary(records)
)
except Exception as error:
failures.append(
ProbeFailure(
path=path,
error_type=error.__class__.__name__,
error_message=_format_probe_error(error),
)
)
summary = build_summary(records, failures=failures)
paths = write_report_bundle(
output_dir=output_dir,
scanned_inputs=input_paths,
records=records,
failures=failures,
summary=summary,
tool_versions=tool_versions,
)
print(f"扫描完成：共发现 {summary.total_files} 个视频文件。")
print(
"扫描完成："
f"共发现 {summary.total_files} 个视频文件，"
f"成功 {summary.successful_files} 个，"
f"失败 {summary.failed_files} 个。"
)
print(f"JSON 报告：{paths['json']}")
print(f"CSV 明细：{paths['csv']}")
print(f"HTML 报告：{paths['html']}")
src/video_metadata_analyzer/models.py
+7
View File
@@ -19,3 +19,10 @@ class VideoMetadata:
video_stream_count: int
audio_stream_count: int
subtitle_stream_count: int
@dataclass(slots=True)
class ProbeFailure:
path: Path
error_type: str
error_message: str
src/video_metadata_analyzer/reporting.py
+112 -5
View File
@@ -9,7 +9,7 @@ from pathlib import Path
from typing import Iterable
from video_metadata_analyzer.analysis import AnalysisSummary, CategoryCount, HistogramBin
from video_metadata_analyzer.models import VideoMetadata
from video_metadata_analyzer.models import ProbeFailure, VideoMetadata
def _record_to_dict(record: VideoMetadata) -> dict[str, object]:
@@ -30,9 +30,19 @@ def _record_to_dict(record: VideoMetadata) -> dict[str, object]:
}
def _failure_to_dict(failure: ProbeFailure) -> dict[str, str]:
return {
"path": failure.path.as_posix(),
"error_type": failure.error_type,
"error_message": failure.error_message,
}
def _summary_to_dict(summary: AnalysisSummary) -> dict[str, object]:
return {
"total_files": summary.total_files,
"successful_files": summary.successful_files,
"failed_files": summary.failed_files,
"total_size_bytes": summary.total_size_bytes,
"total_duration_seconds": summary.total_duration_seconds,
"duration_histogram": [asdict(item) for item in summary.duration_histogram],
@@ -47,6 +57,7 @@ def _write_json(
*,
scanned_inputs: list[Path],
records: list[VideoMetadata],
failures: list[ProbeFailure],
summary: AnalysisSummary,
tool_versions: dict[str, str],
) -> None:
@@ -56,6 +67,7 @@ def _write_json(
"tool_versions": tool_versions,
"summary": _summary_to_dict(summary),
"files": [_record_to_dict(record) for record in records],
"failures": [_failure_to_dict(failure) for failure in failures],
}
output_path.write_text(
json.dumps(payload, ensure_ascii=False, indent=2),
@@ -63,8 +75,39 @@ def _write_json(
)
def _write_csv(output_path: Path, records: list[VideoMetadata]) -> None:
def _build_success_csv_row(record: VideoMetadata) -> dict[str, object]:
return {
"status": "ok",
**_record_to_dict(record),
"error_type": "",
"error_message": "",
}
def _build_failure_csv_row(failure: ProbeFailure) -> dict[str, object]:
return {
"status": "error",
"path": failure.path.as_posix(),
"size_bytes": "",
"duration_seconds": "",
"format_name": "",
"bit_rate": "",
"video_codec": "",
"audio_codec": "",
"width": "",
"height": "",
"frame_rate": "",
"video_stream_count": "",
"audio_stream_count": "",
"subtitle_stream_count": "",
"error_type": failure.error_type,
"error_message": failure.error_message,
}
def _write_csv(output_path: Path, records: list[VideoMetadata], failures: list[ProbeFailure]) -> None:
fieldnames = [
"status",
"path",
"size_bytes",
"duration_seconds",
@@ -78,13 +121,17 @@ def _write_csv(output_path: Path, records: list[VideoMetadata]) -> None:
"video_stream_count",
"audio_stream_count",
"subtitle_stream_count",
"error_type",
"error_message",
]
with output_path.open("w", encoding="utf-8-sig", newline="") as handle:
writer = csv.DictWriter(handle, fieldnames=fieldnames)
writer.writeheader()
for record in records:
writer.writerow(_record_to_dict(record))
writer.writerow(_build_success_csv_row(record))
for failure in failures:
writer.writerow(_build_failure_csv_row(failure))
def _format_size(value: int) -> str:
@@ -144,19 +191,49 @@ def _render_bar_chart(title: str, items: Iterable[HistogramBin | CategoryCount])
return "".join(pieces)
def _render_failures(failures: list[ProbeFailure]) -> str:
if not failures:
return (
'<section class="meta">'
"<h2>错误清单</h2>"
"<p>本次扫描没有探测失败的文件。</p>"
"</section>"
)
items = []
for failure in failures:
items.append(
"<li class=\"error-item\">"
f"<span class=\"error-path\">{escape(failure.path.as_posix())}</span>"
f"<span class=\"error-detail\">{escape(failure.error_type)}: {escape(failure.error_message)}</span>"
"</li>"
)
return (
'<section class="meta">'
"<h2>错误清单</h2>"
'<ul class="error-list">'
f"{''.join(items)}"
"</ul>"
"</section>"
)
def _write_html(
output_path: Path,
*,
scanned_inputs: list[Path],
summary: AnalysisSummary,
tool_versions: dict[str, str],
failures: list[ProbeFailure],
) -> None:
summary_cards = f"""
<section class="cards">
<article class="card"><h2>视频文件数</h2><p>{summary.total_files}</p></article>
<article class="card"><h2>成功探测</h2><p>{summary.successful_files}</p></article>
<article class="card"><h2>探测失败</h2><p>{summary.failed_files}</p></article>
<article class="card"><h2>总大小</h2><p>{escape(_format_size(summary.total_size_bytes))}</p></article>
<article class="card"><h2>总时长</h2><p>{escape(_format_duration(summary.total_duration_seconds))}</p></article>
<article class="card"><h2>扫描目录数</h2><p>{len(scanned_inputs)}</p></article>
</section>
"""
@@ -226,6 +303,31 @@ def _write_html(
font-weight: 700;
}}
ul {{ padding-left: 18px; }}
.error-list {{
list-style: none;
padding: 0;
margin: 0;
display: grid;
gap: 12px;
}}
.error-item {{
border: 1px solid rgba(217, 205, 183, 0.85);
border-radius: 14px;
padding: 14px 16px;
background: rgba(255, 255, 255, 0.55);
}}
.error-path {{
display: block;
color: var(--text);
font-weight: 700;
overflow-wrap: anywhere;
}}
.error-detail {{
display: block;
margin-top: 6px;
color: var(--muted);
overflow-wrap: anywhere;
}}
</style>
</head>
<body>
@@ -233,6 +335,7 @@ def _write_html(
<header>
<h1>视频元数据分析报告</h1>
<p>扫描输入：{escape(", ".join(path.as_posix() for path in scanned_inputs))}</p>
<p>扫描目录数：{len(scanned_inputs)}</p>
</header>
{summary_cards}
<section class="charts">
@@ -241,6 +344,7 @@ def _write_html(
{_render_bar_chart("视频编码分布", summary.video_codecs)}
{_render_bar_chart("分辨率分布", summary.resolutions)}
</section>
{_render_failures(failures)}
<section class="meta">
<h2>工具版本</h2>
<ul>
@@ -260,6 +364,7 @@ def write_report_bundle(
output_dir: Path,
scanned_inputs: list[Path],
records: list[VideoMetadata],
failures: list[ProbeFailure],
summary: AnalysisSummary,
tool_versions: dict[str, str],
) -> dict[str, Path]:
@@ -272,15 +377,17 @@ def write_report_bundle(
json_path,
scanned_inputs=scanned_inputs,
records=records,
failures=failures,
summary=summary,
tool_versions=tool_versions,
)
_write_csv(csv_path, records)
_write_csv(csv_path, records, failures)
_write_html(
html_path,
scanned_inputs=scanned_inputs,
summary=summary,
tool_versions=tool_versions,
failures=failures,
)
return {"json": json_path, "csv": csv_path, "html": html_path}
tests/test_analysis.py
+36 -1
View File
@@ -1,7 +1,7 @@
from pathlib import Path
from video_metadata_analyzer.analysis import build_summary
from video_metadata_analyzer.models import VideoMetadata
from video_metadata_analyzer.models import ProbeFailure, VideoMetadata
def test_build_summary_computes_totals_and_histograms() -> None:
@@ -81,3 +81,38 @@ def test_build_summary_computes_totals_and_histograms() -> None:
("1280x720", 1),
("3840x2160", 1),
]
def test_build_summary_tracks_successful_and_failed_files() -> None:
records = [
VideoMetadata(
path=Path("a.mp4"),
size_bytes=1024,
duration_seconds=10,
format_name="mov,mp4,m4a,3gp,3g2,mj2",
bit_rate=1_000_000,
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 = [
ProbeFailure(
path=Path("broken.mp4"),
error_type="CalledProcessError",
error_message="moov atom not found",
)
]
summary = build_summary(records, failures=failures)
assert summary.total_files == 2
assert summary.successful_files == 1
assert summary.failed_files == 1
assert summary.total_size_bytes == 1024
assert summary.total_duration_seconds == 10
tests/test_cli.py
+64
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import json
import subprocess
from pathlib import Path
from video_metadata_analyzer.cli import main
@@ -50,3 +51,66 @@ def test_main_scans_input_directory_and_writes_reports(tmp_path: Path) -> None:
assert (output_dir / "report.json").exists()
assert (output_dir / "files.csv").exists()
assert (output_dir / "report.html").exists()
def test_main_continues_when_single_file_probe_fails(tmp_path: Path) -> None:
input_dir = tmp_path / "videos"
input_dir.mkdir()
good_sample = input_dir / "good.mp4"
bad_sample = input_dir / "bad.mp4"
good_sample.write_bytes(b"x" * 2048)
bad_sample.write_bytes(b"y" * 1024)
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" and command[-1] == str(good_sample):
return json.dumps(
{
"format": {
"filename": str(good_sample),
"format_name": "mov,mp4,m4a,3gp,3g2,mj2",
"duration": "8.000000",
"bit_rate": "1024000",
"size": "2048",
},
"streams": [
{
"codec_type": "video",
"codec_name": "h264",
"width": 1280,
"height": 720,
"avg_frame_rate": "25/1",
}
],
}
)
if command[0] == "ffprobe" and command[-1] == str(bad_sample):
raise subprocess.CalledProcessError(
returncode=1,
cmd=command,
stderr="moov atom not found",
)
raise AssertionError(f"unexpected command: {command}")
exit_code = main(
[str(input_dir), "--output-dir", str(output_dir)],
command_runner=fake_runner,
)
assert exit_code == 0
payload = json.loads((output_dir / "report.json").read_text(encoding="utf-8"))
assert payload["summary"]["total_files"] == 2
assert payload["summary"]["successful_files"] == 1
assert payload["summary"]["failed_files"] == 1
assert payload["files"][0]["path"] == good_sample.as_posix()
assert payload["failures"] == [
{
"path": bad_sample.as_posix(),
"error_type": "CalledProcessError",
"error_message": "moov atom not found",
}
]
tests/test_report.py
+18 -4
View File
@@ -4,7 +4,7 @@ import json
from pathlib import Path
from video_metadata_analyzer.analysis import build_summary
from video_metadata_analyzer.models import VideoMetadata
from video_metadata_analyzer.models import ProbeFailure, VideoMetadata
from video_metadata_analyzer.reporting import write_report_bundle
@@ -25,12 +25,18 @@ def test_write_report_bundle_creates_json_csv_and_html(tmp_path: Path) -> None:
audio_stream_count=1,
subtitle_stream_count=0,
)
summary = build_summary([record])
failure = ProbeFailure(
path=Path("videos/broken.mp4"),
error_type="CalledProcessError",
error_message="moov atom not found",
)
summary = build_summary([record], failures=[failure])
paths = write_report_bundle(
output_dir=output_dir,
scanned_inputs=[Path("videos")],
records=[record],
failures=[failure],
summary=summary,
tool_versions={"ffprobe": "ffprobe version x", "ffmpeg": "ffmpeg version y"},
)
@@ -41,15 +47,23 @@ def test_write_report_bundle_creates_json_csv_and_html(tmp_path: Path) -> None:
assert paths["html"].exists()
payload = json.loads(paths["json"].read_text(encoding="utf-8"))
assert payload["summary"]["total_files"] == 1
assert payload["summary"]["total_files"] == 2
assert payload["summary"]["successful_files"] == 1
assert payload["summary"]["failed_files"] == 1
assert payload["tool_versions"]["ffmpeg"] == "ffmpeg version y"
assert payload["files"][0]["path"] == "videos/sample.mp4"
assert payload["failures"][0]["path"] == "videos/broken.mp4"
assert payload["failures"][0]["error_message"] == "moov atom not found"
csv_text = paths["csv"].read_text(encoding="utf-8")
assert "path,size_bytes,duration_seconds" in csv_text
assert "status,path,size_bytes,duration_seconds" in csv_text
assert "videos/sample.mp4" in csv_text
assert "videos/broken.mp4" in csv_text
assert "CalledProcessError" in csv_text
html_text = paths["html"].read_text(encoding="utf-8")
assert "视频元数据分析报告" in html_text
assert "时长分布" in html_text
assert "错误清单" in html_text
assert "moov atom not found" in html_text
assert "<svg" in html_text