feat: initialize video metadata analyzer

2026-04-02 18:14:20 +08:00
commit bc242f8a31
20 changed files with 1331 additions and 0 deletions
@@ -0,0 +1,6 @@
 .venv/
 .pytest_cache/
 __pycache__/
 reports/
 tmp_verify_input/
 tmp_verify_output*/
@@ -0,0 +1,87 @@
 # 视频元数据分析器
 基于 Python + `uv` 的命令行工具，用于递归扫描一个或多个目录中的视频文件，调用 `ffprobe` 提取结构化元数据，并调用 `ffmpeg` 记录运行环境能力，最终输出：
 - `report.json`：完整元数据与汇总统计
 - `files.csv`：逐文件明细
 - `report.html`：可直接打开的离线可视化报告
 ## 当前能力
 - 输入一个或多个目录，递归发现常见视频文件
 - 提取视频时长、大小、编码、分辨率、帧率、音频编码等字段
 - 汇总时长分布、大小分布、视频编码分布、分辨率分布
 - 生成适合人工查看的 HTML 报告，以及适合后续二次处理的 JSON / CSV
 ## 依赖要求
 - Python 3.11+
 - `uv`
 - `ffprobe`
 - `ffmpeg`
 要求 `ffprobe` 与 `ffmpeg` 已经在系统 `PATH` 中，或者在运行时通过参数显式指定。
 ## 初始化
 ```powershell
 uv sync --dev
 ```
 ## 运行方式
 最直接的方式：
 ```powershell
 uv run video-metadata-analyzer D:\Videos E:\Archive --output-dir .\reports\latest
 ```
 如果你更习惯模块方式：
 ```powershell
 uv run python -m video_metadata_analyzer D:\Videos --output-dir .\reports\latest
 ```
 如果 `ffprobe` / `ffmpeg` 不在 `PATH` 中，可以显式指定：
 ```powershell
 uv run video-metadata-analyzer D:\Videos `
  --ffprobe-bin C:\ffmpeg\bin\ffprobe.exe `
  --ffmpeg-bin C:\ffmpeg\bin\ffmpeg.exe `
  --output-dir .\reports\latest
 ```
 ## 输出内容
 生成目录内默认包含以下文件：
 - `report.json`
 - `files.csv`
 - `report.html`
 其中 `report.html` 是离线静态页面，可以直接双击打开查看。
 ## 开发与测试
 运行测试：
 ```powershell
 uv run pytest -q
 ```
 查看命令帮助：
 ```powershell
 uv run video-metadata-analyzer --help
 ```
 ## 第一版边界
 当前版本优先保证“扫描、提取、汇总、可视化”这条主链路可用，暂时没有做这些增强项：
 - 损坏视频的深度解码验证
 - 单文件失败后的错误报告面板
 - 更复杂的前端交互图表
 - 增量扫描与数据库缓存
 如果后续要继续做，最值得优先加的是“探测失败文件单独记录，不让整次扫描因为一个坏文件直接中断”。
@@ -0,0 +1,234 @@
 # Video Metadata Analyzer Implementation Plan
 > **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
 **Goal:** 构建一个基于 Python + uv 的命令行视频元数据分析器，递归扫描一个或多个目录，调用 ffprobe/ffmpeg 汇总视频元数据，并输出 JSON、CSV 和静态 HTML 可视化报告。
 **Architecture:** 项目采用 `src/` 布局的 CLI 应用。扫描层负责发现视频文件；探测层调用 ffprobe 获取结构化元数据，并调用 ffmpeg 记录运行环境能力；分析层负责做总量统计、分布分箱和类别汇总；报告层输出机器可读 JSON/CSV 以及带内嵌 SVG 图表的 HTML 报告，避免引入浏览器端框架依赖。
 **Tech Stack:** Python 3.11+、uv、pytest、ffprobe、ffmpeg、标准库 `argparse/json/csv/pathlib/subprocess/statistics/html`
 ---
 ### Task 1: 建立 uv 项目基础
 **Files:**
 - Create: `pyproject.toml`
 - Create: `src/video_metadata_analyzer/__init__.py`
 - Create: `tests/__init__.py`
 **Step 1: Write the failing test**
 ```python
 def test_package_version_is_exposed():
    from video_metadata_analyzer import __version__
    assert __version__
 ```
 **Step 2: Run test to verify it fails**
 Run: `uv run pytest tests/test_package.py::test_package_version_is_exposed -v`
 Expected: FAIL with `ModuleNotFoundError` or missing `__version__`
 **Step 3: Write minimal implementation**
 ```python
 __version__ = "0.1.0"
 ```
 **Step 4: Run test to verify it passes**
 Run: `uv run pytest tests/test_package.py::test_package_version_is_exposed -v`
 Expected: PASS
 **Step 5: Commit**
 ```bash
 git add pyproject.toml src/video_metadata_analyzer/__init__.py tests/test_package.py
 git commit -m "chore: initialize uv project skeleton"
 ```
 ### Task 2: 实现视频文件发现
 **Files:**
 - Create: `tests/test_discovery.py`
 - Create: `src/video_metadata_analyzer/discovery.py`
 **Step 1: Write the failing test**
 ```python
 def test_discover_video_files_recursively_filters_supported_extensions(tmp_path):
    ...
 ```
 **Step 2: Run test to verify it fails**
 Run: `uv run pytest tests/test_discovery.py::test_discover_video_files_recursively_filters_supported_extensions -v`
 Expected: FAIL with missing function
 **Step 3: Write minimal implementation**
 ```python
 def discover_video_files(inputs: list[Path]) -> list[Path]:
    ...
 ```
 **Step 4: Run test to verify it passes**
 Run: `uv run pytest tests/test_discovery.py::test_discover_video_files_recursively_filters_supported_extensions -v`
 Expected: PASS
 **Step 5: Commit**
 ```bash
 git add tests/test_discovery.py src/video_metadata_analyzer/discovery.py
 git commit -m "feat: add recursive video discovery"
 ```
 ### Task 3: 实现 ffprobe 元数据解析
 **Files:**
 - Create: `tests/test_probe.py`
 - Create: `src/video_metadata_analyzer/models.py`
 - Create: `src/video_metadata_analyzer/probe.py`
 **Step 1: Write the failing test**
 ```python
 def test_probe_video_metadata_parses_primary_streams():
    ...
 ```
 **Step 2: Run test to verify it fails**
 Run: `uv run pytest tests/test_probe.py::test_probe_video_metadata_parses_primary_streams -v`
 Expected: FAIL with missing class or parser
 **Step 3: Write minimal implementation**
 ```python
