Merge pull request 'master' (#11) from master into prod

Reviewed-on: #11

.deploy/deploy-dev.yaml

@@ -5,13 +5,12 @@ services:
 
   bot:
     image: mathwave/sprint-repo:roulette-bot
-    command: bot
     environment:
-      TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
       MONGO_HOST: "mongo.develop.sprinthub.ru"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
+      STAGE: "development"
     networks:
-      - net
+      - queues-development
     deploy:
       mode: replicated
       restart_policy:
@@ -21,7 +20,5 @@ services:
         order: start-first
 
 networks:
-  net:
-    driver: overlay
-  common-infra-nginx:
+  queues-development:
     external: true

.deploy/deploy-prod.yaml

@@ -5,14 +5,12 @@ services:
 
   bot:
     image: mathwave/sprint-repo:roulette-bot
-    command: bot
-    networks:
-      - net
-      - common-infra-nginx
     environment:
-      TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
       MONGO_HOST: "mongo.sprinthub.ru"
       MONGO_PASSWORD: $MONGO_PASSWORD_PROD
+      STAGE: "production"
+    networks:
+      - queues
     deploy:
       mode: replicated
       restart_policy:
@@ -22,7 +20,5 @@ services:
         order: start-first
 
 networks:
-  net:
-    driver: overlay
-  common-infra-nginx:
+  queues:
     external: true

.gitea/workflows/deploy-dev.yaml

@@ -0,0 +1,43 @@
+name: Deploy Dev
+
+on:
+  pull_request:
+    branches:
+      - dev
+    types: [closed]
+
+jobs:
+  build:
+    name: Build
+    runs-on: [ dev ]
+    steps:
+      - name: login
+        run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
+      - name: checkout
+        uses: actions/checkout@v4
+        with:
+          ref: dev
+      - name: build
+        run: docker build -t mathwave/sprint-repo:roulette-bot .
+  push:
+    name: Push
+    runs-on: [ dev ]
+    needs: build
+    steps:
+      - name: push
+        run: docker push mathwave/sprint-repo:roulette-bot
+  deploy-dev:
+    name: Deploy dev
+    runs-on: [prod]
+    needs: push
+    steps:
+      - name: login
+        run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
+      - name: checkout
+        uses: actions/checkout@v4
+        with:
+          ref: dev
+      - name: deploy
+        env:
+          MONGO_PASSWORD_DEV: ${{ secrets.MONGO_PASSWORD_DEV }}
+        run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-dev.yaml roulette-bot-development

.gitea/workflows/deploy-prod.yaml

@@ -0,0 +1,43 @@
+name: Deploy Prod
+
+on:
+  pull_request:
+    branches:
+      - prod
+    types: [closed]
+
+jobs:
+  build:
+    name: Build
+    runs-on: [ dev ]
+    steps:
+      - name: login
+        run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
+      - name: checkout
+        uses: actions/checkout@v4
+        with:
+          ref: prod
+      - name: build
+        run: docker build -t mathwave/sprint-repo:roulette-bot .
+  push:
+    name: Push
+    runs-on: [ dev ]
+    needs: build
+    steps:
+      - name: push
+        run: docker push mathwave/sprint-repo:roulette-bot
+  deploy-prod:
+    name: Deploy prod
+    runs-on: [prod]
+    needs: push
+    steps:
+      - name: login
+        run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
+      - name: checkout
+        uses: actions/checkout@v4
+        with:
+            ref: prod
+      - name: deploy
+        env:
+          MONGO_PASSWORD_PROD: ${{ secrets.MONGO_PASSWORD_PROD }}
+        run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-prod.yaml roulette-bot

.gitlab-ci.yml

@@ -1,43 +0,0 @@
-stages:
-  - build
-  - deploy-dev
-  - deploy-prod
-
-build:
-  stage: build
-  tags:
-    - dev
-  before_script:
-    - docker login -u mathwave -p $DOCKERHUB_PASSWORD
-  script:
-    - docker build -t mathwave/sprint-repo:roulette-bot .
-    - docker push mathwave/sprint-repo:roulette-bot
-
-.deploy:
-  before_script:
-    - docker login -u mathwave -p $DOCKERHUB_PASSWORD
-
-deploy-dev:
-  extends:
-    - .deploy
-  stage: deploy-dev
-  tags:
-    - dev
-  rules:
-    - if: '$CI_COMMIT_BRANCH == "main"'
-      when: on_success
-    - when: manual
-  script:
-    - docker stack deploy --with-registry-auth -c ./.deploy/deploy-dev.yaml roulette-bot
-
-deploy-prod:
-  extends:
-    - .deploy
-  stage: deploy-prod
-  tags:
-    - prod
-  only:
-    - main
-  when: manual
-  script:
-    - docker stack deploy --with-registry-auth -c ./.deploy/deploy-prod.yaml roulette-bot

Dockerfile

@@ -5,4 +5,4 @@ COPY requirements.txt requirements.txt
 RUN pip install -r requirements.txt
 COPY . .
 ENV PYTHONUNBUFFERED 1
-ENTRYPOINT ["python", "main.py"]
+ENTRYPOINT ["python", "bot.py"]

bot.py

@@ -1,18 +1,20 @@
-import os
-
-import telebot
+import json
 from telebot.types import Message, ReplyKeyboardRemove
 
-from mongo import mongo
-
-bot = telebot.TeleBot(os.getenv("TELEGRAM_TOKEN"))
+from tools.mongo import mongo
+from tools.queues import TasksHandlerMixin, set_task
 
 
-class Core:
-    def __init__(self, message: Message):
+class Core(TasksHandlerMixin):
+    @property
+    def queue_name(self):
+        return 'roulette_bot_worker'
+
+    def process(self, payload):
+        message: Message = Message.de_json(json.dumps(payload))
         self.message = message
         self.chat_id = message.chat.id
-        self.message_text = message.text
+        self.message_text = message.text or message.caption or ""
         print(f'Handled message from {self.chat_id}: {self.message_text}')
         user = mongo.chats_collection.find_one({"chat_id": message.chat.id})
         if user is None:
@@ -25,8 +27,6 @@ class Core:
             doc = user
         self.doc = doc
         self.state = doc['state']
-
-    def process(self):
         if self.message_text.startswith('/'):
             self.exec_command()
             return
@@ -63,10 +63,42 @@ class Core:
     def handle_state_search(self):
         self.send_message('🤖 Поиски собеседника продолжаются')
 
-    def send_message(self, text, chat_id=None, reply_markup=None, remove_keyboard=True, **kwargs):
+    def send_message(self, text, chat_id=None, reply_markup=None, remove_keyboard=True):
+        if not text:
+            return
         if reply_markup is None and remove_keyboard:
             reply_markup = ReplyKeyboardRemove()
-        bot.send_message(chat_id or self.chat_id, text, reply_markup=reply_markup, **kwargs)
+        body = {
+            'chat_id': chat_id or self.chat_id,
+            'text': text,
+        }
+        if reply_markup:
+            body['reply_markup'] = reply_markup.to_json()
+        set_task(
+            'botalka_mailbox', 
+            {
+                'project': 'roulette-bot', 
+                'name': 'telegram-bot',
+                'method': 'send_message',
+                'body': body,
+            },
+            1,
+        )
+
+    def send(self, chat_id, method, **kwargs):
+        set_task(
+            'botalka_mailbox', 
+            {
+                'project': 'roulette-bot', 
+                'name': 'telegram-bot',
+                'method': method,
+                'body': {
+                    'chat_id': chat_id,
+                    **kwargs,
+                },
+            },
+            1,
+        )
 
     def set_state(self, state, chat_ids=None):
         mongo.chats_collection.update_many({"chat_id": {"$in": chat_ids or [self.chat_id]}}, {"$set": {"state": state}})
@@ -79,13 +111,24 @@ class Core:
 
     def handle_state_dialog(self):
         current_dialog = mongo.get_current_dialog(self.chat_id)
-        mongo.create_message(self.message_text, current_dialog['_id'], self.chat_id)
-        if current_dialog['chat_id_1'] == self.chat_id:
-            self.send_message(self.message_text, current_dialog['chat_id_2'])
-        else:
-            self.send_message(self.message_text, current_dialog['chat_id_1'])
+        chat_to_send = current_dialog['chat_id_2'] if current_dialog['chat_id_1'] == self.chat_id else current_dialog['chat_id_1']
+        if self.message.photo:
+            self.send(chat_to_send, 'send_photo', photo=self.message.photo[-1].file_id)
+        if self.message.sticker:
+            self.send(chat_to_send, 'send_data', data=self.message.sticker.file_id, data_type='sticker')
+        if self.message.voice:
+            self.send(chat_to_send, 'send_voice', voice=self.message.voice.file_id)
+        if self.message.video_note:
+            self.send(chat_to_send, 'send_video_note', data=self.message.video_note.file_id)
+        if self.message.animation:
+            self.send(chat_to_send, 'send_animation', data=self.message.animation.file_id)
+        self.send_message(self.message_text, chat_to_send)
 
     def start_new_dialog(self, chat_ids):
+        for chat in chat_ids:
+            current_dialog = mongo.get_current_dialog(chat)
+            if current_dialog:
+                mongo.finish_dialog(current_dialog['_id'])
         self.set_state('search', chat_ids)
         for chat in chat_ids:
             self.send_message("🤖 Начинаю искать собеседника. Сообщу тебе, когда найду его.", chat)
@@ -98,9 +141,5 @@ class Core:
             self.set_state('dialog', [chat, next_chat['chat_id']])
 
 
-def run_bot():
-    @bot.message_handler()
-    def do_action(message: Message):
-        Core(message).process()
-
-    bot.polling()
+if __name__ == '__main__':
+    Core().poll()

main.py

@@ -1,8 +0,0 @@
-import sys
-
-
-if sys.argv[-1] == "bot":
-    from bot import run_bot
-    run_bot()
-else:
-    raise NotImplementedError

requirements.txt

@@ -1,15 +1,20 @@
+async-timeout==4.0.3
 certifi==2022.12.7
 charset-normalizer==3.0.1
 click==8.1.3
 dnspython==2.3.0
 Flask==2.2.3
 idna==3.4
+importlib-metadata==6.7.0
 itsdangerous==2.1.2
 Jinja2==3.1.2
 MarkupSafe==2.1.2
 minio==7.1.13
 pymongo==4.3.3
 pyTelegramBotAPI==4.1.1
+redis==5.0.4
 requests==2.28.2
+typing_extensions==4.7.1
 urllib3==1.26.14
 Werkzeug==2.2.3
+zipp==3.15.0

settings.py

@@ -3,3 +3,11 @@ import os
 MONGO_USER = os.getenv("MONGO_USER", "mongo")
 MONGO_PASSWORD = os.getenv("MONGO_PASSWORD", "password")
 MONGO_HOST = os.getenv("MONGO_HOST", "localhost")
+
+MINIO_HOST = os.getenv("MINIO_HOST", "localhost") + ":9000"
+MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY", "serviceminioadmin")
+MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY", "minioadmin")
+MINIO_BUCKET_NAME = 'ruletka'
+
+REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
+REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None)

