Merge pull request 'queues' (#5) from queues into dev

Reviewed-on: #5

.deploy/deploy-dev.yaml

@@ -8,8 +8,7 @@ services:
     environment:
       STAGE: "development"
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues-development
     command: poll
     deploy:
       mode: replicated
@@ -26,9 +25,7 @@ services:
       STAGE: "development"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues-development
-      - configurator
     command: worker
     deploy:
       mode: replicated
@@ -43,8 +40,7 @@ services:
     environment:
       STAGE: "development"
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues-development
     command: mailbox
     deploy:
       mode: replicated
@@ -61,8 +57,7 @@ services:
       STAGE: "development"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues-development
     command: fetch
     deploy:
       mode: replicated
@@ -79,8 +74,7 @@ services:
       STAGE: "development"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues-development
     command: notify
     deploy:
       mode: replicated
@@ -99,8 +93,7 @@ services:
       STAGE: "development"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_DEV
-      - queues-development
     command: api
     deploy:
       mode: replicated
@@ -111,9 +104,5 @@ services:
         order: start-first
 
 networks:
-  common-infra-nginx-development:
+  common-infra-nginx:
-    external: true
-  queues-development:
-    external: true
-  configurator:
     external: true

.deploy/deploy-prod.yaml

@@ -8,8 +8,7 @@ services:
     environment:
       STAGE: "production"
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
-      - queues
     command: poll
     deploy:
       mode: replicated
@@ -26,9 +25,7 @@ services:
       STAGE: "production"
       MONGO_PASSWORD: $MONGO_PASSWORD_PROD
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
-      - queues
-      - configurator
     command: worker
     deploy:
       mode: replicated
@@ -43,8 +40,7 @@ services:
     environment:
       STAGE: "production"
       TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
-      - queues
     command: mailbox
     deploy:
       mode: replicated
@@ -62,8 +58,7 @@ services:
       MONGO_PASSWORD: $MONGO_PASSWORD_PROD
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
       DEBUG: "false"
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
-      - queues
     command: fetch
     deploy:
       mode: replicated
@@ -85,8 +80,7 @@ services:
       MONGO_PASSWORD: $MONGO_PASSWORD_PROD
       PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
       DEBUG: "false"
-    networks:
+      QUEUES_TOKEN: $QUEUES_TOKEN_PROD
-      - queues
     command: notify
     deploy:
       mode: replicated
@@ -126,7 +120,3 @@ services:
 networks:
   common-infra-nginx:
     external: true
-  queues:
-    external: true
-  configurator:
-    external: true

.gitea/workflows/deploy-dev.yaml

@@ -28,7 +28,7 @@ jobs:
         run: docker push mathwave/sprint-repo:ruz-bot
   deploy-dev:
     name: Deploy dev
-    runs-on: [prod]
+    runs-on: [dev]
     needs: push
     steps:
       - name: login
@@ -42,4 +42,5 @@ jobs:
           TELEGRAM_TOKEN_DEV: ${{ secrets.TELEGRAM_TOKEN_DEV }}
           MONGO_PASSWORD_DEV: ${{ secrets.MONGO_PASSWORD_DEV }}
           PLATFORM_SECURITY_TOKEN: ${{ secrets.PLATFORM_SECURITY_TOKEN }}
-        run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-dev.yaml ruz-bot-development
+          QUEUES_TOKEN_DEV: ${{ secrets.QUEUES_TOKEN_DEV }}
+        run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-dev.yaml ruz-bot

.gitea/workflows/deploy-prod.yaml

@@ -42,4 +42,5 @@ jobs:
           TELEGRAM_TOKEN_PROD: ${{ secrets.TELEGRAM_TOKEN_PROD }}
           MONGO_PASSWORD_PROD: ${{ secrets.MONGO_PASSWORD_PROD }}
           PLATFORM_SECURITY_TOKEN: ${{ secrets.PLATFORM_SECURITY_TOKEN }}
+          QUEUES_TOKEN_PROD: ${{ secrets.QUEUES_TOKEN_PROD }}
         run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-prod.yaml ruz-bot

daemons/mailbox.py

@@ -5,14 +5,14 @@ from daemons import base
 from utils import queues
 
 
-class Daemon(base.Daemon, queues.TasksHandlerMixin):
+class Daemon(base.BaseDaemon, queues.TasksHandlerMixin):
     def __init__(self):
         super().__init__()
         self.bot = telebot.TeleBot(os.getenv("TELEGRAM_TOKEN"))
 
     @property
     def queue_name(self):
-        return 'ruz_bot_mailbox'
+        return 'pizda_bot_mailbox'
     
     def execute(self):
         self.poll()
@@ -25,7 +25,4 @@ class Daemon(base.Daemon, queues.TasksHandlerMixin):
         reply_markup = payload.get('reply_markup')
         if reply_markup:
             body['reply_markup'] = reply_markup
-        try:
+        self.bot.send_message(**body, parse_mode='Markdown')
-            self.bot.send_message(**body, parse_mode='Markdown')
-        except Exception as exc:
-            print('Error', str(exc))

daemons/poll.py

@@ -6,7 +6,7 @@ from telebot import types
 from utils import queues
 
 
-class Daemon(base.Daemon):
+class Daemon(base.BaseDaemon):
     def execute(self):
         bot = telebot.TeleBot(os.getenv("TELEGRAM_TOKEN"))
         @bot.message_handler()

daemons/worker.py

@@ -5,7 +5,7 @@ import json
 from telebot.types import Message
 
 
-class Daemon(base.Daemon, queues.TasksHandlerMixin):
+class Daemon(base.BaseDaemon, queues.TasksHandlerMixin):
     @property
     def queue_name(self):
         return 'ruz_bot_worker'

helpers/alice.py

@@ -1,9 +1,9 @@
 import logging
 from typing import Optional
 
+from daemons.bot import bot
 from helpers import now
 from helpers.mongo import mongo
-from utils import queues
 
 
 def try_parse(message: str) -> Optional[int]:
@@ -84,7 +84,7 @@ class Processor:
                 }
             else:
                 mongo.users_collection.update_one({"yandex_code": code}, {"$set": {"yandex_id": self.user_id, "yandex_code": None}})
