Blame for processor/app/video_quality.py

"""Post-processing video quality checks.

The checks in this module are deliberately media-only.  They do not try to
decide which hand is left/right or whether an action label is correct.  A
workflow can use them as a gate before the existing ``reviewed``/Approved
state:

* the file opens and sampled frames can be decoded;
* a stream is not materially shorter than the episode timeline;
* sampled frames are not almost entirely black;
* sampled frames do not remain identical for a configurable duration.

An encoded MP4 does not always retain the capture device's original frame
index.  Therefore the frame-count check detects a materially short stream,
while exact capture-time drops require source ``frame_index``/PTS metadata and
are reported separately by future collectors.  The result is fail-closed for
automatic approval, but the old UI still only exposes Reviewing/Approved.
"""

from __future__ import annotations

import asyncio
import os
from pathlib import Path

# 状态写入走 localstore.mutate_episode_state（锁内 RMW），在用到的地方局部导入 ——
# 见 _apply_video_quality_result 的注释。


DEFAULT_SAMPLE_INTERVAL_SEC = 0.5
DEFAULT_MAX_SAMPLES = 240
DEFAULT_FREEZE_MIN_SEC = 2.0
DEFAULT_FREEZE_DIFF = 0.75
DEFAULT_BLACK_MEAN = 5.0
DEFAULT_BLACK_DARK_RATIO = 0.995
DEFAULT_BLACK_MIN_SEC = 0.5
DEFAULT_FRAME_TOLERANCE_RATIO = 0.005
DEFAULT_FRAME_TOLERANCE_MIN = 2
_VIDEO_QC_SEMAPHORE = asyncio.Semaphore(1)


def _float_env(name: str, default: float, minimum: float = 0.0) -> float:
    try:
        return max(minimum, float(os.getenv(name, str(default))))
    except (TypeError, ValueError):
        return default


def _int_env(name: str, default: int, minimum: int = 1) -> int:
    try:
        return max(minimum, int(os.getenv(name, str(default))))
    except (TypeError, ValueError):
        return default


def _raw_videos(batch_dir: Path) -> list[Path]:
    """Return input videos only; never inspect processed run outputs."""
    videos_root = batch_dir / "videos"
    if videos_root.is_dir():
        paths = [p for p in videos_root.rglob("*")
                 if p.is_file() and p.suffix.lower() == ".mp4"]
    else:
        paths = [p for p in batch_dir.rglob("*")
                 if p.is_file() and p.suffix.lower() == ".mp4"
                 and "processed" not in p.parts
                 and "original" not in p.parts
                 and "tmp" not in p.parts]
    return sorted(paths, key=lambda p: str(p).lower())


def _sample_indices(frame_count: int, fps: float) -> tuple[list[int], float]:
    """Build approximately half-second sample positions with a hard cap."""
    count = max(0, int(frame_count))
    if count <= 0:
        return [], 0.0
    interval = _float_env(
        "EGODATA_VIDEO_QC_SAMPLE_INTERVAL_SEC",
        DEFAULT_SAMPLE_INTERVAL_SEC,
        minimum=0.05,
    )
    frame_step = max(1, int(round(max(1.0, fps) * interval)))
    indices = list(range(0, count, frame_step))
    if indices[-1] != count - 1:
        indices.append(count - 1)
    max_samples = _int_env(
        "EGODATA_VIDEO_QC_MAX_SAMPLES", DEFAULT_MAX_SAMPLES, minimum=3)
    if len(indices) > max_samples:
        # Keep samples distributed across the whole stream.  This may miss a
        # very short freeze in a very long video, but it prevents a quality
        # check from turning into a full decode pass on remote storage.
        positions = [round(i * (len(indices) - 1) / (max_samples - 1))
                     for i in range(max_samples)]
        indices = [indices[position] for position in positions]
    return sorted(set(indices)), frame_step / max(1.0, fps)


def _add_freeze_range(ranges: list[list[float]], start: int, end: int,
                      fps: float) -> None:
    if end < start:
        return
    item = [round(start / max(1.0, fps), 3),
            round(end / max(1.0, fps), 3)]
    if ranges and item[0] <= ranges[-1][1] + 0.01:
        ranges[-1][1] = max(ranges[-1][1], item[1])
    else:
        ranges.append(item)


