Merge pull request #8

Refactor Telegram Bot API service and add render service

CHANGELOG.md

@@ -5,11 +5,27 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [1.0.1]
+## [1.3.0]
+
+### Added
+
+- Render service for congratulations text.
 
 ### Changed
 
-- Update `config.example.toml`.
+- Refactor Telegram Bot API connection service.
+
+### Removed
+
+- Message queue support.
+
+---
+
+## [1.2.0]
+
+### Added
+
+- Add employee full names cleaning.
 
 ---
 
@@ -21,6 +37,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ---
 
+## [1.0.1]
+
+### Changed
+
+- Update `config.example.toml`.
+
+---
+
 ## [1.0.0] - 2023-03-04
 
 ### Added

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "employee-birthdays"
-version = "1.2.0"
+version = "1.3.0"
 description = "Notifications about employee's birthdays"
 authors = ["Eldos <[email protected]>"]
 readme = "README.md"

src/http_clients.py

@@ -21,7 +21,7 @@ def closing_dodo_is_connection_http_client(
 ) -> DodoISConnectionHttpClient:
     base_url = f'https://officemanager.dodopizza.{country_code}'
 
-    with httpx.Client(base_url=base_url, **kwargs) as http_client:
+    with httpx.Client(timeout=30, base_url=base_url, **kwargs) as http_client:
         yield DodoISConnectionHttpClient(http_client)
 
 
@@ -30,5 +30,5 @@ def closing_auth_credentials_storage_connection_http_client(
         base_url: str,
         **kwargs,
 ) -> AuthCredentialsStorageConnectionHttpClient:
-    with httpx.Client(base_url=base_url, **kwargs) as http_client:
+    with httpx.Client(timeout=30, base_url=base_url, **kwargs) as http_client:
         yield AuthCredentialsStorageConnectionHttpClient(http_client)

src/main.py

@@ -20,7 +20,8 @@
     closing_dodo_is_connection_http_client,
 )
 from models import EmployeeBirthday
-from telegram import BirthdayNotifier
+from telegram import TelegramBotApiConnection
+from views import render_congratulations
 
 log = structlog.stdlib.get_logger('app')
 
@@ -29,16 +30,14 @@ def main():
     config_file_path = pathlib.Path(__file__).parent.parent / 'config.toml'
     config = load_config_from_file(config_file_path)
 
+    telegram_bot_api_connection = TelegramBotApiConnection(
+        token=config.bot_token.get_secret_value(),
+    )
+
     units = get_units(base_url=str(config.units_storage_base_url))
 
     account_name_to_unit_ids = group_unit_ids_by_account_name(units)
 
-    birthday_notifier = BirthdayNotifier(
-        bot_token=config.bot_token.get_secret_value(),
-        chat_id=config.goretsky_band_chat_id,
-        unit_id_to_name={unit.id: unit.name for unit in units},
-    )
-
     employee_birthdays: list[EmployeeBirthday] = []
 
     with (
@@ -69,17 +68,19 @@ def main():
                 unit_ids=account_name_to_unit_ids[account_cookies.account_name],
             )
 
