CoRE.Vision/backend/app/recorder/indexer.py

"""Индексатор архива: периодически сканирует /records, заносит новые сегменты в БД.

Логика: самый свежий сегмент каждой камеры ещё пишется ffmpeg-ом — его пропускаем.
Все остальные mp4, которых нет в БД, пробуем через ffprobe и добавляем.
Параллельно убираем из БД записи о пропавших файлах.
"""
from __future__ import annotations

import asyncio
import logging
import os
import time
from datetime import datetime, timedelta

from ..config import Config
from ..db import Database
from .ffmpeg import probe_duration

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

SCAN_INTERVAL = 30  # сек


def _parse_started_at(path: str) -> int:
    """Время начала сегмента из имени <cam>/<YYYY-MM-DD_HH-MM-SS>.mp4."""
    try:
        stem = os.path.splitext(os.path.basename(path))[0]
        dt = datetime.strptime(stem, "%Y-%m-%d_%H-%M-%S")
        return int(dt.timestamp())
    except ValueError:
        return int(os.path.getmtime(path))


def _scan_segments(root: str) -> dict[str, list[str]]:
    """Возвращает {camera_id: [пути mp4, отсортированы]}.

    Файлы лежат в подпапке даты <cam>/<YYYY-MM-DD>/*.mp4; старые «плоские»
    <cam>/*.mp4 тоже подхватываем (обратная совместимость). Сортировка по полному
    пути == хронологическая (ISO-даты и в каталоге, и в имени)."""
    result: dict[str, list[str]] = {}
    if not os.path.isdir(root):
        return result
    for cam_id in os.listdir(root):
        cam_dir = os.path.join(root, cam_id)
        if not os.path.isdir(cam_dir):
            continue
        files: list[str] = []
        for entry in os.listdir(cam_dir):
            p = os.path.join(cam_dir, entry)
            if entry.endswith(".mp4") and os.path.isfile(p):
                files.append(p)                       # старый плоский формат
            elif os.path.isdir(p):                    # подпапка даты
                for fn in os.listdir(p):
                    if fn.endswith(".mp4"):
                        files.append(os.path.join(p, fn))
        # сортируем по ИМЕНИ файла (YYYY-MM-DD_HH-MM-SS) — хронологически независимо от того,
        # лежит файл плоско или в подпапке даты (полный путь сортировал бы '/' < '_' неверно,
        # из-за чего пишущийся сейчас файл в подпапке не оказывался бы последним)
        files.sort(key=os.path.basename)
        if files:
            result[cam_id] = files
    return result


class Indexer:
    def __init__(self, config: Config, db: Database):
        self.config = config
        self.db = db
        self._task: asyncio.Task | None = None
        self._stopping = False

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

    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:
            try:
                self._ensure_date_dirs()
                await self.scan_once()
            except Exception:  # noqa: BLE001
                log.exception("indexer scan failed")
            await asyncio.sleep(SCAN_INTERVAL)

    def _ensure_date_dirs(self) -> None:
        """Создаёт каталоги даты (сегодня+завтра) для пишущих камер заранее: ffmpeg
        не умеет -strftime_mkdir, без готового каталога запись встала бы в полночь."""
        today = datetime.now()
        days = [today.strftime("%Y-%m-%d"), (today + timedelta(days=1)).strftime("%Y-%m-%d")]
        for cam in self.config.cameras:
            if not (cam.enabled and cam.record):
                continue
            store = self.config.storage_by_id(cam.storage_id) or self.config.active_storage()
            if store.type == "nas" and not os.path.ismount(store.path):
                continue
            for d in days:
                try:
                    os.makedirs(os.path.join(store.path, cam.id, d), exist_ok=True)
                except OSError:
                    pass

    async def scan_once(self) -> int:
        active_path = self.config.active_storage().path
        known = await self.db.known_paths()
        added = 0

        # сканируем ВСЕ хранилища: после смены активного старые записи не теряются
        for root in self.config.storage_paths():
            by_cam = _scan_segments(root)
            is_active = (root == active_path)
            for cam_id, files in by_cam.items():
                # в активном хранилище последний файл ещё пишется — пропускаем;
                # в неактивных все сегменты завершены, индексируем целиком.
                complete = files[:-1] if is_active else files
                for path in complete:
                    if path in known:
                        continue
                    duration, size = await probe_duration(path)
                    if size is None or size == 0:
                        continue
                    await self.db.add_recording(
                        cam_id, path, _parse_started_at(path), duration, size
                    )
                    added += 1

        # уборка пропавших файлов
        for path in known:
            if not os.path.exists(path):
                await self.db.delete_missing(path)

        if added:
            log.info("indexer: добавлено сегментов %d", added)
        return added