prvni commit

.env.example

@@ -0,0 +1,34 @@
+# Připojení k Hikvision NVR
+NVR_PROTOCOL=https
+NVR_HOST=192.168.1.100
+NVR_PORT=443
+NVR_USERNAME=webcam
+NVR_PASSWORD=zmente-me
+
+# U interního NVR se self-signed certifikátem nastavte false.
+NVR_VERIFY_TLS=false
+NVR_TIMEOUT=15
+
+# Kanál 1 + stream 01 vytvoří ISAPI identifikátor 101.
+NVR_CHANNEL=1
+NVR_STREAM_ID=01
+# Požadované rozlišení výsledného snapshotu.
+NVR_SNAPSHOT_WIDTH=1920
+NVR_SNAPSHOT_HEIGHT=1080
+OUTPUT_PATH=public/webcam.jpg
+
+# Alternativa pro více obrázků (pak se tři řádky výše ignorují):
+# SNAPSHOTS=[{"channel":1,"stream_id":"01","output":"public/vjezd.jpg"},{"channel":2,"stream_id":"01","output":"public/parkoviste.jpg"}]
+
+# Společné denní časové okno. Shodné časy znamenají celý den.
+# Funguje také interval přes půlnoc, např. 20:00 až 06:00.
+TIMEZONE=Europe/Prague
+ACTIVE_FROM=06:00
+ACTIVE_TO=20:00
+
+# Mimo časové okno se tento soubor atomicky zkopíruje do každého OUTPUT_PATH.
+DUMMY_PATH=dummy.jpg
+
+# Endpoint lze změnit, pokud firmware používá jinou cestu.
+NVR_SNAPSHOT_PATH=ISAPI/Streaming/channels/{channel}{stream_id}/picture
+LOG_LEVEL=INFO

.gitignore

@@ -0,0 +1,4 @@
+.env
+*.log
+__pycache__/
+*.py[cod]

README.md

@@ -0,0 +1,59 @@
+# Hikvision NVR webcam snapshot
+
+Jednoduchý skript stáhne aktuální obrázek z kanálu Hikvision NVR a atomicky jej
+publikuje do zadaného souboru na webu. Mimo společné časové okno publikuje
+připravený dummy obrázek. Podporuje jeden i více kanálů.
+
+## Instalace
+
+Požadavek je Python 3.9 nebo novější, bez externích balíčků.
+
+1. Zkopírujte `.env.example` jako `.env`.
+2. Doplňte adresu NVR, samostatného uživatele pouze s potřebným oprávněním,
+   heslo, kanál a cílovou cestu.
+3. Vedle skriptu uložte vlastní `dummy.jpg` (nebo změňte `DUMMY_PATH`).
+4. Jednou skript spusťte ručně:
+
+   ```sh
+   python3 snapshot.py
+   ```
+
+Heslo v `.env` může obsahovat mezery a `#`; uvozovky jsou volitelné. Soubor
+`.env` nepatří do verzovacího systému.
+
+## Pravidelné spouštění
+
+Skript sám čekání ani plánování neprovádí. Na Linuxu jej lze volat každou minutu:
+
+```cron
+* * * * * cd /cesta/k/hik2webcam && /usr/bin/python3 snapshot.py >> snapshot.log 2>&1
+```
+
+Ve Windows použijte Plánovač úloh a jako program nastavte `python.exe`, argument
+`C:\cesta\snapshot.py` a pracovní adresář projektu.
+
+Shodné hodnoty `ACTIVE_FROM` a `ACTIVE_TO` znamenají nepřetržitý provoz.
+Interval může přecházet přes půlnoc. Při chybě NVR zůstane poslední platný
+obrázek zachován a proces skončí kódem `1`; mimo časové okno skončí po úspěšném
+zkopírování dummy obrázku kódem `0`.
+
+## Hikvision endpoint
+
+Výchozí cesta je:
+
+```text
+/ISAPI/Streaming/channels/{channel}{stream_id}/picture
+```
+
+Kanál `1` a stream `01` tedy používají ID `101`. Autentizace zobrazená v NVR
+jako `digest` je podporovaná; skript umí jako zálohu také Basic autentizaci.
+Pro HTTPS se self-signed certifikátem lze použít `NVR_VERIFY_TLS=false`.
+
+Požadované rozlišení se posílá NVR explicitně pomocí ISAPI parametrů
+`videoResolutionWidth` a `videoResolutionHeight`. Nastavuje se globálně přes
+`NVR_SNAPSHOT_WIDTH` a `NVR_SNAPSHOT_HEIGHT`; výchozí hodnota je 1920×1080.
+NVR musí dané rozlišení podporovat, jinak může vrátit nejbližší podporované
+rozlišení nebo chybovou XML odpověď.
+
+Pokud zařízení endpoint odmítne, ověřte, že uživatel smí sledovat daný kanál,
+že je povolené ISAPI a že `NVR_SNAPSHOT_PATH` odpovídá konkrétnímu firmwaru.

