A telegram bot asistant
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Copyright (c) 2023 Lorenzo Carbonell <a.k.a. atareao>

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import asyncio
import logging
import os
import re
from datetime import datetime
from dotenv import load_dotenv
from telegram import Bot
from openmeteo import OpenMeteoClient


async def main(token, chat_id):
    bot = Bot(token)
    logging.debug(await bot.get_me())
    msg = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f")
    logging.debug(await bot.send_message(msg, chat_id))
    msg = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f")
    logging.debug(await bot.send_message(msg, chat_id))
    while True:
        response = await bot.get_updates()
        if response["ok"] and response["result"]:
            for item in response["result"]:
                if "message" in item and "text" in item["message"]:
                    username = item["message"]["from"]["username"]
                    if re.match("^/hora", item["message"]["text"]):
                        hora = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f")
                        msg = f"{username}, son las {hora}"
                        logging.debug(await bot.send_message(msg, chat_id))
                elif "message" in item and "location" in item["message"]:
                    latitude =  item["message"]["location"]["latitude"]
                    longitude =  item["message"]["location"]["longitude"]
                    weather = await OpenMeteoClient.get_current_weather(
                            latitude=latitude, longitude=longitude)
                    if weather and "current_weather" in weather:
                        current = weather["current_weather"]
                        temperature = current["temperature"]
                        icon = OpenMeteoClient.get_icon(current["weathercode"])
                    msg = f"{icon} {temperature} ºC"
                    logging.debug(await bot.send_message(msg, chat_id))


if __name__ == "__main__":
    load_dotenv()
    token = os.getenv("TOKEN")
    chat_id = os.getenv("CHAT_ID")
    log_level = logging.getLevelName(os.getenv("LOG_LEVEL", "DEBUG"))
    logging.basicConfig(
        level=log_level,
        format="%(asctime)s - %(levelname)s: %(message)s"
    )
    asyncio.run(main(token, chat_id))

src/main.py
atareao/asistente
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Copyright (c) 2023 Lorenzo Carbonell <a.k.a. atareao>

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import aiohttp
import logging

logger = logging.getLogger(__name__)


class OpenMeteoClient:
    _url = "https://api.open-meteo.com/v1"

    @staticmethod
    async def get_current_weather(latitude: float,
                                  longitude: float) -> dict:
        logger.info("get_current_weather")
        params = {
            "latitude": latitude,
            "longitude": longitude,
            "current_weather": "true"
        }
        return await OpenMeteoClient._get("forecast", params)

    @staticmethod
    def get_icon(code: int) -> str:
        """
        WMO Weather interpretation codes (WW)
        Code        Description
        0           Clear sky
        1, 2, 3     Mainly clear, partly cloudy, and overcast
        45, 48      Fog and depositing rime fog
        51, 53, 55  Drizzle: Light, moderate, and dense intensity
        56, 57      Freezing Drizzle: Light and dense intensity
        61, 63, 65  Rain: Slight, moderate and heavy intensity
        66, 67      Freezing Rain: Light and heavy intensity
        71, 73, 75  Snow fall: Slight, moderate, and heavy intensity
        77          Snow grains
        80, 81, 82  Rain showers: Slight, moderate, and violent
        85, 86      Snow showers slight and heavy
        95 *        Thunderstorm: Slight or moderate
        96, 99 *    Thunderstorm with slight and heavy hail
        (*) Thunderstorm forecast with hail is only available in Central Europe
        """
        if code == 0:
            # Clear sky
            return "☀️"
        elif code in [1, 2, 3]:
            # Mainly clear, partly cloudy, and overcast
            return "⛅"
        elif code in [45, 58]:
            # Fog and depositing rime fog
            return "🌁"
        elif code in [51, 53, 55]:
            # Drizzle: Light, moderate, and dense intensity
            return "🌦️"
        elif code in [56, 57]:
            # Freezing Drizzle: Light and dense intensity
            pass
        elif code in [61, 63, 65]:
            # Rain: Slight, moderate and heavy intensity
            return "🌧️"
        elif code in [66, 67]:
            # Freezing Rain: Light and heavy intensity
            pass
        elif code in [71, 73, 75]:
            # Snow fall: Slight, moderate, and heavy intensity
            pass
        elif code == 77:
            # 77 Snow grains
            pass
        elif code in [80, 81, 82]:
            return "🌧️"
        elif code in [85, 86]:
            # Snow showers slight and heavy
            pass
        elif code == 95:
            # Thunderstorm: Slight or moderate
            pass
        elif code in [96, 99]:
            # Thunderstorm with slight and heavy hail
            return "⛈️"
        return "🤷"

    @staticmethod
    async def _get(endpoint: str, params: dict = {}) -> dict:
        """Send a generic GET

        Parameters
        ----------
        endpoint : str
            The endpoint
        params : dict
            Params for the query

        Returns
        -------
        dict
            Response from Telegram
        """
        logger.info("_get")
        logger.debug(f"endpoint: {endpoint}")
        logger.debug(f"params: {params}")
        async with aiohttp.ClientSession() as session:
            url = f"{OpenMeteoClient._url}/{endpoint}"
            async with session.get(url, params=params) as response:
                if response.status != 200:
                    content = await response.text()
                    msg = f"Error HTTP {response.status}. {content}"
                    raise Exception(msg)
                response = await response.json()
                logging.debug(f"Response: {response}")
                return response

src/openmeteo.py
atareao/asistente
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Copyright (c) 2023 Lorenzo Carbonell <a.k.a. atareao>

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import aiohttp
import logging
import os
import json

