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

feat: adapt histogram ranges to scanned dataset

Browse Source
This commit is contained in:
kim
2026-04-02 19:00:59 +08:00
parent c5e570f382
commit 915386b778
3 changed files with 155 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1
View File
@@ -12,6 +12,7 @@
- 提取视频时长、大小、编码、分辨率、帧率、音频编码等字段
- 单个视频探测失败时不中断整次扫描，会单独记录错误清单
- 汇总时长分布、大小分布、视频编码分布、分辨率分布
- 时长分布、大小分布会根据当前目录样本范围自动调整分桶粒度，避免结果全部挤在默认大桶里
- 生成适合人工查看的 HTML 报告，以及适合后续二次处理的 JSON / CSV
## 依赖要求
src/video_metadata_analyzer/analysis.py
+91 -20
View File
@@ -1,29 +1,34 @@
from __future__ import annotations
import math
from collections import Counter
from dataclasses import dataclass
from typing import Callable
from video_metadata_analyzer.models import ProbeFailure, VideoMetadata
KILOBYTE = 1024
MEGABYTE = 1024 * 1024
GIGABYTE = 1024 * 1024 * 1024
DURATION_BUCKETS: list[tuple[str, float, float | None]] = [
("0-1 min", 0, 60),
("1-5 min", 60, 300),
("5-15 min", 300, 900),
("15-60 min", 900, 3600),
("60+ min", 3600, None),
]
SIZE_BUCKETS: list[tuple[str, int, int | None]] = [
("0-100 MB", 0, 100 * MEGABYTE),
("100-500 MB", 100 * MEGABYTE, 500 * MEGABYTE),
("500 MB-1 GB", 500 * MEGABYTE, GIGABYTE),
("1-4 GB", GIGABYTE, 4 * GIGABYTE),
("4+ GB", 4 * GIGABYTE, None),
DURATION_STEP_CANDIDATES = [5, 10, 15, 30, 60, 120, 300, 600, 900, 1800, 3600, 7200]
SIZE_STEP_CANDIDATES = [
MEGABYTE,
2 * MEGABYTE,
5 * MEGABYTE,
10 * MEGABYTE,
20 * MEGABYTE,
50 * MEGABYTE,
100 * MEGABYTE,
200 * MEGABYTE,
500 * MEGABYTE,
GIGABYTE,
2 * GIGABYTE,
4 * GIGABYTE,
8 * GIGABYTE,
]
MAX_DYNAMIC_HISTOGRAM_BINS = 8
@dataclass(slots=True)
@@ -53,22 +58,80 @@ class AnalysisSummary:
def _bucketize_numeric(
values: list[float],
buckets: list[tuple[str, float, float | None]],
*,
step_candidates: list[float],
label_formatter: Callable[[float, float, float], str],
) -> list[HistogramBin]:
if not values:
return []
step = _select_histogram_step(max_value=max(values), step_candidates=step_candidates, value_count=len(values))
bin_count = max(1, math.ceil(max(values) / step))
results: list[HistogramBin] = []
for label, lower, upper in buckets:
for index in range(bin_count):
lower = index * step
upper = lower + step
is_last_bin = index == bin_count - 1
count = 0
for value in values:
if value < lower:
continue
if upper is None or value < upper:
if value < upper or (is_last_bin and value <= upper):
count += 1
results.append(HistogramBin(label=label, count=count))
results.append(HistogramBin(label=label_formatter(lower, upper, step), count=count))
return results
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))
def _select_histogram_step(*, max_value: float, step_candidates: list[float], value_count: int) -> float:
if max_value <= 0:
return step_candidates[0]
max_bins = _resolve_histogram_max_bins(value_count)
for candidate in step_candidates:
if math.ceil(max_value / candidate) <= max_bins:
return candidate
step = step_candidates[-1]
while math.ceil(max_value / step) > max_bins:
step *= 2
return step
def _format_duration_bucket_label(lower: float, upper: float, step: float) -> str:
if step >= 3600 and step % 3600 == 0:
return f"{int(lower / 3600)}-{int(upper / 3600)} h"
if step >= 60 and step % 60 == 0:
return f"{int(lower / 60)}-{int(upper / 60)} min"
return f"{int(lower)}-{int(upper)} s"
def _format_size_bucket_label(lower: float, upper: float, step: float) -> str:
if step >= GIGABYTE and step % GIGABYTE == 0:
divisor = GIGABYTE
unit = "GB"
elif step >= MEGABYTE and step % MEGABYTE == 0:
divisor = MEGABYTE
unit = "MB"
elif step >= KILOBYTE and step % KILOBYTE == 0:
divisor = KILOBYTE
unit = "KB"
else:
divisor = 1
unit = "B"
return f"{int(lower / divisor)}-{int(upper / divisor)} {unit}"
def _build_ranked_counts(labels: list[str]) -> list[CategoryCount]:
counts = Counter(labels)
first_seen: dict[str, int] = {}
@@ -100,8 +163,16 @@ def build_summary(
failed_files=len(failures),
total_size_bytes=sum(sizes),
total_duration_seconds=sum(durations),
duration_histogram=_bucketize_numeric(durations, DURATION_BUCKETS),
size_histogram=_bucketize_numeric(sizes, SIZE_BUCKETS),
duration_histogram=_bucketize_numeric(
durations,
step_candidates=DURATION_STEP_CANDIDATES,
label_formatter=_format_duration_bucket_label,
),
size_histogram=_bucketize_numeric(
sizes,
step_candidates=SIZE_STEP_CANDIDATES,
label_formatter=_format_size_bucket_label,
),
video_codecs=_build_ranked_counts(codecs),
resolutions=_build_ranked_counts(resolutions),
)
tests/test_analysis.py
+63 -25
View File
@@ -4,23 +4,27 @@ from video_metadata_analyzer.analysis import build_summary
from video_metadata_analyzer.models import ProbeFailure, VideoMetadata
def _build_record(path_name: str, *, size_mb: int, duration_seconds: float, codec: str = "h264") -> VideoMetadata:
return VideoMetadata(
path=Path(path_name),
size_bytes=size_mb * 1024 * 1024,
duration_seconds=duration_seconds,
format_name="mov,mp4,m4a,3gp,3g2,mj2",
bit_rate=1_000_000,
video_codec=codec,
audio_codec="aac",
width=1920,
height=1080,
frame_rate=30.0,
video_stream_count=1,
audio_stream_count=1,
subtitle_stream_count=0,
)
def test_build_summary_computes_totals_and_histograms() -> None:
records = [
VideoMetadata(
path=Path("a.mp4"),
size_bytes=50 * 1024 * 1024,
duration_seconds=30,
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,
),
_build_record("a.mp4", size_mb=50, duration_seconds=30),
VideoMetadata(
path=Path("b.mkv"),
size_bytes=700 * 1024 * 1024,
@@ -59,18 +63,18 @@ def test_build_summary_computes_totals_and_histograms() -> None:
assert summary.total_size_bytes == 2_933_915_648
assert summary.total_duration_seconds == 4830
assert [(item.label, item.count) for item in summary.duration_histogram] == [
("0-1 min", 1),
("1-5 min", 0),
("5-15 min", 1),
("15-60 min", 0),
("60+ min", 1),
("0-15 min", 2),
("15-30 min", 0),
("30-45 min", 0),
("45-60 min", 0),
("60-75 min", 1),
]
assert [(item.label, item.count) for item in summary.size_histogram] == [
("0-100 MB", 1),
("100-500 MB", 0),
("500 MB-1 GB", 1),
("1-4 GB", 1),
("4+ GB", 0),
("0-500 MB", 1),
("500-1000 MB", 1),
("1000-1500 MB", 0),
("1500-2000 MB", 0),
("2000-2500 MB", 1),
]
assert [(item.label, item.count) for item in summary.video_codecs] == [
("h264", 2),
@@ -83,6 +87,40 @@ def test_build_summary_computes_totals_and_histograms() -> None:
]
def test_build_summary_uses_adaptive_buckets_for_compact_ranges() -> None:
records = [
_build_record("a.mp4", size_mb=10, duration_seconds=10),
_build_record("b.mp4", size_mb=25, duration_seconds=20),
_build_record("c.mp4", size_mb=40, duration_seconds=35),
_build_record("d.mp4", size_mb=55, duration_seconds=55),
_build_record("e.mp4", size_mb=80, duration_seconds=75),
_build_record("f.mp4", size_mb=120, duration_seconds=110),
_build_record("g.mp4", size_mb=160, duration_seconds=150),
_build_record("h.mp4", size_mb=240, duration_seconds=220),
_build_record("i.mp4", size_mb=360, duration_seconds=260),
]
summary = build_summary(records)
assert [(item.label, item.count) for item in summary.duration_histogram] == [
("0-1 min", 4),
("1-2 min", 2),
("2-3 min", 1),
("3-4 min", 1),
("4-5 min", 1),
]
assert [(item.label, item.count) for item in summary.size_histogram] == [
("0-50 MB", 3),
("50-100 MB", 2),
("100-150 MB", 1),
("150-200 MB", 1),
("200-250 MB", 1),
("250-300 MB", 0),
("300-350 MB", 0),
("350-400 MB", 1),
]
def test_build_summary_tracks_successful_and_failed_files() -> None:
records = [
VideoMetadata(