"""Индексатор архива: периодически сканирует /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: """Время начала сегмента из имени /.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]]: """Возвращает {имя_папки: [пути mp4, отсортированы]} (папка = номер слота или легаси-id камеры — сопоставление в камеру делает вызывающий). Файлы лежат в подпапке даты //*.mp4; старые «плоские» /*.mp4 тоже подхватываем (обратная совместимость). Сортировка по полному пути == хронологическая (ISO-даты и в каталоге, и в имени).""" result: dict[str, list[str]] = {} if not os.path.isdir(root): return result for folder in os.listdir(root): cam_dir = os.path.join(root, folder) 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[folder] = 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 sub = self.config.archive_dir(cam.id) # папка-слот, напр. «05» for d in days: try: os.makedirs(os.path.join(store.path, sub, 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_dir = _scan_segments(root) is_active = (root == active_path) for folder, files in by_dir.items(): # Сопоставление папки с камерой по приоритету: # 1) точное совпадение с id камеры — легаси-папка по id (в т.ч. числовому), # иначе папка камеры с id «5» спуталась бы со слотом 5; # 2) папка-слот «NN» → камера в этом слоте (новая схема); # 3) иначе оставляем имя папки (осиротевший архив индексируется как раньше). cam = self.config.camera(folder) or self.config.camera_by_archive_dir(folder) cam_id = cam.id if cam else folder # последний файл ещё пишется только в активной папке-слоте ТЕКУЩЕЙ камеры; # легаси-папки и неактивное хранилище дописаны — индексируем целиком. being_written = ( is_active and cam is not None and self.config.archive_dir(cam.id) == folder ) complete = files[:-1] if being_written 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