feat: notify camera recordings in Google Chat

Author: gdemontalivet

cloud/README.md

@@ -17,3 +17,9 @@ The service must receive camera values from Secret Manager. Do not put camera
 credentials or an ADC file in the image. The recorder has a default hard limit
 of 30 GB of uploaded video per calendar month, leaving headroom in the 50 GB 4G
 plan for interactive viewing and protocol overhead.
+
+`GCHAT_WEBHOOK_URL` is also injected from Secret Manager. The poller sends one
+alert when an external camera session starts the VM. The recorder sends a final
+alert that always states whether an error occurred, whether video was uploaded,
+the elapsed time, and a private Cloud Console link to the session's one-minute
+segments. Notifications are best-effort and never block recording.

cloud/gchat_notifier.py

@@ -0,0 +1,58 @@
+"""Best-effort Google Chat notifications without leaking webhook details."""
+
+from __future__ import annotations
+
+import logging
+import os
+
+import requests
+
+
+LOG = logging.getLogger(__name__)
+
+
+def format_duration(seconds: float) -> str:
+    """Return a compact French duration suitable for an alert."""
+    total_seconds = max(0, round(seconds))
+    hours, remainder = divmod(total_seconds, 3600)
+    minutes, seconds = divmod(remainder, 60)
+    parts: list[str] = []
+    if hours:
+        parts.append(f"{hours} h")
+    if minutes:
+        parts.append(f"{minutes} min")
+    if seconds or not parts:
+        parts.append(f"{seconds} s")
+    return " ".join(parts)
+
+
+def format_bytes(byte_count: int) -> str:
+    """Return a compact binary-size label."""
+    value = float(max(0, byte_count))
+    for unit in ("o", "Kio", "Mio", "Gio"):
+        if value < 1024 or unit == "Gio":
+            if unit == "o":
+                return f"{int(value)} {unit}"
+            return f"{value:.1f} {unit}"
+        value /= 1024
+    raise AssertionError("unreachable")
+
+
+def send_gchat_alert(message: str) -> bool:
+    """Send a message using the webhook supplied by Secret Manager."""
+    webhook_url = os.environ.get("GCHAT_WEBHOOK_URL", "")
+    if not webhook_url:
+        LOG.warning("Google Chat notification skipped: webhook is not configured")
+        return False
+
+    try:
+        response = requests.post(webhook_url, json={"text": message}, timeout=15)
+        response.raise_for_status()
+        return True
+    except requests.RequestException as exc:
+        # Exception strings can contain the complete webhook URL. Log only its
+        # class so the Google Chat token never reaches Cloud Logging.
+        LOG.error(
+            "Google Chat notification failed error_type=%s", type(exc).__name__
+        )
+        return False

cloud/recorder.py

@@ -9,12 +9,14 @@
 import subprocess
 import time
 import uuid
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 from urllib.parse import quote
 
 from google.cloud import storage
 
+from gchat_notifier import format_bytes, format_duration, send_gchat_alert
 from runtime import Tunnel, required_env
 
 
@@ -23,6 +25,16 @@
 RTSP_USERINFO_RE = re.compile(r"rtsp://[^@\s]+@", re.IGNORECASE)
 
 
+@dataclass
+class RecordingState:
+    bucket_name: str = ""
+    project_id: str = ""
+    object_prefix: str = ""
+    uploaded_segments: int = 0
+    uploaded_bytes: int = 0
+    monthly_limit_reached: bool = False
+
+
 def safe_ffmpeg_diagnostic(stderr: str, limit: int = 2000) -> str:
     """Return a short FFmpeg diagnostic with RTSP credentials removed."""
     sanitized = RTSP_USERINFO_RE.sub("rtsp://[REDACTED]@", stderr)
@@ -41,6 +53,44 @@ def monthly_usage_bytes(client: storage.Client, bucket_name: str, prefix: str) -
     )
 
 
