"""单路 RTSP 拉流 worker：负责打开、重连、读帧分发。

从 ``CameraSessionManager._camera_worker`` 抽出，保持同样的行为：
- 打开失败 → 退避 → 重试。
- 连续读帧失败达到阈值 → 释放连接 → 退避 → 重试。
- 读到可用帧后交给上游 ``frame_handler``，由其决定是否推理 / 跳帧。

不知道手术会话、推理结果或数据库。日志中出现 RTSP URL 时会脱敏 user:password。
"""

from __future__ import annotations

import asyncio
import re
from typing import Awaitable, Callable

from loguru import logger

from app.config import Settings
from app.services.video.rtsp_capture import RtspCapture


FrameHandler = Callable[[object], Awaitable[None]]

_RTSP_CRED_RE = re.compile(r"(?P<scheme>rtsp://)(?P<userinfo>[^@/\s]+@)")


def redact_rtsp_url(url: str | None) -> str:
    """把 ``rtsp://user:pwd@host/...`` 脱敏为 ``rtsp://***@host/...``。"""
    if not url:
        return ""
    return _RTSP_CRED_RE.sub(r"\g<scheme>***@", url)


class CameraStreamWorker:
    """以 async 循环封装单路 RTSP 的重连/读帧，交由 handler 处理帧。"""

    def __init__(
        self,
        *,
        settings: Settings,
        surgery_id: str,
        camera_id: str,
        url: str,
    ) -> None:
        self._s = settings
        self._surgery_id = surgery_id
        self._camera_id = camera_id
        self._url = url

    async def run(
        self,
        *,
        stream_ready: asyncio.Event,
        stop_event: asyncio.Event,
        frame_handler: FrameHandler,
    ) -> None:
        cap: RtspCapture | None = None
        consecutive_failures = 0
        first_ready = True
        safe_url = redact_rtsp_url(self._url)

        try:
            while not stop_event.is_set():
                if cap is None:
                    try:
                        cap = RtspCapture(
                            self._url, open_timeout_sec=self._s.video_open_timeout_sec
                        )
                        await asyncio.to_thread(cap.open)
                        consecutive_failures = 0
                        if first_ready:
                            stream_ready.set()
                            first_ready = False
                        logger.info(
                            "RTSP stream opened camera={} surgery={} url={}",
                            self._camera_id,
                            self._surgery_id,
                            safe_url,
                        )
                    except Exception as exc:
                        logger.warning(
                            "RTSP open failed camera={} surgery={} url={}: {}",
                            self._camera_id,
                            self._surgery_id,
                            safe_url,
                            exc,
                        )
                        if cap is not None:
                            await asyncio.to_thread(cap.release)
                            cap = None
                        await asyncio.sleep(self._s.video_reconnect_backoff_seconds)
                        continue

                ok, frame = await asyncio.to_thread(cap.read)
                if not ok or frame is None:
                    consecutive_failures += 1
                    if (
                        consecutive_failures
                        >= self._s.video_read_failure_reconnect_threshold
                    ):
                        logger.warning(
                            "RTSP reconnect camera={} surgery={} url={} after {} read failures",
                            self._camera_id,
                            self._surgery_id,
                            safe_url,
                            consecutive_failures,
                        )
                        await asyncio.to_thread(cap.release)
                        cap = None
                        consecutive_failures = 0
                        await asyncio.sleep(self._s.video_reconnect_backoff_seconds)
                    else:
                        await asyncio.sleep(0.05)
                    continue

                consecutive_failures = 0
                await frame_handler(frame)
        finally:
            if cap is not None:
                await asyncio.to_thread(cap.release)