yuming
b3608f534e
- 新增 analysis/ 模块：MOG2活动检测、YOLO人物跟踪、MediaPipe摔倒检测、Ollama活动打标、缩略图抽取 - 新增 webapp/：Flask事件浏览界面，按需HEVC→H264转码缓存 - 新增 analyze.py：7阶段状态机流水线（scan→backfill→motion→person→fall→activity→thumbnail） - 新增 cleanup_orphans.py：清理文件已删除的数据库孤儿记录 - 修复：motion/person/activity三阶段加单视频异常兜底，避免批次中断 - 修复：ollama_client加健康检查和3次重试，num_ctx从8192降到2048 - 修复：transcode_cache加fcntl文件锁，防止并发转码同一视频 - 修复：SQLite开WAL模式+busy_timeout，支持读写并发 - 修复：sync.py ffmpeg路径改为shutil.which查找，删除automate.py死代码PYTHON - 修复：thumbnail失败加3次重试上限，避免无限重试损坏文件 - 新增：requirements.lock锁定版本、README文档 - 更新：.gitignore排除.playwright-mcp、截图、备份文件
68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
"""检测 recordings/ 目录下内容完全相同（字节级重复）的录像文件。
|
|
|
|
根因：automate.py 的 ADB 自动化下载在滚动/点击时序上有已知问题，会把同一张
|
|
缩略图卡片的下载重复触发，导致同一段录像被存成多份文件名不同的拷贝。
|
|
|
|
先按文件大小分组（成本低，能筛掉绝大多数内容不同的文件），只对同大小的文件
|
|
再计算 MD5 确认是否真的内容相同，避免对全部文件逐个算 MD5 的开销。
|
|
|
|
只生成报告，不删除任何文件、不改动数据库。
|
|
"""
|
|
from collections import defaultdict
|
|
|
|
from analysis.config import PROJECT_ROOT, RECORDINGS_DIR
|
|
from analysis.hashing import compute_md5
|
|
|
|
REPORT_PATH = PROJECT_ROOT / "dedup_report.txt"
|
|
|
|
|
|
def find_duplicates():
|
|
"""返回 {md5: [文件路径, ...]}，只包含重复数≥2的分组。"""
|
|
by_size = defaultdict(list)
|
|
for path in RECORDINGS_DIR.rglob("*.mp4"):
|
|
by_size[path.stat().st_size].append(path)
|
|
|
|
by_md5 = defaultdict(list)
|
|
for size, paths in by_size.items():
|
|
if len(paths) < 2:
|
|
continue
|
|
for path in paths:
|
|
by_md5[compute_md5(path)].append(path)
|
|
|
|
return {md5: paths for md5, paths in by_md5.items() if len(paths) > 1}
|
|
|
|
|
|
def build_report():
|
|
dup_groups = find_duplicates()
|
|
|
|
lines = []
|
|
total_extra_files = 0
|
|
total_wasted_bytes = 0
|
|
for md5, paths in sorted(dup_groups.items(), key=lambda kv: -len(kv[1])):
|
|
paths = sorted(paths, key=lambda p: p.name)
|
|
size = paths[0].stat().st_size
|
|
extra = len(paths) - 1
|
|
total_extra_files += extra
|
|
total_wasted_bytes += extra * size
|
|
|
|
lines.append(f"[{md5}] {len(paths)} 份重复，单份 {size / 1024:.1f} KB")
|
|
for p in paths:
|
|
lines.append(f" {p.relative_to(PROJECT_ROOT)}")
|
|
|
|
summary = (
|
|
f"重复分组数：{len(dup_groups)}\n"
|
|
f"多余拷贝数（可释放的文件数）：{total_extra_files}\n"
|
|
f"可释放空间：{total_wasted_bytes / 1024 / 1024:.1f} MB\n"
|
|
)
|
|
|
|
return summary + "\n" + "\n".join(lines)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("===== 开始检测重复视频 =====")
|
|
report = build_report()
|
|
REPORT_PATH.write_text(report, encoding="utf-8")
|
|
print(report.split("\n\n")[0])
|
|
print(f"→ 详细清单已写入 {REPORT_PATH.relative_to(PROJECT_ROOT)}")
|
|
print("===== 检测完成 =====")
|