CoRE.Vision/backend/app/services/go2rtc.py

"""Синхронизация потоков с go2rtc + рестрим основного потока.

Камеры Hikvision ограничивают основной поток (нельзя открыть его дважды:
запись + live дают RTSP 453). Решение: go2rtc держит ОДНО соединение с камерой
и раздаёт нескольким потребителям. И запись, и live основного потока читают
из go2rtc (rtsp-рестрим на :8554), поэтому камера видит одну сессию.

go2rtc-потоки на камеру:
  <cam>_main — основной поток (101), единственный источник записи и live-HD.
  <cam>      — субпоток (102) для живой сетки; если субпотока нет — ссылка на
               <cam>_main (повторное использование, без второго коннекта к камере).
  <cam>_h264 — субпоток, перекодированный в H.264 (для WebRTC/браузеров без HEVC).
"""
from __future__ import annotations

import asyncio
import json
import logging
import os
import urllib.parse
import urllib.request
from urllib.parse import urlparse

import yaml

from ..config import Camera, Config

log = logging.getLogger("nvr.go2rtc")

GO2RTC_YAML = os.environ.get("GO2RTC_CONFIG", "/app/go2rtc.yaml")
RECONCILE_INTERVAL = 30  # сек — период сверки рантайма go2rtc с конфигом


def _rtsp_host(config: Config) -> str:
    return urlparse(config.live.go2rtc_url).hostname or "127.0.0.1"


def restream_main_url(config: Config, cam_id: str) -> str:
    """RTSP-адрес основного потока камеры через go2rtc (для записи и live-HD)."""
    return f"rtsp://{_rtsp_host(config)}:8554/{cam_id}_main"


def restream_sub_url(config: Config, cam_id: str) -> str:
    """RTSP-адрес живого (суб)потока через go2rtc (для транскода сабпотока)."""
    return f"rtsp://{_rtsp_host(config)}:8554/{cam_id}"


def _streams_for(cam: Camera) -> dict[str, str]:
    """Имя go2rtc-потока -> источник, для одной камеры."""
    out = {f"{cam.id}_main": cam.rtsp_url(cam.main_path)}
    if cam.sub_path:
        out[cam.id] = cam.rtsp_url(cam.sub_path)
    else:
        # нет субпотока — берём основной из go2rtc, не открывая второй коннект к камере
        out[cam.id] = f"rtsp://127.0.0.1:8554/{cam.id}_main"
    out[f"{cam.id}_h264"] = f"ffmpeg:{cam.id}#video=h264#audio=copy"
    return out


def build_streams(config: Config) -> dict:
    streams: dict = {}
    for c in config.cameras:
        if not c.enabled:        # выключенная камера — без потоков (ни live, ни записи)
            continue
        for name, src in _streams_for(c).items():
            streams[name] = [src]
    return streams


def write_go2rtc_yaml(config: Config) -> None:
    data = {
        "api": {"listen": ":1984"},
        "rtsp": {"listen": ":8554"},
        "webrtc": {"listen": ":8555"},
        "log": {"level": "info"},
        "streams": build_streams(config),
    }
    try:
        with open(GO2RTC_YAML, "w", encoding="utf-8") as fh:
            yaml.safe_dump(data, fh, allow_unicode=True, sort_keys=False)
    except OSError as exc:
        log.warning("не удалось записать go2rtc.yaml: %s", exc)


def _http(method: str, url: str) -> int:
    req = urllib.request.Request(url, method=method)
    with urllib.request.urlopen(req, timeout=5) as resp:
        return resp.status


async def _call(method: str, url: str) -> None:
    try:
        await asyncio.to_thread(_http, method, url)
    except Exception as exc:  # noqa: BLE001 — go2rtc может быть недоступен/вернуть 4xx при ro-конфиге
        log.debug("go2rtc API %s -> %s", method, exc)


def _put(base: str, name: str, src: str) -> str:
    return f"{base}/api/streams?" + urllib.parse.urlencode({"name": name, "src": src})


