diff options
Diffstat (limited to 'src/home')
| -rw-r--r-- | src/home/telegram/__init__.py | 1 | ||||
| -rw-r--r-- | src/home/telegram/aio.py | 18 | ||||
| -rw-r--r-- | src/home/telegram/telegram.py | 46 | ||||
| -rw-r--r-- | src/home/util.py | 47 |
4 files changed, 65 insertions, 47 deletions
diff --git a/src/home/telegram/__init__.py b/src/home/telegram/__init__.py new file mode 100644 index 0000000..8565b40 --- /dev/null +++ b/src/home/telegram/__init__.py @@ -0,0 +1 @@ +from .telegram import send_message, send_photo
\ No newline at end of file diff --git a/src/home/telegram/aio.py b/src/home/telegram/aio.py new file mode 100644 index 0000000..fc87c1c --- /dev/null +++ b/src/home/telegram/aio.py @@ -0,0 +1,18 @@ +import functools +import asyncio + +from .telegram import ( + send_message as _send_message_sync, + send_photo as _send_photo_sync +) + + +async def send_message(*args, **kwargs): + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, functools.partial(_send_message_sync, *args, **kwargs)) + + +async def send_photo(*args, **kwargs): + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, functools.partial(_send_photo_sync, *args, **kwargs)) + diff --git a/src/home/telegram/telegram.py b/src/home/telegram/telegram.py new file mode 100644 index 0000000..9c7ea73 --- /dev/null +++ b/src/home/telegram/telegram.py @@ -0,0 +1,46 @@ +import requests +import logging + +from ..config import config + + +_logger = logging.getLogger(__name__) + + +def send_message(text: str, + parse_mode: str = None, + disable_web_page_preview: bool = False): + data, token = _send_telegram_data(text, parse_mode, disable_web_page_preview) + req = requests.post('https://api.telegram.org/bot%s/sendMessage' % token, data=data) + return req.json() + + +def send_photo(filename: str): + data = { + 'chat_id': config['telegram']['chat_id'], + } + token = config['telegram']['token'] + + url = f'https://api.telegram.org/bot{token}/sendPhoto' + with open(filename, "rb") as fd: + req = requests.post(url, data=data, files={"photo": fd}) + return req.json() + + +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 + } + + if parse_mode is not None: + data['parse_mode'] = parse_mode + elif 'parse_mode' in config['telegram']: + data['parse_mode'] = config['telegram']['parse_mode'] + + if disable_web_page_preview or 'disable_web_page_preview' in config['telegram']: + data['disable_web_page_preview'] = 1 + + return data, config['telegram']['token'] diff --git a/src/home/util.py b/src/home/util.py index 4e47f49..9dd84f6 100644 --- a/src/home/util.py +++ b/src/home/util.py @@ -1,17 +1,13 @@ -import functools import json import socket import time -import requests import subprocess import traceback import logging import string import random -import asyncio from enum import Enum -from .config import config from datetime import datetime from typing import Tuple, Optional @@ -100,49 +96,6 @@ def send_datagram(message: str, addr: Addr) -> None: sock.sendto(message.encode(), addr) -def send_telegram(text: str, - parse_mode: str = None, - 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 - } - - if parse_mode is not None: - data['parse_mode'] = parse_mode - elif 'parse_mode' in config['telegram']: - data['parse_mode'] = config['telegram']['parse_mode'] - - if disable_web_page_preview or 'disable_web_page_preview' in config['telegram']: - data['disable_web_page_preview'] = 1 - - return data, config['telegram']['token'] - - def format_tb(exc) -> Optional[list[str]]: tb = traceback.format_tb(exc.__traceback__) if not tb: |