docker-images/linkki-tiedotus/linkki_poster/telegram_api.py

import json
from contextlib import ExitStack
from pathlib import Path

import requests


class TelegramApiError(RuntimeError):
    def __init__(self, description: str, error_code: int | None = None):
        super().__init__(description)
        self.description = description
        self.error_code = error_code


class TelegramAPI:
    def __init__(self, bot_token: str):
        self.base_url = f"https://api.telegram.org/bot{bot_token}"

    def send_media_group(
        self,
        chat_id: str,
        image_paths: list[Path],
        caption: str | None = None,
        parse_mode: str | None = None,
    ) -> list[dict]:
        if not image_paths:
            raise ValueError("send_media_group requires at least one image.")

        media_items = []
        for index, _ in enumerate(image_paths):
            media_item = {
                "type": "photo",
                "media": f"attach://file{index}",
            }
            if index == 0 and caption is not None:
                media_item["caption"] = caption
                if parse_mode is not None:
                    media_item["parse_mode"] = parse_mode
            media_items.append(media_item)

        payload = {"chat_id": chat_id, "media": json.dumps(media_items, ensure_ascii=False)}

        with ExitStack() as stack:
            files = {}
            for index, image_path in enumerate(image_paths):
                files[f"file{index}"] = stack.enter_context(image_path.open("rb"))

            return self._post("sendMediaGroup", data=payload, files=files, timeout=60)

    def get_updates(self, offset: int | None = None, limit: int = 100, timeout: int = 0) -> list[dict]:
        payload = {"limit": limit, "timeout": timeout}
        if offset is not None:
            payload["offset"] = offset
        return self._post("getUpdates", data=payload, timeout=timeout + 10)

    def send_message(
        self,
        chat_id: str,
        text: str,
        parse_mode: str | None = None,
        disable_link_preview: bool = False,
    ) -> dict:
        payload = {"chat_id": chat_id, "text": text}
        if parse_mode is not None:
            payload["parse_mode"] = parse_mode
        if disable_link_preview:
            payload["link_preview_options"] = json.dumps({"is_disabled": True})
        return self._post("sendMessage", data=payload, timeout=30)

    def edit_message_text(
        self,
        chat_id: str,
        message_id: int,
        text: str,
        parse_mode: str | None = None,
        disable_link_preview: bool = False,
    ) -> dict:
        payload = {
            "chat_id": chat_id,
            "message_id": message_id,
            "text": text,
        }
        if parse_mode is not None:
            payload["parse_mode"] = parse_mode
        if disable_link_preview:
            payload["link_preview_options"] = json.dumps({"is_disabled": True})
        return self._post("editMessageText", data=payload, timeout=30)

    def edit_message_caption(
        self,
        chat_id: str,
        message_id: int,
        caption: str,
        parse_mode: str | None = None,
    ) -> dict:
        payload = {
            "chat_id": chat_id,
            "message_id": message_id,
            "caption": caption,
        }
        if parse_mode is not None:
            payload["parse_mode"] = parse_mode
        return self._post("editMessageCaption", data=payload, timeout=30)

    def delete_message(self, chat_id: str, message_id: int) -> bool:
        payload = {
            "chat_id": chat_id,
            "message_id": message_id,
        }
        return bool(self._post("deleteMessage", data=payload, timeout=30))

    def pin_chat_message(self, chat_id: str, message_id: int, disable_notification: bool = True) -> bool:
        payload = {
            "chat_id": chat_id,
            "message_id": message_id,
        }
        if disable_notification:
            payload["disable_notification"] = "true"
        return bool(self._post("pinChatMessage", data=payload, timeout=30))

    def unpin_chat_message(self, chat_id: str, message_id: int) -> bool:
        payload = {
            "chat_id": chat_id,
            "message_id": message_id,
        }
        return bool(self._post("unpinChatMessage", data=payload, timeout=30))

    def _post(self, endpoint: str, data: dict, files: dict | None = None, timeout: int = 30):
        response = requests.post(f"{self.base_url}/{endpoint}", data=data, files=files, timeout=timeout)
        try:
            response.raise_for_status()
        except requests.HTTPError as error:
            description = _extract_error_description(response) or str(error)
            error_code = _extract_error_code(response)
            raise TelegramApiError(description=description, error_code=error_code) from error

        payload = response.json()
        if not payload.get("ok", False):
            raise TelegramApiError(
                description=str(payload.get("description", "Telegram API request failed.")),
                error_code=_to_int_or_none(payload.get("error_code")),
            )
        return payload["result"]


def _extract_error_description(response: requests.Response) -> str | None:
    try:
        payload = response.json()
    except ValueError:
        return None
    description = payload.get("description")
    if description is None:
        return None
    return str(description)


def _extract_error_code(response: requests.Response) -> int | None:
    try:
        payload = response.json()
    except ValueError:
        return None
    return _to_int_or_none(payload.get("error_code"))


def _to_int_or_none(value) -> int | None:
    if isinstance(value, int):
        return value
    if isinstance(value, str) and value.isdigit():
        return int(value)
    return None


def is_message_missing_error(error: TelegramApiError | str) -> bool:
    description = error.description if isinstance(error, TelegramApiError) else str(error)
    lowered = description.lower()
    return "not found" in lowered or "message to delete not found" in lowered or "message to edit not found" in lowered


def is_message_not_modified_error(error: TelegramApiError | str) -> bool:
    description = error.description if isinstance(error, TelegramApiError) else str(error)
    return "message is not modified" in description.lower()


def is_message_not_deletable_error(error: TelegramApiError | str) -> bool:
    description = error.description if isinstance(error, TelegramApiError) else str(error)
    return "can't be deleted" in description.lower()