def _check_stream(path: Path, expected_frames: int = 0,
                  expected_fps: float = 0.0) -> dict:
    """Check one video without loading the complete stream into memory."""
    try:
        import cv2
        import numpy as np
    except Exception as exc:
        return {
            "path": str(path),
            "passed": False,
            "reason": "opencv_unavailable",
            "error": str(exc),
        }

    report: dict = {
        "path": str(path),
        "passed": False,
        "frame_count": 0,
        "expected_frame_count": max(0, int(expected_frames or 0)),
        "fps": 0.0,
        "width": 0,
        "height": 0,
        "sampled_frames": 0,
        "decode_error_count": 0,
        "decode_errors": [],
        "black_frame_count": 0,
        "black_frames": [],
        "black_ranges": [],
        "freeze_ranges": [],
        "dropped_frame_estimate": 0,
        "count_tolerance": 0,
    }
    cap = cv2.VideoCapture(str(path))
    if not cap.isOpened():
        report["reason"] = "video_open_failed"
        cap.release()
        return report

    try:
        frame_count = int(round(float(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)))
        fps = float(cap.get(cv2.CAP_PROP_FPS) or 0.0)
        width = int(round(float(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)))
        height = int(round(float(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)))
        report.update(frame_count=max(0, frame_count), fps=round(max(0.0, fps), 4),
                      width=max(0, width), height=max(0, height))
        if frame_count <= 0 or fps <= 0 or width <= 0 or height <= 0:
            report["reason"] = "video_metadata_invalid"
            return report

        indices, sample_interval = _sample_indices(frame_count, fps)
        report["sample_interval_sec"] = round(sample_interval, 3)
        black_mean = _float_env("EGODATA_VIDEO_QC_BLACK_MEAN",
                                DEFAULT_BLACK_MEAN, minimum=0.0)
        black_dark_ratio = min(
            1.0,
            _float_env("EGODATA_VIDEO_QC_BLACK_DARK_RATIO",
                       DEFAULT_BLACK_DARK_RATIO, minimum=0.0),
        )
        black_min_sec = _float_env("EGODATA_VIDEO_QC_BLACK_MIN_SEC",
                                  DEFAULT_BLACK_MIN_SEC, minimum=0.0)
        freeze_min_sec = _float_env("EGODATA_VIDEO_QC_FREEZE_MIN_SEC",
                                    DEFAULT_FREEZE_MIN_SEC, minimum=0.5)
        freeze_diff = _float_env("EGODATA_VIDEO_QC_FREEZE_DIFF",
                                 DEFAULT_FREEZE_DIFF, minimum=0.0)
        previous_small = None
        previous_index = None
        freeze_start = None
        freeze_last = None
        black_start = None
        black_last = None

        for index in indices:
            # Seeking to sparse positions avoids a full decode pass.  Each
            # seek is still followed by read(), so corrupt samples are caught.
            cap.set(cv2.CAP_PROP_POS_FRAMES, index)
            ok, frame = cap.read()
            if not ok or frame is None:
                report["decode_error_count"] += 1
                if len(report["decode_errors"]) < 20:
                    report["decode_errors"].append(index)
                previous_small = None
                previous_index = None
                freeze_start = None
                freeze_last = None
                black_start = None
                black_last = None
                continue

            report["sampled_frames"] += 1
            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            dark_ratio = float(np.mean(gray <= black_mean))
            gray_mean = float(np.mean(gray))
            if gray_mean <= black_mean and dark_ratio >= black_dark_ratio:
                report["black_frame_count"] += 1
                if black_start is None:
                    black_start = index
                black_last = index
                if len(report["black_frames"]) < 20:
                    report["black_frames"].append({
                        "frame": index,
                        "time_sec": round(index / max(1.0, fps), 3),
                        "mean": round(gray_mean, 3),
                        "dark_ratio": round(dark_ratio, 6),
                    })
            elif black_start is not None:
                if ((black_last - black_start) / max(1.0, fps)
                        >= black_min_sec):
                    _add_freeze_range(report["black_ranges"],
                                      black_start, black_last, fps)
                black_start = None
                black_last = None

            small = cv2.resize(gray, (32, 18), interpolation=cv2.INTER_AREA)
            small = small.astype(np.float32)
            if previous_small is not None and previous_index is not None:
                difference = float(np.mean(np.abs(small - previous_small)))
                if difference <= freeze_diff:
                    if freeze_start is None:
                        freeze_start = previous_index
                    freeze_last = index
                    duration = (index - freeze_start) / max(1.0, fps)
                    if duration >= freeze_min_sec:
                        _add_freeze_range(report["freeze_ranges"],
                                          freeze_start, freeze_last, fps)
                else:
                    freeze_start = None
                    freeze_last = None
            previous_small = small
            previous_index = index

        if black_start is not None and ((black_last - black_start)
                                        / max(1.0, fps) >= black_min_sec):
            _add_freeze_range(report["black_ranges"],
                              black_start, black_last, fps)

        expected = max(0, int(expected_frames or 0))
        if expected > 0:
            tolerance = max(
                DEFAULT_FRAME_TOLERANCE_MIN,
                int(round(expected * _float_env(
                    "EGODATA_VIDEO_QC_FRAME_TOLERANCE_RATIO",
                    DEFAULT_FRAME_TOLERANCE_RATIO,
                    minimum=0.0,
                ))),
            )
            report["count_tolerance"] = tolerance
            deficit = max(0, expected - frame_count)
            report["dropped_frame_estimate"] = deficit
            report["frame_count_mismatch"] = bool(deficit > tolerance)
        else:
            report["frame_count_mismatch"] = False

        reasons: list[str] = []
        if report["decode_error_count"]:
            reasons.append("decode_error")
        if report["black_ranges"]:
            reasons.append("black_screen")
        if report["freeze_ranges"]:
            reasons.append("freeze_suspected")
        if report.get("frame_count_mismatch"):
            reasons.append("frame_count_short")
        report["passed"] = not reasons
        report["reason"] = "passed" if report["passed"] else ",".join(reasons)
        report["policy"] = {
            "black_mean": black_mean,
            "black_dark_ratio": black_dark_ratio,
            "black_min_sec": black_min_sec,
            "freeze_min_sec": freeze_min_sec,
            "freeze_diff": freeze_diff,
        }
        return report
    except Exception as exc:
        report["reason"] = "video_check_failed"
        report["error"] = str(exc)[:240]
        return report
    finally:
        cap.release()


