ipcam_server: telegram notification (not tested... we'll see)

doc/ipcam_server.md

@@ -21,11 +21,15 @@ camera:
 
 motion:
   padding: 2
+  telegram: true
 
 logging:
   verbose: true
-  
-motion_threshold: 1
+
+telegram:
+  token: ""
+  chat_id: ""
+  parse_mode: HTML
 ```
 
 ## Usage

src/home/util.py

@@ -1,3 +1,4 @@
+import functools
 import json
 import socket
 import time
@@ -7,6 +8,7 @@ import traceback
 import logging
 import string
 import random
+import asyncio
 
 from enum import Enum
 from .config import config
@@ -100,8 +102,31 @@ def send_datagram(message: str, addr: Addr) -> None:
 
 def send_telegram(text: str,
                   parse_mode: str = None,
-                  disable_web_page_preview: bool = False,
-                  ):
+                  disable_web_page_preview: bool = False):
+    data, token = _send_telegram_data(text, parse_mode, disable_web_page_preview)
+    r = requests.post('https://api.telegram.org/bot%s/sendMessage' % token, data=data)
+    if r.status_code != 200:
+        logger.error(r.text)
+        raise RuntimeError("telegram returned %d" % r.status_code)
+
+
+async def send_telegram_aio(text: str,
+                            parse_mode: str = None,
+                            disable_web_page_preview: bool = False):
+    loop = asyncio.get_event_loop()
+    data, token = _send_telegram_data(text, parse_mode, disable_web_page_preview)
+    r = await loop.run_in_executor(None,
+                                   functools.partial(requests.post,
+                                                     'https://api.telegram.org/bot%s/sendMessage' % token,
+                                                     data=data))
+    if r.status_code != 200:
+        logger.error(r.text)
+        raise RuntimeError("telegram returned %d" % r.status_code)
+
+
+def _send_telegram_data(text: str,
+                        parse_mode: str = None,
+                        disable_web_page_preview: bool = False) -> tuple[dict, str]:
     data = {
         'chat_id': config['telegram']['chat_id'],
         'text': text
@@ -115,11 +140,7 @@ def send_telegram(text: str,
     if disable_web_page_preview or 'disable_web_page_preview' in config['telegram']:
         data['disable_web_page_preview'] = 1
 
-    r = requests.post('https://api.telegram.org/bot%s/sendMessage' % config['telegram']['token'], data=data)
-
-    if r.status_code != 200:
-        logger.error(r.text)
-        raise RuntimeError("telegram returned %d" % r.status_code)
+    return data, config['telegram']['token']
 
 
 def format_tb(exc) -> Optional[list[str]]:

src/ipcam_server.py

@@ -7,7 +7,7 @@ import time
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 
 from home.config import config
-from home.util import parse_addr
+from home.util import parse_addr, send_telegram_aio
 from home import http
 from home.database.sqlite import SQLiteBase
 from home.camera import util as camutil
@@ -22,6 +22,11 @@ class TimeFilterType(Enum):
     MOTION = 'motion'
 
 
+class TelegramLinkType(Enum):
+    FRAGMENT = 'fragment'
+    ORIGINAL_FILE = 'original_file'
+
+
 def valid_recording_name(filename: str) -> bool:
     return filename.startswith('record_') and filename.endswith('.mp4')
 
@@ -325,6 +330,46 @@ async def process_fragments(camera: int,
                                  start_pos=start,
                                  duration=duration)
 
+    try:
+        if fragments and config['motion']['telegram']:
+            asyncio.ensure_future(motion_notify_tg(camera, filename, fragments))
+    except KeyError:
+        pass
+
+
+async def motion_notify_tg(camera: int,
+                           filename: str,
+                           fragments: list[tuple[int, int]]):
+    dt_file = filename_to_datetime(filename)
+    fmt = '%H:%M:%S'
+
+    text = f'Camera: <b>{camera}</b>\n'
+    text += f'Original file: <b>{filename}</b> '
+    text += _tg_links(TelegramLinkType.ORIGINAL_FILE, camera, filename)
+
+    for start, end in fragments:
+        duration = end - start
+        if duration < 0:
+            duration = 0
+
+        dt1 = dt_file + timedelta(seconds=start)
+        dt2 = dt_file + timedelta(seconds=end)
+
+        text += f'\nFragment: <b>{duration}s</b>, {dt1.strftime(fmt)} - {dt2.strftime(fmt)} '
+        text += _tg_links(TelegramLinkType.FRAGMENT, camera, f'{dt1.strftime(datetime_format)}__{dt2.strftime(datetime_format)}.mp4')
+
+    await send_telegram_aio(text)
+
+
+def _tg_links(link_type: TelegramLinkType,
+              camera: int,
+              file: str) -> str:
+    links = []
+    for link_name, link_template in config['telegram'][f'{link_type.value}_url_templates']:
+        link = link_template.replace('{camera}', str(camera)).replace('{file}', file)
+        links.append(f'<a href="{link}">{link_name}</a>')
+    return ' '.join(links)
+
 
 async def fix_job() -> None:
     global fix_job_running