109 lines
3.2 KiB
Python
109 lines
3.2 KiB
Python
"""
|
|
notifier.py — Send Telegram notifications with video and caption.
|
|
Uses httpx directly (no python-telegram-bot dependency).
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TG_API = "https://api.telegram.org"
|
|
|
|
def _proxy() -> str | None:
|
|
return os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy") or None
|
|
|
|
|
|
def _format_caption(
|
|
camera_name: str,
|
|
event_type: str,
|
|
event_time: datetime,
|
|
description: str,
|
|
) -> str:
|
|
"""Build Telegram message caption (≤1024 chars for video)."""
|
|
tz_label = "UTC"
|
|
time_str = event_time.strftime("%d.%m.%Y %H:%M:%S")
|
|
type_labels = {
|
|
"motion": "🔴 Движение",
|
|
"person": "🚶 Человек",
|
|
"vehicle": "🚗 Транспорт",
|
|
"pet": "🐾 Животное",
|
|
"baby_cry": "👶 Плач",
|
|
}
|
|
type_display = type_labels.get(event_type.lower(), f"⚡ {event_type}")
|
|
|
|
caption = (
|
|
f"📷 *{camera_name}*\n"
|
|
f"{type_display} · {time_str} {tz_label}\n\n"
|
|
f"{description}"
|
|
)
|
|
# Telegram caption limit: 1024 chars
|
|
return caption[:1024]
|
|
|
|
|
|
def send_video(
|
|
bot_token: str,
|
|
chat_id: str,
|
|
video_path: str,
|
|
camera_name: str,
|
|
event_type: str,
|
|
event_time: datetime,
|
|
description: str,
|
|
timeout: int = 60,
|
|
) -> bool:
|
|
"""
|
|
Send a video file to Telegram with a formatted caption.
|
|
|
|
Returns True on success, False on failure.
|
|
"""
|
|
caption = _format_caption(camera_name, event_type, event_time, description)
|
|
url = f"{TG_API}/bot{bot_token}/sendVideo"
|
|
|
|
try:
|
|
with open(video_path, "rb") as video_file:
|
|
with httpx.Client(timeout=timeout, proxy=_proxy()) as client:
|
|
response = client.post(
|
|
url,
|
|
data={
|
|
"chat_id": chat_id,
|
|
"caption": caption,
|
|
"parse_mode": "Markdown",
|
|
"supports_streaming": "true",
|
|
},
|
|
files={"video": (Path(video_path).name, video_file, "video/mp4")},
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
logger.info("Telegram notification sent for %s", camera_name)
|
|
return True
|
|
else:
|
|
logger.error(
|
|
"Telegram API error %s: %s",
|
|
response.status_code,
|
|
response.text[:500],
|
|
)
|
|
return False
|
|
|
|
except httpx.TimeoutException:
|
|
logger.error("Telegram upload timed out for %s", video_path)
|
|
return False
|
|
except Exception as exc:
|
|
logger.error("Telegram send failed: %s", exc)
|
|
return False
|
|
|
|
|
|
def send_text(bot_token: str, chat_id: str, text: str) -> bool:
|
|
"""Send a plain text message (for errors / startup notifications)."""
|
|
url = f"{TG_API}/bot{bot_token}/sendMessage"
|
|
try:
|
|
with httpx.Client(timeout=10, proxy=_proxy()) as client:
|
|
r = client.post(url, json={"chat_id": chat_id, "text": text})
|
|
return r.status_code == 200
|
|
except Exception as exc:
|
|
logger.error("send_text failed: %s", exc)
|
|
return False
|