def check_video_quality(batch_dir: str | Path, expected_frames: int = 0,
                        expected_fps: float = 0.0) -> dict:
    """Run the bounded post-processing video quality check for a batch."""
    root = Path(batch_dir)
    try:
        videos = _raw_videos(root)
    except Exception as exc:
        return {
            "passed": False,
            "reason": "video_scan_failed",
            "error": str(exc)[:240],
            "streams": [],
            "expected_frame_count": max(0, int(expected_frames or 0)),
        }
    if not videos:
        return {
            "passed": False,
            "reason": "no_input_video",
            "streams": [],
            "expected_frame_count": max(0, int(expected_frames or 0)),
        }

    streams = [_check_stream(path, expected_frames, expected_fps)
               for path in videos]
    passed = all(bool(stream.get("passed")) for stream in streams)
    return {
        "passed": passed,
        "reason": "passed" if passed else "video_quality_failed",
        "streams": streams,
        "stream_count": len(streams),
        "expected_frame_count": max(0, int(expected_frames or 0)),
        "expected_fps": round(max(0.0, float(expected_fps or 0.0)), 4),
    }


async def check_video_quality_async(batch_dir: str | Path,
                                   expected_frames: int = 0,
                                   expected_fps: float = 0.0) -> dict:
    """Serialize bounded media scans so multiple completions do not flood IO."""
    async with _VIDEO_QC_SEMAPHORE:
        return await asyncio.to_thread(
            check_video_quality, batch_dir, expected_frames, expected_fps)


def _apply_video_quality_result(episode_id: str, report: dict) -> None:
    """Persist the result while preserving the two-state review UI."""
    from app.localstore import mutate_episode_state
    from app.processing.cleaning.store import is_approval_blocked

    def _apply(state: dict) -> None:
        state["video_quality_status"] = "passed" if report.get("passed") else "failed"
        state["video_quality_report"] = report
        current = str(state.get("status") or "")
        if report.get("passed"):
            # 视频干净 ≠ 数据干净。数据质检判 FAIL/ERROR 的批次留了粘性标记，
            # 这里必须让路 —— 否则质检刚推回 to_review，又被本函数置成 reviewed。
            if (current in {"to_review", "completed", "reviewed"}
                    and not is_approval_blocked(state)):
                state["status"] = "reviewed"
                if not state.get("approved_at"):
                    state["approved_at"] = _utcnow()
        elif current in {"processing", "completed", "reviewed", "approved", "failed"}:
            state["status"] = "to_review"
            state["approved_at"] = None

    # 锁内 RMW：本函数会改 ``status``，而数据质检任务在同一个 run 完成后并发
    # 写同一份 state。分开 read/write 的话，质检那边写回的旧快照会把这里的
    # to_review 冲掉 —— 失败批次看上去像已通过。
    mutate_episode_state(episode_id, _apply)


def _utcnow() -> str:
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).isoformat()


async def run_video_quality_review(episode_id: str, batch_dir: str | Path,
                                  expected_frames: int = 0,
                                  expected_fps: float = 0.0) -> dict:
    """Run media checks off the event loop and persist the review decision."""
    try:
        report = await check_video_quality_async(
            batch_dir, expected_frames, expected_fps)
    except Exception as exc:
        # A background quality task must never become an unhandled asyncio
        # exception and accidentally leave an old Approved state in place.
        report = {
            "passed": False,
            "reason": "video_check_failed",
            "error": str(exc)[:240],
            "streams": [],
        }
    _apply_video_quality_result(episode_id, report)
    return report