mastoposter-oss_images/mastoposter/integrations/telegram.py

257 lines
8.1 KiB
Python
Raw Normal View History

from configparser import SectionProxy
2022-08-24 08:09:41 +03:00
from dataclasses import dataclass
2022-11-01 13:37:47 +03:00
from logging import getLogger
from typing import Any, List, Mapping, Optional
2022-08-24 08:09:41 +03:00
from httpx import AsyncClient
2022-08-31 16:19:39 +03:00
from jinja2 import Template
2022-08-24 08:28:18 +03:00
from mastoposter.integrations.base import BaseIntegration
from mastoposter.types import Attachment, Poll, Status
2022-08-31 16:19:39 +03:00
from emoji import emojize
2022-08-24 08:09:41 +03:00
2022-11-01 14:33:47 +03:00
logger = getLogger("integrations.telegram")
2022-11-01 13:37:47 +03:00
2022-08-24 08:09:41 +03:00
@dataclass
class TGResponse:
ok: bool
params: dict
result: Optional[Any] = None
error: Optional[str] = None
@classmethod
def from_dict(cls, data: dict, params: dict) -> "TGResponse":
return cls(
ok=data["ok"],
params=params,
result=data.get("result"),
error=data.get("description"),
)
2022-08-24 08:09:41 +03:00
2022-08-31 16:19:39 +03:00
API_URL: str = "https://api.telegram.org/bot{}/{}"
MEDIA_COMPATIBILITY: Mapping[str, set] = {
"image": {"image", "video"},
"video": {"image", "video"},
"gifv": {"gifv"},
"audio": {"audio"},
"unknown": {"unknown"},
}
MEDIA_MAPPING: Mapping[str, str] = {
"image": "photo",
"video": "video",
"gifv": "animation",
"audio": "audio",
"unknown": "document",
}
2023-01-20 12:59:42 +03:00
MEDIA_SPOILER_SUPPORT: Mapping[str, bool] = {
"image": True,
"video": True,
"gifv": True,
"audio": False,
"unknown": False,
}
2022-08-31 16:19:39 +03:00
DEFAULT_TEMPLATE: str = """\
{% if status.reblog %}\
Boost from <a href="{{status.reblog.account.url}}">\
{{status.reblog.account.name}}</a>\
{% endif %}\
{% if status.reblog_or_status.spoiler_text %}\
{{status.reblog_or_status.spoiler_text}}
<tg-spoiler>{% endif %}{{ status.reblog_or_status.content_flathtml }}\
{% if status.reblog_or_status.spoiler_text %}</tg-spoiler>{% endif %}
2022-08-31 16:19:39 +03:00
<a href="{{status.link}}">Link to post</a>"""
2022-08-24 08:09:41 +03:00
2022-08-31 16:19:39 +03:00
class TelegramIntegration(BaseIntegration):
def __init__(
self,
token: str,
chat_id: str,
template: Optional[Template] = None,
silent: bool = True,
):
self.token = token
self.chat_id = chat_id
self.silent = silent
if template is None:
self.template = Template(emojize(DEFAULT_TEMPLATE))
else:
self.template = template
@classmethod
def from_section(cls, section: SectionProxy) -> "TelegramIntegration":
return cls(
token=section["token"],
chat_id=section["chat"],
silent=section.getboolean("silent", True),
template=Template(
emojize(section.get("template", DEFAULT_TEMPLATE))
),
2022-08-31 18:36:11 +03:00
)
2022-08-24 08:09:41 +03:00
async def _tg_request(self, method: str, **kwargs) -> TGResponse:
2022-08-31 16:19:39 +03:00
url = API_URL.format(self.token, method)
2022-08-24 08:09:41 +03:00
async with AsyncClient() as client:
2022-11-02 20:18:31 +03:00
logger.debug("TG request: %s(%r)", method, kwargs)
response = TGResponse.from_dict(
2022-08-24 08:09:41 +03:00
(await client.post(url, json=kwargs)).json(), kwargs
)
if not response.ok:
logger.error("TG error: %r", response.error)
2022-11-01 14:33:47 +03:00
logger.error("parameters: %r", kwargs)
2022-11-02 20:18:31 +03:00
else:
logger.debug("Result: %r", response.result)
return response
2022-08-24 08:09:41 +03:00
async def _post_plaintext(self, text: str) -> TGResponse:
2022-11-01 13:37:47 +03:00
logger.debug("Sending HTML message: %r", text)
2022-08-24 08:09:41 +03:00
return await self._tg_request(
"sendMessage",
parse_mode="HTML",
disable_notification=self.silent,
2022-08-24 08:09:41 +03:00
disable_web_page_preview=True,
chat_id=self.chat_id,
text=text,
)
2023-01-20 12:59:42 +03:00
async def _post_media(
self, text: str, media: Attachment, spoiler: bool = False
) -> TGResponse:
2022-08-24 08:09:41 +03:00
# Just to be safe
2022-08-31 16:19:39 +03:00
if media.type not in MEDIA_MAPPING:
2022-11-01 13:37:47 +03:00
logger.warning(
"Media %r has unknown type, falling back to plaintext", media
)
2022-08-24 08:09:41 +03:00
return await self._post_plaintext(text)
return await self._tg_request(
2022-08-31 16:19:39 +03:00
"send%s" % MEDIA_MAPPING[media.type].title(),
2022-08-24 08:09:41 +03:00
parse_mode="HTML",
disable_notification=self.silent,
2022-08-24 08:09:41 +03:00
disable_web_page_preview=True,
chat_id=self.chat_id,
caption=text,
2022-08-31 16:19:39 +03:00
**{MEDIA_MAPPING[media.type]: media.url},
2023-01-20 13:22:23 +03:00
**(
{"has_spoiler": spoiler}
if MEDIA_SPOILER_SUPPORT.get(media.type, False)
else {}
),
2022-08-24 08:09:41 +03:00
)
async def _post_mediagroup(
2023-01-20 12:59:42 +03:00
self, text: str, media: List[Attachment], spoiler: bool = False
) -> TGResponse:
2022-11-01 13:37:47 +03:00
logger.debug("Sendind media group: %r (text=%r)", media, text)
2022-08-24 08:09:41 +03:00
media_list: List[dict] = []
allowed_medias = {"image", "gifv", "video", "audio", "unknown"}
for attachment in media:
if attachment.type not in allowed_medias:
continue
2022-08-31 16:19:39 +03:00
if attachment.type not in MEDIA_COMPATIBILITY:
2022-11-01 14:33:47 +03:00
logger.warning(
"attachment %r is not in %r",
attachment.type,
MEDIA_COMPATIBILITY,
)
2022-08-24 08:09:41 +03:00
continue
2022-08-31 16:19:39 +03:00
allowed_medias &= MEDIA_COMPATIBILITY[attachment.type]
2022-08-24 08:09:41 +03:00
media_list.append(
{
2022-08-31 16:19:39 +03:00
"type": MEDIA_MAPPING[attachment.type],
2022-08-24 08:09:41 +03:00
"media": attachment.url,
2023-01-20 12:59:42 +03:00
**(
{"has_spoiler": spoiler}
if MEDIA_SPOILER_SUPPORT.get(attachment.type, False)
else {}
),
2022-08-24 08:09:41 +03:00
}
)
if len(media_list) == 1:
media_list[0].update(
{
"caption": text,
"parse_mode": "HTML",
}
)
return await self._tg_request(
"sendMediaGroup",
disable_notification=self.silent,
2022-08-24 08:09:41 +03:00
disable_web_page_preview=True,
chat_id=self.chat_id,
media=media_list,
)
async def _post_poll(
self, poll: Poll, reply_to: Optional[str] = None
) -> TGResponse:
2022-11-01 13:37:47 +03:00
logger.debug("Sending poll: %r", poll)
return await self._tg_request(
"sendPoll",
disable_notification=self.silent,
disable_web_page_preview=True,
chat_id=self.chat_id,
question=f"Poll:{poll.id}",
reply_to_message_id=reply_to,
2022-09-23 18:21:31 +03:00
allows_multiple_answers=poll.multiple,
options=[opt.title for opt in poll.options],
)
async def __call__(self, status: Status) -> Optional[str]:
2022-08-24 08:09:41 +03:00
source = status.reblog or status
has_spoiler = source.sensitive
text = self.template.render({"status": status})
2022-08-24 08:09:41 +03:00
ids = []
2022-08-24 08:09:41 +03:00
if not source.media_attachments:
if (res := await self._post_plaintext(text)).ok:
if res.result:
ids.append(res.result["message_id"])
2022-08-24 08:09:41 +03:00
elif len(source.media_attachments) == 1:
if (
res := await self._post_media(
2023-01-20 12:59:42 +03:00
text, source.media_attachments[0], has_spoiler
)
).ok and res.result is not None:
ids.append(res.result["message_id"])
2022-08-24 08:09:41 +03:00
else:
if (
res := await self._post_mediagroup(
2023-01-20 12:59:42 +03:00
text, source.media_attachments, has_spoiler
)
).ok and res.result is not None:
ids.append(res.result["message_id"])
if source.poll:
if (
res := await self._post_poll(
source.poll, reply_to=ids[0] if ids else None
)
).ok and res.result:
ids.append(res.result["message_id"])
2022-08-24 08:09:41 +03:00
return str.join(",", map(str, ids))
2022-08-24 08:09:41 +03:00
def __repr__(self) -> str:
2022-11-01 13:37:47 +03:00
bot_uid, key = self.token.split(":")
2022-08-24 08:09:41 +03:00
return (
"<TelegramIntegration "
"chat_id={chat!r} "
"template={template!r} "
2022-11-01 13:37:47 +03:00
"token={bot_uid}:{key} "
"silent={silent!r}>"
2022-11-01 13:37:47 +03:00
).format(
chat=self.chat_id,
silent=self.silent,
template=self.template,
bot_uid=bot_uid,
key=str.join("", ("X" for _ in key)),
)