Video Screenshot

# OCR 提供者（local = RapidOCR 离线，目前仅支持 local）
OCR_PROVIDER=local

# RapidOCR 为本地 OCR，无需 API 配置
# 安装: pip install rapidocr-onnxruntime

MIT License

Copyright (c) 2026

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = [
#   "Pillow>=10.0.0",
#   "rapidocr-onnxruntime",
# ]
# ///

"""video-screenshot 视频截图提取工具。

从录屏视频中抽取关键帧、去重并保存为图片文件。
"""

from __future__ import annotations

import argparse
import json
import logging
import re
import shutil
import sys
import tempfile
import time
from datetime import datetime
from hashlib import sha256
from pathlib import Path

# 将 scripts/ 同级目录加入搜索路径以便导入 lib
sys.path.insert(0, str(Path(__file__).resolve().parent))

from lib import (
    DedupState,
    ExtractParams,
    FFProbeInfo,
    calc_blur_score,
    calc_capture_time,
    calc_content_quality,
    calc_dhash_hex,
    calc_scroll_image,
    calc_thumb_bytes,
    check_ocr_similarity,
    collect_frame_files,
    create_ocr_engine,
    crop_for_ocr_bytes_with_range,
    find_tool,
    is_frame_duplicate,
    ocr_extract_text,
    probe_video,
    run_ffmpeg_extract,
    shingles,
)

logger = logging.getLogger("video-screenshot")


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="视频取证关键帧提取工具",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""\
示例:
  # 场景检测 + 图像去重（默认）
  uv run scripts/extract.py -i recording.mp4

  # 固定间隔，每 0.5 秒一帧
  uv run scripts/extract.py -i recording.mp4 -s interval --interval 0.5

  # 场景检测 + OCR 去重（适合聊天录屏）
  uv run scripts/extract.py -i recording.mp4 --ocr-dedup

  # 关键帧提取，不去重
  uv run scripts/extract.py -i recording.mp4 -s keyframe -d 0
""",
    )
    p.add_argument("-i", "--input", required=True, help="输入视频文件路径")
    p.add_argument("-o", "--output", default=None, help="输出目录（默认: <视频名>_frames/）")
    p.add_argument("-s", "--strategy", default="scene",
                   choices=["scene", "keyframe", "interval", "smart"],
                   help="抽帧策略（默认: scene）")
    p.add_argument("--interval", type=float, default=1.0, help="间隔秒数（interval 模式，默认: 1.0）")
    p.add_argument("--scene-threshold", type=float, default=0.10, help="场景变化阈值（scene 模式，默认: 0.10）")
    p.add_argument("--sample-interval", type=float, default=2.0, help="定期采样间隔秒数（scene 模式保底，默认: 2.0，0=禁用）")
    p.add_argument("-d", "--dedup-threshold", type=int, default=4, help="dHash 汉明距离阈值（0=禁用，默认: 4）")
    p.add_argument("--content-crop-top", type=float, default=0.12, help="内容区顶部裁剪比例（默认: 0.12）")
    p.add_argument("--content-crop-bottom", type=float, default=0.12, help="内容区底部裁剪比例（默认: 0.12）")
    p.add_argument("--content-crop-left", type=float, default=0.04, help="内容区左侧裁剪比例（默认: 0.04）")
    p.add_argument("--content-crop-right", type=float, default=0.04, help="内容区右侧裁剪比例（默认: 0.04）")
    p.add_argument("--ssim-threshold", type=float, default=0.93, help="SSIM 结构相似度阈值（0=禁用，默认: 0.93）")
    p.add_argument("--scroll-merge", action="store_true", default=False, help="滚动帧合并（默认关闭）")
    p.add_argument("--no-scroll-merge", action="store_false", dest="scroll_merge", help="禁用滚动帧合并")
    p.add_argument("--scroll-diff-threshold", type=float, default=32.0, help="滚动重叠平均像素差阈值（默认: 32.0）")
    p.add_argument("--ocr-dedup", action="store_true", help="启用 OCR 文本去重")
    p.add_argument("--ocr-threshold", type=float, default=0.92, help="OCR 相似度阈值（默认: 0.92）")
    p.add_argument("--ocr-min-new", type=int, default=8, help="OCR 最少新字符数（默认: 8）")
    p.add_argument("--max-size", type=int, default=0, help="输出最长边像素限制（0=保持原始分辨率，默认: 0）")
    p.add_argument("-q", "--quality", type=int, default=2, help="JPEG 输出质量 1-31，越小越清晰（默认: 2）")
    p.add_argument("--timeout", type=float, default=1800, help="超时秒数（默认: 1800）")
    p.add_argument("--filter-blur", action="store_true", help="启用模糊帧过滤")
    p.add_argument("--blur-threshold", type=float, default=50.0, help="模糊阈值，Laplacian 方差低于此值视为模糊（默认: 50.0）")
    p.add_argument("--filter-quality", action="store_true", default=True, help="内容质量过滤（默认开启，过滤空白页、启动画面、过渡帧）")
    p.add_argument("--no-filter-quality", action="store_false", dest="filter_quality", help="禁用内容质量过滤")
    p.add_argument("--min-gap", type=float, default=0.5, help="保留帧之间的最小时间间隔秒数（默认: 0.5）")
    p.add_argument(
        "--keep-drop-candidates",
        action="store_true",
        help="保存被去重或过滤丢弃的候选帧，供多模态复核",
    )
    p.add_argument(
        "--drop-candidate-limit",
        type=int,
        default=200,
        help="最多保存多少张丢弃候选帧（默认: 200，0=不限）",
    )
    p.add_argument("--keep-temp", action="store_true", help="保留临时 ffmpeg 输出文件")
    return p.parse_args(argv)


def format_timestamp(seconds: float | None) -> str:
    if seconds is None:
        return "00m00s"
    total = int(max(0, seconds))
    h, rem = divmod(total, 3600)
    m, s = divmod(rem, 60)
    if h > 0:
        return f"{h:02d}h{m:02d}m{s:02d}s"
    return f"{m:02d}m{s:02d}s"


def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
        datefmt="%H:%M:%S",
    )

    args = parse_args()

    # 验证输入
    video_path = str(Path(args.input).resolve())
    if not Path(video_path).exists():
        print(f"错误: 视频文件不存在: {video_path}", file=sys.stderr)
        sys.exit(1)

    # 验证 ffmpeg
    if not find_tool("ffmpeg"):
        print("错误: 未检测到 ffmpeg，请先安装: brew install ffmpeg", file=sys.stderr)
        sys.exit(1)

    # 确定输出目录（默认在视频文件同级目录下）
    video_stem = Path(video_path).stem
    output_dir = args.output or str(Path(video_path).parent / f"{video_stem}_frames")
    output_dir = str(Path(output_dir).resolve())
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    cleanup_stats = _clean_output_dir(output_dir)
    if cleanup_stats["stale_deleted_count"]:
        print(f"  已清理旧输出文件: {cleanup_stats['stale_deleted_count']}")

    # 探测视频
    print(f"探测视频: {video_path}")
    try:
        info = probe_video(video_path)
    except Exception as e:
        print(f"错误: {e}", file=sys.stderr)
        sys.exit(1)
    print(f"  时长: {info.duration_seconds:.1f}s")

    # 参数
    params = ExtractParams(
        interval_seconds=args.interval,
        strategy=args.strategy,
        dedup_threshold=args.dedup_threshold,
        ocr_similarity_threshold=args.ocr_threshold,
        ocr_min_new_chars=args.ocr_min_new,
        content_crop_top=args.content_crop_top,
        content_crop_bottom=args.content_crop_bottom,
        content_crop_left=args.content_crop_left,
        content_crop_right=args.content_crop_right,
        ssim_threshold=args.ssim_threshold,
        scroll_merge=args.scroll_merge,
        scroll_diff_threshold=args.scroll_diff_threshold,
    )

    # OCR 引擎
    ocr_engine = None
    if args.ocr_dedup:
        ocr_engine = create_ocr_engine()
        if ocr_engine is None:
            print("警告: RapidOCR 未安装，OCR 去重已禁用。安装: pip install rapidocr-onnxruntime", file=sys.stderr)
            args.ocr_dedup = False
        else:
            print("  OCR: RapidOCR (本地)（SSIM 去重已自动禁用）")
            params.ssim_threshold = 0

    # 创建临时目录
    started_at = time.monotonic()
    tmpdir = tempfile.mkdtemp(prefix="video-screenshot-")
    try:
        # FFmpeg 抽帧
        interval_based = params.strategy == "interval"
        output_pattern = str(
            Path(tmpdir) / ("frame_%06d.jpg" if interval_based else "frame_%010d.jpg")
        )
        print(f"抽帧中 (策略: {params.strategy})...")
        ffmpeg_timeout = max(30.0, args.timeout - 5.0)

        for kv in run_ffmpeg_extract(
            video_path=video_path,
            output_pattern=output_pattern,
            strategy=params.strategy,
            interval_seconds=params.interval_seconds,
            scene_threshold=args.scene_threshold,
            max_size=args.max_size,
            quality=args.quality,
            timeout_seconds=ffmpeg_timeout,
            frame_rate_fps=info.frame_rate_fps,
            sample_interval=args.sample_interval,
        ):
            if "out_time_ms" in kv:
                try:
                    out_us = int(kv["out_time_ms"])
                    out_s = out_us / 1_000_000.0
                    pct = int(out_s * 100 / info.duration_seconds) if info.duration_seconds else 0
                    pct = min(max(pct, 0), 99)
                    print(f"\r  进度: {pct}% ({out_s:.1f}s/{info.duration_seconds:.1f}s)", end="", flush=True)
                except Exception:
                    pass
        print()  # 换行

        # 收集帧文件
        frame_files = collect_frame_files(tmpdir)
        total_extracted = len(frame_files)
        print(f"提取帧数: {total_extracted}")

        if not frame_files:
            print("警告: 未提取到任何帧", file=sys.stderr)
            _write_report(
                output_dir, video_path, info, params, 0, DedupState(), [], cleanup_stats,
                [], args.keep_drop_candidates, args.drop_candidate_limit,
            )
            return

        # 去重
        state = DedupState()
        window = 20
        pixel_diff_threshold = 8.0
        frames_meta: list[dict] = []
        drop_candidates_meta: list[dict] = []
        last_kept_time: float | None = None

        print("去重中...")
        for idx, frame_path in enumerate(frame_files, 1):
            state.total_count += 1

            with open(frame_path, "rb") as fp:
                content = fp.read()

            digest = sha256(content).hexdigest()
            dhash_hex = calc_dhash_hex(
                content,
                crop_top_ratio=params.content_crop_top,
                crop_bottom_ratio=params.content_crop_bottom,
                crop_left_ratio=params.content_crop_left,
                crop_right_ratio=params.content_crop_right,
            )

            capture_time = calc_capture_time(frame_path, idx, params, info)

            # 图像去重
            is_dup, drop_reason, thumb, ssim_thumb, scroll_image = is_frame_duplicate(
                content, digest, dhash_hex, state, params, window, pixel_diff_threshold,
            )
            if is_dup:
                _record_drop_candidate(
                    output_dir,
                    frame_path,
                    idx,
                    drop_reason,
                    capture_time,
                    digest,
                    drop_candidates_meta,
                    enabled=args.keep_drop_candidates,
                    limit=args.drop_candidate_limit,
                )
                continue

            # 最小时间间隔过滤
            if args.min_gap > 0 and last_kept_time is not None and capture_time is not None:
                if capture_time - last_kept_time < args.min_gap:
                    state.min_gap_drops += 1
                    _record_drop_candidate(
                        output_dir,
                        frame_path,
                        idx,
                        "min_gap",
                        capture_time,
                        digest,
                        drop_candidates_meta,
                        enabled=args.keep_drop_candidates,
                        limit=args.drop_candidate_limit,
                    )
                    continue

            # 内容质量过滤（空白页、启动画面、过渡帧）
            if args.filter_quality:
                quality = calc_content_quality(content)
                if quality["label"]:
                    state.quality_drops += 1
                    _record_drop_candidate(
                        output_dir,
                        frame_path,
                        idx,
                        f"quality_{quality['label']}",
                        capture_time,
                        digest,
                        drop_candidates_meta,
                        enabled=args.keep_drop_candidates,
                        limit=args.drop_candidate_limit,
                        extra={"quality": quality},
                    )
                    continue

            # 模糊帧过滤
            if args.filter_blur:
                blur_score = calc_blur_score(content)
                if blur_score < args.blur_threshold:
                    state.blur_drops += 1
                    _record_drop_candidate(
                        output_dir,
                        frame_path,
                        idx,
                        "blur",
                        capture_time,
                        digest,
                        drop_candidates_meta,
                        enabled=args.keep_drop_candidates,
                        limit=args.drop_candidate_limit,
                        extra={"blur_score": blur_score},
                    )
                    continue

            # OCR 去重
            if args.ocr_dedup and ocr_engine is not None:
                crop_bytes, crop_range = crop_for_ocr_bytes_with_range(content)
                ocr_text = ""
                if crop_bytes and crop_range >= 18:
                    ocr_text = ocr_extract_text(ocr_engine, crop_bytes)
                    ocr_text = re.sub(r"\s+", "", ocr_text or "")
                    ocr_text = re.sub(r"[^\w一-鿿]+", "", ocr_text)

                if ocr_text and check_ocr_similarity(
                    ocr_text,
                    state.kept_ocr_texts,
                    state.kept_ocr_shingles,
                    params.ocr_similarity_threshold,
                    params.ocr_min_new_chars,
                ):
                    state.ocr_dups += 1
                    _record_drop_candidate(
                        output_dir,
                        frame_path,
                        idx,
                        "ocr_duplicate",
                        capture_time,
                        digest,
                        drop_candidates_meta,
                        enabled=args.keep_drop_candidates,
                        limit=args.drop_candidate_limit,
                    )
                    continue
            else:
                ocr_text = ""

            # 保留帧
            state.kept_count += 1
            state.seen_sha256.add(digest)
            if dhash_hex:
                state.kept_dhashes.append(dhash_hex)
            if not thumb:
                thumb = calc_thumb_bytes(
                    content,
                    crop_top_ratio=params.content_crop_top,
                    crop_bottom_ratio=params.content_crop_bottom,
                    crop_left_ratio=params.content_crop_left,
                    crop_right_ratio=params.content_crop_right,
                )
            if thumb:
                state.kept_thumbs.append(thumb)
            if not ssim_thumb and params.ssim_threshold and params.ssim_threshold > 0:
                ssim_thumb = calc_thumb_bytes(
                    content,
                    size=32,
                    crop_top_ratio=params.content_crop_top,
                    crop_bottom_ratio=params.content_crop_bottom,
                    crop_left_ratio=params.content_crop_left,
                    crop_right_ratio=params.content_crop_right,
                    autocontrast=True,
                )
            if ssim_thumb:
                state.kept_ssim_thumbs.append(ssim_thumb)
            if not scroll_image and params.scroll_merge:
                scroll_image = calc_scroll_image(
                    content,
                    crop_top_ratio=params.content_crop_top,
                    crop_bottom_ratio=params.content_crop_bottom,
                    crop_left_ratio=params.content_crop_left,
                    crop_right_ratio=params.content_crop_right,
                )
            if scroll_image:
                state.kept_scroll_images.append(scroll_image)
            if ocr_text:
                state.kept_ocr_texts.append(ocr_text)
                state.kept_ocr_shingles.append(shingles(ocr_text))

            # 复制到输出目录
            ts = format_timestamp(capture_time)
            out_name = f"frame_{state.kept_count:03d}_{ts}.jpg"
            out_path = str(Path(output_dir) / out_name)
            shutil.copy2(frame_path, out_path)
            last_kept_time = capture_time

            frames_meta.append({
                "index": state.kept_count,
                "filename": out_name,
                "capture_time_seconds": capture_time,
                "sha256": digest,
            })

            if idx % 50 == 0 or idx == total_extracted:
                print(
                    f"\r  已处理: {idx}/{total_extracted}, "
                    f"保留: {state.kept_count}, "
                    f"去重: {idx - state.kept_count}",
                    end="", flush=True,
                )
        print()  # 换行

        # 写入报告
        _write_report(
            output_dir, video_path, info, params, total_extracted, state, frames_meta, cleanup_stats,
            drop_candidates_meta, args.keep_drop_candidates, args.drop_candidate_limit,
        )

        # 归档
        archive_dir = _archive_result(
            output_dir, video_path, info, params, args, state, frames_meta, cleanup_stats,
            drop_candidates_meta,
            elapsed_seconds=time.monotonic() - started_at,
        )

        # 汇总
        print(f"\n完成!")
        print(f"  输出目录: {output_dir}")
        print(f"  提取帧: {total_extracted}")
        print(f"  保留帧: {state.kept_count}")
        print(f"  去重统计:")
        print(f"    SHA256 重复: {state.sha256_dups}")
        print(f"    dHash 重复:  {state.dhash_dups}")
        print(f"    像素重复:    {state.pixel_dups}")
        print(f"    SSIM 重复:    {state.ssim_dups}")
        print(f"    滚动合并:    {state.scroll_dups}")
        print(f"    OCR 重复:    {state.ocr_dups}")
        if state.blur_drops:
            print(f"    模糊过滤:    {state.blur_drops}")
        if state.quality_drops:
            print(f"    质量过滤:    {state.quality_drops}")
        if state.min_gap_drops:
            print(f"    时间间隔过滤: {state.min_gap_drops}")
        if args.keep_drop_candidates:
            print(f"  复核候选帧: {len(drop_candidates_meta)}")
        if archive_dir:
            print(f"  归档: {archive_dir}")

    finally:
        if not args.keep_temp:
            shutil.rmtree(tmpdir, ignore_errors=True)
        else:
            print(f"  临时文件: {tmpdir}")


def _clean_output_dir(output_dir: str) -> dict[str, object]:
    """清理本工具生成的旧输出文件，避免本次结果混入残留帧。"""
    root = Path(output_dir)
    stale_files: list[Path] = []
    stale_files.extend(root.glob("frame_*.jpg"))
    stale_files.extend(root.glob("frame_*.jpeg"))
    for name in ("_report.json", "extraction_meta.json"):
        p = root / name
        if p.exists() and p.is_file():
            stale_files.append(p)

    stale_dirs: list[Path] = []
    candidates_dir = root / "_review_candidates"
    if candidates_dir.exists() and candidates_dir.is_dir():
        stale_dirs.append(candidates_dir)

    deleted: list[str] = []
    seen: set[Path] = set()
    for p in stale_files:
        if p in seen or not p.is_file():
            continue
        seen.add(p)
        p.unlink()
        deleted.append(p.name)
    for p in stale_dirs:
        shutil.rmtree(p, ignore_errors=True)
        deleted.append(p.name + "/")

    return {
        "stale_deleted_count": len(deleted),
        "stale_deleted_files": deleted,
    }


def _record_drop_candidate(
    output_dir: str,
    frame_path: str,
    source_index: int,
    reason: str,
    capture_time: float | None,
    digest: str,
    candidates: list[dict],
    *,
    enabled: bool,
    limit: int,
    extra: dict[str, object] | None = None,
) -> None:
    """保存被算法丢弃的候选帧，供后续视觉复核。"""
    if not enabled:
        return
    if limit > 0 and len(candidates) >= limit:
        return

    candidates_dir = Path(output_dir) / "_review_candidates"
    candidates_dir.mkdir(parents=True, exist_ok=True)
    safe_reason = re.sub(r"[^a-z0-9_]+", "_", (reason or "drop").lower()).strip("_") or "drop"
    seq = len(candidates) + 1
    filename = f"candidate_{seq:03d}_{safe_reason}_{format_timestamp(capture_time)}.jpg"
    dst = candidates_dir / filename
    shutil.copy2(frame_path, dst)

    item: dict[str, object] = {
        "index": seq,
        "filename": str(Path("_review_candidates") / filename),
        "source_frame_index": source_index,
        "source_temp_filename": Path(frame_path).name,
        "reason": reason,
        "capture_time_seconds": capture_time,
        "sha256": digest,
    }
    if extra:
        item.update(extra)
    candidates.append(item)


def _write_report(
    output_dir: str,
    video_path: str,
    info: FFProbeInfo,
    params: ExtractParams,
    total_extracted: int,
    state: DedupState,
    frames: list[dict],
    cleanup_stats: dict[str, object],
    drop_candidates: list[dict],
    keep_drop_candidates: bool,
    drop_candidate_limit: int,
) -> None:
    report = {
        "input": video_path,
        "duration_seconds": info.duration_seconds,
        "strategy": params.strategy,
        "options": {
            "interval_seconds": params.interval_seconds,
            "dedup_threshold": params.dedup_threshold,
            "ocr_similarity_threshold": params.ocr_similarity_threshold,
            "ocr_min_new_chars": params.ocr_min_new_chars,
            "content_crop": {
                "top": params.content_crop_top,
                "bottom": params.content_crop_bottom,
                "left": params.content_crop_left,
                "right": params.content_crop_right,
            },
            "ssim_threshold": params.ssim_threshold,
            "scroll_merge": params.scroll_merge,
            "scroll_diff_threshold": params.scroll_diff_threshold,
        },
        "total_extracted": total_extracted,
        "kept_after_dedup": state.kept_count,
        "cleanup": cleanup_stats,
        "review": {
            "drop_candidates_enabled": keep_drop_candidates,
            "drop_candidate_limit": drop_candidate_limit,
            "drop_candidate_count": len(drop_candidates),
            "drop_candidates": drop_candidates,
            "vision_review_status": "not_run",
            "vision_review_note": "脚本只导出复核候选帧；是否执行多模态复核由当前模型能力决定。",
        },
        "dedup_stats": {
            "sha256_duplicates": state.sha256_dups,
            "dhash_duplicates": state.dhash_dups,
            "pixel_duplicates": state.pixel_dups,
            "ssim_duplicates": state.ssim_dups,
            "scroll_duplicates": state.scroll_dups,
            "ocr_duplicates": state.ocr_dups,
            "blur_drops": state.blur_drops,
            "quality_drops": state.quality_drops,
            "min_gap_drops": state.min_gap_drops,
        },
        "frames": frames,
    }
    report_path = str(Path(output_dir) / "_report.json")
    with open(report_path, "w", encoding="utf-8") as fp:
        json.dump(report, fp, ensure_ascii=False, indent=2)


def _build_archive_subdir(video_path: str) -> Path:
    """创建 archive 子目录，命名格式: YYYYMMDD_HHMMSS_{视频名}"""
    skill_root = Path(__file__).resolve().parent.parent
    archive_root = skill_root / "archive"
    video_stem = Path(video_path).stem
    # 截断过长的文件名
    if len(video_stem) > 60:
        video_stem = video_stem[:60]
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    archive_dir = archive_root / f"{ts}_{video_stem}"
    archive_dir.mkdir(parents=True, exist_ok=True)
    return archive_dir


def _archive_result(
    output_dir: str,
    video_path: str,
    info: FFProbeInfo,
    params: ExtractParams,
    args: argparse.Namespace,
    state: DedupState,
    frames_meta: list[dict],
    cleanup_stats: dict[str, object],
    drop_candidates: list[dict],
    elapsed_seconds: float,
) -> Path | None:
    """将分析结果归档到 archive/ 目录。"""
    archive_dir = _build_archive_subdir(video_path)

    frames_dir = archive_dir / "frames"
    frames_dir.mkdir(exist_ok=True)
    expected_names = [str(frame["filename"]) for frame in frames_meta]
    for name in expected_names:
        src = Path(output_dir) / name
        if not src.exists():
            raise FileNotFoundError(f"归档失败，报告帧不存在: {src}")
        shutil.copy2(src, frames_dir / name)

    report_src = Path(output_dir) / "_report.json"
    if not report_src.exists():
        raise FileNotFoundError(f"归档失败，报告文件不存在: {report_src}")
    shutil.copy2(report_src, archive_dir / "_report.json")

    actual_names = sorted(p.name for p in frames_dir.glob("*.jpg"))
    expected_sorted = sorted(expected_names)
    if actual_names != expected_sorted:
        extra = sorted(set(actual_names) - set(expected_sorted))
        missing = sorted(set(expected_sorted) - set(actual_names))
        raise RuntimeError(
            "归档一致性校验失败: "
            f"expected={len(expected_sorted)}, actual={len(actual_names)}, "
            f"extra={extra[:5]}, missing={missing[:5]}"
        )

    review_dir = archive_dir / "_review_candidates"
    if drop_candidates:
        review_dir.mkdir(exist_ok=True)
        for item in drop_candidates:
            rel_name = str(item.get("filename") or "")
            src = Path(output_dir) / rel_name
            if src.exists() and src.is_file():
                shutil.copy2(src, review_dir / src.name)

    meta = {
        "source_file": video_path,
        "archive_path": str(archive_dir),
        "timestamp": datetime.now().isoformat(),
        "elapsed_seconds": round(elapsed_seconds, 1),
        "video_info": {
            "duration_seconds": info.duration_seconds,
            "time_base_seconds": info.time_base_seconds,
            "frame_rate_fps": info.frame_rate_fps,
        },
        "options": {
            "strategy": params.strategy,
            "interval_seconds": params.interval_seconds,
            "scene_threshold": args.scene_threshold,
            "dedup_threshold": params.dedup_threshold,
            "content_crop": {
                "top": params.content_crop_top,
                "bottom": params.content_crop_bottom,
                "left": params.content_crop_left,
                "right": params.content_crop_right,
            },
            "ssim_threshold": params.ssim_threshold,
            "scroll_merge": params.scroll_merge,
            "scroll_diff_threshold": params.scroll_diff_threshold,
            "ocr_dedup": args.ocr_dedup,
            "ocr_similarity_threshold": params.ocr_similarity_threshold,
            "ocr_min_new_chars": params.ocr_min_new_chars,
            "max_size": args.max_size,
            "quality": args.quality,
            "filter_blur": args.filter_blur,
            "blur_threshold": args.blur_threshold,
            "filter_quality": args.filter_quality,
            "keep_drop_candidates": args.keep_drop_candidates,
            "drop_candidate_limit": args.drop_candidate_limit,
        },
        "cleanup": cleanup_stats,
        "archive_validation": {
            "frames_match_report": True,
            "expected_frame_count": len(expected_sorted),
            "actual_frame_count": len(actual_names),
        },
        "review": {
            "drop_candidate_count": len(drop_candidates),
            "drop_candidates_archived": bool(drop_candidates),
            "vision_review_status": "not_run",
        },
        "result": {
            "total_extracted": state.total_count,
            "kept_after_dedup": state.kept_count,
            "dedup_stats": {
                "sha256_duplicates": state.sha256_dups,
                "dhash_duplicates": state.dhash_dups,
                "pixel_duplicates": state.pixel_dups,
                "ssim_duplicates": state.ssim_dups,
                "scroll_duplicates": state.scroll_dups,
                "ocr_duplicates": state.ocr_dups,
                "blur_drops": state.blur_drops,
                "quality_drops": state.quality_drops,
                "min_gap_drops": state.min_gap_drops,
            },
        },
        "frame_count": state.kept_count,
    }
    with open(archive_dir / "extraction_meta.json", "w", encoding="utf-8") as fp:
        json.dump(meta, fp, ensure_ascii=False, indent=2)

    return archive_dir


if __name__ == "__main__":
    main()

"""video-screenshot 共享工具函数。

从 fachuan chat_records/services/ 移植，去除 Django 依赖。
"""

from __future__ import annotations

import contextlib
import io
import json
import logging
import re
import select
import shutil
import subprocess
import time
from dataclasses import dataclass, field
from difflib import SequenceMatcher
from pathlib import Path
from typing import Any, cast

from PIL import Image, ImageChops, ImageEnhance, ImageFilter, ImageOps, ImageStat

logger = logging.getLogger("video-screenshot")

_LANCZOS: Any = getattr(Image, "Resampling", Image).LANCZOS

# ======================================================================
# A. FFmpeg 工具
# ======================================================================


@dataclass(frozen=True)
class FFProbeInfo:
    duration_seconds: float
    time_base_seconds: float | None = None
    frame_rate_fps: float | None = None


def _parse_rate(value: Any) -> float | None:
    text = str(value or "")
    if not text or text == "0/0":
        return None
    try:
        if "/" in text:
            n, d = text.split("/", 1)
            denom = float(d)
            return float(n) / denom if denom else None
        rate = float(text)
        return rate if rate > 0 else None
    except Exception:
        return None


def find_tool(name: str) -> str | None:
    p = shutil.which(name)
    if p:
        return p
    for root in ("/usr/local/bin", "/opt/homebrew/bin", "/usr/bin"):
        candidate = str(Path(root) / name)
        if Path(candidate).exists() and Path(candidate).stat().st_mode & 0o111:
            return candidate
    return None


def probe_video(video_path: str) -> FFProbeInfo:
    if not video_path or not Path(video_path).exists():
        raise FileNotFoundError(f"视频文件不存在: {video_path}")

    duration = 0.0
    time_base_seconds: float | None = None
    frame_rate_fps: float | None = None
    ffprobe = find_tool("ffprobe")
    if ffprobe:
        cmd = [
            ffprobe, "-v", "error",
            "-select_streams", "v:0",
            "-show_entries", "format=duration:stream=time_base,avg_frame_rate,r_frame_rate",
            "-of", "json",
            video_path,
        ]
        try:
            result = subprocess.run(cmd, timeout=10, check=True, capture_output=True, text=True)
            data = json.loads(result.stdout or "{}")
            duration = float((data.get("format") or {}).get("duration") or 0.0)
            streams = data.get("streams") or []
            if streams:
                tb = str((streams[0] or {}).get("time_base") or "")
                if "/" in tb:
                    n, d = tb.split("/", 1)
                    time_base_seconds = float(n) / float(d) if float(d) else None
                frame_rate_fps = (
                    _parse_rate((streams[0] or {}).get("avg_frame_rate"))
                    or _parse_rate((streams[0] or {}).get("r_frame_rate"))
                )
        except Exception:
            logger.exception("ffprobe 解析失败: %s", video_path)
            duration = 0.0
            frame_rate_fps = None
    else:
        duration = _probe_duration_by_ffmpeg(video_path)
        frame_rate_fps = None

    if duration <= 0:
        raise RuntimeError(f"无法解析视频时长: {video_path}")

    return FFProbeInfo(
        duration_seconds=duration,
        time_base_seconds=time_base_seconds,
        frame_rate_fps=frame_rate_fps,
    )


def _probe_duration_by_ffmpeg(video_path: str) -> float:
    ffmpeg = find_tool("ffmpeg")
    if not ffmpeg:
        return 0.0
    cmd = [ffmpeg, "-hide_banner", "-i", video_path]
    try:
        result = subprocess.run(cmd, timeout=10, check=False, capture_output=True, text=True)
    except Exception:
        return 0.0
    text = (result.stderr or "") + "\n" + (result.stdout or "")
    m = re.search(r"Duration:\s*(\d+):(\d+):(\d+\.\d+)", text)
    if not m:
        return 0.0
    return int(m.group(1)) * 3600 + int(m.group(2)) * 60 + float(m.group(3))


def build_ffmpeg_filter_args(
    strategy: str,
    interval_seconds: float,
    scene_threshold: float,
    max_size: int = 0,
    frame_rate_fps: float | None = None,
    sample_interval: float = 5.0,
) -> tuple[list[str], str, list[str]]:
    """构建 ffmpeg 滤镜参数，返回 (input_args, vf, extra_args)。"""
    # max_size=0 时保持原始分辨率，不缩放
    if max_size and max_size > 0:
        scale = (
            f"scale='if(gt(iw,ih),min({max_size},iw),-2)':"
            f"'if(gt(iw,ih),-2,min({max_size},ih))'"
        )
    else:
        scale = ""
    vfr_args = ["-vsync", "vfr", "-frame_pts", "1"]

    fmt = ",format=yuvj420p"
    # 构建 scale 部分的滤镜链（可能为空）
    scale_part = f",{scale}" if scale else ""

    # scene 策略：场景检测 + 定期采样保底（确保静态画面也有覆盖）
    scene_expr = f"gt(scene,{float(scene_threshold)})"
    if frame_rate_fps and frame_rate_fps > 0 and sample_interval and sample_interval > 0:
        n_frames = max(1, round(frame_rate_fps * sample_interval))
        scene_expr = f"{scene_expr}+not(mod(n\\,{n_frames}))"
    scene_vf = f"select='{scene_expr}'{scale_part}{fmt}"

    strategy_map: dict[str, tuple[list[str], str, list[str]]] = {
        "scene": ([], scene_vf, vfr_args),
        "keyframe": (["-skip_frame", "nokey"], f"{scale_part[1:]},mpdecimate{fmt}" if scale_part else f"mpdecimate{fmt}", vfr_args),
        "smart": ([], f"{scale_part[1:]},mpdecimate{fmt}" if scale_part else f"mpdecimate{fmt}", vfr_args),
    }

    if strategy in strategy_map:
        return strategy_map[strategy]

    fps = 1.0 / interval_seconds
    return [], f"fps={fps}{scale_part},mpdecimate{fmt}", []


def _force_kill_proc(proc: subprocess.Popen[str]) -> None:
    with contextlib.suppress(Exception):
        proc.terminate()
    try:
        proc.wait(timeout=2)
    except Exception:
        with contextlib.suppress(Exception):
            proc.kill()


def _read_progress_lines(
    proc: subprocess.Popen[str],
    timeout_seconds: float | None,
    started: float,
) -> Any:
    if proc.stdout is None:
        return
    while True:
        if timeout_seconds is not None and time.monotonic() - started > timeout_seconds:
            _force_kill_proc(proc)
            raise RuntimeError("ffmpeg 抽帧超时")
        if proc.poll() is not None:
            break
        rlist, _, _ = select.select([proc.stdout], [], [], 0.2)
        if not rlist:
            continue
        line = proc.stdout.readline()
        if not line:
            break
        line = (line or "").strip()
        if not line or "=" not in line:
            continue
        k, v = line.split("=", 1)
        yield {k: v}


def _check_exit(proc: subprocess.Popen[str]) -> None:
    try:
        rc = proc.wait(timeout=5)
    except Exception:
        with contextlib.suppress(Exception):
            proc.kill()
        rc = proc.wait()
    if rc != 0:
        err = ""
        try:
            if proc.stderr is not None:
                err = proc.stderr.read() or ""
        except Exception:
            err = ""
        err = (err or "").strip()
        if err:
            tail = "\n".join(err.splitlines()[-12:])
            raise RuntimeError(f"ffmpeg 抽帧失败:\n{tail}")
        raise RuntimeError("ffmpeg 抽帧失败，请检查视频文件或 ffmpeg 安装")


def run_ffmpeg_extract(
    *,
    video_path: str,
    output_pattern: str,
    strategy: str = "scene",
    interval_seconds: float = 1.0,
    scene_threshold: float = 0.25,
    max_size: int = 1280,
    quality: int = 6,
    timeout_seconds: float | None = None,
    frame_rate_fps: float | None = None,
    sample_interval: float = 5.0,
) -> Any:
    """运行 ffmpeg 抽帧，yield 进度字典。"""
    ffmpeg = find_tool("ffmpeg")
    if not ffmpeg:
        raise RuntimeError("未检测到 ffmpeg，请先安装 (brew install ffmpeg)")

    input_args, vf, extra_args = build_ffmpeg_filter_args(
        strategy, interval_seconds, scene_threshold, max_size,
        frame_rate_fps=frame_rate_fps,
        sample_interval=sample_interval,
    )

    cmd = [
        ffmpeg, "-hide_banner", "-nostats",
        "-loglevel", "error",
        "-progress", "pipe:1",
        *input_args,
        "-i", video_path,
        "-vf", vf,
        *extra_args,
        "-q:v", str(quality),
        output_pattern,
    ]
    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    started = time.monotonic()

    yield from _read_progress_lines(proc, timeout_seconds, started)
    _check_exit(proc)


# ======================================================================
# B. 图像处理
# ======================================================================


def _content_crop_box(
    width: int,
    height: int,
    *,
    top_ratio: float = 0.12,
    bottom_ratio: float = 0.12,
    left_ratio: float = 0.04,
    right_ratio: float = 0.04,
) -> tuple[int, int, int, int]:
    top = int(max(0, min(height - 1, round(height * top_ratio))))
    bottom_cut = int(max(0, min(height - 1, round(height * bottom_ratio))))
    left = int(max(0, min(width - 1, round(width * left_ratio))))
    right_cut = int(max(0, min(width - 1, round(width * right_ratio))))
    bottom = max(top + 1, height - bottom_cut)
    right = max(left + 1, width - right_cut)
    return left, top, right, bottom


def _crop_content(
    img: Image.Image,
    *,
    top_ratio: float = 0.12,
    bottom_ratio: float = 0.12,
    left_ratio: float = 0.04,
    right_ratio: float = 0.04,
) -> Image.Image:
    w, h = img.size
    if w <= 0 or h <= 0:
        return img
    return img.crop(
        _content_crop_box(
            w,
            h,
            top_ratio=top_ratio,
            bottom_ratio=bottom_ratio,
            left_ratio=left_ratio,
            right_ratio=right_ratio,
        )
    )


def calc_dhash_hex(
    image_bytes: bytes,
    *,
    hash_size: int = 8,
    crop_top_ratio: float = 0.12,
    crop_bottom_ratio: float = 0.12,
    crop_left_ratio: float = 0.04,
    crop_right_ratio: float = 0.04,
) -> str:
    if not image_bytes or hash_size <= 0:
        return ""
    img = Image.open(io.BytesIO(image_bytes))
    img = img.convert("L")
    img = _crop_content(
        img,
        top_ratio=crop_top_ratio,
        bottom_ratio=crop_bottom_ratio,
        left_ratio=crop_left_ratio,
        right_ratio=crop_right_ratio,
    )
    img = img.resize((hash_size + 1, hash_size), _LANCZOS)
    pixels = list(img.getdata())
    bits = 0
    for row in range(hash_size):
        row_start = row * (hash_size + 1)
        for col in range(hash_size):
            left = pixels[row_start + col]
            right = pixels[row_start + col + 1]
            if left > right:
                bits |= 1 << (row * hash_size + col)
    hex_len = (hash_size * hash_size) // 4
    return f"{bits:0{hex_len}x}"


def hamming_distance_hex(a: str, b: str) -> int | None:
    if not a or not b:
        return None
    try:
        x = int(a, 16)
        y = int(b, 16)
    except Exception:
        return None
    return (x ^ y).bit_count()


def calc_thumb_bytes(
    image_bytes: bytes,
    *,
    size: int = 48,
    crop_top_ratio: float = 0.12,
    crop_bottom_ratio: float = 0.12,
    crop_left_ratio: float = 0.04,
    crop_right_ratio: float = 0.04,
    autocontrast: bool = False,
) -> bytes:
    if not image_bytes or size <= 0:
        return b""
    img = Image.open(io.BytesIO(image_bytes))
    img = img.convert("L")
    w, h = img.size
    if w <= 0 or h <= 0:
        return b""
    img = _crop_content(
        img,
        top_ratio=crop_top_ratio,
        bottom_ratio=crop_bottom_ratio,
        left_ratio=crop_left_ratio,
        right_ratio=crop_right_ratio,
    )
    if autocontrast:
        img = ImageOps.autocontrast(img)
    img = img.resize((size, size), _LANCZOS)
    return cast(bytes, img.tobytes())


def mean_abs_diff(a: bytes, b: bytes) -> float | None:
    if not a or not b or len(a) != len(b):
        return None
    total = 0
    for x, y in zip(a, b):
        total += x - y if x >= y else y - x
    return total / float(len(a))


def ssim_bytes(a: bytes, b: bytes) -> float | None:
    """计算两个等长灰度缩略图的全局 SSIM。"""
    if not a or not b or len(a) != len(b):
        return None
    n = len(a)
    mean_a = sum(a) / float(n)
    mean_b = sum(b) / float(n)
    denom = max(n - 1, 1)
    var_a = sum((x - mean_a) ** 2 for x in a) / float(denom)
    var_b = sum((y - mean_b) ** 2 for y in b) / float(denom)
    cov = sum((x - mean_a) * (y - mean_b) for x, y in zip(a, b)) / float(denom)
    c1 = (0.01 * 255) ** 2
    c2 = (0.03 * 255) ** 2
    divisor = (mean_a * mean_a + mean_b * mean_b + c1) * (var_a + var_b + c2)
    if not divisor:
        return None
    return ((2 * mean_a * mean_b + c1) * (2 * cov + c2)) / divisor


def calc_scroll_image(
    image_bytes: bytes,
    *,
    width: int = 96,
    height: int = 160,
    crop_top_ratio: float = 0.12,
    crop_bottom_ratio: float = 0.12,
    crop_left_ratio: float = 0.04,
    crop_right_ratio: float = 0.04,
) -> Image.Image | None:
    if not image_bytes or width <= 0 or height <= 0:
        return None
    img = Image.open(io.BytesIO(image_bytes))
    img = img.convert("L")
    img = _crop_content(
        img,
        top_ratio=crop_top_ratio,
        bottom_ratio=crop_bottom_ratio,
        left_ratio=crop_left_ratio,
        right_ratio=crop_right_ratio,
    )
    img = ImageOps.autocontrast(img)
    return img.resize((width, height), _LANCZOS)


def _shifted_mean_abs_diff(a: Image.Image, b: Image.Image, shift: int) -> tuple[float, float]:
    width, height = a.size
    if b.size != a.size or width <= 0 or height <= 0:
        return 999.0, 0.0
    if shift >= 0:
        box_a = (0, shift, width, height)
        box_b = (0, 0, width, height - shift)
    else:
        box_a = (0, 0, width, height + shift)
        box_b = (0, -shift, width, height)
    crop_a = a.crop(box_a)
    crop_b = b.crop(box_b)
    if crop_a.size[1] <= 0 or crop_b.size[1] <= 0:
        return 999.0, 0.0
    diff = ImageChops.difference(crop_a, crop_b)
    return ImageStat.Stat(diff).mean[0], crop_a.size[1] / float(height)


def scroll_overlap_duplicate(
    current: Image.Image,
    previous_images: list[Image.Image],
    *,
    threshold: float,
    min_shift: int = 4,
    max_shift_ratio: float = 0.35,
    min_overlap_ratio: float = 0.70,
    step: int = 4,
) -> bool:
    """检测当前帧是否只是最近保留帧的轻微纵向滚动版本。"""
    if current is None or not previous_images or threshold <= 0:
        return False
    width, height = current.size
    if width <= 0 or height <= 0:
        return False
    max_shift = max(min_shift, int(round(height * max_shift_ratio)))
    for prev in reversed(previous_images):
        if prev.size != current.size:
            continue
        best_diff = 999.0
        best_shift = 0
        best_overlap = 0.0
        for shift in range(-max_shift, max_shift + 1, step):
            diff, overlap = _shifted_mean_abs_diff(prev, current, shift)
            if overlap < min_overlap_ratio:
                continue
            if diff < best_diff:
                best_diff = diff
                best_shift = shift
                best_overlap = overlap
        if abs(best_shift) >= min_shift and best_overlap >= min_overlap_ratio and best_diff <= threshold:
            return True
    return False


# 3x3 Laplacian 卷积核（用于模糊检测）
_LAPLACIAN = ImageFilter.Kernel((3, 3), [0, 1, 0, 1, -4, 1, 0, 1, 0], scale=1, offset=0)


def calc_blur_score(image_bytes: bytes, *, size: int = 128) -> float:
    """计算帧的 Laplacian 方差，值越低越模糊。"""
    if not image_bytes:
        return 0.0
    img = Image.open(io.BytesIO(image_bytes))
    img = img.convert("L")
    img = img.resize((size, size), _LANCZOS)
    filtered = img.filter(_LAPLACIAN)
    return ImageStat.Stat(filtered).var[0]


def calc_content_quality(image_bytes: bytes) -> dict[str, Any]:
    """分析帧的内容质量，返回指标字典。"""
    if not image_bytes:
        return {"label": "empty", "content_std": 0.0, "white_ratio": 0.0, "grid_flat": 9}
    img = Image.open(io.BytesIO(image_bytes)).convert("L")
    w, h = img.size
    if w <= 0 or h <= 0:
        return {"label": "empty", "content_std": 0.0, "white_ratio": 0.0, "grid_flat": 9}

    # 直方图分析
    hist = img.histogram()
    total = w * h
    white_ratio = sum(hist[240:]) / total
    black_ratio = sum(hist[:15]) / total

    # 内容区域（排除顶部 8% 和底部 8% 的状态栏）
    content_crop = img.crop((0, int(h * 0.08), w, int(h * 0.92)))
    content_std = ImageStat.Stat(content_crop).stddev[0]

    # 3×3 网格分析
    grid_stds: list[float] = []
    for row in range(3):
        for col in range(3):
            y1, y2 = row * h // 3, (row + 1) * h // 3
            x1, x2 = col * w // 3, (col + 1) * w // 3
            grid_stds.append(ImageStat.Stat(img.crop((x1, y1, x2, y2))).stddev[0])

    grid_flat = sum(1 for s in grid_stds if s < 10)
    grid_high = sum(1 for s in grid_stds if s > 50)
    grid_spread = max(grid_stds) - min(grid_stds)

    # 分类
    label = ""
    if content_std < 10 or white_ratio > 0.95 or black_ratio > 0.95:
        label = "blank"
    elif content_std < 35 and grid_high <= 2:
        label = "startup"
    elif grid_spread > 45 and grid_flat >= 2:
        label = "transition"

    return {
        "label": label,
        "content_std": content_std,
        "white_ratio": white_ratio,
        "black_ratio": black_ratio,
        "grid_flat": grid_flat,
        "grid_high": grid_high,
        "grid_spread": grid_spread,
    }


def crop_for_ocr_bytes_with_range(
    image_bytes: bytes,
    *,
    crop_top_ratio: float = 0.16,
    crop_bottom_ratio: float = 0.14,
    crop_left_ratio: float = 0.06,
    crop_right_ratio: float = 0.06,
    max_width: int = 720,
) -> tuple[bytes, int]:
    if not image_bytes:
        return (b"", 0)
    img = Image.open(io.BytesIO(image_bytes))
    w, h = img.size
    if w <= 0 or h <= 0:
        return (b"", 0)

    top = int(max(0, min(h - 1, round(h * crop_top_ratio))))
    bottom_cut = int(max(0, min(h - 1, round(h * crop_bottom_ratio))))
    bottom = max(top + 1, h - bottom_cut)
    left = int(max(0, min(w - 1, round(w * crop_left_ratio))))
    right_cut = int(max(0, min(w - 1, round(w * crop_right_ratio))))
    right = max(left + 1, w - right_cut)
    img = img.crop((left, top, right, bottom))

    if max_width and img.size[0] > max_width:
        new_w = int(max_width)
        new_h = round(img.size[1] * (new_w / float(img.size[0])))
        img = img.resize((new_w, max(1, new_h)), _LANCZOS)

    img = img.convert("L")
    img = ImageOps.autocontrast(img)
    img = ImageEnhance.Contrast(img).enhance(1.35)
    img = ImageEnhance.Sharpness(img).enhance(1.15)

    dynamic_range = 0
    try:
        extrema = img.getextrema()
        if isinstance(extrema, tuple) and len(extrema) == 2:
            lo_val, hi_val = extrema
            if isinstance(lo_val, (int, float)) and isinstance(hi_val, (int, float)):
                dynamic_range = int(hi_val) - int(lo_val)
    except Exception:
        pass

    buf = io.BytesIO()
    img.save(buf, format="PNG", optimize=True)
    return (buf.getvalue(), max(0, dynamic_range))


# ======================================================================
# C. 去重逻辑
# ======================================================================


def shingles(s: str, n: int = 3) -> set[str]:
    s = s or ""
    if not s:
        return set()
    if len(s) <= n:
        return {s}
    return {s[i : i + n] for i in range(0, len(s) - n + 1)}


def jaccard_sets(sa: set[str], sb: set[str]) -> float:
    if not sa or not sb:
        return 0.0
    inter = len(sa & sb)
    union = len(sa | sb)
    return float(inter) / float(union) if union else 0.0


@dataclass
class ExtractParams:
    interval_seconds: float = 1.0
    strategy: str = "scene"
    dedup_threshold: int = 8
    ocr_similarity_threshold: float = 0.92
    ocr_min_new_chars: int = 8
    content_crop_top: float = 0.12
    content_crop_bottom: float = 0.12
    content_crop_left: float = 0.04
    content_crop_right: float = 0.04
    ssim_threshold: float = 0.93
    scroll_merge: bool = True
    scroll_diff_threshold: float = 32.0


@dataclass
class DedupState:
    seen_sha256: set[str] = field(default_factory=set)
    kept_dhashes: list[str] = field(default_factory=list)
    kept_thumbs: list[bytes] = field(default_factory=list)
    kept_ssim_thumbs: list[bytes] = field(default_factory=list)
    kept_scroll_images: list[Image.Image] = field(default_factory=list)
    kept_ocr_texts: list[str] = field(default_factory=list)
    kept_ocr_shingles: list[set[str]] = field(default_factory=list)
    sha256_dups: int = 0
    dhash_dups: int = 0
    pixel_dups: int = 0
    ssim_dups: int = 0
    scroll_dups: int = 0
    ocr_dups: int = 0
    blur_drops: int = 0
    quality_drops: int = 0
    min_gap_drops: int = 0
    kept_count: int = 0
    total_count: int = 0


def is_dhash_duplicate(
    dhash_hex: str,
    kept_dhashes: list[str],
    window: int,
    threshold: int,
) -> bool:
    for prev in kept_dhashes[-window:]:
        dist = hamming_distance_hex(prev, dhash_hex)
        if dist is not None and dist <= threshold:
            return True
    return False


def is_pixel_duplicate(
    thumb: bytes,
    kept_thumbs: list[bytes],
    window: int,
    threshold: float,
) -> bool:
    for prev_thumb in kept_thumbs[-window:]:
        diff = mean_abs_diff(prev_thumb, thumb)
        if diff is not None and diff <= threshold:
            return True
    return False


def is_ssim_duplicate(
    thumb: bytes,
    kept_thumbs: list[bytes],
    window: int,
    threshold: float,
) -> bool:
    for prev_thumb in kept_thumbs[-window:]:
        sim = ssim_bytes(prev_thumb, thumb)
        if sim is not None and sim >= threshold:
            return True
    return False


def check_ocr_similarity(
    ocr_text: str,
    kept_ocr_texts: list[str],
    kept_ocr_shingles: list[set[str]],
    ocr_similarity_threshold: float,
    ocr_min_new_chars: int,
) -> bool:
    """检查 OCR 文本是否与最近帧重复，返回 True 表示重复应跳过。"""
    if not ocr_text or not kept_ocr_texts:
        return False
    cur_set = shingles(ocr_text)
    for prev_text, prev_set in zip(
        kept_ocr_texts[-4:],
        kept_ocr_shingles[-4:],
    ):
        if not prev_text:
            continue
        seq_sim = float(SequenceMatcher(None, prev_text, ocr_text).ratio())
        jac_sim = jaccard_sets(prev_set, cur_set)
        sim = max(seq_sim, jac_sim)
        new_tokens = len(cur_set - prev_set) if prev_set else len(cur_set)
        if sim >= ocr_similarity_threshold and new_tokens < ocr_min_new_chars:
            return True
    return False


def is_frame_duplicate(
    content: bytes,
    digest: str,
    dhash_hex: str,
    state: DedupState,
    params: ExtractParams,
    window: int = 20,
    pixel_diff_threshold: float = 8.0,
) -> tuple[bool, str, bytes, bytes, Image.Image | None]:
    """图像层级去重，返回 (is_dup, reason, pixel_thumb, ssim_thumb, scroll_image)。"""
    if digest in state.seen_sha256:
        state.sha256_dups += 1
        return True, "duplicate_sha256", b"", b"", None

    if (
        params.dedup_threshold
        and state.kept_dhashes
        and is_dhash_duplicate(dhash_hex, state.kept_dhashes, window, params.dedup_threshold)
    ):
        state.dhash_dups += 1
        return True, "duplicate_dhash", b"", b"", None

    thumb = b""
    if pixel_diff_threshold and state.kept_thumbs:
        thumb = calc_thumb_bytes(
            content,
            crop_top_ratio=params.content_crop_top,
            crop_bottom_ratio=params.content_crop_bottom,
            crop_left_ratio=params.content_crop_left,
            crop_right_ratio=params.content_crop_right,
        )
        if thumb and is_pixel_duplicate(thumb, state.kept_thumbs, window, pixel_diff_threshold):
            state.pixel_dups += 1
            return True, "duplicate_pixel", thumb, b"", None

    ssim_thumb = b""
    if params.ssim_threshold and params.ssim_threshold > 0 and state.kept_ssim_thumbs:
        ssim_thumb = calc_thumb_bytes(
            content,
            size=32,
            crop_top_ratio=params.content_crop_top,
            crop_bottom_ratio=params.content_crop_bottom,
            crop_left_ratio=params.content_crop_left,
            crop_right_ratio=params.content_crop_right,
            autocontrast=True,
        )
        if ssim_thumb and is_ssim_duplicate(ssim_thumb, state.kept_ssim_thumbs, window, params.ssim_threshold):
            state.ssim_dups += 1
            return True, "duplicate_ssim", thumb, ssim_thumb, None

    scroll_image = None
    if params.scroll_merge and params.scroll_diff_threshold > 0 and state.kept_scroll_images:
        scroll_image = calc_scroll_image(
            content,
            crop_top_ratio=params.content_crop_top,
            crop_bottom_ratio=params.content_crop_bottom,
            crop_left_ratio=params.content_crop_left,
            crop_right_ratio=params.content_crop_right,
        )
        if scroll_image and scroll_overlap_duplicate(
            scroll_image,
            state.kept_scroll_images[-8:],
            threshold=params.scroll_diff_threshold,
        ):
            state.scroll_dups += 1
            return True, "duplicate_scroll", thumb, ssim_thumb, scroll_image

    return False, "", thumb, ssim_thumb, scroll_image


def calc_capture_time(
    path: str,
    index: int,
    params: ExtractParams,
    info: FFProbeInfo,
) -> float | None:
    interval_based = params.strategy in ("interval",)
    if not interval_based and (info.time_base_seconds or info.frame_rate_fps):
        m = re.search(r"(\d+)", Path(path).name)
        if not m:
            return None
        pts = int(m.group(1))
        if info.frame_rate_fps and info.frame_rate_fps > 0:
            fps_time = float(pts) / float(info.frame_rate_fps)
            if 0 <= fps_time <= info.duration_seconds * 1.1:
                return fps_time
        return float(pts * float(info.time_base_seconds)) if info.time_base_seconds else None
    return float(index - 1) * float(params.interval_seconds)


def collect_frame_files(tmpdir: str) -> list[str]:
    frame_files = [
        str(Path(tmpdir) / f.name)
        for f in Path(tmpdir).iterdir()
        if f.name.lower().endswith((".jpg", ".jpeg", ".png"))
    ]
    frame_files.sort()
    return frame_files


# ======================================================================
# D. OCR 集成
# ======================================================================

_ocr_engine = None


def create_ocr_engine():
    """创建本地 RapidOCR 引擎（懒加载单例）。"""
    global _ocr_engine
    if _ocr_engine is not None:
        return _ocr_engine
    try:
        from rapidocr_onnxruntime import RapidOCR
        _ocr_engine = RapidOCR()
        return _ocr_engine
    except ImportError:
        logger.warning(
            "rapidocr-onnxruntime 未安装，OCR 去重不可用。"
            "安装方式: pip install rapidocr-onnxruntime"
        )
        return None


def ocr_extract_text(ocr_engine: Any, image_bytes: bytes) -> str:
    """调用 OCR 提取文本。"""
    if ocr_engine is None:
        return ""
    try:
        result, _ = ocr_engine(image_bytes)
        if result:
            return "|".join(line[-1] for line in result)
    except Exception:
        logger.debug("OCR 识别失败", exc_info=True)
    return ""