feat: build tgvideo damage detection cli

This commit is contained in:
kim
2026-03-30 09:34:36 +08:00
commit 3b2d3a5df4
32 changed files with 3661 additions and 0 deletions
+13
View File
@@ -0,0 +1,13 @@
.venv/
.pytest_cache/
.pytest_tmp*/
.tmp/
__pycache__/
*.pyc
*.pyo
*.pyd
*.db
*.sqlite
*.sqlite3
*.jsonl
*.csv
+699
View File
@@ -0,0 +1,699 @@
# TG Video Damage Detector Implementation Plan
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
**Goal:** 构建一个专门面向 Telegram 导出场景的批量视频损坏检测工具,重点识别“点击跳过后文件未完整下载”的半截视频,并在机械硬盘上尽量减少无效寻道和全量深度解码带来的吞吐损失。
**Architecture:** 采用独立的 Python CLI 工具,使用 `uv` 管理环境,通过 `ffprobe` 做第一轮容器级快速校验,通过 `ffmpeg` 做第二轮定向深度验证。执行模型采用“分阶段、低并发、可恢复”的流水线:目录枚举与阶段一尽量顺序化,阶段二只处理阶段一失败或可疑的子集,并将所有中间状态持久化到 SQLite 以支持断点续跑和大规模批处理。
**Tech Stack:** Python 3.12+, `uv`, `ffprobe`, `ffmpeg`, `sqlite3`, `typer`, `orjson`, `pytest`, `pytest-xdist`(仅测试时可选)
---
## 1. 问题背景与核心判断
目标场景不是通用“坏视频”检测,而是高度特化的 TG 导出异常场景:
- 用户在 Telegram 导出期间点击“跳过”。
- 目标目录中留下了扩展名正确、文件名正常、但内容不完整的视频文件。
- 损坏主因通常是文件截断,而不是编码器随机写坏数据。
- 数据位于机械硬盘,文件量较大,若对所有文件做完整解码,I/O 寻道与进程启动开销会迅速失控。
因此本项目的优化目标必须明确:
- 优先提升“单位时间筛出坏文件的数量”,而不是追求通用媒体法医级完整性验证。
- 优先减少不必要的全量 `ffmpeg` 深度扫描。
- 优先保证长时间批处理过程可恢复、可中断、可复跑,而不是一次性跑完后丢失上下文。
## 2. 方案比较与推荐
### 方案 A: 全量 `ffmpeg` 深度验证
思路:
- 对所有视频直接执行类似 `ffmpeg -v error -xerror -i <file> -f null -` 的验证。
优点:
- 结果直观。
- 发现中后段流损坏的能力最强。
缺点:
- 对机械硬盘最不友好。
- 对大量正常视频的成本极高。
- 进程启动和全量读取会成为主要瓶颈。
结论:
- 不作为默认方案。
### 方案 B: 两轮检测,第一轮 `ffprobe`,第二轮定向 `ffmpeg`
思路:
- 第一轮对所有视频做容器级快速校验。
- 第二轮只处理第一轮失败或被标记为可疑的文件。
- 第二轮不默认全量解码,而是优先做“定向采样验证”,必要时再升级为完整验证。
优点:
- 与实际损坏模式匹配。
- 能以较低成本抓住大量截断文件。
- 适合机械硬盘和大批量数据。
缺点:
- 需要认真设计“可疑”判定规则。
- 结果分类比单轮方案复杂。
结论:
- **推荐方案,也是本计划采用的方案。**
## 3. 范围与非目标
### 范围
- 选择一个或多个目录作为输入。
- 递归扫描常见视频扩展名。
- 第一轮对所有候选视频执行容器级快速校验。
- 第二轮只对剩余可疑文件做更重验证。
- 针对机械硬盘提供保守并发与顺序调度策略。
- 面向大量文件提供断点续跑、结果落盘、批量输出。
### 非目标
- 不做 GUI,首版只做 CLI。
- 不尝试修复视频。
- 不尝试恢复 Telegram 元数据。
- 不做跨机器分布式扫描。
- 不做全媒体库去重、转码、预览图生成。
## 4. 推荐系统架构
首版采用单机、单进程协调、多子进程调用 `ffprobe`/`ffmpeg` 的结构。
组件划分:
1. `cli`
- 解析参数。
- 加载 profile。
- 启动阶段流程。
2. `discovery`
- 递归枚举目录。
- 过滤扩展名。
- 产出基础元数据:路径、大小、修改时间、扩展名。
3. `store`
- 将扫描记录持久化到 SQLite。
- 记录阶段状态,支持恢复。
- 记录错误原因、命令返回码、耗时。
4. `stage1_probe`
- 调用 `ffprobe`。
- 生成容器级结果和可疑标记。
5. `stage2_verify`
- 调用 `ffmpeg` 做更重验证。
- 优先执行定向采样验证,而非对所有可疑文件完整解码。
6. `scheduler`
- 负责 HDD 友好的阶段切换与低并发调度。
- 控制 worker 数、队列长度、重试与中断。
7. `reporting`
- 输出 SQLite、JSONL、CSV 摘要。
- 提供按状态和原因分类的报告。
## 5. 数据模型与状态机
建议使用 SQLite 单文件数据库,例如 `scan_state.db`。
核心表:
- `scan_runs`
- `id`
- `created_at`
- `profile`
- `root_paths_json`
- `status`
- `files`
- `id`
- `run_id`
- `path`
- `size`
- `mtime_ns`
- `extension`
- `discovery_order`
- `stage1_status`
- `stage2_status`
- `final_status`
- `stage1_results`
- `file_id`
- `exit_code`
- `stderr_text`
- `format_name`
- `duration`
- `has_video_stream`
- `warning_flags_json`
- `elapsed_ms`
- `stage2_results`
- `file_id`
- `mode`
- `sample_points_json`
- `exit_code`
- `stderr_text`
- `reason_code`
- `elapsed_ms`
最终状态建议统一为:
- `confirmed_broken`
- `suspicious`
- `passed_stage1`
- `passed_stage2`
- `skipped`
- `command_error`
状态流转建议:
- 发现文件后进入 `pending_stage1`
- 第一轮失败直接进入 `confirmed_broken`
- 第一轮通过但命中可疑规则,进入 `pending_stage2`
- 第二轮通过进入 `passed_stage2`
- 第二轮失败进入 `confirmed_broken`
- 第二轮无法判定则保留 `suspicious`
## 6. 第一轮: 容器级快速校验设计
### 目标
- 用尽量低的 I/O 成本筛掉绝大多数明显坏文件。
- 尽量不做深度解码。
### 推荐命令
建议首版基于如下 `ffprobe` 级别的信息:
```bash
ffprobe -v error -show_entries format=format_name,duration,size:stream=index,codec_type,codec_name,width,height -of json <file>
```
说明:
- `-v error` 只保留错误输出。
- `show_entries` 只拿最小必要字段,避免无意义输出。
- 不启用 `-count_frames`、`-show_packets` 这类重操作。
### 第一轮判定规则
直接判定 `confirmed_broken`:
- `ffprobe` 非零退出。
- 无法识别容器格式。
- 没有视频流。
- 错误信息命中高置信度关键词,例如:
- `moov atom not found`
- `error reading header`
- `EBML header parsing failed`
- `Invalid data found when processing input`
判定 `pending_stage2`:
- `ffprobe` 成功,但 `stderr` 非空且命中软告警关键词。
- `duration` 缺失、为 `N/A`、非数字或 `<= 0`。
- 容器属于 TG 场景高风险集合,且启用了 `tg-hdd` profile。
建议首版高风险集合:
- `mp4`
- `mov`
- `m4v`
- `mkv`
- `webm`
判定 `passed_stage1`:
- `ffprobe` 成功。
- 存在视频流。
- 关键字段齐全。
- 未命中可疑规则。
### 为什么第一轮只做容器级校验
因为你的场景里,大量半截文件会在这一层就暴露:
- MP4/MOV 缺尾时常直接在 `moov`/header 层失败。
- MKV/WebM 容器不完整时常直接在 EBML 层失败。
这一步的价值是先把“最便宜能抓到的大头”清掉。
## 7. 第二轮: 更重验证设计
### 目标
- 不对所有文件做完整解码。
- 只对第一轮失败边界附近的文件做更重验证。
- 重点针对“头部还能读,但内容后段已截断”的视频。
### 推荐验证策略
第二轮默认不做全量解码,采用“定向采样 + 必要时升级”的两级策略。
#### 模式 1: 定向采样验证
推荐用 `ffmpeg` 在少数时间点解码短窗口,例如:
- 片头附近
- 中段
- 尾段附近
示例思路:
```bash
ffmpeg -v error -xerror -ss <offset> -i <file> -t 3 -map 0:v:0 -f null -
```
采样点建议:
- 时长 <= 60s: `10%`, `85%`
- 60s < 时长 <= 10min: `10%`, `50%`, `90%`
- 时长 > 10min: `10%`, `60%`, `92%`
尾段点位最关键,因为 TG 半截文件最常见的问题就是尾部缺失。
#### 模式 2: 完整解码确认
只对以下文件升级:
- 定向采样失败但错误原因不够稳定。
- 用户手动指定 `--full-decode-confirm`。
- 需要给出更高置信度结论的少量边界样本。
示例命令:
```bash
ffmpeg -v error -xerror -i <file> -f null -
```
### 第二轮判定规则
直接判定 `confirmed_broken`:
- 任一采样窗口退出非零。
- 错误信息命中以下关键词:
- `Truncating packet`
- `Invalid NAL unit size`
- `decode_slice_header error`
- `corrupt input packet in stream`
保留为 `suspicious`:
- 第二轮命令执行异常,但不是明确媒体损坏。
- 不同采样窗口结果相互矛盾。
判定 `passed_stage2`:
- 所有采样窗口通过。
## 8. 机械硬盘优化策略
这是本项目最重要的工程约束之一,不能事后再补。
### 调度原则
- 不让第一轮和第二轮并发运行。
- 默认采用阶段式执行:
- 目录发现 + 第一轮
- 第一轮结束后再跑第二轮
原因:
- 在同一块 HDD 上同时跑 `ffprobe` 和 `ffmpeg`,会把顺序读打碎成大量随机寻道。
### 默认并发策略
建议提供 `--storage-profile`:
- `hdd`
- `ssd`
- `auto`
首版默认建议:
- `hdd`: `stage1_workers=1`, `stage2_workers=1`
- `ssd`: `stage1_workers=4`, `stage2_workers=2`
- `auto`: Windows 下默认保守映射到 `hdd`
理由:
- HDD 上盲目开线程,经常只会让总吞吐下降。
- 多线程对 CPU 友好,不代表对磁盘友好。
### 队列与背压
- 使用有界队列。
- 不允许一次性把几十万条待处理任务全部压进内存。
- 队列深度建议默认不超过 `64`。
### 文件顺序
- 发现顺序和处理顺序尽量保持一致。
- 默认按目录遍历顺序写入 `discovery_order`。
- 阶段一按 `discovery_order` 处理。
- 阶段二按可疑文件的 `discovery_order` 再处理。
### 数据落盘
- SQLite 写入采用批量提交,例如每 `100` 或 `500` 条一次事务提交。
- 结果边写边落盘,不等全流程结束。
## 9. 大量视频处理优化
### 必做优化
1. **断点续跑**
- 再次运行时,若 `path + size + mtime` 未变化,直接复用上一轮结果。
2. **结果流式写入**
- 不把全部结果留在内存。
- 允许处理百万级文件清单。
3. **分阶段输出**
- 第一轮结束后立即产出“明确损坏 + 待二轮复核”列表。
4. **最小日志**
- 默认只输出阶段进度与统计摘要。
- 不为每个文件打印一行日志。
5. **可中断**
- 捕获 `Ctrl+C`,将状态安全落盘。
### 建议优化
1. **扩展名白名单**
- 首版只扫常见视频扩展名。
- 减少误把非视频文件交给 `ffprobe`。
2. **目录级切分**
- 对超大目录,允许后续增加“按一级目录分批执行”的操作模式。
3. **结果摘要缓存**
- 对重复查看的报告生成固定 JSONL/CSV,不每次重新聚合。
## 10. CLI 设计
建议提供如下命令:
```bash
uv run tgvideo-check scan D:\Exports\ChatA D:\Exports\ChatB
```
首版参数建议:
- `--profile tg-hdd`
- `--recursive / --no-recursive`
- `--extensions mp4,mov,mkv,webm,avi`
- `--db .\scan_state.db`
- `--resume`
- `--stage1-workers 1`
- `--stage2-workers 1`
- `--sample-duration 3`
- `--full-decode-confirm`
- `--export-jsonl .\results.jsonl`
- `--export-csv .\results.csv`
建议内置 profile:
- `tg-hdd`
- 针对当前用户场景的默认推荐档。
- 保守并发。
- 高风险容器进入第二轮。
- `tg-fast`
- 第一轮尽可能快。
- 第二轮更严格地缩小范围。
- `deep-verify`
- 面向人工复核的小样本。
- 允许完整解码确认。
## 11. 项目结构建议
建议在 `plan-for-tgvideo` 下实现为独立 Python 工具:
```text
plan-for-tgvideo/
pyproject.toml
uv.lock
README.md
src/
tgvideo_check/
__init__.py
cli.py
config.py
discovery.py
scheduler.py
store.py
models.py
ffprobe_runner.py
ffmpeg_runner.py
classifiers.py
reporting.py
profiles.py
tests/
test_profiles.py
test_stage1_classification.py
test_stage2_classification.py
test_resume_logic.py
test_integration_truncated_mp4.py
test_integration_truncated_mkv.py
samples/
valid/
broken/
```
## 12. 分任务实施计划
### Task 1: 初始化独立项目骨架
**Files:**
- Create: `plan-for-tgvideo/pyproject.toml`
- Create: `plan-for-tgvideo/README.md`
- Create: `plan-for-tgvideo/src/tgvideo_check/__init__.py`
- Create: `plan-for-tgvideo/src/tgvideo_check/cli.py`
**Steps:**
1. 用 `uv init` 初始化项目。
2. 设定 CLI 入口为 `tgvideo-check`。
3. 增加最小依赖:`typer`、`orjson`。
4. 跑 `uv run tgvideo-check --help`,确认骨架可执行。
### Task 2: 建立配置与 profile 系统
**Files:**
- Create: `plan-for-tgvideo/src/tgvideo_check/config.py`
- Create: `plan-for-tgvideo/src/tgvideo_check/profiles.py`
- Test: `plan-for-tgvideo/tests/test_profiles.py`
**Steps:**
1. 定义全局配置模型。
2. 实现 `tg-hdd`、`tg-fast`、`deep-verify` 三个 profile。
3. 明确 `hdd` 默认 worker 数与采样策略。
4. 编写配置解析测试。
### Task 3: 实现目录发现与候选文件过滤
**Files:**
- Create: `plan-for-tgvideo/src/tgvideo_check/discovery.py`
- Create: `plan-for-tgvideo/src/tgvideo_check/models.py`
- Test: `plan-for-tgvideo/tests/test_discovery.py`
**Steps:**
1. 实现多目录递归枚举。
2. 只保留目标扩展名。
3. 记录 `path / size / mtime / extension / discovery_order`。
4. 保证发现顺序稳定。
### Task 4: 实现 SQLite 状态存储与恢复
**Files:**
- Create: `plan-for-tgvideo/src/tgvideo_check/store.py`
- Test: `plan-for-tgvideo/tests/test_resume_logic.py`
**Steps:**
1. 初始化数据库 schema。
2. 支持批量 upsert 文件记录。
3. 以 `path + size + mtime` 作为恢复依据。
4. 确保中断后可恢复到上次阶段状态。
### Task 5: 实现第一轮 `ffprobe` 执行器
**Files:**
- Create: `plan-for-tgvideo/src/tgvideo_check/ffprobe_runner.py`
- Test: `plan-for-tgvideo/tests/test_stage1_runner.py`
**Steps:**
1. 封装 `ffprobe` 命令构建。
2. 限制输出字段。
3. 捕获 `stdout`、`stderr`、退出码与耗时。
4. 对超时和命令不存在给出明确错误状态。
### Task 6: 实现第一轮分类器
**Files:**
- Create: `plan-for-tgvideo/src/tgvideo_check/classifiers.py`
- Test: `plan-for-tgvideo/tests/test_stage1_classification.py`
**Steps:**
1. 定义高置信度错误关键词。
2. 定义软告警关键词和可疑规则。
3. 加入 TG 高风险容器升级规则。
4. 输出 `confirmed_broken / pending_stage2 / passed_stage1` 三种核心结果。
### Task 7: 实现第二轮 `ffmpeg` 定向验证
**Files:**
- Create: `plan-for-tgvideo/src/tgvideo_check/ffmpeg_runner.py`
- Test: `plan-for-tgvideo/tests/test_stage2_classification.py`
**Steps:**
1. 根据时长生成采样点。
2. 实现定向短窗口验证命令。
3. 支持按需升级到完整解码确认。
4. 归类典型流损坏错误。
### Task 8: 实现 HDD 友好的调度器
**Files:**
- Create: `plan-for-tgvideo/src/tgvideo_check/scheduler.py`
- Test: `plan-for-tgvideo/tests/test_scheduler_hdd.py`
**Steps:**
1. 强制阶段一和阶段二不重叠运行。
2. 用有界队列限制内存和并发。
3. 默认 `hdd` profile 使用单 worker。
4. 保证顺序处理和中断恢复。
### Task 9: 实现报告导出
**Files:**
- Create: `plan-for-tgvideo/src/tgvideo_check/reporting.py`
- Test: `plan-for-tgvideo/tests/test_reporting.py`
**Steps:**
1. 输出 JSONL。
2. 输出 CSV 摘要。
3. 支持按原因统计。
4. 支持导出“建议重新下载”清单。
### Task 10: 集成测试与样本验证
**Files:**
- Create: `plan-for-tgvideo/tests/test_integration_truncated_mp4.py`
- Create: `plan-for-tgvideo/tests/test_integration_truncated_mkv.py`
- Create: `plan-for-tgvideo/tests/test_integration_resume.py`
- Create: `plan-for-tgvideo/samples/README.md`
**Steps:**
1. 准备有效视频样本。
2. 人工截断生成坏样本。
3. 验证第一轮能抓住明显容器损坏。
4. 验证第二轮能抓住头部可读但中后段损坏的样本。
### Task 11: 性能基线与调优
**Files:**
- Create: `plan-for-tgvideo/tests/test_perf_smoke.py`
- Modify: `plan-for-tgvideo/README.md`
**Steps:**
1. 在 HDD 场景下跑小规模基线测试。
2. 比较 `stage1_workers=1` 与 `2` 的总耗时。
3. 比较“定向采样”与“完整解码”的吞吐差异。
4. 固化默认 profile 参数。
## 13. 测试策略
### 单元测试
- profile 解析
- SQLite 恢复逻辑
- 第一轮分类规则
- 第二轮错误归类
- 采样点计算
### 集成测试
- 正常 MP4
- 截断 MP4
- 正常 MKV
- 截断 MKV
- 命令缺失
- 中断后恢复
### 手工验证
- 在真实 TG 导出目录上跑 `tg-hdd` profile
- 抽样核对 `confirmed_broken`
- 抽样核对 `passed_stage2`
- 对边界文件再用播放器人工抽查
## 14. 风险与缓解
### 风险 1: 第二轮样本点过少,漏掉局部损坏
缓解:
- 保留 `--full-decode-confirm`。
- 允许 profile 提高采样点数量。
### 风险 2: 对 HDD 开过多并发导致更慢
缓解:
- `tg-hdd` 默认单 worker。
- 基线测试后再决定是否允许提高默认值。
### 风险 3: 结果解释不清,用户不知道哪些文件该重下
缓解:
- 报告中明确区分 `confirmed_broken` 和 `suspicious`。
- 输出可读原因码与建议动作。
### 风险 4: 边界格式太多,首版支持面过宽
缓解:
- 首版只优先优化 `mp4/mov/m4v/mkv/webm/avi`。
## 15. 验收标准
满足以下条件即可认为首版达标:
- 能选择一个或多个目录运行。
- 第一轮完成全量容器级快速校验。
- 第二轮只处理第一轮失败或可疑文件。
- 具备 `tg-hdd` profile,并默认采用 HDD 友好的保守并发。
- 支持 SQLite 落盘与断点续跑。
- 能输出明确的坏文件清单与原因分类。
- 在真实机械硬盘数据集上,默认模式明显快于“对所有文件完整解码”的朴素方案。
## 16. 首版执行建议
建议严格按以下顺序推进:
1. 先把 SQLite 恢复和第一轮 `ffprobe` 跑通。
2. 再实现分类器,确保“坏 / 可疑 / 通过”三分法稳定。
3. 然后实现第二轮定向采样验证。
4. 最后再做 HDD 调度细化和性能调优。
不要一开始就做“全量完整解码 + 高并发”,那是最容易把项目做成一个在真实机械硬盘上难用的慢工具的路线。
+1
View File
@@ -0,0 +1 @@
3.12
+43
View File
@@ -0,0 +1,43 @@
## TG Video Check
针对 Telegram 导出场景的视频损坏检测 CLI,优先识别被中断下载导致的半截视频文件。
### 当前能力
- 递归扫描一个或多个目录中的常见视频扩展名
- 使用 SQLite 落盘扫描状态,并基于 `path + size + mtime` 复用旧结果
- stage1 使用 `ffprobe` 做容器级快速校验
- stage2 只对可疑文件使用 `ffmpeg` 做定向采样或完整确认
- 输出 JSONL / CSV 报告
### 基本用法
```powershell
uv run tgvideo-check scan D:\Exports\ChatA D:\Exports\ChatB --profile tg-hdd
```
常用参数:
- `--db`:指定 SQLite 状态库
- `--export-jsonl` / `--export-csv`:导出结果报告
- `--stage1-workers` / `--stage2-workers`:覆盖 profile 默认 worker 数
- `--queue-limit`:限制单阶段批次大小,避免把待处理文件一次性堆进内存
- `--full-decode-confirm`:对 stage2 强制使用完整解码确认
### 性能 Smoke
仓库内提供了 `tests/test_perf_smoke.py`,默认跳过,不把环境波动硬编码成普通单测。
手动运行方式:
```powershell
$env:TGVIDEO_RUN_PERF_SMOKE='1'
uv run pytest tests/test_perf_smoke.py -q
```
这个 smoke 目前只做两件事:
- 生成一批最小测试视频
- 分别跑 `stage1_workers=1` 和 `stage1_workers=2`,确认基线采集流程可执行
它不是严格 benchmark;真正的默认参数取舍,仍应基于真实 HDD 数据集人工复核。
+22
View File
@@ -0,0 +1,22 @@
[project]
name = "plan-for-tgvideo"
version = "0.1.0"
description = "Telegram export video damage detector"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"orjson>=3.11.3",
"typer>=0.19.2",
]
[project.scripts]
tgvideo-check = "tgvideo_check.cli:main"
[dependency-groups]
dev = [
"pytest>=8.4.2",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
+18
View File
@@ -0,0 +1,18 @@
# Sample Strategy
当前仓库不直接提交二进制视频样本,原因很简单:
- 媒体文件会迅速放大仓库体积。
- 截断样本本质上可以通过脚本化方式稳定生成。
- 当前测试环境已经有 `ffmpeg` / `ffprobe`,更适合在测试中即时构造最小样本。
当前集成测试采用两类动态样本:
- 截断 `mp4`:验证 stage1 容器级快速校验能直接识别明显损坏。
- 截断 `mkv`:验证 stage1 先升级到 stage2,随后由 `ffmpeg` 定向验证抓住尾段损坏。
如果后续需要保留人工复核样本,建议采用以下策略:
- 仅放入极小尺寸、极短时长的媒体文件。
- 同时保留生成命令和截断规则,避免样本来源不可追溯。
- 将“真实用户样本”与“测试生成样本”分开管理,不要混到同一目录。
@@ -0,0 +1,2 @@
"""Telegram video damage detector package."""
@@ -0,0 +1,118 @@
from __future__ import annotations
from dataclasses import dataclass
HARD_ERROR_KEYWORDS = (
"moov atom not found",
"error reading header",
"ebml header parsing failed",
"invalid data found when processing input",
)
SOFT_WARNING_KEYWORDS = (
"partial file",
"estimating duration from bitrate",
"missing key frame",
)
@dataclass(frozen=True)
class Stage1Classification:
status: str
reason_code: str
warning_flags: tuple[str, ...] = ()
def classify_stage1_result(
*,
exit_code: int | None,
stderr_text: str,
probe_payload: dict[str, object] | None,
extension: str,
high_risk_extensions: tuple[str, ...],
) -> Stage1Classification:
normalized_stderr = stderr_text.lower()
if exit_code not in (0, None):
return Stage1Classification(
status="confirmed_broken",
reason_code="ffprobe_exit_nonzero",
)
for keyword in HARD_ERROR_KEYWORDS:
if keyword in normalized_stderr:
return Stage1Classification(
status="confirmed_broken",
reason_code="hard_error_keyword",
warning_flags=(keyword,),
)
format_info = _as_dict(probe_payload.get("format") if probe_payload else None)
streams = probe_payload.get("streams") if probe_payload else None
if not format_info.get("format_name"):
return Stage1Classification(
status="confirmed_broken",
reason_code="missing_format_name",
)
if not _has_video_stream(streams):
return Stage1Classification(
status="confirmed_broken",
reason_code="missing_video_stream",
)
soft_hits = tuple(
keyword for keyword in SOFT_WARNING_KEYWORDS if keyword in normalized_stderr
)
if soft_hits:
return Stage1Classification(
status="pending_stage2",
reason_code="soft_warning_keyword",
warning_flags=soft_hits,
)
if not _is_valid_duration(format_info.get("duration")):
return Stage1Classification(
status="pending_stage2",
reason_code="invalid_duration",
)
if extension in high_risk_extensions:
return Stage1Classification(
status="pending_stage2",
reason_code="high_risk_extension",
)
return Stage1Classification(
status="passed_stage1",
reason_code="passed_stage1",
)
def _as_dict(value: object) -> dict[str, object]:
if isinstance(value, dict):
return value
return {}
def _has_video_stream(streams: object) -> bool:
if not isinstance(streams, list):
return False
for stream in streams:
if isinstance(stream, dict) and stream.get("codec_type") == "video":
return True
return False
def _is_valid_duration(value: object) -> bool:
if not isinstance(value, str):
return False
if value in {"N/A", ""}:
return False
try:
return float(value) > 0
except ValueError:
return False
+362
View File
@@ -0,0 +1,362 @@
from __future__ import annotations
from collections import Counter
from pathlib import Path
import orjson
import typer
from tgvideo_check.classifiers import classify_stage1_result
from tgvideo_check.config import ScanConfig, normalize_extension
from tgvideo_check.discovery import discover_files
from tgvideo_check.ffmpeg_runner import generate_sample_offsets, run_ffmpeg_verification
from tgvideo_check.ffprobe_runner import run_ffprobe
from tgvideo_check.profiles import resolve_config
from tgvideo_check.reporting import ReportRow, export_csv_summary, export_jsonl
from tgvideo_check.scheduler import (
build_stage_plan,
execute_stage_batch,
select_pending_stage1_files,
select_pending_stage2_files,
)
from tgvideo_check.store import ScanStore
app = typer.Typer(
add_completion=False,
help="Batch detector for truncated or damaged Telegram-exported videos.",
)
@app.callback()
def root() -> None:
"""TG video damage detector CLI."""
@app.command()
def scan(
paths: list[Path] = typer.Argument(
...,
exists=True,
file_okay=False,
dir_okay=True,
readable=True,
help="One or more root directories to scan.",
),
profile: str = typer.Option("tg-hdd", "--profile"),
recursive: bool = typer.Option(True, "--recursive/--no-recursive"),
extensions: str = typer.Option(
"mp4,mov,m4v,mkv,webm,avi",
"--extensions",
help="Comma-separated video extensions.",
),
db_path: Path = typer.Option(Path("scan_state.db"), "--db"),
resume: bool = typer.Option(True, "--resume/--no-resume"),
stage1_workers: int | None = typer.Option(None, "--stage1-workers"),
stage2_workers: int | None = typer.Option(None, "--stage2-workers"),
queue_limit: int = typer.Option(64, "--queue-limit"),
sample_duration: int | None = typer.Option(None, "--sample-duration"),
full_decode_confirm: bool = typer.Option(False, "--full-decode-confirm"),
export_jsonl_path: Path | None = typer.Option(None, "--export-jsonl"),
export_csv_path: Path | None = typer.Option(None, "--export-csv"),
) -> None:
"""Scan directories for suspicious video files."""
config = resolve_config(
ScanConfig(
roots=tuple(paths),
profile=profile,
recursive=recursive,
extensions=_parse_extensions(extensions),
db_path=db_path,
resume=resume,
stage1_workers=stage1_workers,
stage2_workers=stage2_workers,
queue_limit=queue_limit,
sample_duration=sample_duration,
full_decode_confirm=full_decode_confirm,
export_jsonl=export_jsonl_path,
export_csv=export_csv_path,
)
)
rows = _execute_scan(config)
summary = Counter(row.final_status for row in rows)
summary_text = " ".join(
f"{status}={summary[status]}" for status in sorted(summary.keys())
)
typer.echo(f"Scanned {len(rows)} file(s). {summary_text}".strip())
def main() -> None:
app()
def _execute_scan(config: ScanConfig) -> list[ReportRow]:
store = ScanStore(config.db_path)
store.initialize()
run_id = store.create_run(profile=config.profile, root_paths=config.roots)
reasons: dict[int, str] = {}
stage1_payloads: dict[int, dict[str, object]] = {}
try:
candidates = discover_files(
config.roots,
extensions=config.extensions,
recursive=config.recursive,
)
stored_files = store.upsert_files(
run_id,
candidates,
reuse_previous_results=config.resume,
)
stage1_batches = build_stage_plan(
stage1_files=select_pending_stage1_files(stored_files),
stage2_files=[],
stage1_workers=config.stage1_workers or 1,
stage2_workers=config.stage2_workers or 1,
queue_limit=config.queue_limit,
)
for batch in stage1_batches:
probe_results = execute_stage_batch(
batch,
lambda item: run_ffprobe(item.path),
)
for stored_file, probe_result in zip(batch.files, probe_results):
store.record_stage1_result(
stored_file.file_id,
exit_code=probe_result.exit_code,
stdout_text=probe_result.stdout_text,
stderr_text=probe_result.stderr_text,
elapsed_ms=probe_result.elapsed_ms,
)
if probe_result.status != "completed":
store.record_stage1_status(
stored_file.file_id,
stage1_status="command_error",
final_status="command_error",
reason_code=probe_result.status,
)
reasons[stored_file.file_id] = probe_result.status
continue
payload = _parse_probe_payload(probe_result.stdout_text)
if probe_result.stdout_text and payload is None:
store.record_stage1_status(
stored_file.file_id,
stage1_status="command_error",
final_status="command_error",
reason_code="invalid_probe_output",
)
reasons[stored_file.file_id] = "invalid_probe_output"
continue
classification = classify_stage1_result(
exit_code=probe_result.exit_code,
stderr_text=probe_result.stderr_text,
probe_payload=payload,
extension=stored_file.extension,
high_risk_extensions=config.high_risk_extensions,
)
final_status = (
classification.status
if classification.status in {"confirmed_broken", "passed_stage1"}
else None
)
store.record_stage1_status(
stored_file.file_id,
stage1_status=classification.status,
final_status=final_status,
reason_code=classification.reason_code,
)
reasons[stored_file.file_id] = classification.reason_code
if payload is not None:
stage1_payloads[stored_file.file_id] = payload
stored_files = store.list_files(run_id)
stage2_batches = build_stage_plan(
stage1_files=[],
stage2_files=select_pending_stage2_files(stored_files),
stage1_workers=config.stage1_workers or 1,
stage2_workers=config.stage2_workers or 1,
queue_limit=config.queue_limit,
)
for batch in stage2_batches:
stage2_outputs = execute_stage_batch(
batch,
lambda item: _run_stage2_for_file(
item.path,
payload=stage1_payloads.get(item.file_id)
or _parse_probe_payload(store.get_stage1_stdout_text(item.file_id) or "")
or _reload_probe_payload(item.path),
sample_duration=config.sample_duration or 3,
full_decode_confirm=config.full_decode_confirm,
),
)
for stored_file, (stage2_status, reason_code, stage2_result) in zip(
batch.files, stage2_outputs
):
store.record_stage2_result(
stored_file.file_id,
mode=stage2_result["mode"],
sample_points_json=stage2_result["sample_points_json"],
exit_code=stage2_result["exit_code"],
stderr_text=stage2_result["stderr_text"],
reason_code=reason_code,
elapsed_ms=stage2_result["elapsed_ms"],
)
store.record_stage2_status(
stored_file.file_id,
stage2_status=stage2_status,
final_status=stage2_status,
reason_code=reason_code,
)
reasons[stored_file.file_id] = reason_code
rows = _build_report_rows(store.list_files(run_id), reasons)
if config.export_jsonl is not None:
export_jsonl(rows, config.export_jsonl)
if config.export_csv is not None:
export_csv_summary(rows, config.export_csv)
store.update_run_status(run_id, status="completed")
return rows
except Exception:
store.update_run_status(run_id, status="failed")
raise
def _build_report_rows(
stored_files: list,
reasons: dict[int, str],
) -> list[ReportRow]:
rows: list[ReportRow] = []
for stored_file in stored_files:
final_status = stored_file.final_status or stored_file.stage1_status
rows.append(
ReportRow(
path=stored_file.path,
final_status=final_status,
reason_code=stored_file.reason_code
or reasons.get(stored_file.file_id, final_status or "unknown_reason"),
size=stored_file.size,
)
)
return rows
def _run_stage2_for_file(
path: Path,
*,
payload: dict[str, object] | None,
sample_duration: int,
full_decode_confirm: bool,
) -> tuple[str, str, dict[str, int | str | None]]:
duration = _extract_duration(payload)
if full_decode_confirm or duration is None:
sample_points: tuple[float, ...] = ()
mode = "full_decode"
executions = [
run_ffmpeg_verification(
path,
full_decode=True,
)
]
else:
sample_points = generate_sample_offsets(duration)
mode = "sample"
executions = [
run_ffmpeg_verification(
path,
offset_seconds=offset,
sample_duration=sample_duration,
)
for offset in sample_points
]
total_elapsed = sum(item.elapsed_ms for item in executions)
stderr_text = "\n".join(item.stderr_text for item in executions if item.stderr_text)
for execution in executions:
if execution.status != "completed":
return (
"command_error",
execution.reason_code or execution.status,
{
"mode": mode,
"sample_points_json": _dump_json(sample_points),
"exit_code": execution.exit_code,
"stderr_text": stderr_text,
"elapsed_ms": total_elapsed,
},
)
if execution.exit_code not in (0, None):
return (
"confirmed_broken",
execution.reason_code or "ffmpeg_exit_nonzero",
{
"mode": mode,
"sample_points_json": _dump_json(sample_points),
"exit_code": execution.exit_code,
"stderr_text": stderr_text,
"elapsed_ms": total_elapsed,
},
)
return (
"passed_stage2",
"passed_stage2",
{
"mode": mode,
"sample_points_json": _dump_json(sample_points),
"exit_code": 0,
"stderr_text": stderr_text,
"elapsed_ms": total_elapsed,
},
)
def _reload_probe_payload(path: Path) -> dict[str, object] | None:
probe_result = run_ffprobe(path)
if probe_result.status != "completed":
return None
return _parse_probe_payload(probe_result.stdout_text)
def _parse_probe_payload(stdout_text: str) -> dict[str, object] | None:
if not stdout_text.strip():
return None
payload = orjson.loads(stdout_text)
if isinstance(payload, dict):
return payload
return None
def _extract_duration(payload: dict[str, object] | None) -> float | None:
if payload is None:
return None
format_info = payload.get("format")
if not isinstance(format_info, dict):
return None
duration_value = format_info.get("duration")
if not isinstance(duration_value, str):
return None
try:
duration = float(duration_value)
except ValueError:
return None
return duration if duration > 0 else None
def _parse_extensions(raw_extensions: str) -> tuple[str, ...]:
return tuple(
normalize_extension(item)
for item in raw_extensions.split(",")
if item.strip()
)
def _dump_json(values: tuple[float, ...]) -> str:
return orjson.dumps(values).decode("utf-8")
@@ -0,0 +1,35 @@
from __future__ import annotations
from dataclasses import dataclass, replace
from pathlib import Path
def normalize_extension(value: str) -> str:
return value.lower().lstrip(".")
@dataclass(frozen=True)
class ScanConfig:
roots: tuple[Path, ...]
profile: str = "tg-hdd"
recursive: bool = True
extensions: tuple[str, ...] = ("mp4", "mov", "m4v", "mkv", "webm", "avi")
db_path: Path = Path("scan_state.db")
resume: bool = True
stage1_workers: int | None = None
stage2_workers: int | None = None
queue_limit: int = 64
sample_duration: int | None = None
full_decode_confirm: bool = False
export_jsonl: Path | None = None
export_csv: Path | None = None
high_risk_extensions: tuple[str, ...] = ()
def normalized(self) -> "ScanConfig":
return replace(
self,
extensions=tuple(normalize_extension(item) for item in self.extensions),
high_risk_extensions=tuple(
normalize_extension(item) for item in self.high_risk_extensions
),
)
@@ -0,0 +1,62 @@
from __future__ import annotations
from pathlib import Path
from tgvideo_check.config import normalize_extension
from tgvideo_check.models import FileCandidate
def discover_files(
roots: tuple[Path, ...],
*,
extensions: tuple[str, ...],
recursive: bool = True,
) -> list[FileCandidate]:
normalized_extensions = {normalize_extension(item) for item in extensions}
discovered: list[FileCandidate] = []
for root in roots:
_discover_root(
root=root,
recursive=recursive,
extensions=normalized_extensions,
discovered=discovered,
)
return discovered
def _discover_root(
*,
root: Path,
recursive: bool,
extensions: set[str],
discovered: list[FileCandidate],
) -> None:
entries = sorted(root.iterdir(), key=lambda item: (not item.is_file(), item.name.lower()))
for entry in entries:
if entry.is_dir():
if recursive:
_discover_root(
root=entry,
recursive=recursive,
extensions=extensions,
discovered=discovered,
)
continue
extension = normalize_extension(entry.suffix)
if extension not in extensions:
continue
stat = entry.stat()
discovered.append(
FileCandidate(
path=entry,
size=stat.st_size,
mtime_ns=stat.st_mtime_ns,
extension=extension,
discovery_order=len(discovered),
)
)
@@ -0,0 +1,148 @@
from __future__ import annotations
import subprocess
import time
from dataclasses import dataclass
from pathlib import Path
ERROR_REASON_PATTERNS: tuple[tuple[str, str], ...] = (
("truncating packet", "truncating_packet"),
("invalid nal unit size", "invalid_nal_unit_size"),
("decode_slice_header error", "decode_slice_header_error"),
("corrupt input packet in stream", "corrupt_input_packet"),
)
@dataclass(frozen=True)
class FfmpegExecutionResult:
status: str
command: tuple[str, ...]
exit_code: int | None
stdout_text: str
stderr_text: str
elapsed_ms: int
reason_code: str | None
def generate_sample_offsets(duration_seconds: float) -> tuple[float, ...]:
if duration_seconds <= 60:
ratios = (0.10, 0.85)
elif duration_seconds <= 600:
ratios = (0.10, 0.50, 0.90)
else:
ratios = (0.10, 0.60, 0.92)
return tuple(round(duration_seconds * ratio, 3) for ratio in ratios)
def build_ffmpeg_command(
path: Path,
*,
offset_seconds: float | None = None,
sample_duration: int = 3,
full_decode: bool = False,
executable: str = "ffmpeg",
launcher_args: tuple[str, ...] = (),
) -> list[str]:
command = [executable, *launcher_args, "-v", "error", "-xerror"]
if not full_decode:
if offset_seconds is None:
raise ValueError("offset_seconds is required for sample verification")
command.extend(["-ss", _format_time_value(offset_seconds)])
command.extend(["-i", str(path)])
if not full_decode:
command.extend(["-t", str(sample_duration), "-map", "0:v:0"])
command.extend(["-f", "null", "-"])
return command
def run_ffmpeg_verification(
path: Path,
*,
offset_seconds: float | None = None,
sample_duration: int = 3,
full_decode: bool = False,
executable: str = "ffmpeg",
launcher_args: tuple[str, ...] = (),
timeout_seconds: float = 30.0,
) -> FfmpegExecutionResult:
command = tuple(
build_ffmpeg_command(
path,
offset_seconds=offset_seconds,
sample_duration=sample_duration,
full_decode=full_decode,
executable=executable,
launcher_args=launcher_args,
)
)
start = time.perf_counter()
try:
completed = subprocess.run(
command,
capture_output=True,
text=True,
encoding="utf-8",
timeout=timeout_seconds,
check=False,
)
except FileNotFoundError as exc:
return FfmpegExecutionResult(
status="missing_command",
command=command,
exit_code=None,
stdout_text="",
stderr_text=str(exc),
elapsed_ms=_elapsed_ms(start),
reason_code="missing_command",
)
except subprocess.TimeoutExpired as exc:
return FfmpegExecutionResult(
status="timeout",
command=command,
exit_code=None,
stdout_text=_coerce_process_output(exc.stdout),
stderr_text=_coerce_process_output(exc.stderr),
elapsed_ms=_elapsed_ms(start),
reason_code="timeout",
)
return FfmpegExecutionResult(
status="completed",
command=command,
exit_code=completed.returncode,
stdout_text=completed.stdout,
stderr_text=completed.stderr,
elapsed_ms=_elapsed_ms(start),
reason_code=classify_ffmpeg_error(completed.stderr),
)
def classify_ffmpeg_error(stderr_text: str) -> str | None:
normalized = stderr_text.lower()
for pattern, reason_code in ERROR_REASON_PATTERNS:
if pattern in normalized:
return reason_code
return None
def _elapsed_ms(start: float) -> int:
return int((time.perf_counter() - start) * 1000)
def _coerce_process_output(output: str | bytes | None) -> str:
if output is None:
return ""
if isinstance(output, bytes):
return output.decode("utf-8", errors="replace")
return output
def _format_time_value(value: float) -> str:
return f"{value:.3f}".rstrip("0").rstrip(".")
@@ -0,0 +1,107 @@
from __future__ import annotations
import subprocess
import time
from dataclasses import dataclass
from pathlib import Path
MINIMAL_SHOW_ENTRIES = (
"format=format_name,duration,size:"
"stream=index,codec_type,codec_name,width,height"
)
@dataclass(frozen=True)
class ProbeExecutionResult:
status: str
command: tuple[str, ...]
exit_code: int | None
stdout_text: str
stderr_text: str
elapsed_ms: int
def build_ffprobe_command(
path: Path,
*,
executable: str = "ffprobe",
launcher_args: tuple[str, ...] = (),
) -> list[str]:
return [
executable,
*launcher_args,
"-v",
"error",
"-show_entries",
MINIMAL_SHOW_ENTRIES,
"-of",
"json",
str(path),
]
def run_ffprobe(
path: Path,
*,
executable: str = "ffprobe",
launcher_args: tuple[str, ...] = (),
timeout_seconds: float = 30.0,
) -> ProbeExecutionResult:
command = tuple(
build_ffprobe_command(
path,
executable=executable,
launcher_args=launcher_args,
)
)
start = time.perf_counter()
try:
completed = subprocess.run(
command,
capture_output=True,
text=True,
encoding="utf-8",
timeout=timeout_seconds,
check=False,
)
except FileNotFoundError as exc:
return ProbeExecutionResult(
status="missing_command",
command=command,
exit_code=None,
stdout_text="",
stderr_text=str(exc),
elapsed_ms=_elapsed_ms(start),
)
except subprocess.TimeoutExpired as exc:
return ProbeExecutionResult(
status="timeout",
command=command,
exit_code=None,
stdout_text=_coerce_process_output(exc.stdout),
stderr_text=_coerce_process_output(exc.stderr),
elapsed_ms=_elapsed_ms(start),
)
return ProbeExecutionResult(
status="completed",
command=command,
exit_code=completed.returncode,
stdout_text=completed.stdout,
stderr_text=completed.stderr,
elapsed_ms=_elapsed_ms(start),
)
def _elapsed_ms(start: float) -> int:
return int((time.perf_counter() - start) * 1000)
def _coerce_process_output(output: str | bytes | None) -> str:
if output is None:
return ""
if isinstance(output, bytes):
return output.decode("utf-8", errors="replace")
return output
@@ -0,0 +1,29 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class FileCandidate:
path: Path
size: int
mtime_ns: int
extension: str
discovery_order: int
@dataclass(frozen=True)
class StoredFile:
file_id: int
run_id: int
path: Path
size: int
mtime_ns: int
extension: str
discovery_order: int
stage1_status: str
stage2_status: str | None
final_status: str | None
resumed_from_run_id: int | None
reason_code: str | None = None
@@ -0,0 +1,58 @@
from __future__ import annotations
from dataclasses import dataclass, replace
from tgvideo_check.config import ScanConfig
@dataclass(frozen=True)
class Profile:
name: str
stage1_workers: int
stage2_workers: int
sample_duration: int
high_risk_extensions: tuple[str, ...]
PROFILES: dict[str, Profile] = {
"tg-hdd": Profile(
name="tg-hdd",
stage1_workers=1,
stage2_workers=1,
sample_duration=3,
high_risk_extensions=("mp4", "mov", "m4v", "mkv", "webm"),
),
"tg-fast": Profile(
name="tg-fast",
stage1_workers=2,
stage2_workers=1,
sample_duration=3,
high_risk_extensions=("mp4", "mov", "m4v", "mkv", "webm"),
),
"deep-verify": Profile(
name="deep-verify",
stage1_workers=1,
stage2_workers=1,
sample_duration=5,
high_risk_extensions=("mp4", "mov", "m4v", "mkv", "webm", "avi"),
),
}
def get_profile(name: str) -> Profile:
try:
return PROFILES[name]
except KeyError as exc:
raise ValueError(f"Unknown profile: {name}") from exc
def resolve_config(config: ScanConfig) -> ScanConfig:
profile = get_profile(config.profile)
normalized = config.normalized()
return replace(
normalized,
stage1_workers=normalized.stage1_workers or profile.stage1_workers,
stage2_workers=normalized.stage2_workers or profile.stage2_workers,
sample_duration=normalized.sample_duration or profile.sample_duration,
high_risk_extensions=profile.high_risk_extensions,
)
@@ -0,0 +1,61 @@
from __future__ import annotations
import csv
from dataclasses import asdict, dataclass
from pathlib import Path
import orjson
@dataclass(frozen=True)
class ReportRow:
path: Path
final_status: str
reason_code: str
size: int
def summarize_reason_counts(rows: list[ReportRow]) -> dict[str, int]:
summary: dict[str, int] = {}
for row in rows:
summary[row.reason_code] = summary.get(row.reason_code, 0) + 1
return summary
def collect_redownload_candidates(rows: list[ReportRow]) -> list[Path]:
return [
row.path
for row in rows
if row.final_status in {"confirmed_broken", "suspicious"}
]
def export_jsonl(rows: list[ReportRow], destination: Path) -> None:
lines = [
orjson.dumps(
{
**asdict(row),
"path": str(row.path),
}
).decode("utf-8")
for row in rows
]
destination.write_text("\n".join(lines) + ("\n" if lines else ""), encoding="utf-8")
def export_csv_summary(rows: list[ReportRow], destination: Path) -> None:
with destination.open("w", encoding="utf-8", newline="") as handle:
writer = csv.DictWriter(
handle,
fieldnames=["final_status", "reason_code", "path", "size"],
)
writer.writeheader()
for row in rows:
writer.writerow(
{
"final_status": row.final_status,
"reason_code": row.reason_code,
"path": str(row.path),
"size": row.size,
}
)
@@ -0,0 +1,90 @@
from __future__ import annotations
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
from typing import Callable, TypeVar
from tgvideo_check.models import StoredFile
T = TypeVar("T")
@dataclass(frozen=True)
class StageBatch:
stage_name: str
worker_count: int
files: tuple[StoredFile, ...]
def select_pending_stage1_files(files: list[StoredFile]) -> list[StoredFile]:
return sorted(
(item for item in files if item.stage1_status == "pending_stage1"),
key=lambda item: item.discovery_order,
)
def select_pending_stage2_files(files: list[StoredFile]) -> list[StoredFile]:
return sorted(
(
item
for item in files
if item.stage1_status == "pending_stage2" and item.stage2_status is None
),
key=lambda item: item.discovery_order,
)
def build_stage_plan(
*,
stage1_files: list[StoredFile],
stage2_files: list[StoredFile],
stage1_workers: int,
stage2_workers: int,
queue_limit: int,
) -> tuple[StageBatch, ...]:
if queue_limit <= 0:
raise ValueError("queue_limit must be positive")
batches: list[StageBatch] = []
for chunk in _chunk_by_limit(stage1_files, queue_limit):
batches.append(
StageBatch(
stage_name="stage1",
worker_count=stage1_workers,
files=chunk,
)
)
for chunk in _chunk_by_limit(stage2_files, queue_limit):
batches.append(
StageBatch(
stage_name="stage2",
worker_count=stage2_workers,
files=chunk,
)
)
return tuple(batches)
def execute_stage_batch(
batch: StageBatch,
worker: Callable[[StoredFile], T],
) -> tuple[T, ...]:
if len(batch.files) <= 1 or batch.worker_count <= 1:
return tuple(worker(item) for item in batch.files)
max_workers = min(batch.worker_count, len(batch.files))
with ThreadPoolExecutor(max_workers=max_workers) as executor:
return tuple(executor.map(worker, batch.files))
def _chunk_by_limit(
files: list[StoredFile], queue_limit: int
) -> tuple[tuple[StoredFile, ...], ...]:
ordered = sorted(files, key=lambda item: item.discovery_order)
return tuple(
tuple(ordered[index : index + queue_limit])
for index in range(0, len(ordered), queue_limit)
)
+448
View File
@@ -0,0 +1,448 @@
from __future__ import annotations
import sqlite3
from dataclasses import dataclass
from pathlib import Path
import orjson
from tgvideo_check.models import FileCandidate, StoredFile
@dataclass(frozen=True)
class ScanRun:
run_id: int
profile: str
root_paths_json: str
status: str
class ScanStore:
def __init__(self, db_path: Path) -> None:
self.db_path = db_path
def initialize(self) -> None:
with self._connect() as connection:
connection.executescript(
"""
CREATE TABLE IF NOT EXISTS scan_runs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
profile TEXT NOT NULL,
root_paths_json TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'running'
);
CREATE TABLE IF NOT EXISTS files (
id INTEGER PRIMARY KEY AUTOINCREMENT,
run_id INTEGER NOT NULL,
path TEXT NOT NULL,
size INTEGER NOT NULL,
mtime_ns INTEGER NOT NULL,
extension TEXT NOT NULL,
discovery_order INTEGER NOT NULL,
stage1_status TEXT NOT NULL,
stage2_status TEXT,
final_status TEXT,
resumed_from_run_id INTEGER,
reason_code TEXT,
UNIQUE (run_id, path),
FOREIGN KEY (run_id) REFERENCES scan_runs(id)
);
CREATE TABLE IF NOT EXISTS stage1_results (
file_id INTEGER PRIMARY KEY,
exit_code INTEGER,
stdout_text TEXT,
stderr_text TEXT,
elapsed_ms INTEGER,
FOREIGN KEY (file_id) REFERENCES files(id)
);
CREATE TABLE IF NOT EXISTS stage2_results (
file_id INTEGER PRIMARY KEY,
mode TEXT,
sample_points_json TEXT,
exit_code INTEGER,
stderr_text TEXT,
reason_code TEXT,
elapsed_ms INTEGER,
FOREIGN KEY (file_id) REFERENCES files(id)
);
CREATE INDEX IF NOT EXISTS idx_files_identity
ON files(path, size, mtime_ns);
CREATE INDEX IF NOT EXISTS idx_files_run_order
ON files(run_id, discovery_order);
"""
)
self._ensure_column(connection, "files", "reason_code", "TEXT")
def create_run(self, *, profile: str, root_paths: tuple[Path, ...]) -> int:
root_paths_json = orjson.dumps([str(path) for path in root_paths]).decode("utf-8")
with self._connect() as connection:
cursor = connection.execute(
"""
INSERT INTO scan_runs(profile, root_paths_json, status)
VALUES (?, ?, 'running')
""",
(profile, root_paths_json),
)
return int(cursor.lastrowid)
def upsert_files(
self,
run_id: int,
candidates: list[FileCandidate],
*,
reuse_previous_results: bool = True,
) -> list[StoredFile]:
with self._connect() as connection:
for candidate in candidates:
previous = (
self._find_previous_match(connection, run_id, candidate)
if reuse_previous_results
else None
)
stage1_status = previous["stage1_status"] if previous else "pending_stage1"
stage2_status = previous["stage2_status"] if previous else None
final_status = previous["final_status"] if previous else None
resumed_from_run_id = previous["run_id"] if previous else None
reason_code = previous["reason_code"] if previous else None
connection.execute(
"""
INSERT INTO files(
run_id,
path,
size,
mtime_ns,
extension,
discovery_order,
stage1_status,
stage2_status,
final_status,
resumed_from_run_id,
reason_code
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(run_id, path) DO UPDATE SET
size = excluded.size,
mtime_ns = excluded.mtime_ns,
extension = excluded.extension,
discovery_order = excluded.discovery_order,
stage1_status = excluded.stage1_status,
stage2_status = excluded.stage2_status,
final_status = excluded.final_status,
resumed_from_run_id = excluded.resumed_from_run_id,
reason_code = excluded.reason_code
""",
(
run_id,
str(candidate.path),
candidate.size,
candidate.mtime_ns,
candidate.extension,
candidate.discovery_order,
stage1_status,
stage2_status,
final_status,
resumed_from_run_id,
reason_code,
),
)
current_file_id = connection.execute(
"""
SELECT id
FROM files
WHERE run_id = ? AND path = ?
""",
(run_id, str(candidate.path)),
).fetchone()["id"]
if previous is not None:
self._copy_previous_results(
connection,
previous_file_id=previous["file_id"],
current_file_id=current_file_id,
)
rows = connection.execute(
"""
SELECT
id,
run_id,
path,
size,
mtime_ns,
extension,
discovery_order,
stage1_status,
stage2_status,
final_status,
resumed_from_run_id,
reason_code
FROM files
WHERE run_id = ?
ORDER BY discovery_order ASC
""",
(run_id,),
).fetchall()
return self._rows_to_stored_files(rows)
def record_stage1_status(
self,
file_id: int,
*,
stage1_status: str,
final_status: str | None,
reason_code: str | None = None,
) -> None:
with self._connect() as connection:
connection.execute(
"""
UPDATE files
SET stage1_status = ?, final_status = ?, reason_code = ?
WHERE id = ?
""",
(stage1_status, final_status, reason_code, file_id),
)
def record_stage1_result(
self,
file_id: int,
*,
exit_code: int | None,
stdout_text: str,
stderr_text: str,
elapsed_ms: int,
) -> None:
with self._connect() as connection:
connection.execute(
"""
INSERT INTO stage1_results(file_id, exit_code, stdout_text, stderr_text, elapsed_ms)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(file_id) DO UPDATE SET
exit_code = excluded.exit_code,
stdout_text = excluded.stdout_text,
stderr_text = excluded.stderr_text,
elapsed_ms = excluded.elapsed_ms
""",
(file_id, exit_code, stdout_text, stderr_text, elapsed_ms),
)
def record_stage2_status(
self,
file_id: int,
*,
stage2_status: str,
final_status: str | None,
reason_code: str | None = None,
) -> None:
with self._connect() as connection:
connection.execute(
"""
UPDATE files
SET stage2_status = ?, final_status = ?, reason_code = ?
WHERE id = ?
""",
(stage2_status, final_status, reason_code, file_id),
)
def record_stage2_result(
self,
file_id: int,
*,
mode: str,
sample_points_json: str,
exit_code: int | None,
stderr_text: str,
reason_code: str | None,
elapsed_ms: int,
) -> None:
with self._connect() as connection:
connection.execute(
"""
INSERT INTO stage2_results(
file_id,
mode,
sample_points_json,
exit_code,
stderr_text,
reason_code,
elapsed_ms
)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(file_id) DO UPDATE SET
mode = excluded.mode,
sample_points_json = excluded.sample_points_json,
exit_code = excluded.exit_code,
stderr_text = excluded.stderr_text,
reason_code = excluded.reason_code,
elapsed_ms = excluded.elapsed_ms
""",
(
file_id,
mode,
sample_points_json,
exit_code,
stderr_text,
reason_code,
elapsed_ms,
),
)
def update_run_status(self, run_id: int, *, status: str) -> None:
with self._connect() as connection:
connection.execute(
"""
UPDATE scan_runs
SET status = ?
WHERE id = ?
""",
(status, run_id),
)
def get_stage1_stdout_text(self, file_id: int) -> str | None:
with self._connect() as connection:
row = connection.execute(
"""
SELECT stdout_text
FROM stage1_results
WHERE file_id = ?
""",
(file_id,),
).fetchone()
if row is None:
return None
return row["stdout_text"]
def list_files(self, run_id: int) -> list[StoredFile]:
with self._connect() as connection:
rows = connection.execute(
"""
SELECT
id,
run_id,
path,
size,
mtime_ns,
extension,
discovery_order,
stage1_status,
stage2_status,
final_status,
resumed_from_run_id,
reason_code
FROM files
WHERE run_id = ?
ORDER BY discovery_order ASC
""",
(run_id,),
).fetchall()
return self._rows_to_stored_files(rows)
def _rows_to_stored_files(self, rows: list[sqlite3.Row]) -> list[StoredFile]:
return [
StoredFile(
file_id=row["id"],
run_id=row["run_id"],
path=Path(row["path"]),
size=row["size"],
mtime_ns=row["mtime_ns"],
extension=row["extension"],
discovery_order=row["discovery_order"],
stage1_status=row["stage1_status"],
stage2_status=row["stage2_status"],
final_status=row["final_status"],
resumed_from_run_id=row["resumed_from_run_id"],
reason_code=row["reason_code"],
)
for row in rows
]
def _connect(self) -> sqlite3.Connection:
connection = sqlite3.connect(self.db_path)
connection.row_factory = sqlite3.Row
return connection
def _find_previous_match(
self,
connection: sqlite3.Connection,
run_id: int,
candidate: FileCandidate,
) -> sqlite3.Row | None:
return connection.execute(
"""
SELECT id AS file_id, run_id, stage1_status, stage2_status, final_status, reason_code
FROM files
WHERE path = ? AND size = ? AND mtime_ns = ? AND run_id <> ?
ORDER BY run_id DESC
LIMIT 1
""",
(str(candidate.path), candidate.size, candidate.mtime_ns, run_id),
).fetchone()
def _copy_previous_results(
self,
connection: sqlite3.Connection,
*,
previous_file_id: int,
current_file_id: int,
) -> None:
connection.execute(
"""
INSERT INTO stage1_results(file_id, exit_code, stdout_text, stderr_text, elapsed_ms)
SELECT ?, exit_code, stdout_text, stderr_text, elapsed_ms
FROM stage1_results
WHERE file_id = ?
ON CONFLICT(file_id) DO UPDATE SET
exit_code = excluded.exit_code,
stdout_text = excluded.stdout_text,
stderr_text = excluded.stderr_text,
elapsed_ms = excluded.elapsed_ms
""",
(current_file_id, previous_file_id),
)
connection.execute(
"""
INSERT INTO stage2_results(
file_id,
mode,
sample_points_json,
exit_code,
stderr_text,
reason_code,
elapsed_ms
)
SELECT ?, mode, sample_points_json, exit_code, stderr_text, reason_code, elapsed_ms
FROM stage2_results
WHERE file_id = ?
ON CONFLICT(file_id) DO UPDATE SET
mode = excluded.mode,
sample_points_json = excluded.sample_points_json,
exit_code = excluded.exit_code,
stderr_text = excluded.stderr_text,
reason_code = excluded.reason_code,
elapsed_ms = excluded.elapsed_ms
""",
(current_file_id, previous_file_id),
)
def _ensure_column(
self,
connection: sqlite3.Connection,
table_name: str,
column_name: str,
column_definition: str,
) -> None:
columns = {
row["name"]
for row in connection.execute(f"PRAGMA table_info({table_name})").fetchall()
}
if column_name in columns:
return
connection.execute(
f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_definition}"
)
+236
View File
@@ -0,0 +1,236 @@
from typer.testing import CliRunner
from tgvideo_check.cli import app
from tgvideo_check.ffmpeg_runner import FfmpegExecutionResult
from tgvideo_check.ffprobe_runner import ProbeExecutionResult
runner = CliRunner()
def test_help_shows_scan_command() -> None:
result = runner.invoke(app, ["--help"])
assert result.exit_code == 0
assert "scan" in result.stdout
def test_scan_command_processes_stage1_results_and_exports_reports(
tmp_path, monkeypatch
) -> None:
root = tmp_path / "videos"
root.mkdir()
broken = root / "broken.mp4"
passed = root / "passed.avi"
broken.write_bytes(b"broken")
passed.write_bytes(b"passed")
jsonl_path = tmp_path / "results.jsonl"
csv_path = tmp_path / "results.csv"
db_path = tmp_path / "scan_state.db"
def fake_run_ffprobe(path, **kwargs):
if path == broken:
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=1,
stdout_text="",
stderr_text="moov atom not found",
elapsed_ms=1,
)
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"avi","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_run_ffprobe)
result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
"--export-jsonl",
str(jsonl_path),
"--export-csv",
str(csv_path),
],
)
assert result.exit_code == 0
assert "Scanned 2 file(s)" in result.stdout
assert "confirmed_broken=1" in result.stdout
assert "passed_stage1=1" in result.stdout
assert jsonl_path.exists()
assert csv_path.exists()
def test_scan_command_runs_stage2_for_pending_files(tmp_path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
risky = root / "risky.mp4"
risky.write_bytes(b"risky")
def fake_run_ffprobe(path, **kwargs):
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"mov,mp4,m4a,3gp,3g2,mj2","duration":"20.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
def fake_run_ffmpeg(path, **kwargs):
return FfmpegExecutionResult(
status="completed",
command=("ffmpeg", str(path)),
exit_code=0,
stdout_text="frame=10",
stderr_text="",
elapsed_ms=2,
reason_code=None,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_run_ffprobe)
monkeypatch.setattr("tgvideo_check.cli.run_ffmpeg_verification", fake_run_ffmpeg)
result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(tmp_path / "scan_state.db"),
],
)
assert result.exit_code == 0
assert "Scanned 1 file(s)" in result.stdout
assert "passed_stage2=1" in result.stdout
def test_scan_resume_preserves_reason_code_in_exported_report(tmp_path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
broken = root / "broken.mp4"
broken.write_bytes(b"broken")
db_path = tmp_path / "scan_state.db"
first_jsonl = tmp_path / "first.jsonl"
second_jsonl = tmp_path / "second.jsonl"
def fake_run_ffprobe(path, **kwargs):
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=1,
stdout_text="",
stderr_text="moov atom not found",
elapsed_ms=1,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_run_ffprobe)
first_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
"--export-jsonl",
str(first_jsonl),
],
)
assert first_result.exit_code == 0
assert '"reason_code":"ffprobe_exit_nonzero"' in first_jsonl.read_text(encoding="utf-8")
def fail_ffprobe(*args, **kwargs):
raise AssertionError("resume run should not call ffprobe")
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fail_ffprobe)
second_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
"--export-jsonl",
str(second_jsonl),
],
)
assert second_result.exit_code == 0
assert '"reason_code":"ffprobe_exit_nonzero"' in second_jsonl.read_text(encoding="utf-8")
def test_scan_no_resume_forces_probe_execution(tmp_path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
video = root / "video.avi"
video.write_bytes(b"video")
db_path = tmp_path / "scan_state.db"
def first_run_ffprobe(path, **kwargs):
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"avi","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", first_run_ffprobe)
first_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
],
)
assert first_result.exit_code == 0
probe_calls = 0
def second_run_ffprobe(path, **kwargs):
nonlocal probe_calls
probe_calls += 1
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"avi","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", second_run_ffprobe)
second_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
"--no-resume",
],
)
assert second_result.exit_code == 0
assert probe_calls == 1
+40
View File
@@ -0,0 +1,40 @@
from pathlib import Path
from tgvideo_check.discovery import discover_files
def test_discovery_filters_extensions_and_preserves_order(tmp_path: Path) -> None:
root_a = tmp_path / "A"
root_b = tmp_path / "B"
root_a.mkdir()
root_b.mkdir()
keep_1 = root_a / "001.mp4"
skip = root_a / "note.txt"
nested = root_a / "nested"
nested.mkdir()
keep_2 = nested / "002.MKV"
keep_3 = root_b / "003.webm"
for file_path in (keep_1, skip, keep_2, keep_3):
file_path.write_bytes(b"data")
discovered = discover_files((root_a, root_b), extensions=("mp4", "mkv", "webm"))
assert [item.path for item in discovered] == [keep_1, keep_2, keep_3]
assert [item.discovery_order for item in discovered] == [0, 1, 2]
def test_discovery_can_disable_recursion(tmp_path: Path) -> None:
root = tmp_path / "root"
root.mkdir()
direct = root / "001.mp4"
nested_dir = root / "nested"
nested_dir.mkdir()
nested = nested_dir / "002.mp4"
direct.write_bytes(b"a")
nested.write_bytes(b"b")
discovered = discover_files((root,), extensions=("mp4",), recursive=False)
assert [item.path for item in discovered] == [direct]
@@ -0,0 +1,72 @@
import shutil
import subprocess
from pathlib import Path
import pytest
from typer.testing import CliRunner
from tgvideo_check.cli import app
runner = CliRunner()
HAS_FFMPEG = shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None
@pytest.mark.skipif(not HAS_FFMPEG, reason="ffmpeg/ffprobe not installed")
def test_scan_reuses_previous_results_without_rerunning_probes(
tmp_path: Path, monkeypatch
) -> None:
root = tmp_path / "videos"
root.mkdir()
video_path = root / "video.mp4"
_generate_test_video(video_path, duration=2)
db_path = tmp_path / "scan_state.db"
first_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
],
)
assert first_result.exit_code == 0
assert "passed_stage2=1" in first_result.stdout
def fail_ffprobe(*args, **kwargs):
raise AssertionError("run_ffprobe should not be called during resume")
def fail_ffmpeg(*args, **kwargs):
raise AssertionError("run_ffmpeg_verification should not be called during resume")
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fail_ffprobe)
monkeypatch.setattr("tgvideo_check.cli.run_ffmpeg_verification", fail_ffmpeg)
second_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
],
)
assert second_result.exit_code == 0
assert "passed_stage2=1" in second_result.stdout
def _generate_test_video(path: Path, *, duration: int) -> None:
command = [
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
f"testsrc=size=64x64:duration={duration}:rate=25",
"-pix_fmt",
"yuv420p",
str(path),
]
subprocess.run(command, check=True, capture_output=True)
@@ -0,0 +1,68 @@
import shutil
import subprocess
from pathlib import Path
import pytest
from typer.testing import CliRunner
from tgvideo_check.cli import app
from tgvideo_check.store import ScanStore
runner = CliRunner()
HAS_FFMPEG = shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None
@pytest.mark.skipif(not HAS_FFMPEG, reason="ffmpeg/ffprobe not installed")
def test_scan_detects_truncated_mkv_during_stage2(tmp_path: Path) -> None:
root = tmp_path / "videos"
root.mkdir()
valid_path = root / "valid.mkv"
broken_path = root / "broken.mkv"
_generate_test_video(valid_path, duration=8)
_truncate_copy(
valid_path,
broken_path,
trim_bytes=max(1024, valid_path.stat().st_size * 4 // 5),
)
db_path = tmp_path / "scan_state.db"
result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
],
)
assert result.exit_code == 0
store = ScanStore(db_path)
rows = store.list_files(1)
broken_row = next(row for row in rows if row.path.name == "broken.mkv")
assert broken_row.stage1_status == "pending_stage2"
assert broken_row.stage2_status == "confirmed_broken"
assert broken_row.final_status == "confirmed_broken"
def _generate_test_video(path: Path, *, duration: int) -> None:
command = [
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
f"testsrc=size=64x64:duration={duration}:rate=25",
"-pix_fmt",
"yuv420p",
str(path),
]
subprocess.run(command, check=True, capture_output=True)
def _truncate_copy(source: Path, destination: Path, *, trim_bytes: int) -> None:
data = source.read_bytes()
destination.write_bytes(data[: max(1, len(data) - trim_bytes)])
@@ -0,0 +1,67 @@
import shutil
import subprocess
from pathlib import Path
import orjson
import pytest
from tgvideo_check.cli import app
from typer.testing import CliRunner
runner = CliRunner()
HAS_FFMPEG = shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None
@pytest.mark.skipif(not HAS_FFMPEG, reason="ffmpeg/ffprobe not installed")
def test_scan_detects_truncated_mp4_during_stage1(tmp_path: Path) -> None:
root = tmp_path / "videos"
root.mkdir()
valid_path = root / "valid.mp4"
broken_path = root / "broken.mp4"
_generate_test_video(valid_path, container="mp4", duration=2)
_truncate_copy(valid_path, broken_path, trim_bytes=512)
jsonl_path = tmp_path / "results.jsonl"
result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(tmp_path / "scan_state.db"),
"--export-jsonl",
str(jsonl_path),
],
)
assert result.exit_code == 0
rows = _read_jsonl(jsonl_path)
broken_row = next(row for row in rows if row["path"].endswith("broken.mp4"))
assert broken_row["final_status"] == "confirmed_broken"
assert broken_row["reason_code"] in {"ffprobe_exit_nonzero", "hard_error_keyword"}
def _generate_test_video(path: Path, *, container: str, duration: int) -> None:
command = [
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
f"testsrc=size=64x64:duration={duration}:rate=25",
"-pix_fmt",
"yuv420p",
str(path),
]
subprocess.run(command, check=True, capture_output=True)
def _truncate_copy(source: Path, destination: Path, *, trim_bytes: int) -> None:
data = source.read_bytes()
destination.write_bytes(data[: max(1, len(data) - trim_bytes)])
def _read_jsonl(path: Path) -> list[dict[str, object]]:
return [orjson.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
+67
View File
@@ -0,0 +1,67 @@
import os
import shutil
import subprocess
import time
from pathlib import Path
import pytest
from tgvideo_check.cli import app
from typer.testing import CliRunner
runner = CliRunner()
HAS_FFMPEG = shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None
RUN_PERF_SMOKE = os.environ.get("TGVIDEO_RUN_PERF_SMOKE") == "1"
@pytest.mark.skipif(
not HAS_FFMPEG or not RUN_PERF_SMOKE,
reason="set TGVIDEO_RUN_PERF_SMOKE=1 and install ffmpeg/ffprobe to run",
)
def test_perf_smoke_collects_baseline_for_stage1_worker_counts(tmp_path: Path) -> None:
root = tmp_path / "videos"
root.mkdir()
for index in range(6):
_generate_test_video(root / f"video-{index}.mp4", duration=2)
single_worker_seconds = _run_scan(root, tmp_path / "scan-1.db", stage1_workers=1)
multi_worker_seconds = _run_scan(root, tmp_path / "scan-2.db", stage1_workers=2)
assert single_worker_seconds > 0
assert multi_worker_seconds > 0
def _run_scan(root: Path, db_path: Path, *, stage1_workers: int) -> float:
start = time.perf_counter()
result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
"--stage1-workers",
str(stage1_workers),
"--stage2-workers",
"1",
],
)
elapsed = time.perf_counter() - start
assert result.exit_code == 0
return elapsed
def _generate_test_video(path: Path, *, duration: int) -> None:
command = [
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
f"testsrc=size=64x64:duration={duration}:rate=25",
"-pix_fmt",
"yuv420p",
str(path),
]
subprocess.run(command, check=True, capture_output=True)
+29
View File
@@ -0,0 +1,29 @@
from tgvideo_check.config import ScanConfig
from tgvideo_check.profiles import get_profile, resolve_config
def test_tg_hdd_profile_uses_conservative_defaults() -> None:
profile = get_profile("tg-hdd")
assert profile.name == "tg-hdd"
assert profile.stage1_workers == 1
assert profile.stage2_workers == 1
assert profile.sample_duration == 3
assert profile.high_risk_extensions == ("mp4", "mov", "m4v", "mkv", "webm")
def test_resolve_config_applies_profile_overrides() -> None:
config = resolve_config(
ScanConfig(
roots=(),
profile="tg-hdd",
stage1_workers=3,
stage2_workers=None,
extensions=("MP4", ".MKV"),
)
)
assert config.stage1_workers == 3
assert config.stage2_workers == 1
assert config.extensions == ("mp4", "mkv")
+63
View File
@@ -0,0 +1,63 @@
from pathlib import Path
from tgvideo_check.reporting import (
ReportRow,
collect_redownload_candidates,
export_csv_summary,
export_jsonl,
summarize_reason_counts,
)
def sample_rows() -> list[ReportRow]:
return [
ReportRow(
path=Path("a.mp4"),
final_status="confirmed_broken",
reason_code="ffprobe_exit_nonzero",
size=100,
),
ReportRow(
path=Path("b.mkv"),
final_status="suspicious",
reason_code="timeout",
size=200,
),
ReportRow(
path=Path("c.avi"),
final_status="passed_stage1",
reason_code="passed_stage1",
size=300,
),
]
def test_summarize_reason_counts_groups_rows_by_reason() -> None:
summary = summarize_reason_counts(sample_rows())
assert summary == {
"ffprobe_exit_nonzero": 1,
"timeout": 1,
"passed_stage1": 1,
}
def test_collect_redownload_candidates_returns_broken_and_suspicious() -> None:
candidates = collect_redownload_candidates(sample_rows())
assert candidates == [Path("a.mp4"), Path("b.mkv")]
def test_export_jsonl_and_csv_write_expected_content(tmp_path: Path) -> None:
jsonl_path = tmp_path / "results.jsonl"
csv_path = tmp_path / "results.csv"
export_jsonl(sample_rows(), jsonl_path)
export_csv_summary(sample_rows(), csv_path)
jsonl_text = jsonl_path.read_text(encoding="utf-8")
csv_text = csv_path.read_text(encoding="utf-8")
assert '"path":"a.mp4"' in jsonl_text
assert "final_status,reason_code,path,size" in csv_text
assert "confirmed_broken,ffprobe_exit_nonzero,a.mp4,100" in csv_text
+100
View File
@@ -0,0 +1,100 @@
import sqlite3
from pathlib import Path
from tgvideo_check.models import FileCandidate
from tgvideo_check.store import ScanStore
def make_candidate(path: Path, *, order: int, size: int = 10, mtime_ns: int = 100) -> FileCandidate:
path.write_bytes(b"x" * size)
return FileCandidate(
path=path,
size=size,
mtime_ns=mtime_ns,
extension=path.suffix.lower().lstrip("."),
discovery_order=order,
)
def test_initialize_creates_expected_tables(tmp_path: Path) -> None:
store = ScanStore(tmp_path / "scan_state.db")
store.initialize()
with sqlite3.connect(store.db_path) as connection:
table_names = {
row[0]
for row in connection.execute(
"SELECT name FROM sqlite_master WHERE type = 'table'"
)
}
assert {"scan_runs", "files", "stage1_results", "stage2_results"} <= table_names
def test_upsert_files_persists_candidates_in_discovery_order(tmp_path: Path) -> None:
store = ScanStore(tmp_path / "scan_state.db")
store.initialize()
run_id = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
first = make_candidate(tmp_path / "001.mp4", order=0)
second = make_candidate(tmp_path / "002.mkv", order=1)
store.upsert_files(run_id, [first, second])
rows = store.list_files(run_id)
assert [row.path for row in rows] == [first.path, second.path]
assert [row.discovery_order for row in rows] == [0, 1]
assert all(row.stage1_status == "pending_stage1" for row in rows)
def test_upsert_files_reuses_previous_status_when_identity_matches(tmp_path: Path) -> None:
store = ScanStore(tmp_path / "scan_state.db")
store.initialize()
first_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
file_path = tmp_path / "clip.mp4"
candidate = make_candidate(file_path, order=0, size=12, mtime_ns=999)
[stored] = store.upsert_files(first_run, [candidate])
store.record_stage1_status(
stored.file_id,
stage1_status="passed_stage1",
final_status="passed_stage1",
)
second_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
[resumed] = store.upsert_files(second_run, [candidate])
assert resumed.stage1_status == "passed_stage1"
assert resumed.final_status == "passed_stage1"
assert resumed.resumed_from_run_id == first_run
def test_upsert_files_does_not_reuse_changed_file(tmp_path: Path) -> None:
store = ScanStore(tmp_path / "scan_state.db")
store.initialize()
first_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
file_path = tmp_path / "clip.mp4"
original = make_candidate(file_path, order=0, size=12, mtime_ns=999)
[stored] = store.upsert_files(first_run, [original])
store.record_stage1_status(
stored.file_id,
stage1_status="confirmed_broken",
final_status="confirmed_broken",
)
second_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
changed = FileCandidate(
path=file_path,
size=13,
mtime_ns=1000,
extension="mp4",
discovery_order=0,
)
[fresh] = store.upsert_files(second_run, [changed])
assert fresh.stage1_status == "pending_stage1"
assert fresh.final_status is None
assert fresh.resumed_from_run_id is None
@@ -0,0 +1,146 @@
from pathlib import Path
import threading
import time
from tgvideo_check.models import StoredFile
from tgvideo_check.scheduler import (
build_stage_plan,
execute_stage_batch,
select_pending_stage1_files,
select_pending_stage2_files,
)
def make_stored_file(
*,
file_id: int,
discovery_order: int,
stage1_status: str,
stage2_status: str | None = None,
) -> StoredFile:
return StoredFile(
file_id=file_id,
run_id=1,
path=Path(f"video-{file_id}.mp4"),
size=100,
mtime_ns=1000,
extension="mp4",
discovery_order=discovery_order,
stage1_status=stage1_status,
stage2_status=stage2_status,
final_status=None,
resumed_from_run_id=None,
)
def test_select_pending_stage1_files_only_keeps_stage1_work() -> None:
files = [
make_stored_file(file_id=1, discovery_order=2, stage1_status="pending_stage1"),
make_stored_file(file_id=2, discovery_order=1, stage1_status="passed_stage1"),
make_stored_file(file_id=3, discovery_order=0, stage1_status="pending_stage1"),
]
selected = select_pending_stage1_files(files)
assert [item.file_id for item in selected] == [3, 1]
def test_select_pending_stage2_files_skips_completed_or_irrelevant_items() -> None:
files = [
make_stored_file(
file_id=1,
discovery_order=2,
stage1_status="pending_stage2",
stage2_status=None,
),
make_stored_file(
file_id=2,
discovery_order=1,
stage1_status="pending_stage2",
stage2_status="passed_stage2",
),
make_stored_file(
file_id=3,
discovery_order=0,
stage1_status="passed_stage1",
stage2_status=None,
),
]
selected = select_pending_stage2_files(files)
assert [item.file_id for item in selected] == [1]
def test_build_stage_plan_keeps_stage1_and_stage2_separate() -> None:
stage1_files = [
make_stored_file(file_id=1, discovery_order=0, stage1_status="pending_stage1"),
make_stored_file(file_id=2, discovery_order=1, stage1_status="pending_stage1"),
make_stored_file(file_id=3, discovery_order=2, stage1_status="pending_stage1"),
]
stage2_files = [
make_stored_file(file_id=4, discovery_order=3, stage1_status="pending_stage2"),
make_stored_file(file_id=5, discovery_order=4, stage1_status="pending_stage2"),
]
plan = build_stage_plan(
stage1_files=stage1_files,
stage2_files=stage2_files,
stage1_workers=1,
stage2_workers=1,
queue_limit=2,
)
assert [batch.stage_name for batch in plan] == [
"stage1",
"stage1",
"stage2",
]
assert [len(batch.files) for batch in plan] == [2, 1, 2]
assert [[item.file_id for item in batch.files] for batch in plan] == [
[1, 2],
[3],
[4, 5],
]
def test_execute_stage_batch_preserves_input_order() -> None:
batch = build_stage_plan(
stage1_files=[
make_stored_file(file_id=1, discovery_order=0, stage1_status="pending_stage1"),
make_stored_file(file_id=2, discovery_order=1, stage1_status="pending_stage1"),
],
stage2_files=[],
stage1_workers=2,
stage2_workers=1,
queue_limit=64,
)[0]
results = execute_stage_batch(batch, lambda item: item.file_id * 10)
assert results == (10, 20)
def test_execute_stage_batch_uses_multiple_threads_when_workers_allow() -> None:
batch = build_stage_plan(
stage1_files=[
make_stored_file(file_id=1, discovery_order=0, stage1_status="pending_stage1"),
make_stored_file(file_id=2, discovery_order=1, stage1_status="pending_stage1"),
make_stored_file(file_id=3, discovery_order=2, stage1_status="pending_stage1"),
],
stage2_files=[],
stage1_workers=3,
stage2_workers=1,
queue_limit=64,
)[0]
thread_ids: set[int] = set()
def worker(item: StoredFile) -> int:
thread_ids.add(threading.get_ident())
time.sleep(0.02)
return item.file_id
execute_stage_batch(batch, worker)
assert len(thread_ids) >= 2
@@ -0,0 +1,78 @@
from tgvideo_check.classifiers import classify_stage1_result
def test_stage1_marks_nonzero_exit_as_confirmed_broken() -> None:
result = classify_stage1_result(
exit_code=1,
stderr_text="moov atom not found",
probe_payload=None,
extension="mp4",
high_risk_extensions=("mp4", "mkv"),
)
assert result.status == "confirmed_broken"
assert result.reason_code == "ffprobe_exit_nonzero"
def test_stage1_marks_missing_video_stream_as_confirmed_broken() -> None:
result = classify_stage1_result(
exit_code=0,
stderr_text="",
probe_payload={
"format": {"format_name": "mov,mp4,m4a,3gp,3g2,mj2", "duration": "12.3"},
"streams": [{"codec_type": "audio"}],
},
extension="mp4",
high_risk_extensions=(),
)
assert result.status == "confirmed_broken"
assert result.reason_code == "missing_video_stream"
def test_stage1_marks_invalid_duration_as_pending_stage2() -> None:
result = classify_stage1_result(
exit_code=0,
stderr_text="",
probe_payload={
"format": {"format_name": "matroska,webm", "duration": "N/A"},
"streams": [{"codec_type": "video"}],
},
extension="webm",
high_risk_extensions=(),
)
assert result.status == "pending_stage2"
assert result.reason_code == "invalid_duration"
def test_stage1_marks_profile_high_risk_extension_as_pending_stage2() -> None:
result = classify_stage1_result(
exit_code=0,
stderr_text="",
probe_payload={
"format": {"format_name": "mov,mp4,m4a,3gp,3g2,mj2", "duration": "21.0"},
"streams": [{"codec_type": "video"}],
},
extension="mp4",
high_risk_extensions=("mp4", "mkv"),
)
assert result.status == "pending_stage2"
assert result.reason_code == "high_risk_extension"
def test_stage1_marks_clean_non_risky_file_as_passed() -> None:
result = classify_stage1_result(
exit_code=0,
stderr_text="",
probe_payload={
"format": {"format_name": "avi", "duration": "21.0"},
"streams": [{"codec_type": "video"}],
},
extension="avi",
high_risk_extensions=("mp4", "mkv"),
)
assert result.status == "passed_stage1"
assert result.reason_code == "passed_stage1"
@@ -0,0 +1,65 @@
import sys
from pathlib import Path
from tgvideo_check.ffprobe_runner import build_ffprobe_command, run_ffprobe
def test_build_ffprobe_command_requests_minimal_fields(tmp_path: Path) -> None:
video_path = tmp_path / "clip.mp4"
command = build_ffprobe_command(video_path)
assert command[:3] == ["ffprobe", "-v", "error"]
assert "format=format_name,duration,size:stream=index,codec_type,codec_name,width,height" in command
assert command[-1] == str(video_path)
def test_run_ffprobe_captures_stdout_stderr_and_exit_code(tmp_path: Path) -> None:
script = tmp_path / "fake_ffprobe.py"
script.write_text(
"import sys\n"
"print('{\"format\": {\"duration\": \"1.0\"}}')\n"
"print('soft warning', file=sys.stderr)\n",
encoding="utf-8",
)
result = run_ffprobe(
tmp_path / "clip.mp4",
executable=sys.executable,
launcher_args=(str(script),),
)
assert result.status == "completed"
assert result.exit_code == 0
assert "\"duration\": \"1.0\"" in result.stdout_text
assert "soft warning" in result.stderr_text
assert result.elapsed_ms >= 0
def test_run_ffprobe_reports_missing_command(tmp_path: Path) -> None:
result = run_ffprobe(
tmp_path / "clip.mp4",
executable="ffprobe-command-that-does-not-exist",
)
assert result.status == "missing_command"
assert result.exit_code is None
def test_run_ffprobe_reports_timeout(tmp_path: Path) -> None:
script = tmp_path / "slow_ffprobe.py"
script.write_text(
"import time\n"
"time.sleep(0.3)\n",
encoding="utf-8",
)
result = run_ffprobe(
tmp_path / "clip.mp4",
executable=sys.executable,
launcher_args=(str(script),),
timeout_seconds=0.05,
)
assert result.status == "timeout"
assert result.exit_code is None
@@ -0,0 +1,95 @@
import sys
from pathlib import Path
from tgvideo_check.ffmpeg_runner import (
build_ffmpeg_command,
classify_ffmpeg_error,
generate_sample_offsets,
run_ffmpeg_verification,
)
def test_generate_sample_offsets_for_short_clip() -> None:
assert generate_sample_offsets(30.0) == (3.0, 25.5)
def test_generate_sample_offsets_for_medium_clip() -> None:
assert generate_sample_offsets(300.0) == (30.0, 150.0, 270.0)
def test_build_ffmpeg_command_for_sample_window(tmp_path: Path) -> None:
video_path = tmp_path / "clip.mp4"
command = build_ffmpeg_command(
video_path,
offset_seconds=30.0,
sample_duration=3,
)
assert command[:4] == ["ffmpeg", "-v", "error", "-xerror"]
assert "-ss" in command
assert "-t" in command
assert "-map" in command
assert command[-1] == "-"
def test_build_ffmpeg_command_for_full_decode_mode(tmp_path: Path) -> None:
video_path = tmp_path / "clip.mp4"
command = build_ffmpeg_command(
video_path,
full_decode=True,
)
assert "-ss" not in command
assert "-t" not in command
assert command[-1] == "-"
def test_classify_ffmpeg_error_detects_truncation() -> None:
assert (
classify_ffmpeg_error("Truncating packet of size 4096 to 1024")
== "truncating_packet"
)
def test_run_ffmpeg_verification_captures_process_result(tmp_path: Path) -> None:
script = tmp_path / "fake_ffmpeg.py"
script.write_text(
"import sys\n"
"print('frame=12 fps=0.0')\n"
"print('decode warning', file=sys.stderr)\n",
encoding="utf-8",
)
result = run_ffmpeg_verification(
tmp_path / "clip.mp4",
offset_seconds=3.0,
executable=sys.executable,
launcher_args=(str(script),),
)
assert result.status == "completed"
assert result.exit_code == 0
assert "frame=12" in result.stdout_text
assert "decode warning" in result.stderr_text
def test_run_ffmpeg_verification_reports_timeout(tmp_path: Path) -> None:
script = tmp_path / "slow_ffmpeg.py"
script.write_text(
"import time\n"
"time.sleep(0.3)\n",
encoding="utf-8",
)
result = run_ffmpeg_verification(
tmp_path / "clip.mp4",
offset_seconds=3.0,
executable=sys.executable,
launcher_args=(str(script),),
timeout_seconds=0.05,
)
assert result.status == "timeout"
assert result.exit_code is None
+219
View File
@@ -0,0 +1,219 @@
version = 1
revision = 3
requires-python = ">=3.12"
[[package]]
name = "annotated-doc"
version = "0.0.4"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/57/ba/046ceea27344560984e26a590f90bc7f4a75b06701f653222458922b558c/annotated_doc-0.0.4.tar.gz", hash = "sha256:fbcda96e87e9c92ad167c2e53839e57503ecfda18804ea28102353485033faa4", size = 7288, upload-time = "2025-11-10T22:07:42.062Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/1e/d3/26bf1008eb3d2daa8ef4cacc7f3bfdc11818d111f7e2d0201bc6e3b49d45/annotated_doc-0.0.4-py3-none-any.whl", hash = "sha256:571ac1dc6991c450b25a9c2d84a3705e2ae7a53467b5d111c24fa8baabbed320", size = 5303, upload-time = "2025-11-10T22:07:40.673Z" },
]
[[package]]
name = "click"
version = "8.3.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/3d/fa/656b739db8587d7b5dfa22e22ed02566950fbfbcdc20311993483657a5c0/click-8.3.1.tar.gz", hash = "sha256:12ff4785d337a1bb490bb7e9c2b1ee5da3112e94a8622f26a6c77f5d2fc6842a", size = 295065, upload-time = "2025-11-15T20:45:42.706Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/98/78/01c019cdb5d6498122777c1a43056ebb3ebfeef2076d9d026bfe15583b2b/click-8.3.1-py3-none-any.whl", hash = "sha256:981153a64e25f12d547d3426c367a4857371575ee7ad18df2a6183ab0545b2a6", size = 108274, upload-time = "2025-11-15T20:45:41.139Z" },
]
[[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 = "markdown-it-py"
version = "4.0.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "mdurl" },
]
sdist = { url = "https://files.pythonhosted.org/packages/5b/f5/4ec618ed16cc4f8fb3b701563655a69816155e79e24a17b651541804721d/markdown_it_py-4.0.0.tar.gz", hash = "sha256:cb0a2b4aa34f932c007117b194e945bd74e0ec24133ceb5bac59009cda1cb9f3", size = 73070, upload-time = "2025-08-11T12:57:52.854Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/94/54/e7d793b573f298e1c9013b8c4dade17d481164aa517d1d7148619c2cedbf/markdown_it_py-4.0.0-py3-none-any.whl", hash = "sha256:87327c59b172c5011896038353a81343b6754500a08cd7a4973bb48c6d578147", size = 87321, upload-time = "2025-08-11T12:57:51.923Z" },
]
[[package]]
name = "mdurl"
version = "0.1.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" },
]
[[package]]
name = "orjson"
version = "3.11.7"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/53/45/b268004f745ede84e5798b48ee12b05129d19235d0e15267aa57dcdb400b/orjson-3.11.7.tar.gz", hash = "sha256:9b1a67243945819ce55d24a30b59d6a168e86220452d2c96f4d1f093e71c0c49", size = 6144992, upload-time = "2026-02-02T15:38:49.29Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/80/bf/76f4f1665f6983385938f0e2a5d7efa12a58171b8456c252f3bae8a4cf75/orjson-3.11.7-cp312-cp312-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:bd03ea7606833655048dab1a00734a2875e3e86c276e1d772b2a02556f0d895f", size = 228545, upload-time = "2026-02-02T15:37:46.376Z" },
{ url = "https://files.pythonhosted.org/packages/79/53/6c72c002cb13b5a978a068add59b25a8bdf2800ac1c9c8ecdb26d6d97064/orjson-3.11.7-cp312-cp312-macosx_15_0_arm64.whl", hash = "sha256:89e440ebc74ce8ab5c7bc4ce6757b4a6b1041becb127df818f6997b5c71aa60b", size = 125224, upload-time = "2026-02-02T15:37:47.697Z" },
{ url = "https://files.pythonhosted.org/packages/2c/83/10e48852865e5dd151bdfe652c06f7da484578ed02c5fca938e3632cb0b8/orjson-3.11.7-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:5ede977b5fe5ac91b1dffc0a517ca4542d2ec8a6a4ff7b2652d94f640796342a", size = 128154, upload-time = "2026-02-02T15:37:48.954Z" },
{ url = "https://files.pythonhosted.org/packages/6e/52/a66e22a2b9abaa374b4a081d410edab6d1e30024707b87eab7c734afe28d/orjson-3.11.7-cp312-cp312-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:b7b1dae39230a393df353827c855a5f176271c23434cfd2db74e0e424e693e10", size = 123548, upload-time = "2026-02-02T15:37:50.187Z" },
{ url = "https://files.pythonhosted.org/packages/de/38/605d371417021359f4910c496f764c48ceb8997605f8c25bf1dfe58c0ebe/orjson-3.11.7-cp312-cp312-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:ed46f17096e28fb28d2975834836a639af7278aa87c84f68ab08fbe5b8bd75fa", size = 129000, upload-time = "2026-02-02T15:37:51.426Z" },
{ url = "https://files.pythonhosted.org/packages/44/98/af32e842b0ffd2335c89714d48ca4e3917b42f5d6ee5537832e069a4b3ac/orjson-3.11.7-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:3726be79e36e526e3d9c1aceaadbfb4a04ee80a72ab47b3f3c17fefb9812e7b8", size = 141686, upload-time = "2026-02-02T15:37:52.607Z" },
{ url = "https://files.pythonhosted.org/packages/96/0b/fc793858dfa54be6feee940c1463370ece34b3c39c1ca0aa3845f5ba9892/orjson-3.11.7-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:0724e265bc548af1dedebd9cb3d24b4e1c1e685a343be43e87ba922a5c5fff2f", size = 130812, upload-time = "2026-02-02T15:37:53.944Z" },
{ url = "https://files.pythonhosted.org/packages/dc/91/98a52415059db3f374757d0b7f0f16e3b5cd5976c90d1c2b56acaea039e6/orjson-3.11.7-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e7745312efa9e11c17fbd3cb3097262d079da26930ae9ae7ba28fb738367cbad", size = 133440, upload-time = "2026-02-02T15:37:55.615Z" },
{ url = "https://files.pythonhosted.org/packages/dc/b6/cb540117bda61791f46381f8c26c8f93e802892830a6055748d3bb1925ab/orjson-3.11.7-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:f904c24bdeabd4298f7a977ef14ca2a022ca921ed670b92ecd16ab6f3d01f867", size = 138386, upload-time = "2026-02-02T15:37:56.814Z" },
{ url = "https://files.pythonhosted.org/packages/63/1a/50a3201c334a7f17c231eee5f841342190723794e3b06293f26e7cf87d31/orjson-3.11.7-cp312-cp312-musllinux_1_2_armv7l.whl", hash = "sha256:b9fc4d0f81f394689e0814617aadc4f2ea0e8025f38c226cbf22d3b5ddbf025d", size = 408853, upload-time = "2026-02-02T15:37:58.291Z" },
{ url = "https://files.pythonhosted.org/packages/87/cd/8de1c67d0be44fdc22701e5989c0d015a2adf391498ad42c4dc589cd3013/orjson-3.11.7-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:849e38203e5be40b776ed2718e587faf204d184fc9a008ae441f9442320c0cab", size = 144130, upload-time = "2026-02-02T15:38:00.163Z" },
{ url = "https://files.pythonhosted.org/packages/0f/fe/d605d700c35dd55f51710d159fc54516a280923cd1b7e47508982fbb387d/orjson-3.11.7-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:4682d1db3bcebd2b64757e0ddf9e87ae5f00d29d16c5cdf3a62f561d08cc3dd2", size = 134818, upload-time = "2026-02-02T15:38:01.507Z" },
{ url = "https://files.pythonhosted.org/packages/e4/e4/15ecc67edb3ddb3e2f46ae04475f2d294e8b60c1825fbe28a428b93b3fbd/orjson-3.11.7-cp312-cp312-win32.whl", hash = "sha256:f4f7c956b5215d949a1f65334cf9d7612dde38f20a95f2315deef167def91a6f", size = 127923, upload-time = "2026-02-02T15:38:02.75Z" },
{ url = "https://files.pythonhosted.org/packages/34/70/2e0855361f76198a3965273048c8e50a9695d88cd75811a5b46444895845/orjson-3.11.7-cp312-cp312-win_amd64.whl", hash = "sha256:bf742e149121dc5648ba0a08ea0871e87b660467ef168a3a5e53bc1fbd64bb74", size = 125007, upload-time = "2026-02-02T15:38:04.032Z" },
{ url = "https://files.pythonhosted.org/packages/68/40/c2051bd19fc467610fed469dc29e43ac65891571138f476834ca192bc290/orjson-3.11.7-cp312-cp312-win_arm64.whl", hash = "sha256:26c3b9132f783b7d7903bf1efb095fed8d4a3a85ec0d334ee8beff3d7a4749d5", size = 126089, upload-time = "2026-02-02T15:38:05.297Z" },
{ url = "https://files.pythonhosted.org/packages/89/25/6e0e52cac5aab51d7b6dcd257e855e1dec1c2060f6b28566c509b4665f62/orjson-3.11.7-cp313-cp313-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:1d98b30cc1313d52d4af17d9c3d307b08389752ec5f2e5febdfada70b0f8c733", size = 228390, upload-time = "2026-02-02T15:38:06.8Z" },
{ url = "https://files.pythonhosted.org/packages/a5/29/a77f48d2fc8a05bbc529e5ff481fb43d914f9e383ea2469d4f3d51df3d00/orjson-3.11.7-cp313-cp313-macosx_15_0_arm64.whl", hash = "sha256:d897e81f8d0cbd2abb82226d1860ad2e1ab3ff16d7b08c96ca00df9d45409ef4", size = 125189, upload-time = "2026-02-02T15:38:08.181Z" },
{ url = "https://files.pythonhosted.org/packages/89/25/0a16e0729a0e6a1504f9d1a13cdd365f030068aab64cec6958396b9969d7/orjson-3.11.7-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:814be4b49b228cfc0b3c565acf642dd7d13538f966e3ccde61f4f55be3e20785", size = 128106, upload-time = "2026-02-02T15:38:09.41Z" },
{ url = "https://files.pythonhosted.org/packages/66/da/a2e505469d60666a05ab373f1a6322eb671cb2ba3a0ccfc7d4bc97196787/orjson-3.11.7-cp313-cp313-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:d06e5c5fed5caedd2e540d62e5b1c25e8c82431b9e577c33537e5fa4aa909539", size = 123363, upload-time = "2026-02-02T15:38:10.73Z" },
{ url = "https://files.pythonhosted.org/packages/23/bf/ed73f88396ea35c71b38961734ea4a4746f7ca0768bf28fd551d37e48dd0/orjson-3.11.7-cp313-cp313-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:31c80ce534ac4ea3739c5ee751270646cbc46e45aea7576a38ffec040b4029a1", size = 129007, upload-time = "2026-02-02T15:38:12.138Z" },
{ url = "https://files.pythonhosted.org/packages/73/3c/b05d80716f0225fc9008fbf8ab22841dcc268a626aa550561743714ce3bf/orjson-3.11.7-cp313-cp313-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:f50979824bde13d32b4320eedd513431c921102796d86be3eee0b58e58a3ecd1", size = 141667, upload-time = "2026-02-02T15:38:13.398Z" },
{ url = "https://files.pythonhosted.org/packages/61/e8/0be9b0addd9bf86abfc938e97441dcd0375d494594b1c8ad10fe57479617/orjson-3.11.7-cp313-cp313-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:9e54f3808e2b6b945078c41aa8d9b5834b28c50843846e97807e5adb75fa9705", size = 130832, upload-time = "2026-02-02T15:38:14.698Z" },
{ url = "https://files.pythonhosted.org/packages/c9/ec/c68e3b9021a31d9ec15a94931db1410136af862955854ed5dd7e7e4f5bff/orjson-3.11.7-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a12b80df61aab7b98b490fe9e4879925ba666fccdfcd175252ce4d9035865ace", size = 133373, upload-time = "2026-02-02T15:38:16.109Z" },
{ url = "https://files.pythonhosted.org/packages/d2/45/f3466739aaafa570cc8e77c6dbb853c48bf56e3b43738020e2661e08b0ac/orjson-3.11.7-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:996b65230271f1a97026fd0e6a753f51fbc0c335d2ad0c6201f711b0da32693b", size = 138307, upload-time = "2026-02-02T15:38:17.453Z" },
{ url = "https://files.pythonhosted.org/packages/e1/84/9f7f02288da1ffb31405c1be07657afd1eecbcb4b64ee2817b6fe0f785fa/orjson-3.11.7-cp313-cp313-musllinux_1_2_armv7l.whl", hash = "sha256:ab49d4b2a6a1d415ddb9f37a21e02e0d5dbfe10b7870b21bf779fc21e9156157", size = 408695, upload-time = "2026-02-02T15:38:18.831Z" },
{ url = "https://files.pythonhosted.org/packages/18/07/9dd2f0c0104f1a0295ffbe912bc8d63307a539b900dd9e2c48ef7810d971/orjson-3.11.7-cp313-cp313-musllinux_1_2_i686.whl", hash = "sha256:390a1dce0c055ddf8adb6aa94a73b45a4a7d7177b5c584b8d1c1947f2ba60fb3", size = 144099, upload-time = "2026-02-02T15:38:20.28Z" },
{ url = "https://files.pythonhosted.org/packages/a5/66/857a8e4a3292e1f7b1b202883bcdeb43a91566cf59a93f97c53b44bd6801/orjson-3.11.7-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:1eb80451a9c351a71dfaf5b7ccc13ad065405217726b59fdbeadbcc544f9d223", size = 134806, upload-time = "2026-02-02T15:38:22.186Z" },
{ url = "https://files.pythonhosted.org/packages/0a/5b/6ebcf3defc1aab3a338ca777214966851e92efb1f30dc7fc8285216e6d1b/orjson-3.11.7-cp313-cp313-win32.whl", hash = "sha256:7477aa6a6ec6139c5cb1cc7b214643592169a5494d200397c7fc95d740d5fcf3", size = 127914, upload-time = "2026-02-02T15:38:23.511Z" },
{ url = "https://files.pythonhosted.org/packages/00/04/c6f72daca5092e3117840a1b1e88dfc809cc1470cf0734890d0366b684a1/orjson-3.11.7-cp313-cp313-win_amd64.whl", hash = "sha256:b9f95dcdea9d4f805daa9ddf02617a89e484c6985fa03055459f90e87d7a0757", size = 124986, upload-time = "2026-02-02T15:38:24.836Z" },
{ url = "https://files.pythonhosted.org/packages/03/ba/077a0f6f1085d6b806937246860fafbd5b17f3919c70ee3f3d8d9c713f38/orjson-3.11.7-cp313-cp313-win_arm64.whl", hash = "sha256:800988273a014a0541483dc81021247d7eacb0c845a9d1a34a422bc718f41539", size = 126045, upload-time = "2026-02-02T15:38:26.216Z" },
{ url = "https://files.pythonhosted.org/packages/e9/1e/745565dca749813db9a093c5ebc4bac1a9475c64d54b95654336ac3ed961/orjson-3.11.7-cp314-cp314-macosx_10_15_x86_64.macosx_11_0_arm64.macosx_10_15_universal2.whl", hash = "sha256:de0a37f21d0d364954ad5de1970491d7fbd0fb1ef7417d4d56a36dc01ba0c0a0", size = 228391, upload-time = "2026-02-02T15:38:27.757Z" },
{ url = "https://files.pythonhosted.org/packages/46/19/e40f6225da4d3aa0c8dc6e5219c5e87c2063a560fe0d72a88deb59776794/orjson-3.11.7-cp314-cp314-macosx_15_0_arm64.whl", hash = "sha256:c2428d358d85e8da9d37cba18b8c4047c55222007a84f97156a5b22028dfbfc0", size = 125188, upload-time = "2026-02-02T15:38:29.241Z" },
{ url = "https://files.pythonhosted.org/packages/9d/7e/c4de2babef2c0817fd1f048fd176aa48c37bec8aef53d2fa932983032cce/orjson-3.11.7-cp314-cp314-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:3c4bc6c6ac52cdaa267552544c73e486fecbd710b7ac09bc024d5a78555a22f6", size = 128097, upload-time = "2026-02-02T15:38:30.618Z" },
{ url = "https://files.pythonhosted.org/packages/eb/74/233d360632bafd2197f217eee7fb9c9d0229eac0c18128aee5b35b0014fe/orjson-3.11.7-cp314-cp314-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:bd0d68edd7dfca1b2eca9361a44ac9f24b078de3481003159929a0573f21a6bf", size = 123364, upload-time = "2026-02-02T15:38:32.363Z" },
{ url = "https://files.pythonhosted.org/packages/79/51/af79504981dd31efe20a9e360eb49c15f06df2b40e7f25a0a52d9ae888e8/orjson-3.11.7-cp314-cp314-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:623ad1b9548ef63886319c16fa317848e465a21513b31a6ad7b57443c3e0dcf5", size = 129076, upload-time = "2026-02-02T15:38:33.68Z" },
{ url = "https://files.pythonhosted.org/packages/67/e2/da898eb68b72304f8de05ca6715870d09d603ee98d30a27e8a9629abc64b/orjson-3.11.7-cp314-cp314-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:6e776b998ac37c0396093d10290e60283f59cfe0fc3fccbd0ccc4bd04dd19892", size = 141705, upload-time = "2026-02-02T15:38:34.989Z" },
{ url = "https://files.pythonhosted.org/packages/c5/89/15364d92acb3d903b029e28d834edb8780c2b97404cbf7929aa6b9abdb24/orjson-3.11.7-cp314-cp314-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:652c6c3af76716f4a9c290371ba2e390ede06f6603edb277b481daf37f6f464e", size = 130855, upload-time = "2026-02-02T15:38:36.379Z" },
{ url = "https://files.pythonhosted.org/packages/c2/8b/ecdad52d0b38d4b8f514be603e69ccd5eacf4e7241f972e37e79792212ec/orjson-3.11.7-cp314-cp314-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a56df3239294ea5964adf074c54bcc4f0ccd21636049a2cf3ca9cf03b5d03cf1", size = 133386, upload-time = "2026-02-02T15:38:37.704Z" },
{ url = "https://files.pythonhosted.org/packages/b9/0e/45e1dcf10e17d0924b7c9162f87ec7b4ca79e28a0548acf6a71788d3e108/orjson-3.11.7-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:bda117c4148e81f746655d5a3239ae9bd00cb7bc3ca178b5fc5a5997e9744183", size = 138295, upload-time = "2026-02-02T15:38:39.096Z" },
{ url = "https://files.pythonhosted.org/packages/63/d7/4d2e8b03561257af0450f2845b91fbd111d7e526ccdf737267108075e0ba/orjson-3.11.7-cp314-cp314-musllinux_1_2_armv7l.whl", hash = "sha256:23d6c20517a97a9daf1d48b580fcdc6f0516c6f4b5038823426033690b4d2650", size = 408720, upload-time = "2026-02-02T15:38:40.634Z" },
{ url = "https://files.pythonhosted.org/packages/78/cf/d45343518282108b29c12a65892445fc51f9319dc3c552ceb51bb5905ed2/orjson-3.11.7-cp314-cp314-musllinux_1_2_i686.whl", hash = "sha256:8ff206156006da5b847c9304b6308a01e8cdbc8cce824e2779a5ba71c3def141", size = 144152, upload-time = "2026-02-02T15:38:42.262Z" },
{ url = "https://files.pythonhosted.org/packages/a9/3a/d6001f51a7275aacd342e77b735c71fa04125a3f93c36fee4526bc8c654e/orjson-3.11.7-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:962d046ee1765f74a1da723f4b33e3b228fe3a48bd307acce5021dfefe0e29b2", size = 134814, upload-time = "2026-02-02T15:38:43.627Z" },
{ url = "https://files.pythonhosted.org/packages/1d/d3/f19b47ce16820cc2c480f7f1723e17f6d411b3a295c60c8ad3aa9ff1c96a/orjson-3.11.7-cp314-cp314-win32.whl", hash = "sha256:89e13dd3f89f1c38a9c9eba5fbf7cdc2d1feca82f5f290864b4b7a6aac704576", size = 127997, upload-time = "2026-02-02T15:38:45.06Z" },
{ url = "https://files.pythonhosted.org/packages/12/df/172771902943af54bf661a8d102bdf2e7f932127968080632bda6054b62c/orjson-3.11.7-cp314-cp314-win_amd64.whl", hash = "sha256:845c3e0d8ded9c9271cd79596b9b552448b885b97110f628fb687aee2eed11c1", size = 124985, upload-time = "2026-02-02T15:38:46.388Z" },
{ url = "https://files.pythonhosted.org/packages/6f/1c/f2a8d8a1b17514660a614ce5f7aac74b934e69f5abc2700cc7ced882a009/orjson-3.11.7-cp314-cp314-win_arm64.whl", hash = "sha256:4a2e9c5be347b937a2e0203866f12bba36082e89b402ddb9e927d5822e43088d", size = 126038, upload-time = "2026-02-02T15:38:47.703Z" },
]
[[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 = "plan-for-tgvideo"
version = "0.1.0"
source = { editable = "." }
dependencies = [
{ name = "orjson" },
{ name = "typer" },
]
[package.dev-dependencies]
dev = [
{ name = "pytest" },
]
[package.metadata]
requires-dist = [
{ name = "orjson", specifier = ">=3.11.3" },
{ name = "typer", specifier = ">=0.19.2" },
]
[package.metadata.requires-dev]
dev = [{ name = "pytest", specifier = ">=8.4.2" }]
[[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.19.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/b0/77/a5b8c569bf593b0140bde72ea885a803b82086995367bf2037de0159d924/pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887", size = 4968631, upload-time = "2025-06-21T13:39:12.283Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" },
]
[[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 = "rich"
version = "14.3.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "markdown-it-py" },
{ name = "pygments" },
]
sdist = { url = "https://files.pythonhosted.org/packages/b3/c6/f3b320c27991c46f43ee9d856302c70dc2d0fb2dba4842ff739d5f46b393/rich-14.3.3.tar.gz", hash = "sha256:b8daa0b9e4eef54dd8cf7c86c03713f53241884e814f4e2f5fb342fe520f639b", size = 230582, upload-time = "2026-02-19T17:23:12.474Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/14/25/b208c5683343959b670dc001595f2f3737e051da617f66c31f7c4fa93abc/rich-14.3.3-py3-none-any.whl", hash = "sha256:793431c1f8619afa7d3b52b2cdec859562b950ea0d4b6b505397612db8d5362d", size = 310458, upload-time = "2026-02-19T17:23:13.732Z" },
]
[[package]]
name = "shellingham"
version = "1.5.4"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/58/15/8b3609fd3830ef7b27b655beb4b4e9c62313a4e8da8c676e142cc210d58e/shellingham-1.5.4.tar.gz", hash = "sha256:8dbca0739d487e5bd35ab3ca4b36e11c4078f3a234bfce294b0a0291363404de", size = 10310, upload-time = "2023-10-24T04:13:40.426Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e0/f9/0595336914c5619e5f28a1fb793285925a8cd4b432c9da0a987836c7f822/shellingham-1.5.4-py2.py3-none-any.whl", hash = "sha256:7ecfff8f2fd72616f7481040475a65b2bf8af90a56c89140852d1120324e8686", size = 9755, upload-time = "2023-10-24T04:13:38.866Z" },
]
[[package]]
name = "typer"
version = "0.24.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "annotated-doc" },
{ name = "click" },
{ name = "rich" },
{ name = "shellingham" },
]
sdist = { url = "https://files.pythonhosted.org/packages/f5/24/cb09efec5cc954f7f9b930bf8279447d24618bb6758d4f6adf2574c41780/typer-0.24.1.tar.gz", hash = "sha256:e39b4732d65fbdcde189ae76cf7cd48aeae72919dea1fdfc16593be016256b45", size = 118613, upload-time = "2026-02-21T16:54:40.609Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/4a/91/48db081e7a63bb37284f9fbcefda7c44c277b18b0e13fbc36ea2335b71e6/typer-0.24.1-py3-none-any.whl", hash = "sha256:112c1f0ce578bfb4cab9ffdabc68f031416ebcc216536611ba21f04e9aa84c9e", size = 56085, upload-time = "2026-02-21T16:54:41.616Z" },
]