@dataclass
 class VideoMetadata:
    ...
 ```
 **Step 4: Run test to verify it passes**
 Run: `uv run pytest tests/test_probe.py::test_probe_video_metadata_parses_primary_streams -v`
 Expected: PASS
 **Step 5: Commit**
 ```bash
 git add tests/test_probe.py src/video_metadata_analyzer/models.py src/video_metadata_analyzer/probe.py
 git commit -m "feat: parse metadata from ffprobe output"
 ```
 ### Task 4: 实现聚合分析与图表数据
 **Files:**
 - Create: `tests/test_analysis.py`
 - Create: `src/video_metadata_analyzer/analysis.py`
 **Step 1: Write the failing test**
 ```python
 def test_build_summary_computes_totals_and_histograms():
    ...
 ```
 **Step 2: Run test to verify it fails**
 Run: `uv run pytest tests/test_analysis.py::test_build_summary_computes_totals_and_histograms -v`
 Expected: FAIL with missing function
 **Step 3: Write minimal implementation**
 ```python
 def build_summary(records: list[VideoMetadata]) -> AnalysisSummary:
    ...
 ```
 **Step 4: Run test to verify it passes**
 Run: `uv run pytest tests/test_analysis.py::test_build_summary_computes_totals_and_histograms -v`
 Expected: PASS
 **Step 5: Commit**
 ```bash
 git add tests/test_analysis.py src/video_metadata_analyzer/analysis.py
 git commit -m "feat: add aggregate analysis summaries"
 ```
 ### Task 5: 实现报告输出与 CLI
 **Files:**
 - Create: `tests/test_report.py`
 - Create: `tests/test_cli.py`
 - Create: `src/video_metadata_analyzer/reporting.py`
 - Create: `src/video_metadata_analyzer/cli.py`
 **Step 1: Write the failing test**
 ```python
 def test_write_reports_creates_json_csv_and_html(tmp_path):
    ...
 ```
 **Step 2: Run test to verify it fails**
 Run: `uv run pytest tests/test_report.py::test_write_reports_creates_json_csv_and_html -v`
 Expected: FAIL with missing writer
 **Step 3: Write minimal implementation**
 ```python
 def write_report_bundle(output_dir: Path, ...):
    ...
 ```
 **Step 4: Run test to verify it passes**
 Run: `uv run pytest tests/test_report.py::test_write_reports_creates_json_csv_and_html -v`
 Expected: PASS
 **Step 5: Commit**
 ```bash
 git add tests/test_report.py tests/test_cli.py src/video_metadata_analyzer/reporting.py src/video_metadata_analyzer/cli.py
 git commit -m "feat: add report writers and CLI entrypoint"
 ```
 ### Task 6: 文档与验收
 **Files:**
 - Create: `README.md`
 **Step 1: Write the failing test**
 ```python
 def test_readme_mentions_uv_run_command():
    ...
 ```
 **Step 2: Run test to verify it fails**
 Run: `uv run pytest tests/test_cli.py::test_readme_mentions_uv_run_command -v`
 Expected: FAIL because README or command example is absent
 **Step 3: Write minimal implementation**
 补充中文 README，写明依赖、安装、运行方式、输出文件说明和常见问题。
 **Step 4: Run test to verify it passes**
 Run: `uv run pytest -q`
 Expected: PASS
 **Step 5: Commit**
 ```bash
 git add README.md
 git commit -m "docs: add usage guide for video metadata analyzer"
 ```
@@ -0,0 +1,23 @@
 [build-system]
 requires = ["hatchling>=1.27.0"]
 build-backend = "hatchling.build"
 [project]
 name = "video-metadata-analyzer"
 version = "0.1.0"
 description = "递归扫描视频目录并生成元数据分析报告的 CLI 工具"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = []
 [project.scripts]
 video-metadata-analyzer = "video_metadata_analyzer.cli:main"
 [dependency-groups]
 dev = [
    "pytest>=8.4.2",
 ]
 [tool.pytest.ini_options]
 pythonpath = ["src"]
 testpaths = ["tests"]
@@ -0,0 +1 @@
 __version__ = "0.1.0"
@@ -0,0 +1,5 @@
 from video_metadata_analyzer.cli import main
 if __name__ == "__main__":
    raise SystemExit(main())
@@ -0,0 +1,98 @@
 from __future__ import annotations
 from collections import Counter
 from dataclasses import dataclass
 from video_metadata_analyzer.models import VideoMetadata
 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),
 ]
@dataclass(slots=True)
 class HistogramBin:
    label: str
    count: int
@dataclass(slots=True)
 class CategoryCount:
    label: str
    count: int
@dataclass(slots=True)
 class AnalysisSummary:
    total_files: int
    total_size_bytes: int
    total_duration_seconds: float
    duration_histogram: list[HistogramBin]
    size_histogram: list[HistogramBin]
    video_codecs: list[CategoryCount]
    resolutions: list[CategoryCount]
 def _bucketize_numeric(
    values: list[float],
    buckets: list[tuple[str, float, float | None]],
 ) -> list[HistogramBin]:
    results: list[HistogramBin] = []
    for label, lower, upper in buckets:
        count = 0
        for value in values:
            if value < lower:
                continue
            if upper is None or value < upper:
                count += 1
        results.append(HistogramBin(label=label, count=count))
    return results
 def _build_ranked_counts(labels: list[str]) -> list[CategoryCount]:
    counts = Counter(labels)
    first_seen: dict[str, int] = {}
    for index, label in enumerate(labels):
        first_seen.setdefault(label, index)
    ordered = sorted(counts.items(), key=lambda item: (-item[1], first_seen[item[0]]))
    return [CategoryCount(label=label, count=count) for label, count in ordered]
 def build_summary(records: list[VideoMetadata]) -> AnalysisSummary:
    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]
    resolutions = [
        f"{record.width}x{record.height}"
        for record in records
        if record.width is not None and record.height is not None
    ]
    return AnalysisSummary(
        total_files=len(records),
        total_size_bytes=sum(sizes),
        total_duration_seconds=sum(durations),
        duration_histogram=_bucketize_numeric(durations, DURATION_BUCKETS),
        size_histogram=_bucketize_numeric(sizes, SIZE_BUCKETS),
        video_codecs=_build_ranked_counts(codecs),
        resolutions=_build_ranked_counts(resolutions),
    )
@@ -0,0 +1,71 @@
 from __future__ import annotations
 import argparse
 from datetime import datetime
 from pathlib import Path
 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.probe import CommandRunner, probe_tool_versions, probe_video_metadata
 from video_metadata_analyzer.reporting import write_report_bundle
 def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="video-metadata-analyzer",
        description="递归扫描目录中的视频文件并生成元数据分析报告。",
    )
    parser.add_argument("inputs", nargs="+", help="一个或多个待扫描的目录或视频文件路径")
    parser.add_argument(
        "--output-dir",
        help="报告输出目录，默认写入 reports/<timestamp>",
    )
    parser.add_argument("--ffprobe-bin", default="ffprobe", help="ffprobe 可执行文件路径")
    parser.add_argument("--ffmpeg-bin", default="ffmpeg", help="ffmpeg 可执行文件路径")
    return parser
 def main(
    argv: Sequence[str] | None = None,
    *,
    command_runner: CommandRunner | None = None,
 ) -> int:
    parser = build_parser()
    args = parser.parse_args(list(argv) if argv is not None else None)
    input_paths = [Path(item) for item in args.inputs]
    output_dir = (
        Path(args.output_dir)
        if args.output_dir
        else Path("reports") / datetime.now().strftime("%Y%m%d-%H%M%S")
    )
    tool_versions = probe_tool_versions(
        ffprobe_bin=args.ffprobe_bin,
        ffmpeg_bin=args.ffmpeg_bin,
        runner=command_runner,
    )
    video_files = discover_video_files(input_paths)
    records = [
        probe_video_metadata(
            path,
            ffprobe_bin=args.ffprobe_bin,
            runner=command_runner,
        )
        for path in video_files
    ]
    summary = build_summary(records)
    paths = write_report_bundle(
        output_dir=output_dir,
        scanned_inputs=input_paths,
        records=records,
        summary=summary,
        tool_versions=tool_versions,
    )
    print(f"扫描完成：共发现 {summary.total_files} 个视频文件。")
    print(f"JSON 报告：{paths['json']}")
    print(f"CSV 明细：{paths['csv']}")
    print(f"HTML 报告：{paths['html']}")
    return 0
@@ -0,0 +1,39 @@
 from __future__ import annotations
 from pathlib import Path
 VIDEO_EXTENSIONS = {
    ".3gp",
    ".avi",
    ".flv",
    ".m4v",
    ".mkv",
    ".mov",
    ".mp4",
    ".mpeg",
    ".mpg",
    ".mts",
    ".m2ts",
    ".ts",
    ".webm",
    ".wmv",
 }
 def discover_video_files(inputs: list[Path]) -> list[Path]:
    discovered: set[Path] = set()
    for input_path in inputs:
        if input_path.is_file() and input_path.suffix.lower() in VIDEO_EXTENSIONS:
            discovered.add(input_path)
            continue
        if not input_path.is_dir():
            continue
        for candidate in input_path.rglob("*"):
            if candidate.is_file() and candidate.suffix.lower() in VIDEO_EXTENSIONS:
                discovered.add(candidate)
    return sorted(discovered)
@@ -0,0 +1,21 @@
 from __future__ import annotations
 from dataclasses import dataclass
 from pathlib import Path
@dataclass(slots=True)
 class VideoMetadata:
    path: Path
    size_bytes: int
    duration_seconds: float | None
    format_name: str | None
    bit_rate: int | None
    video_codec: str | None
    audio_codec: str | None
    width: int | None
    height: int | None
    frame_rate: float | None
    video_stream_count: int
    audio_stream_count: int
    subtitle_stream_count: int
@@ -0,0 +1,104 @@
 from __future__ import annotations
 import json
 import subprocess
 from pathlib import Path
 from typing import Callable
 from video_metadata_analyzer.models import VideoMetadata
 CommandRunner = Callable[[list[str]], str]
 def _run_command(command: list[str]) -> str:
    completed = subprocess.run(
        command,
        check=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    return completed.stdout
 def _parse_optional_int(value: object) -> int | None:
    if value in (None, "", "N/A"):
        return None
    return int(value)
 def _parse_optional_float(value: object) -> float | None:
    if value in (None, "", "N/A"):
        return None
    return float(value)
 def _parse_frame_rate(value: object) -> float | None:
    if value in (None, "", "0/0", "N/A"):
        return None
    if isinstance(value, str) and "/" in value:
        numerator, denominator = value.split("/", maxsplit=1)
        if denominator == "0":
            return None
        return float(numerator) / float(denominator)
    return float(value)
 def probe_video_metadata(
    path: Path,
    *,
    ffprobe_bin: str = "ffprobe",
    runner: CommandRunner | None = None,
 ) -> VideoMetadata:
    command = [
        ffprobe_bin,
        "-v",
        "quiet",
        "-print_format",
        "json",
        "-show_format",
        "-show_streams",
        str(path),
    ]
    raw_output = (runner or _run_command)(command)
    payload = json.loads(raw_output)
    format_info = payload.get("format", {})
    streams = payload.get("streams", [])
    video_streams = [stream for stream in streams if stream.get("codec_type") == "video"]
    audio_streams = [stream for stream in streams if stream.get("codec_type") == "audio"]
    subtitle_streams = [stream for stream in streams if stream.get("codec_type") == "subtitle"]
    primary_video = video_streams[0] if video_streams else {}
    primary_audio = audio_streams[0] if audio_streams else {}
    return VideoMetadata(
        path=path,
        size_bytes=_parse_optional_int(format_info.get("size")) or path.stat().st_size,
        duration_seconds=_parse_optional_float(format_info.get("duration")),
        format_name=format_info.get("format_name"),
        bit_rate=_parse_optional_int(format_info.get("bit_rate")),
        video_codec=primary_video.get("codec_name"),
        audio_codec=primary_audio.get("codec_name"),
        width=_parse_optional_int(primary_video.get("width")),
        height=_parse_optional_int(primary_video.get("height")),
        frame_rate=_parse_frame_rate(primary_video.get("avg_frame_rate")),
        video_stream_count=len(video_streams),
        audio_stream_count=len(audio_streams),
        subtitle_stream_count=len(subtitle_streams),
    )
 def probe_tool_versions(
    *,
    ffprobe_bin: str = "ffprobe",
    ffmpeg_bin: str = "ffmpeg",
    runner: CommandRunner | None = None,
 ) -> dict[str, str]:
    active_runner = runner or _run_command
    ffprobe_output = active_runner([ffprobe_bin, "-version"])
    ffmpeg_output = active_runner([ffmpeg_bin, "-version"])
    return {
        "ffprobe": ffprobe_output.splitlines()[0].strip() if ffprobe_output else "unknown",
        "ffmpeg": ffmpeg_output.splitlines()[0].strip() if ffmpeg_output else "unknown",
    }
@@ -0,0 +1,286 @@
 from __future__ import annotations
 import csv
 import json
 from dataclasses import asdict
 from datetime import datetime
 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.models import VideoMetadata
 def _record_to_dict(record: VideoMetadata) -> dict[str, object]:
    return {
        "path": record.path.as_posix(),
        "size_bytes": record.size_bytes,
        "duration_seconds": record.duration_seconds,
        "format_name": record.format_name,
        "bit_rate": record.bit_rate,
        "video_codec": record.video_codec,
        "audio_codec": record.audio_codec,
        "width": record.width,
        "height": record.height,
        "frame_rate": record.frame_rate,
        "video_stream_count": record.video_stream_count,
        "audio_stream_count": record.audio_stream_count,
        "subtitle_stream_count": record.subtitle_stream_count,
    }
 def _summary_to_dict(summary: AnalysisSummary) -> dict[str, object]:
    return {
        "total_files": summary.total_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],
        "size_histogram": [asdict(item) for item in summary.size_histogram],
        "video_codecs": [asdict(item) for item in summary.video_codecs],
        "resolutions": [asdict(item) for item in summary.resolutions],
    }
 def _write_json(
    output_path: Path,
    *,
    scanned_inputs: list[Path],
    records: list[VideoMetadata],
    summary: AnalysisSummary,
    tool_versions: dict[str, str],
 ) -> None:
    payload = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "scanned_inputs": [path.as_posix() for path in scanned_inputs],
        "tool_versions": tool_versions,
        "summary": _summary_to_dict(summary),
        "files": [_record_to_dict(record) for record in records],
    }
    output_path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
 def _write_csv(output_path: Path, records: list[VideoMetadata]) -> None:
    fieldnames = [
        "path",
        "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",
    ]
    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))
 def _format_size(value: int) -> str:
    units = ["B", "KB", "MB", "GB", "TB"]
    size = float(value)
    for unit in units:
        if size < 1024 or unit == units[-1]:
            return f"{size:.2f} {unit}"
        size /= 1024
    return f"{size:.2f} TB"
 def _format_duration(value: float) -> str:
    total_seconds = int(round(value))
    hours, remainder = divmod(total_seconds, 3600)
    minutes, seconds = divmod(remainder, 60)
    if hours:
        return f"{hours}h {minutes}m {seconds}s"
    if minutes:
        return f"{minutes}m {seconds}s"
    return f"{seconds}s"
 def _render_bar_chart(title: str, items: Iterable[HistogramBin | CategoryCount]) -> str:
    rows = list(items)
    width = 680
    height = 320
    margin_left = 110
    margin_right = 24
    margin_top = 24
    margin_bottom = 48
    plot_width = width - margin_left - margin_right
    plot_height = height - margin_top - margin_bottom
    bar_gap = 12
    bar_width = plot_width / max(len(rows), 1) - bar_gap
    max_count = max((item.count for item in rows), default=1)
    pieces = [
        f'<section class="chart"><h2>{escape(title)}</h2>',
        f'<svg viewBox="0 0 {width} {height}" role="img" aria-label="{escape(title)}">',
        f'<line x1="{margin_left}" y1="{margin_top + plot_height}" x2="{width - margin_right}" y2="{margin_top + plot_height}" class="axis" />',
    ]
    for index, item in enumerate(rows):
        bar_height = 0 if max_count == 0 else plot_height * (item.count / max_count)
        x = margin_left + index * (bar_width + bar_gap) + bar_gap / 2
        y = margin_top + plot_height - bar_height
        text_x = x + bar_width / 2
        pieces.append(
            f'<rect x="{x:.2f}" y="{y:.2f}" width="{bar_width:.2f}" height="{bar_height:.2f}" class="bar" />'
        )
        pieces.append(f'<text x="{text_x:.2f}" y="{y - 8:.2f}" class="count">{item.count}</text>')
        pieces.append(
            f'<text x="{text_x:.2f}" y="{margin_top + plot_height + 20:.2f}" class="label">{escape(item.label)}</text>'
        )
    pieces.append("</svg></section>")
    return "".join(pieces)
 def _write_html(
    output_path: Path,
    *,
    scanned_inputs: list[Path],
    summary: AnalysisSummary,
    tool_versions: dict[str, str],
 ) -> None:
    summary_cards = f"""
    <section class="cards">
      <article class="card"><h2>视频文件数</h2><p>{summary.total_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>
    """
    html = f"""<!DOCTYPE html>
 <html lang="zh-CN">
 <head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>视频元数据分析报告</title>
  <style>
    :root {{
      color-scheme: light;
      --bg: #f4efe6;
      --panel: #fffdf8;
      --line: #d9cdb7;
      --text: #1b1b18;
      --muted: #625d53;
      --accent: #1d6b5b;
      --accent-soft: #d6efe8;
    }}
    * {{ box-sizing: border-box; }}
    body {{
      margin: 0;
      font-family: "Segoe UI", "PingFang SC", "Microsoft YaHei", sans-serif;
      background:
        radial-gradient(circle at top right, rgba(29,107,91,0.14), transparent 28%),
        linear-gradient(180deg, #fbf7ef 0%, var(--bg) 100%);
      color: var(--text);
    }}
    main {{ max-width: 1180px; margin: 0 auto; padding: 40px 24px 64px; }}
    header {{ margin-bottom: 28px; }}
    h1 {{ margin: 0 0 8px; font-size: 2.4rem; }}
    p, li {{ color: var(--muted); line-height: 1.6; }}
    .cards {{
      display: grid;
      grid-template-columns: repeat(auto-fit, minmax(210px, 1fr));
      gap: 16px;
      margin: 24px 0 36px;
    }}
    .card, .chart, .meta {{
      background: rgba(255, 253, 248, 0.88);
      border: 1px solid rgba(217, 205, 183, 0.85);
      border-radius: 18px;
      padding: 20px;
      box-shadow: 0 16px 40px rgba(36, 31, 24, 0.05);
      backdrop-filter: blur(10px);
    }}
    .card h2, .chart h2, .meta h2 {{ margin: 0 0 10px; font-size: 1rem; }}
    .card p {{ margin: 0; font-size: 1.6rem; color: var(--text); }}
    .charts {{
      display: grid;
      grid-template-columns: repeat(auto-fit, minmax(340px, 1fr));
      gap: 18px;
    }}
    svg {{ width: 100%; height: auto; overflow: visible; }}
    .bar {{ fill: var(--accent); opacity: 0.9; }}
    .axis {{ stroke: var(--line); stroke-width: 1.5; }}
    .label {{
      fill: var(--muted);
      font-size: 10px;
      text-anchor: middle;
    }}
    .count {{
      fill: var(--text);
      font-size: 11px;
      text-anchor: middle;
      font-weight: 700;
    }}
    ul {{ padding-left: 18px; }}
  </style>
 </head>
 <body>
  <main>
    <header>
      <h1>视频元数据分析报告</h1>
      <p>扫描输入：{escape(", ".join(path.as_posix() for path in scanned_inputs))}</p>
    </header>
    {summary_cards}
    <section class="charts">
      {_render_bar_chart("时长分布", summary.duration_histogram)}
      {_render_bar_chart("大小分布", summary.size_histogram)}
      {_render_bar_chart("视频编码分布", summary.video_codecs)}
      {_render_bar_chart("分辨率分布", summary.resolutions)}
    </section>
    <section class="meta">
      <h2>工具版本</h2>
      <ul>
        <li>ffprobe: {escape(tool_versions.get("ffprobe", "unknown"))}</li>
        <li>ffmpeg: {escape(tool_versions.get("ffmpeg", "unknown"))}</li>
      </ul>
    </section>
  </main>
 </body>
 </html>
 """
    output_path.write_text(html, encoding="utf-8")
 def write_report_bundle(
    *,
    output_dir: Path,
    scanned_inputs: list[Path],
    records: list[VideoMetadata],
    summary: AnalysisSummary,
    tool_versions: dict[str, str],
 ) -> dict[str, Path]:
    output_dir.mkdir(parents=True, exist_ok=True)
    json_path = output_dir / "report.json"
    csv_path = output_dir / "files.csv"
    html_path = output_dir / "report.html"
    _write_json(
        json_path,
        scanned_inputs=scanned_inputs,
        records=records,
        summary=summary,
        tool_versions=tool_versions,
    )
    _write_csv(csv_path, records)
    _write_html(
        html_path,
        scanned_inputs=scanned_inputs,
        summary=summary,
        tool_versions=tool_versions,
    )
    return {"json": json_path, "csv": csv_path, "html": html_path}
@@ -0,0 +1 @@
@@ -0,0 +1,83 @@
 from pathlib import Path
 from video_metadata_analyzer.analysis import build_summary
 from video_metadata_analyzer.models import VideoMetadata
 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,
        ),
        VideoMetadata(
            path=Path("b.mkv"),
            size_bytes=700 * 1024 * 1024,
            duration_seconds=600,
            format_name="matroska,webm",
            bit_rate=2_000_000,
            video_codec="h264",
            audio_codec="aac",
            width=1280,
            height=720,
            frame_rate=24.0,
            video_stream_count=1,
            audio_stream_count=1,
            subtitle_stream_count=0,
        ),
        VideoMetadata(
            path=Path("c.mov"),
            size_bytes=2 * 1024 * 1024 * 1024,
            duration_seconds=4200,
            format_name="mov,mp4,m4a,3gp,3g2,mj2",
            bit_rate=8_000_000,
            video_codec="hevc",
            audio_codec="aac",
            width=3840,
            height=2160,
            frame_rate=60.0,
            video_stream_count=1,
            audio_stream_count=1,
            subtitle_stream_count=0,
        ),
    ]
    summary = build_summary(records)
    assert summary.total_files == 3
    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),
    ]
    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),
    ]
    assert [(item.label, item.count) for item in summary.video_codecs] == [
        ("h264", 2),
        ("hevc", 1),
    ]
    assert [(item.label, item.count) for item in summary.resolutions] == [
        ("1920x1080", 1),
        ("1280x720", 1),
        ("3840x2160", 1),
    ]
@@ -0,0 +1,52 @@
 from __future__ import annotations
 import json
 from pathlib import Path
 from video_metadata_analyzer.cli import main
 def test_main_scans_input_directory_and_writes_reports(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": "8.000000",
                        "bit_rate": "1024000",
                        "size": "2048",
                    },
                    "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)],
        command_runner=fake_runner,
    )
    assert exit_code == 0
    assert (output_dir / "report.json").exists()
    assert (output_dir / "files.csv").exists()
    assert (output_dir / "report.html").exists()
@@ -0,0 +1,24 @@
 from pathlib import Path
 from video_metadata_analyzer.discovery import discover_video_files
 def test_discover_video_files_recursively_filters_supported_extensions(tmp_path: Path) -> None:
    source_a = tmp_path / "source-a"
    source_b = tmp_path / "source-b"
    nested = source_a / "nested"
    nested.mkdir(parents=True)
    source_b.mkdir(parents=True)
    keep_1 = source_a / "movie.mp4"
    keep_2 = nested / "clip.MKV"
    keep_3 = source_b / "recording.mov"
    skip_1 = source_a / "notes.txt"
    skip_2 = nested / "thumb.jpg"
    for path in [keep_1, keep_2, keep_3, skip_1, skip_2]:
        path.write_bytes(b"sample")
    discovered = discover_video_files([source_a, source_b])
    assert discovered == sorted([keep_1, keep_2, keep_3])
@@ -0,0 +1,5 @@
 from video_metadata_analyzer import __version__
 def test_package_version_is_exposed() -> None:
    assert __version__
@@ -0,0 +1,57 @@
 from __future__ import annotations
 import json
 from pathlib import Path
 from video_metadata_analyzer.probe import probe_video_metadata
 def test_probe_video_metadata_parses_primary_streams(tmp_path: Path) -> None:
    sample = tmp_path / "sample.mp4"
    sample.write_bytes(b"x" * 4096)
    captured_command: list[str] = []
    def fake_runner(command: list[str]) -> str:
        captured_command[:] = command
        return json.dumps(
            {
                "format": {
                    "filename": str(sample),
                    "format_name": "mov,mp4,m4a,3gp,3g2,mj2",
                    "duration": "12.500000",
                    "bit_rate": "2621440",
                    "size": "4096",
                },
                "streams": [
                    {
                        "codec_type": "video",
                        "codec_name": "h264",
                        "width": 1920,
                        "height": 1080,
                        "avg_frame_rate": "30000/1001",
                    },
                    {
                        "codec_type": "audio",
                        "codec_name": "aac",
                    },
                ],
            }
        )
    result = probe_video_metadata(sample, runner=fake_runner)
    assert captured_command[:4] == ["ffprobe", "-v", "quiet", "-print_format"]
    assert result.path == sample
    assert result.size_bytes == 4096
    assert result.duration_seconds == 12.5
    assert result.format_name == "mov,mp4,m4a,3gp,3g2,mj2"
    assert result.bit_rate == 2621440
    assert result.video_codec == "h264"
    assert result.audio_codec == "aac"
    assert result.width == 1920
    assert result.height == 1080
    assert result.frame_rate == 30000 / 1001
    assert result.video_stream_count == 1
    assert result.audio_stream_count == 1
    assert result.subtitle_stream_count == 0
@@ -0,0 +1,55 @@
 from __future__ import annotations
 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.reporting import write_report_bundle
 def test_write_report_bundle_creates_json_csv_and_html(tmp_path: Path) -> None:
    output_dir = tmp_path / "report"
    record = VideoMetadata(
        path=Path("videos/sample.mp4"),
        size_bytes=50 * 1024 * 1024,
        duration_seconds=120,
        format_name="mov,mp4,m4a,3gp,3g2,mj2",
        bit_rate=1_500_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,
    )
    summary = build_summary([record])
    paths = write_report_bundle(
        output_dir=output_dir,
        scanned_inputs=[Path("videos")],
        records=[record],
        summary=summary,
        tool_versions={"ffprobe": "ffprobe version x", "ffmpeg": "ffmpeg version y"},
    )
    assert set(paths) == {"json", "csv", "html"}
    assert paths["json"].exists()
    assert paths["csv"].exists()
    assert paths["html"].exists()
    payload = json.loads(paths["json"].read_text(encoding="utf-8"))
    assert payload["summary"]["total_files"] == 1
    assert payload["tool_versions"]["ffmpeg"] == "ffmpeg version y"
    assert payload["files"][0]["path"] == "videos/sample.mp4"
    csv_text = paths["csv"].read_text(encoding="utf-8")
    assert "path,size_bytes,duration_seconds" in csv_text
    assert "videos/sample.mp4" in csv_text
    html_text = paths["html"].read_text(encoding="utf-8")
    assert "视频元数据分析报告" in html_text
    assert "时长分布" in html_text
    assert "<svg" in html_text
@@ -0,0 +1,79 @@
 version = 1
 revision = 3
 requires-python = ">=3.11"
 [[package]]
 name = "colorama"
 version = "0.4.6"
 source = { registry = "https://pypi.org/simple" }
 sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" }
 wheels = [
    { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" },
 ]
 [[package]]
 name = "iniconfig"
 version = "2.3.0"
 source = { registry = "https://pypi.org/simple" }
 sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" }
 wheels = [
    { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" },
 ]
 [[package]]
 name = "packaging"
 version = "26.0"
 source = { registry = "https://pypi.org/simple" }
 sdist = { url = "https://files.pythonhosted.org/packages/65/ee/299d360cdc32edc7d2cf530f3accf79c4fca01e96ffc950d8a52213bd8e4/packaging-26.0.tar.gz", hash = "sha256:00243ae351a257117b6a241061796684b084ed1c516a08c48a3f7e147a9d80b4", size = 143416, upload-time = "2026-01-21T20:50:39.064Z" }
 wheels = [
    { url = "https://files.pythonhosted.org/packages/b7/b9/c538f279a4e237a006a2c98387d081e9eb060d203d8ed34467cc0f0b9b53/packaging-26.0-py3-none-any.whl", hash = "sha256:b36f1fef9334a5588b4166f8bcd26a14e521f2b55e6b9de3aaa80d3ff7a37529", size = 74366, upload-time = "2026-01-21T20:50:37.788Z" },
 ]
 [[package]]
 name = "pluggy"
 version = "1.6.0"
 source = { registry = "https://pypi.org/simple" }
 sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" }
 wheels = [
    { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" },
 ]
 [[package]]
 name = "pygments"
 version = "2.20.0"
 source = { registry = "https://pypi.org/simple" }
 sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" }
 wheels = [
    { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" },
 ]
 [[package]]
 name = "pytest"
 version = "9.0.2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "colorama", marker = "sys_platform == 'win32'" },
    { name = "iniconfig" },
    { name = "packaging" },
    { name = "pluggy" },
    { name = "pygments" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/d1/db/7ef3487e0fb0049ddb5ce41d3a49c235bf9ad299b6a25d5780a89f19230f/pytest-9.0.2.tar.gz", hash = "sha256:75186651a92bd89611d1d9fc20f0b4345fd827c41ccd5c299a868a05d70edf11", size = 1568901, upload-time = "2025-12-06T21:30:51.014Z" }
 wheels = [
    { url = "https://files.pythonhosted.org/packages/3b/ab/b3226f0bd7cdcf710fbede2b3548584366da3b19b5021e74f5bde2a8fa3f/pytest-9.0.2-py3-none-any.whl", hash = "sha256:711ffd45bf766d5264d487b917733b453d917afd2b0ad65223959f59089f875b", size = 374801, upload-time = "2025-12-06T21:30:49.154Z" },
 ]
 [[package]]
 name = "video-metadata-analyzer"
 version = "0.1.0"
 source = { editable = "." }
 [package.dev-dependencies]
 dev = [
    { name = "pytest" },
 ]
 [package.metadata]
 [package.metadata.requires-dev]
 dev = [{ name = "pytest", specifier = ">=8.4.2" }]