logger = logging.getLogger(__name__)

CURDIR = os.path.realpath(os.path.dirname(__file__))
CONFIG = os.path.join(CURDIR, "config.json")


class Bot:
    """A Telegram Bot"""

    def __init__(self, token: str, update_timeout: int = 120) -> None:
        """Init the bot

        Parameters
        ----------
        token : str
            Token of the bot
        update_offset : int
            First uptate to get
        update_timeout : int
            Timeout between calls
        """
        logger.debug("__init__")
        self._url = f"https://api.telegram.org/bot{token}"
        self._update_timeout = update_timeout
        self._read_config()

    def _read_config(self) -> None:
        if os.path.exists(CONFIG):
            with open(CONFIG, "r") as fr:
                config = json.load(fr)
                self._update_offset = config["offset"]
        else:
            self._update_offset = 0

    def _save_config(self) -> None:
        with open(CONFIG, "w") as fw:
            config = {
                "offset": self._update_offset
            }
            json.dump(config, fw)

    async def get_me(self) -> dict:
        """Get info about the bot

        Returns
        -------
        dict
            Info about the bot
        """
        logger.info("get_me")
        return await self._get("getMe")

    async def get_updates(self) -> dict:
        """Get updates

        Returns
        -------
        dict
            Updates
        """
        logger.info("get_updates")
        params = {
            "offset": self._update_offset,
            "timeout": self._update_timeout
        }
        response = await self._get("getUpdates", params)
        logger.debug(f"Response: {response}")
        if response["ok"] and response["result"]:
            offset = max([item["update_id"] for item in response["result"]])
            self._update_offset = offset + 1
            self._save_config()
        return response

    async def send_message(self, text: str, chat_id: int,
                           thread_id: int = 0) -> dict:
        """Send a message

        Parameters
        ----------
        text : str
            The message
        chat_id : int
            The chat_it
        thread_id : int
            The thread_id if any

        Returns
        -------
        dict
            The response
        """
        logger.info("send_message")
        data = {
            "chat_id": chat_id,
            "text": text
        }
        if thread_id > 0:
            data.update({"message_thread_id": thread_id})
        return await self._post("sendMessage", data)

    async def _get(self, endpoint: str, params: dict = {}) -> dict:
        """Send a generic GET

        Parameters
        ----------
        endpoint : str
            The endpoint
        params : dict
            Params for the query

        Returns
        -------
        dict
            Response from Telegram
        """
        logger.info("_get")
        logger.debug(f"endpoint: {endpoint}")
        logger.debug(f"params: {params}")
        async with aiohttp.ClientSession() as session:
            url = f"{self._url}/{endpoint}"
            async with session.get(url, params=params) as response:
                if response.status != 200:
                    content = await response.text()
                    msg = f"Error HTTP {response.status}. {content}"
                    raise Exception(msg)
                response = await response.json()
                logging.debug(f"Response: {response}")
                return response

    async def _post(self, endpoint: str, data: dict = {}) -> dict:
        """Send a generic POST

        Parameters
        ----------
        endpoint : str
            The endpoint
        data : dict
            Data to send

        Returns
        -------
        dict
            Response from Telegram
        """
        logger.info("_post")
        logger.debug(f"endpoint: {endpoint}")
        logger.debug(f"data: {data}")
        async with aiohttp.ClientSession() as session:
            url = f"{self._url}/{endpoint}"
            async with session.post(url, json=data) as response:
                if response.status != 200:
                    content = await response.text()
                    msg = f"Error HTTP {response.status}. {content}"
                    raise Exception(msg)
                response = await response.json()
                logging.debug(f"Response: {response}")
                return response

src/telegram.py
atareao/asistente
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Copyright (c) 2023 Lorenzo Carbonell <a.k.a. atareao>

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import logging
import os
import sys

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s - %(levelname)s: %(message)s"
)

PROJECT_PATH = os.getcwd()
SOURCE_PATH = os.path.join(PROJECT_PATH, "src")
sys.path.append(SOURCE_PATH)

tests/__init__.py
atareao/asistente
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Copyright (c) 2023 Lorenzo Carbonell <a.k.a. atareao>

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.


import unittest

from openmeteo import OpenMeteoClient


class TestOpenMeteoClient(unittest.IsolatedAsyncioTestCase):

    async def test_get_current_weather(self):
        response = await OpenMeteoClient.get_current_weather(39.3626, -0.4117)
        self.assertIsNotNone(response)


if __name__ == "__main__":
    unittest.main()

tests/openmeteo_tests.py
atareao/asistente
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Copyright (c) 2023 Lorenzo Carbonell <a.k.a. atareao>

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.


import os
import unittest
from dotenv import load_dotenv

from telegram import Bot


class TestTelegramBot(unittest.IsolatedAsyncioTestCase):

    def setUp(self) -> None:
        load_dotenv()
        token = os.getenv("TOKEN", "")
        self.chat_id = int(os.getenv("CHAT_ID", 0))
        update_offset = 0

        self.bot = Bot(token, update_offset)

    async def test_get_me(self):
        response = await self.bot.get_me()
        self.assertIsNotNone(response)

    async def test_get_updates(self):
        response = await self.bot.get_updates()
        self.assertIsNotNone(response)

    async def test_send_message(self):
        text = "Este es un mensaje de prueba"
        response = await self.bot.send_message(text, self.chat_id)
        self.assertIsNotNone(response)


if __name__ == "__main__":
    unittest.main()

tests/telegram_tests.py
atareao/asistente