tools/mongo.py

@@ -16,9 +16,16 @@ class Mongo:
             ('state', 1)
         ])
         self.dialogs_collection.create_index([
-            ("chat_id", 1),
+            ("chat_id_1", 1),
             ('finished_at', 1)
         ])
+        self.dialogs_collection.create_index([
+            ("chat_id_2", 1),
+            ('finished_at', 1)
+        ])
+        self.messages_collection.create_index([
+            ('dialog_id', 1)
+        ])
 
     def __getitem__(self, item):
         return self.database.get_collection(item)
@@ -44,8 +51,9 @@ class Mongo:
     def get_current_dialog(self, chat_id):
         return self.dialogs_collection.find_one({'$or': [{'chat_id_1': chat_id}, {'chat_id_2': chat_id}], 'finished_at': None})
 
-    def create_message(self, text, dialog_id, sender):
-        self.messages_collection.insert_one({
+    def create_message(self, message_type, text, dialog_id, sender):
+        return self.messages_collection.insert_one({
+            'message_type': message_type,
             'dialog_id': dialog_id,
             'text': text,
             'sender': sender,

tools/queues.py

@@ -0,0 +1,52 @@
+import os
+import requests
+import time
+
+
+stage = os.getenv("STAGE", 'local')
+if stage == 'local':
+    QUEUES_URL = 'http://localhost:1239'
+else:
+    QUEUES_URL = 'http://queues:1239'
+
+
+class QueuesException(Exception):
+    ...
+
+
+class TasksHandlerMixin:
+    def poll(self):
+        while True:
+            response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name}).json()
+            task = response.get('task')
+            if not task:
+                time.sleep(0.2)
+                continue
+            try:
+                self.process(task['payload'])
+            except Exception as exc:
+                print(f'Error processing message id={task["id"]}, payload={task["payload"]}, exc={exc}')
+                continue
+            try:
+                resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': task['id']})
+                if resp.status_code != 202:
+                    raise QueuesException
+            except:
+                print(f'Failed to finish task id={task["id"]}')
+
+    @property
+    def queue_name(self):
+        raise NotImplemented
+
+    def process(self, payload):
+        raise NotImplemented
+
+
+def set_task(queue_name: str, payload: dict, seconds_to_execute: int, delay: int|None = None):
+    resp = requests.post(f'{QUEUES_URL}/api/v1/put', headers={'queue': queue_name}, json={
+        'payload': payload,
+        'seconds_to_execute': seconds_to_execute,
+        'delay': delay,
+    })
+    if resp.status_code != 202:
+        raise QueuesException