Add telegram client

app/config.py.example

@@ -1,8 +1,17 @@
 import os
 
 from fastapi_mail import ConnectionConfig
+from pydantic import BaseSettings
+
+
+class Settings(BaseSettings):
+    app_name: str = "PyLander"
+    bot_api: str = "BOT_API"
+    webhook_url: str = "WEBHOOK_URL"
+
+    class Config:
+        env_file = ".env"
 
-# flake8: noqa
 
 # general
 DOMAIN = 'Our-Domain'

app/database/models.py

@@ -30,6 +30,7 @@ class User(Base):
     full_name = Column(String)
     description = Column(String, default="Happy new user!")
     avatar = Column(String, default="profile.png")
+    telegram_id = Column(String, unique=True)
     is_active = Column(Boolean, default=False)
 
     events = relationship("UserEvent", back_populates="participants")

app/dependencies.py

@@ -1,3 +1,4 @@
+from functools import lru_cache
 import os
 
 from fastapi.templating import Jinja2Templates
@@ -11,3 +12,8 @@
 TEMPLATES_PATH = os.path.join(APP_PATH, "templates")
 
 templates = Jinja2Templates(directory=TEMPLATES_PATH)
+
+
+@lru_cache()
+def get_settings():
+    return config.Settings()

app/main.py

@@ -5,7 +5,8 @@
 from app.database.database import engine
 from app.dependencies import (
     MEDIA_PATH, STATIC_PATH, templates)
-from app.routers import agenda, event, profile, email, invitation
+from app.telegram.pylander import pylander
+from app.routers import agenda, email, event, invitation, profile, telegram
 
 models.Base.metadata.create_all(bind=engine)
 
@@ -16,9 +17,12 @@
 app.include_router(profile.router)
 app.include_router(event.router)
 app.include_router(agenda.router)
+app.include_router(telegram.router)
 app.include_router(email.router)
 app.include_router(invitation.router)
 
+pylander.set_webhook()
+
 
 @app.get("/")
 async def home(request: Request):

app/routers/profile.py

@@ -26,6 +26,7 @@ def get_placeholder_user():
         email='[email protected]',
         password='1a2s3d4f5g6',
         full_name='My Name',
+        telegram_id=''
     )
 
 