def _del(base: str, name: str) -> str:
    return f"{base}/api/streams?" + urllib.parse.urlencode({"src": name})


async def add_camera_streams(config: Config, cam: Camera) -> None:
    base = config.live.go2rtc_url.rstrip("/")
    for name, src in _streams_for(cam).items():
        await _call("PUT", _put(base, name, src))
    write_go2rtc_yaml(config)


async def remove_camera_streams(config: Config, cam_id: str) -> None:
    base = config.live.go2rtc_url.rstrip("/")
    for name in (f"{cam_id}_main", cam_id, f"{cam_id}_h264"):
        await _call("DELETE", _del(base, name))
    write_go2rtc_yaml(config)


async def sync_all(config: Config) -> None:
    """При старте: записать конфиг и добавить все потоки в go2rtc через API."""
    write_go2rtc_yaml(config)
    base = config.live.go2rtc_url.rstrip("/")
    for cam in config.cameras:
        if not cam.enabled:
            continue
        for name, src in _streams_for(cam).items():
            await _call("PUT", _put(base, name, src))


def _get_stream_names(base: str) -> set[str]:
    """Имена потоков, реально загруженных в рантайм go2rtc (/api/streams)."""
    req = urllib.request.Request(f"{base}/api/streams", method="GET")
    with urllib.request.urlopen(req, timeout=5) as resp:
        data = json.loads(resp.read().decode("utf-8") or "{}")
    return set(data.keys())


async def reconcile(config: Config) -> int:
    """Дотягивает в go2rtc недостающие потоки (самовосстановление).

    go2rtc читает go2rtc.yaml только при старте, а рантайм-таблицу держит через API.
    Если её опустошить (DELETE при чистке камер, перезапуск go2rtc со старым/пустым
    файлом, молча провалившийся стартовый sync_all) — рестрим начинает отдавать 404,
    и live, и запись «слепнут». Здесь сверяем желаемый набор с /api/streams и PUT-им
    ТОЛЬКО отсутствующие — существующие не трогаем, чтобы не рвать активных
    producer'ов/consumer'ов. Возвращает число восстановленных потоков."""
    if not config.cameras:
        return 0
    base = config.live.go2rtc_url.rstrip("/")
    try:
        existing = await asyncio.to_thread(_get_stream_names, base)
    except Exception as exc:  # noqa: BLE001 — go2rtc может быть недоступен/перезапускаться
        log.debug("go2rtc reconcile: /api/streams недоступен: %s", exc)
        return 0
    added = 0
    for cam in config.cameras:
        if not cam.enabled:        # выключенные камеры в go2rtc не держим
            continue
        for name, src in _streams_for(cam).items():
            if name not in existing:
                await _call("PUT", _put(base, name, src))
                added += 1
    if added:
        log.info("go2rtc reconcile: восстановлено потоков %d", added)
        write_go2rtc_yaml(config)  # и файл приводим в соответствие — для будущего рестарта
    return added


class Reconciler:
    """Фоновая сверка потоков go2rtc с конфигом каждые RECONCILE_INTERVAL секунд.

    Держит ссылку на тот же объект Config, что и остальной рантайм (его .cameras
    мутируется in-place при add/clear), поэтому всегда видит актуальный список."""

    def __init__(self, config: Config, interval: int = RECONCILE_INTERVAL):
        self.config = config
        self.interval = interval
        self._task: asyncio.Task | None = None
        self._stopping = False

    def start(self) -> None:
        self._stopping = False
        self._task = asyncio.create_task(self._loop(), name="go2rtc-reconcile")

    async def stop(self) -> None:
        self._stopping = True
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass

    async def _loop(self) -> None:
        while not self._stopping:
            await asyncio.sleep(self.interval)
            try:
                await reconcile(self.config)
            except Exception:  # noqa: BLE001
                log.exception("go2rtc reconcile failed")