diff options
| -rw-r--r-- | doc/ipcam_server.md | 8 | ||||
| -rw-r--r-- | src/home/util.py | 35 | ||||
| -rwxr-xr-x | src/ipcam_server.py | 47 |
3 files changed, 80 insertions, 10 deletions
diff --git a/doc/ipcam_server.md b/doc/ipcam_server.md index f443094..a22c239 100644 --- a/doc/ipcam_server.md +++ b/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 diff --git a/src/home/util.py b/src/home/util.py index a6ac906..4e47f49 100644 --- a/src/home/util.py +++ b/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]]: diff --git a/src/ipcam_server.py b/src/ipcam_server.py index 56b6e18..6856621 100755 --- a/src/ipcam_server.py +++ b/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 |