@@ -43,8 +44,6 @@ async def profile(
         session.commit()
         user = session.query(User).filter_by(id=1).first()
 
-    session.close()
-
     return templates.TemplateResponse("profile.html", {
         "request": request,
         "user": user,
@@ -64,11 +63,8 @@ async def update_user_fullname(
     user.full_name = new_fullname
     session.commit()
 
-    session.close()
-
     url = router.url_path_for("profile")
-    response = RedirectResponse(url=url, status_code=HTTP_302_FOUND)
-    return response
+    return RedirectResponse(url=url, status_code=HTTP_302_FOUND)
 
 
 @router.post("/update_user_email")
@@ -83,8 +79,6 @@ async def update_user_email(
     user.email = new_email
     session.commit()
 
-    session.close()
-
     url = router.url_path_for("profile")
     return RedirectResponse(url=url, status_code=HTTP_302_FOUND)
 
@@ -101,8 +95,6 @@ async def update_profile(
     user.description = new_description
     session.commit()
 
-    session.close()
-
     url = router.url_path_for("profile")
     return RedirectResponse(url=url, status_code=HTTP_302_FOUND)
 
@@ -118,14 +110,27 @@ async def upload_user_photo(
         # Save to database
         user.avatar = await process_image(pic, user)
         session.commit()
-
     finally:
-        session.close()
-
         url = router.url_path_for("profile")
         return RedirectResponse(url=url, status_code=HTTP_302_FOUND)
 
 
+@router.post("/update_telegram_id")
+async def update_telegram_id(
+        request: Request, session=Depends(get_db)):
+
+    user = session.query(User).filter_by(id=1).first()
+    data = await request.form()
+    new_telegram_id = data['telegram_id']
+
+    # Update database
+    user.telegram_id = new_telegram_id
+    session.commit()
+
+    url = router.url_path_for("profile")
+    return RedirectResponse(url=url, status_code=HTTP_302_FOUND)
+
+
 async def process_image(image, user):
     img = Image.open(io.BytesIO(image))
     width, height = img.size

app/routers/telegram.py

@@ -0,0 +1,33 @@
+from fastapi import APIRouter, Body, Depends, Request
+
+from app.database.database import get_db
+from app.database.models import User
+from app.telegram.handlers import MessageHandler, reply_unknown_user
+from app.telegram.models import Chat
+
+
+router = APIRouter(
+    prefix="/telegram",
+    tags=["telegram"],
+    responses={404: {"description": "Not found"}},
+)
+
+
+@router.get("/")
+async def telegram(request: Request, session=Depends(get_db)):
+
+    # todo: Add templating
+    return "Start using PyLander telegram bot!"
+
+
+@router.post("/")
+def bot_client(req: dict = Body(...), session=Depends(get_db)):
+    chat = Chat(req)
+
+    # Check if current chatter is registered to use the bot
+    user = session.query(User).filter_by(telegram_id=chat.user_id).first()
+    if user is None:
+        return reply_unknown_user(chat)
+
+    message = MessageHandler(chat, user)
+    return message.process_callback()

app/telegram/handlers.py

@@ -0,0 +1,111 @@
+import datetime
+
+from .keyboards import (
+    DATE_FORMAT, gen_inline_keyboard, get_this_week_buttons, show_events_kb)
+from .models import Chat
+from .pylander import pylander
+from app.database.models import User
+
+
+class MessageHandler:
+    def __init__(self, chat: Chat, user: User):
+        self.chat = chat
+        self.user = user
+        self.handlers = {}
+        self.handlers['/start'] = self.start_handler
+        self.handlers['/show_events'] = self.show_events_handler
+        self.handlers['Today'] = self.today_handler
+        self.handlers['This week'] = self.this_week_handler
+
+        # Add next 6 days to handlers dict
+        for row in get_this_week_buttons():
+            for button in row:
+                self.handlers[button['text']] = self.chosen_day_handler
+
+    def process_callback(self):
+        if self.chat.message in self.handlers:
+            return self.handlers[self.chat.message]()
+        return self.default_handler()
+
+    def default_handler(self):
+        answer = "Unknown command."
+        pylander.send_message(chat_id=self.chat.user_id, text=answer)
+        return answer
+
+    def start_handler(self):
+        answer = f'''Hello, {self.chat.first_name}!
+Welcome to Pylander telegram client!'''
+        pylander.send_message(chat_id=self.chat.user_id, text=answer)
+        return answer
+
+    def show_events_handler(self):
+        answer = 'Choose events day.'
+        pylander.send_message(
+            chat_id=self.chat.user_id,
+            text=answer,
+            reply_markup=show_events_kb)
+        return answer
+
+    def today_handler(self):
+        today = datetime.date.today()
+        events = [
+            event for event in self.user.events
+            if event.start <= today <= event.end]
+
+        answer = f"{today.strftime('%B %d')}, {today.strftime('%A')} Events:\n"
+
+        if not len(events):
+            answer = "There're no events today."
+
+        for event in events:
+            answer += f'\n\n{event.title}: from {event.start} to {event.ends}.'
+
+        pylander.send_message(chat_id=self.chat.user_id, text=answer)
+        return answer
+
+    def this_week_handler(self):
+        answer = 'Choose a day.'
+        this_week_kb = gen_inline_keyboard(get_this_week_buttons())
+
+        pylander.send_message(
+            chat_id=self.chat.user_id,
+            text=answer,
+            reply_markup=this_week_kb)
+        return answer
+
+    def chosen_day_handler(self):
+        # Convert chosen day (string) to datetime format
+        chosen_date = datetime.datetime.strptime(
+            self.chat.message, DATE_FORMAT)
+
+        events = [
+            event for event in self.user.events
+            if event.start <= chosen_date <= event.end]
+
+        answer = f"{chosen_date.strftime('%B %d')}, \
+{chosen_date.strftime('%A')} Events:\n"
+
+        if not len(events):
+            answer = f"There're no events on {chosen_date.strftime('%B %d')}."
+
+        for event in events:
+            answer += f'\n\n{event.title}: from {event.start} to {event.ends}.'
+
+        pylander.send_message(chat_id=self.chat.user_id, text=answer)
+        return answer
+
+
+def reply_unknown_user(chat):
+    answer = f'''
+Hello, {chat.first_name}!
+
+To use PyLander Bot you have to register
+your Telegram Id in your profile page.
+
+Your Id is {chat.user_id}
+Keep it secret!
+
+https://calendar.pythonic.guru/profile/
+'''
+    pylander.send_message(chat_id=chat.user_id, text=answer)
+    return answer

app/telegram/keyboards.py

@@ -0,0 +1,49 @@
+import datetime
+import json
+from typing import Any, Dict, List
+
+
+show_events_buttons = [
+    [
+        {'text': 'Today', 'callback_data': 'Today'},
+        {'text': 'This week', 'callback_data': 'This week'}
+    ]
+]
+
+DATE_FORMAT = '%d %b %Y'
+
+
+def get_this_week_buttons() -> List[List[Any]]:
+    today = datetime.date.today()
+    day1 = today + datetime.timedelta(days=1)
+    day2 = today + datetime.timedelta(days=2)
+    day3 = today + datetime.timedelta(days=3)
+    day4 = today + datetime.timedelta(days=4)
+    day5 = today + datetime.timedelta(days=5)
+    day6 = today + datetime.timedelta(days=6)
+
+    return [
+        [
+            {'text': day1.strftime(DATE_FORMAT),
+                'callback_data': day1.strftime(DATE_FORMAT)},
+            {'text': day2.strftime(DATE_FORMAT),
+                'callback_data': day2.strftime(DATE_FORMAT)},
+            {'text': day3.strftime(DATE_FORMAT),
+                'callback_data': day3.strftime(DATE_FORMAT)}
+        ],
+        [
+            {'text': day4.strftime(DATE_FORMAT),
+                'callback_data': day4.strftime(DATE_FORMAT)},
+            {'text': day5.strftime(DATE_FORMAT),
+                'callback_data': day5.strftime(DATE_FORMAT)},
+            {'text': day6.strftime(DATE_FORMAT),
+                'callback_data': day6.strftime(DATE_FORMAT)}
+        ]
+    ]
+
+
+def gen_inline_keyboard(buttons: List[List[Any]]) -> Dict[str, Any]:
+    return {'reply_markup': json.dumps({'inline_keyboard': buttons})}
+
+
+show_events_kb = gen_inline_keyboard(show_events_buttons)

app/telegram/models.py

@@ -0,0 +1,59 @@
+from typing import Any, Dict, Optional
+
+import requests
+
+
+class Chat:
+    def __init__(self, data):
+        self.message = self._get_message_content(data)
+        self.user_id = self._get_user_id(data)
+        self.first_name = self._get_first_name(data)
+
+    def _get_message_content(self, data):
+        if 'callback_query' in data:
+            return data['callback_query']['data']
+        return data['message']['text']
+
+    def _get_user_id(self, data):
+        if 'callback_query' in data:
+            return data['callback_query']['from']['id']
+        return data['message']['from']['id']
+
+    def _get_first_name(self, data):
+        if 'callback_query' in data:
+            return data['callback_query']['from']['first_name']
+        return data['message']['from']['first_name']
+
+
+class Bot:
+    def __init__(self, bot_api: str, webhook_url: str):
+        self.base = self._set_base_url(bot_api)
+        self.webhook_setter_url = self._set_webhook_setter_url(webhook_url)
+
+    def _set_base_url(self, bot_api):
+        return f'https://api.telegram.org/bot{bot_api}/'
+
+    def _set_webhook_setter_url(self, webhook_url: str):
+        return f'{self.base}setWebhook?url={webhook_url}/telegram/'
+
+    def set_webhook(self):
+        return requests.get(self.webhook_setter_url)
+
+    def drop_webhook(self):
+        data = {
+            'drop_pending_updates': True
+        }
+        return requests.get(
+            url=f'{self.base}deleteWebhook', data=data)
+
+    def send_message(
+            self, chat_id: str,
+            text: str,
+            reply_markup: Optional[Dict[str, Any]] = None):
+        message = {
+            'chat_id': chat_id,
+            'text': text,
+            'reply_markup': None}
+        if reply_markup:
+            message.update(reply_markup)
+        return requests.post(f'{self.base}sendMessage', data=message)