35 lines
911 B
Python
35 lines
911 B
Python
import json
|
|
import logging
|
|
|
|
from settings import TELEGRAM_API_TOKEN, TELEGRAM_CHAT_ID
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TELEGRAM_API_URL = 'https://api.telegram.org/bot'
|
|
|
|
class Telegram:
|
|
def __init__(self, aiosession):
|
|
if TELEGRAM_API_TOKEN is None:
|
|
raise Exception("TELEGRAM_API_TOKEN is required")
|
|
|
|
self.aiosession = aiosession
|
|
|
|
async def post(self, request, data):
|
|
async with self.aiosession.post('{}{}/{}'.format(
|
|
TELEGRAM_API_URL,
|
|
TELEGRAM_API_TOKEN,
|
|
request
|
|
), json=data) as response:
|
|
if not response.ok:
|
|
logger.error('{} {}'.format(request, response.status))
|
|
raise Exception('{} {}'.format(request, response.status))
|
|
|
|
return await response.json()
|
|
|
|
async def message_send(self, text):
|
|
data = {
|
|
'chat_id': TELEGRAM_CHAT_ID,
|
|
'text': text
|
|
}
|
|
|
|
return await self.post('sendMessage', data=data)
|