-                queues.set_task('ruz_bot_mailbox', {'text': "Алиса успешно подключена!", 'chat_id': user["chat_id"]}, 1)
+                bot.send_message(user['chat_id'], "Алиса успешно подключена!")
                 lesson = self.get_lesson_for_user(user['chat_id'])
                 if lesson is None:
                     return {
@@ -94,4 +94,4 @@ class Processor:
                 return {
                     "text": f'Отлично, теперь я могу подсказывать тебе расписание. Твое ближайшее занятие в {lesson["begin"].strftime("%A %d %B %H:%M")}: {lesson["discipline"].replace("(рус)", "").replace("(анг)", "")}',
                     "end_session": True
-                }
+                }

helpers/sprint_platform.py

@@ -15,8 +15,11 @@ class PlatformClient:
         self.stage = stage
         self.configs = configs
         self.experiments = experiments
-        self.endpoint = 'http://configurator/'
+        self.endpoint = 'https://platform.sprinthub.ru/'
-        self.fetch_url = urllib.parse.urljoin(self.endpoint, '/api/v1/fetch')
+        self.configs_url = urllib.parse.urljoin(self.endpoint, 'configs/get')
+        self.experiments_url = urllib.parse.urljoin(self.endpoint, 'experiments/get')
+        self.staff_url = urllib.parse.urljoin(self.endpoint, 'is_staff')
+        self.fetch_url = urllib.parse.urljoin(self.endpoint, 'fetch')
         self.config_storage = {}
         self.experiment_storage = {}
         self.staff_storage = {}
@@ -41,6 +44,7 @@ class PlatformClient:
             try:
                 response = get(
                     url,
+                    headers={'X-Security-Token': self.platform_security_token},
                     params=params
                 )
                 if response.status_code == 200:

utils/queues.py

@@ -1,13 +1,18 @@
+import json
 import os
 import requests
 import time
 
 
 stage = os.getenv("STAGE", 'local')
-if stage == 'local':
+if stage == 'development':
-    QUEUES_URL = 'http://localhost:1239'
+    QUEUES_URL = 'https://queues.develop.sprinthub.ru'
+elif stage == 'production':
+    QUEUES_URL = 'https://queues.sprinthub.ru'
 else:
-    QUEUES_URL = 'http://queues:1239'
+    QUEUES_URL = None
+
+token = os.getenv('QUEUES_TOKEN')
 
 
 class QueuesException(Exception):
@@ -17,22 +22,29 @@ 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()
+            if QUEUES_URL is None:
-            task = response.get('task')
+                data = {'payload': json.loads(input('Input message: '))}
-            if not task:
+            else:
-                time.sleep(0.2)
+                response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name, 'X-Queues-Token': token})
-                continue
+                if response.status_code == 404:
+                    time.sleep(0.2)
+                    continue
+                if response.status_code == 403:
+                    raise NotImplemented('QUEUE_TOKEN is incorrect')
+                data = response.json()
             try:
-                self.process(task['payload'])
+                self.process(data['payload'])
             except Exception as exc:
-                print(f'Error processing message id={task["id"]}, payload={task["payload"]}, exc={exc}')
+                print(f'Error processing message id={data["id"]}, payload={data["payload"]}, exc={exc}')
+                continue
+            if QUEUES_URL is None:
                 continue
             try:
-                resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': task['id']})
+                resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': data['id']}, headers={'X-Queues-Token': token})
                 if resp.status_code != 202:
                     raise QueuesException
             except:
-                print(f'Failed to finish task id={task["id"]}')
+                print(f'Failed to finish task id={data["id"]}')
 
     @property
     def queue_name(self):
@@ -43,7 +55,7 @@ class TasksHandlerMixin:
 
 
 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={
+    resp = requests.post(f'{QUEUES_URL}/api/v1/put', headers={'queue': queue_name, 'X-Queues-Token': token}, json={
         'payload': payload,
         'seconds_to_execute': seconds_to_execute,
         'delay': delay,