-    units_employee_birthdays = filter_employee_birthdays_in_blacklist(
+    employee_birthdays = filter_employee_birthdays_in_blacklist(
         employee_birthdays=employee_birthdays,
         employees_blacklist=config.employees_blacklist,
     )
 
-    try:
-        birthday_notifier.send_notifications(units_employee_birthdays)
-    except Exception:
-        log.exception(
-            'Failed to send birthday notifications to the Goretsky Band channel'
-        )
+    congratulations_text = render_congratulations(
+        employee_birthdays=employee_birthdays,
+        unit_id_to_name={unit.id: unit.name for unit in units},
+    )
+    telegram_bot_api_connection.send_message(
+        chat_id=config.goretsky_band_chat_id,
+        text=congratulations_text,
+    )
 
 
 if __name__ == '__main__':

src/telegram.py

@@ -1,77 +1,57 @@
-from collections.abc import Iterable
+import json
 
 import httpx
 import structlog.stdlib
+from structlog.contextvars import bound_contextvars
 
-from models import EmployeeBirthday
-
-__all__ = ('BirthdayNotifier',)
+__all__ = ('TelegramBotApiConnection',)
 
 log = structlog.stdlib.get_logger('app')
 
 
-class BirthdayNotifier:
+class TelegramBotApiConnection:
 
     def __init__(
             self,
             *,
-            chat_id: int,
-            unit_id_to_name: dict[int, str],
-            bot_token: str,
+            token: str,
     ):
-        self.__chat_id = chat_id
-        self.__unit_id_to_name = unit_id_to_name
-        self.__bot_token = bot_token
+        self.__token = token
 
     @property
     def base_url(self) -> str:
-        return f'https://api.telegram.org/bot{self.__bot_token}'
-
-    def _format_message(
-            self,
-            *,
-            unit_id: int,
-            employee_full_name: str,
-    ) -> str:
-        unit_name = self.__unit_id_to_name[unit_id]
-        return (
-            f'Банда, сегодня свой день рождения празднует'
-            f' {employee_full_name} из пиццерии {unit_name} 🎇🎇🎇\n'
-            'Поздравляем тебя и желаем всего самого наилучшего 🥳'
-        )
+        return f'https://api.telegram.org/bot{self.__token}'
+
+    def send_message(self, *, chat_id: int, text: str) -> None:
+        url = f'{self.base_url}/sendMessage'
+        request_data = {'chat_id': chat_id, 'text': text, 'parse_mode': 'HTML'}
+
+        # No need to create connection pool, because we have only one request
+        # to the Telegram Bot API in whole app.
+        response = httpx.post(url=url, json=request_data)
+
+        with bound_contextvars(
+                request_data=request_data,
+                status=response.status_code,
+        ):
+
+            try:
+                response_data = response.json()
+            except json.JSONDecodeError:
+                log.error(
+                    'Telegram Bot API connection: Failed to decode JSON',
+                    response_text=response.text,
+                )
+                return
 
-    def _send_notification(
-            self,
-            *,
-            http_client: httpx.Client,
-            unit_id: int,
-            employee_full_name: str,
-    ):
-        response = http_client.post('/sendMessage', json={
-            'chat_id': self.__chat_id,
-            'text': self._format_message(
-                unit_id=unit_id,
-                employee_full_name=employee_full_name,
-            ),
-        })
+            if not response_data.get('ok', False):
+                log.error(
+                    'Telegram Bot API connection: Failed to send message',
+                    response_data=response_data,
+                )
+                return
 
-        if response.is_error:
-            log.error(
-                'Failed to send a message to the'
-                ' Goretsky Band channel',
-                response=response.text,
+            log.info(
+                'Telegram Bot API connection: Message sent',
+                response_data=response_data,
             )
-        else:
-            log.info('Sent a message to the Goretsky Band channel')
-
-    def send_notifications(
-            self,
-            employee_birthdays: Iterable[EmployeeBirthday],
-    ):
-        with httpx.Client(base_url=self.base_url) as http_client:
-            for employee_birthday in employee_birthdays:
-                self._send_notification(
-                    http_client=http_client,
-                    unit_id=employee_birthday.unit_id,
-                    employee_full_name=employee_birthday.full_name,
-                )

src/views.py

@@ -0,0 +1,96 @@
+import random
+from collections.abc import Iterable
+from typing import Final
+
+from models import EmployeeBirthday
+
+__all__ = ('render_congratulations',)
+
+CONGRATULATIONS_PHRASES: Final[tuple[str, ...]] = (
+    'Поздравляем с важным днем!'
+    ' Желаем, чтобы этот год принес вам новые возможности и радостные события.',
+
+    'С наилучшими пожеланиями в этот особый день.'
+    ' Пусть солнце всегда сияет ярко над вашим путем.',
+
+    'Желаем вам здоровья, счастья и успехов на всех путях вашей жизни.'
+    ' Пусть каждый день приносит вам радость и улыбку.',
+
+    'С днем рождения! Желаем вам много новых и ярких впечатлений,'
+    ' незабываемых моментов и радостных встреч.',
+
+    'Поздравляем с праздником! Пусть ваш день рождения будет наполнен любовью,'
+    ' счастьем и душевным теплом.',
+
+    'Желаем вам реализации всех задуманных целей,'
+    ' смелости преодолевать преграды и мудрости в принятии важных решений.',
+
+    'Пусть каждый ваш день будет абсолютно волшебным,'
+    ' наполненным счастьем, добротой и улыбками.',
+
+    'Сегодня ваш день рождения, и пусть весь мир радуется этому с вами!'
+    ' Желаем вам все самое лучшее в жизни.',
+
+    'От всего сердца поздравляем с днем рождения!'
+    ' Пусть он станет началом новых возможностей и великих достижений.',
+
+    'С днем рождения! Желаем вам любви, радости и благополучия.'
+    ' Пусть ваша жизнь будет яркой и насыщенной счастливыми моментами.',
+
+    'Поздравляем с особым днем! Желаем,'
+    ' чтобы каждый момент вашей жизни был наполнен радостью и счастьем.',
+
+    'С днем рождения! Пусть этот день будет началом новой главы в вашей жизни,'
+    ' полной интересных приключений.',
+
+    'Желаем вам прекрасных мгновений и ни одной тени на вашем пути.'
+    ' Пусть каждый день принесет вам радость и улыбку.',
+
+    'Поздравляем с важным событием! Желаем, чтобы ваш день рождения'
+    ' стал трепетным моментом, который запомнится надолго.',
+
+    'Пусть солнце всегда светит в вашей жизни,'
+    ' а дождь проливается только для полива вашего счастья. С днем рождения!',
+
+    'С наилучшими пожеланиями в этот радостный день! Пусть звезды всегда'
+    ' благосклонны к вам, а удача сопутствует во всех начинаниях.',
+
+    'Желаем вам сияющего дня рождения, наполненного любовью и красотой.'
+    ' Пусть он станет началом самых ярких моментов вашей жизни.',
+
+    'Поздравляем с этим волшебным днем! Желаем, чтобы ваше сердце всегда было'
+    ' наполнено радостью и благодарностью за все, что у вас есть.',
+
+    'Пусть каждый год вашей жизни будет насыщен впечатлениями, достижениями'
+    ' и счастьем. С днем рождения!',
+
+    'Счастливого дня рождения! Желаем вам мощного вдохновения,'
+    ' ярких моментов и замечательных людей в вашей жизни.',
+)
+
+CONGRATULATIONS_EMOJIS: Final[tuple[str, ...]] = ('🥳', '🎉', '🎊')
+
+
+def render_congratulations(
+        *,
+        employee_birthdays: Iterable[EmployeeBirthday],
+        unit_id_to_name: dict[int, str],
+) -> str:
+    lines = ['<b>Банда, сегодня свой день рождения празднуют:</b>\n']
+
+    emoji = random.choice(CONGRATULATIONS_EMOJIS)
+
+    for employee_birthday in employee_birthdays:
+        unit_name = unit_id_to_name.get(employee_birthday.unit_id)
+        employee_name = employee_birthday.full_name
+
+        if unit_name is None:
+            lines.append(f'{emoji} {employee_name}')
+        else:
+            lines.append(f'{emoji} {employee_name} из пиццерии {unit_name}')
+
+    congratulation_phrase = random.choice(CONGRATULATIONS_PHRASES)
+
+    lines.append(f'\n{congratulation_phrase}')
+
+    return '\n'.join(lines)