+def storage_console_url(state: RecordingState) -> str:
+    """Build a private Cloud Console link to this recording session."""
+    if not state.bucket_name or not state.object_prefix:
+        return ""
+    path = quote(
+        f"{state.bucket_name}/{state.object_prefix.rstrip('/')}", safe="/"
+    )
+    project = quote(state.project_id, safe="")
+    return f"https://console.cloud.google.com/storage/browser/{path}?project={project}"
+
+
+def recording_summary(
+    state: RecordingState, elapsed_seconds: float, error_name: str | None = None
+) -> str:
+    """Build the structured end-of-session notification requested by the user."""
+    recorded = state.uploaded_segments > 0
+    title = "❌ *Session caméra interrompue*" if error_name else "✅ *Session caméra terminée*"
+    lines = [
+        title,
+        f"Erreur : {'oui (' + error_name + ')' if error_name else 'non'}",
+        f"Vidéo enregistrée : {'oui' if recorded else 'non'}",
+        f"Durée : {format_duration(elapsed_seconds)}",
+    ]
+    link = storage_console_url(state) if recorded else ""
+    if link:
+        lines.append(f"Lien : <{link}|ouvrir les vidéos>")
+    else:
+        lines.append("Lien : aucun")
+    if recorded:
+        lines.append(
+            f"Fichiers : {state.uploaded_segments} segment(s), "
+            f"{format_bytes(state.uploaded_bytes)}"
+        )
+    if state.monthly_limit_reached:
+        lines.append("Limite mensuelle de 30 Go atteinte.")
+    return "\n".join(lines)
+
+
 def capture_segment(rtsp_url: str, destination: Path, duration: int) -> None:
     command = [
         "ffmpeg",
@@ -93,14 +143,12 @@ def capture_segment(rtsp_url: str, destination: Path, duration: int) -> None:
         raise RuntimeError("ffmpeg produced insufficient video data")
 
 
-def main() -> None:
-    signal.signal(signal.SIGTERM, _stop_on_signal)
-    signal.signal(signal.SIGINT, _stop_on_signal)
-
+def record(state: RecordingState) -> None:
     serial = required_env("CAMERA_SERIAL")
     username = required_env("CAMERA_USERNAME")
     password = required_env("CAMERA_PASSWORD")
-    bucket_name = required_env("GCS_BUCKET")
+    state.bucket_name = required_env("GCS_BUCKET")
+    state.project_id = os.environ.get("PROJECT_ID", "calving-monitor-20260819")
     duration_seconds = int(os.environ.get("RECORD_DURATION_SECONDS", "600"))
     segment_seconds = int(os.environ.get("SEGMENT_SECONDS", "60"))
     monthly_limit = int(os.environ.get("MONTHLY_RECORDING_LIMIT_BYTES", "30000000000"))
@@ -111,9 +159,14 @@ def main() -> None:
 
     now = datetime.now(timezone.utc)
     monthly_prefix = f"raw/{now:%Y/%m}/"
+    session_id = f"{now:%Y%m%dT%H%M%SZ}_{uuid.uuid4().hex[:8]}"
+    state.object_prefix = f"raw/{now:%Y/%m/%d}/{session_id}/"
     storage_client = storage.Client()
-    used_bytes = monthly_usage_bytes(storage_client, bucket_name, monthly_prefix)
+    used_bytes = monthly_usage_bytes(
+        storage_client, state.bucket_name, monthly_prefix
+    )
     if used_bytes >= monthly_limit:
+        state.monthly_limit_reached = True
         LOG.warning("Monthly recording limit already reached; recording skipped")
         return
 
@@ -122,10 +175,8 @@ def main() -> None:
     rtsp_url = (
         f"rtsp://{escaped_user}:{escaped_password}@127.0.0.1:{local_port}{stream_path}"
     )
-    bucket = storage_client.bucket(bucket_name)
+    bucket = storage_client.bucket(state.bucket_name)
     deadline = time.monotonic() + duration_seconds
-    uploaded = 0
-
     with Tunnel(
         serial=serial,
         local_port=local_port,
@@ -143,7 +194,7 @@ def main() -> None:
             try:
                 capture_segment(rtsp_url, local_path, current_duration)
                 size = local_path.stat().st_size
-                object_name = f"raw/{timestamp:%Y/%m/%d}/{filename}"
+                object_name = f"{state.object_prefix}{filename}"
                 blob = bucket.blob(object_name)
                 blob.upload_from_filename(
                     local_path,
@@ -152,12 +203,40 @@ def main() -> None:
                     timeout=180,
                 )
                 used_bytes += size
-                uploaded += 1
+                state.uploaded_segments += 1
+                state.uploaded_bytes += size
                 LOG.info("Uploaded segment object=%s bytes=%d", object_name, size)
             finally:
                 local_path.unlink(missing_ok=True)
 
-    LOG.info("Recording job completed segments=%d monthly_bytes=%d", uploaded, used_bytes)
+    if used_bytes >= monthly_limit:
+        state.monthly_limit_reached = True
+    LOG.info(
+        "Recording job completed segments=%d monthly_bytes=%d",
+        state.uploaded_segments,
+        used_bytes,
+    )
+
+
+def main() -> None:
+    started_at = time.monotonic()
+    state = RecordingState()
+    signal.signal(signal.SIGTERM, _stop_on_signal)
+    signal.signal(signal.SIGINT, _stop_on_signal)
+
+    try:
+        record(state)
+    except BaseException as exc:
+        send_gchat_alert(
+            recording_summary(
+                state,
+                time.monotonic() - started_at,
+                error_name=type(exc).__name__,
+            )
+        )
+        raise
+
+    send_gchat_alert(recording_summary(state, time.monotonic() - started_at))
 
 
 if __name__ == "__main__":

cloud/runtime.py

@@ -13,6 +13,8 @@
 import requests
 from requests.auth import HTTPDigestAuth
 
+from gchat_notifier import send_gchat_alert
+
 
 LOG = logging.getLogger(__name__)
 ACTIVE_USER_RE = re.compile(r"^users\[(\d+)]\.([^=]+)=(.*)$", re.IGNORECASE)
@@ -245,6 +247,11 @@ def poll_once() -> dict[str, Any]:
     if session_types and os.environ.get("TRIGGER_RECORDING", "true").casefold() == "true":
         result["recorder_operation"] = start_recorder()
         result["recorder_started"] = True
+        session_label = ", ".join(session_types)
+        send_gchat_alert(
+            "🐄 *Caméra allumée* — utilisation détectée "
+            f"({session_label}). L'enregistrement GCP vient de démarrer."
+        )
 
     # Deliberately log only counts and client types, never usernames or addresses.
     LOG.info(

cloud/test_runtime.py

@@ -1,7 +1,8 @@
 import unittest
 from unittest.mock import patch
 
-from recorder import safe_ffmpeg_diagnostic
+from gchat_notifier import format_bytes, format_duration, send_gchat_alert
+from recorder import RecordingState, recording_summary, safe_ffmpeg_diagnostic
 from runtime import (
     active_session_types,
     execution_is_active,
@@ -12,6 +13,44 @@
 
 
 class RuntimeTests(unittest.TestCase):
+    def test_formats_chat_metrics(self):
+        self.assertEqual(format_duration(0.2), "0 s")
+        self.assertEqual(format_duration(65.2), "1 min 5 s")
+        self.assertEqual(format_duration(3661), "1 h 1 min 1 s")
+        self.assertEqual(format_bytes(2_399_827), "2.3 Mio")
+
+    def test_recording_summary_contains_status_duration_and_private_link(self):
+        state = RecordingState(
+            bucket_name="private-bucket",
+            project_id="camera-project",
+            object_prefix="raw/2026/08/19/session/",
+            uploaded_segments=2,
+            uploaded_bytes=2_399_827,
+        )
+        summary = recording_summary(state, 65.2)
+        self.assertIn("Erreur : non", summary)
+        self.assertIn("Vidéo enregistrée : oui", summary)
+        self.assertIn("Durée : 1 min 5 s", summary)
+        self.assertIn("private-bucket/raw/2026/08/19/session", summary)
+        self.assertIn("project=camera-project", summary)
+
+    def test_failed_recording_summary_says_when_no_video_exists(self):
+        summary = recording_summary(
+            RecordingState(), 4.8, error_name="TimeoutError"
+        )
+        self.assertIn("Erreur : oui (TimeoutError)", summary)
+        self.assertIn("Vidéo enregistrée : non", summary)
+        self.assertIn("Lien : aucun", summary)
+
+    @patch("gchat_notifier.requests.post")
+    @patch.dict("os.environ", {"GCHAT_WEBHOOK_URL": "https://example.invalid/token"})
+    def test_chat_notification_uses_json_payload(self, post):
+        post.return_value.raise_for_status.return_value = None
+        self.assertTrue(send_gchat_alert("test"))
+        post.assert_called_once_with(
+            "https://example.invalid/token", json={"text": "test"}, timeout=15
+        )
+
     def test_parse_and_ignore_own_cgi_session(self):
         payload = "\r\n".join(
             [
@@ -72,6 +111,19 @@ def test_transient_probe_timeout_is_unavailable(self, _active, _probe):
             },
         )
 
+    @patch("runtime.send_gchat_alert")
+    @patch("runtime.start_recorder", return_value="operation")
+    @patch("runtime.probe_active_users", return_value=([{}], ["Mobile"]))
+    @patch("runtime.recorder_is_active", return_value=False)
+    def test_notifies_after_starting_recorder(
+        self, _active, _probe, start, notify
+    ):
+        result = poll_once()
+        self.assertTrue(result["recorder_started"])
+        start.assert_called_once_with()
+        notify.assert_called_once()
+        self.assertIn("Caméra allumée", notify.call_args.args[0])
+
 
 if __name__ == "__main__":
     unittest.main()

vm/startup.sh

@@ -51,11 +51,16 @@ docker pull "${recorder_image}"
 
 umask 077
 env_file=$(mktemp /run/calving-monitor.XXXXXX.env)
+gchat_webhook=$(secret_value calving-google-chat-webhook || true)
 {
     printf 'CAMERA_SERIAL=%s\n' "$(secret_value calving-camera-serial)"
     printf 'CAMERA_USERNAME=%s\n' "$(secret_value calving-camera-username)"
     printf 'CAMERA_PASSWORD=%s\n' "$(secret_value calving-camera-password)"
+    if [ -n "${gchat_webhook}" ]; then
+        printf 'GCHAT_WEBHOOK_URL=%s\n' "${gchat_webhook}"
+    fi
     printf 'GCS_BUCKET=%s\n' "${bucket}"
+    printf 'PROJECT_ID=%s\n' "${project}"
     printf 'RECORD_DURATION_SECONDS=%s\n' "${record_duration}"
     printf 'SEGMENT_SECONDS=60\n'
     printf 'MONTHLY_RECORDING_LIMIT_BYTES=30000000000\n'