snapshot.py

@@ -0,0 +1,278 @@
+#!/usr/bin/env python3
+"""Download current images from a Hikvision NVR or publish a placeholder."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import shutil
+import ssl
+import sys
+import tempfile
+from dataclasses import dataclass
+from datetime import datetime, time
+from pathlib import Path
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode, urljoin
+from urllib.request import (
+    HTTPBasicAuthHandler,
+    HTTPDigestAuthHandler,
+    HTTPSHandler,
+    HTTPPasswordMgrWithDefaultRealm,
+    Request,
+    build_opener,
+)
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+
+LOG = logging.getLogger("nvr-snapshot")
+
+
+class ConfigError(ValueError):
+    pass
+
+
+@dataclass(frozen=True)
+class Snapshot:
+    channel: int
+    output: Path
+    stream_id: str = "01"
+
+
+def load_dotenv(path: Path) -> None:
+    """Load a small, conventional subset of .env syntax without dependencies."""
+    if not path.exists():
+        return
+    for number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
+        line = raw_line.strip()
+        if not line or line.startswith("#"):
+            continue
+        if line.startswith("export "):
+            line = line[7:].lstrip()
+        if "=" not in line:
+            raise ConfigError(f"{path}:{number}: očekávám KEY=VALUE")
+        key, value = line.split("=", 1)
+        key, value = key.strip(), value.strip()
+        if not key:
+            raise ConfigError(f"{path}:{number}: prázdný název proměnné")
+        if len(value) >= 2 and value[0] == value[-1] and value[0] in "'\"":
+            value = value[1:-1]
+        os.environ.setdefault(key, value)
+
+
+def env(name: str, default: str | None = None, *, required: bool = False) -> str:
+    value = os.environ.get(name, default)
+    if required and not value:
+        raise ConfigError(f"Chybí povinná proměnná {name}")
+    assert value is not None
+    return value
+
+
+def env_bool(name: str, default: bool) -> bool:
+    raw = env(name, "true" if default else "false").lower()
+    if raw in {"1", "true", "yes", "on"}:
+        return True
+    if raw in {"0", "false", "no", "off"}:
+        return False
+    raise ConfigError(f"{name} musí být true/false")
+
+
+def parse_clock(value: str, name: str) -> time:
+    try:
+        return time.fromisoformat(value)
+    except ValueError as exc:
+        raise ConfigError(f"{name} musí být ve formátu HH:MM nebo HH:MM:SS") from exc
+
+
+def is_active(now: datetime, start: time, end: time) -> bool:
+    current = now.timetz().replace(tzinfo=None)
+    if start == end:
+        return True
+    if start < end:
+        return start <= current < end
+    return current >= start or current < end
+
+
+def snapshots_from_env(base_dir: Path) -> list[Snapshot]:
+    raw = os.environ.get("SNAPSHOTS")
+    if raw:
+        try:
+            items = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise ConfigError(f"SNAPSHOTS není platné JSON: {exc}") from exc
+        if not isinstance(items, list) or not items:
+            raise ConfigError("SNAPSHOTS musí být neprázdné JSON pole")
+    else:
+        items = [{
+            "channel": int(env("NVR_CHANNEL", "1")),
+            "stream_id": env("NVR_STREAM_ID", "01"),
+            "output": env("OUTPUT_PATH", required=True),
+        }]
+
+    result = []
+    for index, item in enumerate(items):
+        if not isinstance(item, dict):
+            raise ConfigError(f"SNAPSHOTS[{index}] musí být objekt")
+        try:
+            channel = int(item["channel"])
+            output = Path(str(item["output"]))
+            stream_id = str(item.get("stream_id", "01"))
+        except (KeyError, TypeError, ValueError) as exc:
+            raise ConfigError(
+                f"SNAPSHOTS[{index}] vyžaduje channel (číslo) a output"
+            ) from exc
+        if channel < 1 or not stream_id.isdigit():
+            raise ConfigError(f"Neplatný kanál/stream v SNAPSHOTS[{index}]")
+        if not output.is_absolute():
+            output = base_dir / output
+        result.append(Snapshot(channel, output.resolve(), stream_id))
+    return result
+
+
+def atomic_copy(source: Path, destination: Path) -> None:
+    if not source.is_file():
+        raise ConfigError(f"Dummy soubor neexistuje: {source}")
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    fd, temp_name = tempfile.mkstemp(
+        prefix=f".{destination.name}.", suffix=".tmp", dir=destination.parent
+    )
+    os.close(fd)
+    temp = Path(temp_name)
+    try:
+        shutil.copyfile(source, temp)
+        os.replace(temp, destination)
+    finally:
+        temp.unlink(missing_ok=True)
+
+
+def atomic_write(data: bytes, destination: Path) -> None:
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    fd, temp_name = tempfile.mkstemp(
+        prefix=f".{destination.name}.", suffix=".tmp", dir=destination.parent
+    )
+    temp = Path(temp_name)
+    try:
+        with os.fdopen(fd, "wb") as handle:
+            handle.write(data)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temp, destination)
+    finally:
+        temp.unlink(missing_ok=True)
+
+
+def make_opener(base_url: str, username: str, password: str, verify_tls: bool):
+    manager = HTTPPasswordMgrWithDefaultRealm()
+    manager.add_password(None, base_url, username, password)
+    context = ssl.create_default_context()
+    if not verify_tls:
+        context.check_hostname = False
+        context.verify_mode = ssl.CERT_NONE
+    return build_opener(
+        HTTPDigestAuthHandler(manager),
+        HTTPBasicAuthHandler(manager),
+        HTTPSHandler(context=context),
+    )
+
+
+def validate_image(data: bytes, content_type: str) -> None:
+    signatures = (
+        data.startswith(b"\xff\xd8\xff"),
+        data.startswith(b"\x89PNG\r\n\x1a\n"),
+        data.startswith((b"GIF87a", b"GIF89a")),
+        data.startswith(b"BM"),
+    )
+    if not data or not any(signatures):
+        detail = data[:160].decode("utf-8", errors="replace").replace("\n", " ")
+        raise RuntimeError(
+            f"NVR nevrátil podporovaný obrázek (Content-Type {content_type!r}): {detail}"
+        )
+
+
+def run() -> int:
+    script_dir = Path(__file__).resolve().parent
+    dotenv_path = Path(os.environ.get("ENV_FILE", script_dir / ".env"))
+    load_dotenv(dotenv_path)
+
+    log_level = env("LOG_LEVEL", "INFO").upper()
+    logging.basicConfig(
+        level=getattr(logging, log_level, logging.INFO),
+        format="%(asctime)s %(levelname)s %(message)s",
+    )
+
+    protocol = env("NVR_PROTOCOL", "https").lower()
+    if protocol not in {"http", "https"}:
+        raise ConfigError("NVR_PROTOCOL musí být http nebo https")
+    host = env("NVR_HOST", required=True)
+    port = int(env("NVR_PORT", "443" if protocol == "https" else "80"))
+    base_url = f"{protocol}://{host}:{port}/"
+    username = env("NVR_USERNAME", required=True)
+    password = env("NVR_PASSWORD", required=True)
+    timeout = float(env("NVR_TIMEOUT", "15"))
+    verify_tls = env_bool("NVR_VERIFY_TLS", True)
+    path_template = env(
+        "NVR_SNAPSHOT_PATH", "ISAPI/Streaming/channels/{channel}{stream_id}/picture"
+    )
+    snapshot_width = int(env("NVR_SNAPSHOT_WIDTH", "1920"))
+    snapshot_height = int(env("NVR_SNAPSHOT_HEIGHT", "1080"))
+    if snapshot_width < 1 or snapshot_height < 1:
+        raise ConfigError("NVR_SNAPSHOT_WIDTH a NVR_SNAPSHOT_HEIGHT musí být kladné")
+
+    try:
+        timezone = ZoneInfo(env("TIMEZONE", "Europe/Prague"))
+    except ZoneInfoNotFoundError as exc:
+        raise ConfigError("Neznámá TIMEZONE; použijte např. Europe/Prague") from exc
+    start = parse_clock(env("ACTIVE_FROM", "00:00"), "ACTIVE_FROM")
+    end = parse_clock(env("ACTIVE_TO", "00:00"), "ACTIVE_TO")
+    snapshots = snapshots_from_env(script_dir)
+    now = datetime.now(timezone)
+
+    dummy = Path(env("DUMMY_PATH", str(script_dir / "dummy.jpg")))
+    if not dummy.is_absolute():
+        dummy = script_dir / dummy
+
+    if not is_active(now, start, end):
+        for snapshot in snapshots:
+            atomic_copy(dummy, snapshot.output)
+            LOG.info("Mimo časové okno: dummy -> %s", snapshot.output)
+        return 0
+
+    opener = make_opener(base_url, username, password, verify_tls)
+    failures = 0
+    for snapshot in snapshots:
+        relative_path = path_template.format(
+            channel=snapshot.channel, stream_id=snapshot.stream_id
+        ).lstrip("/")
+        url = urljoin(base_url, relative_path)
+        query = urlencode({
+            "videoResolutionWidth": snapshot_width,
+            "videoResolutionHeight": snapshot_height,
+            "snapShotImageType": "JPEG",
+        })
+        url = f"{url}{'&' if '?' in url else '?'}{query}"
+        request = Request(url, headers={"Accept": "image/jpeg,image/*;q=0.9"})
+        try:
+            with opener.open(request, timeout=timeout) as response:
+                data = response.read()
+                content_type = response.headers.get("Content-Type", "")
+            validate_image(data, content_type)
+            atomic_write(data, snapshot.output)
+            LOG.info("Staženo %s (%d B) -> %s", url, len(data), snapshot.output)
+        except (HTTPError, URLError, OSError, RuntimeError) as exc:
+            failures += 1
+            LOG.error("Stažení %s selhalo: %s", url, exc)
+    return 1 if failures else 0
+
+
+def main() -> None:
+    try:
+        raise SystemExit(run())
+    except (ConfigError, ValueError) as exc:
+        logging.basicConfig(level=logging.ERROR, format="%(levelname)s %(message)s")
+        LOG.error("Chyba konfigurace: %s", exc)
+        raise SystemExit(2)
+
+
+if __name__ == "__main__":
+    main()

tests/test_snapshot.py

@@ -0,0 +1,39 @@
+import tempfile
+import unittest
+from datetime import datetime, time
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import snapshot
+
+
+class ScheduleTests(unittest.TestCase):
+    def test_day_window(self):
+        now = datetime(2026, 7, 30, 12, 0, tzinfo=ZoneInfo("Europe/Prague"))
+        self.assertTrue(snapshot.is_active(now, time(6), time(20)))
+        self.assertFalse(snapshot.is_active(now, time(13), time(20)))
+
+    def test_overnight_window(self):
+        late = datetime(2026, 7, 30, 23, 0, tzinfo=ZoneInfo("Europe/Prague"))
+        noon = datetime(2026, 7, 30, 12, 0, tzinfo=ZoneInfo("Europe/Prague"))
+        self.assertTrue(snapshot.is_active(late, time(20), time(6)))
+        self.assertFalse(snapshot.is_active(noon, time(20), time(6)))
+
+    def test_equal_times_mean_all_day(self):
+        now = datetime(2026, 7, 30, 12, 0, tzinfo=ZoneInfo("Europe/Prague"))
+        self.assertTrue(snapshot.is_active(now, time(0), time(0)))
+
+
+class FileTests(unittest.TestCase):
+    def test_atomic_copy_replaces_destination(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "dummy.jpg"
+            destination = root / "web" / "camera.jpg"
+            source.write_bytes(b"dummy")
+            snapshot.atomic_copy(source, destination)
+            self.assertEqual(destination.read_bytes(), b"dummy")
+
+
+if __name__ == "__main__":
+    unittest.main()