Merge pull request 'queues' (#15) from queues into prod

Reviewed-on: #15
This commit is contained in:
emmatveev 2024-11-25 00:07:14 +03:00
commit 26af411bf4
17 changed files with 347 additions and 180 deletions

View File

@ -3,15 +3,49 @@ version: "3.4"
services:
bot:
poll:
image: mathwave/sprint-repo:ruz-bot
environment:
STAGE: "development"
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
networks:
- queues-development
command: poll
deploy:
mode: replicated
restart_policy:
condition: any
update_config:
parallelism: 1
order: start-first
worker:
image: mathwave/sprint-repo:ruz-bot
environment:
MONGO_HOST: "mongo.develop.sprinthub.ru"
STAGE: "development"
MONGO_PASSWORD: $MONGO_PASSWORD_DEV
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
command: bot
networks:
- queues-development
- configurator
command: worker
deploy:
mode: replicated
restart_policy:
condition: any
update_config:
parallelism: 1
order: start-first
mailbox:
image: mathwave/sprint-repo:ruz-bot
environment:
STAGE: "development"
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
networks:
- queues-development
command: mailbox
deploy:
mode: replicated
restart_policy:
@ -26,8 +60,9 @@ services:
MONGO_HOST: "mongo.develop.sprinthub.ru"
STAGE: "development"
MONGO_PASSWORD: $MONGO_PASSWORD_DEV
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
networks:
- queues-development
command: fetch
deploy:
mode: replicated
@ -43,8 +78,9 @@ services:
MONGO_HOST: "mongo.develop.sprinthub.ru"
STAGE: "development"
MONGO_PASSWORD: $MONGO_PASSWORD_DEV
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
networks:
- queues-development
command: notify
deploy:
mode: replicated
@ -62,8 +98,9 @@ services:
MONGO_HOST: "mongo.develop.sprinthub.ru"
STAGE: "development"
MONGO_PASSWORD: $MONGO_PASSWORD_DEV
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
networks:
- queues-development
command: api
deploy:
mode: replicated
@ -74,5 +111,9 @@ services:
order: start-first
networks:
common-infra-nginx:
common-infra-nginx-development:
external: true
queues-development:
external: true
configurator:
external: true

View File

@ -3,24 +3,53 @@ version: "3.4"
services:
bot:
poll:
image: mathwave/sprint-repo:ruz-bot
environment:
STAGE: "production"
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
networks:
- queues
command: poll
deploy:
mode: replicated
restart_policy:
condition: any
update_config:
parallelism: 1
order: start-first
worker:
image: mathwave/sprint-repo:ruz-bot
environment:
MONGO_HOST: "mongo.sprinthub.ru"
STAGE: "production"
MONGO_PASSWORD: $MONGO_PASSWORD_PROD
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
DEBUG: "false"
command: bot
networks:
- queues
- configurator
command: worker
deploy:
mode: replicated
restart_policy:
condition: any
update_config:
parallelism: 1
order: start-first
mailbox:
image: mathwave/sprint-repo:ruz-bot
environment:
STAGE: "production"
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
networks:
- queues
command: mailbox
deploy:
mode: replicated
restart_policy:
condition: any
placement:
constraints:
- node.role == worker
- node.labels.zone == ru
update_config:
parallelism: 1
order: start-first
@ -31,9 +60,10 @@ services:
MONGO_HOST: "mongo.sprinthub.ru"
STAGE: "production"
MONGO_PASSWORD: $MONGO_PASSWORD_PROD
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
DEBUG: "false"
networks:
- queues
command: fetch
deploy:
mode: replicated
@ -53,9 +83,10 @@ services:
MONGO_HOST: "mongo.sprinthub.ru"
STAGE: "production"
MONGO_PASSWORD: $MONGO_PASSWORD_PROD
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
DEBUG: "false"
networks:
- queues
command: notify
deploy:
mode: replicated
@ -77,7 +108,6 @@ services:
MONGO_HOST: "mongo.sprinthub.ru"
STAGE: "production"
MONGO_PASSWORD: $MONGO_PASSWORD_PROD
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
DEBUG: "false"
command: api
@ -96,3 +126,7 @@ services:
networks:
common-infra-nginx:
external: true
queues:
external: true
configurator:
external: true

View File

@ -28,7 +28,7 @@ jobs:
run: docker push mathwave/sprint-repo:ruz-bot
deploy-dev:
name: Deploy dev
runs-on: [dev]
runs-on: [prod]
needs: push
steps:
- name: login
@ -42,4 +42,4 @@ 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
run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-dev.yaml ruz-bot-development

View File

@ -3,73 +3,75 @@ from flask import Flask, request
import settings
from helpers.alice import Processor
from helpers.mongo import mongo
from daemons import base
def api():
app = Flask(__name__)
class Daemon(base.Daemon):
def execute(self):
app = Flask(__name__)
@app.route('/stats/json', methods=['GET'])
def stats_json():
all_users = mongo.users_collection.count_documents({})
teachers = mongo.users_collection.count_documents({"is_teacher": True})
return {
"Всего пользователей": all_users,
"Пользователей прошедших регистрацию": mongo.users_collection.count_documents({'email': {'$ne': None}}),
"Преподавателей": teachers,
"Отписались от уведомлений": mongo.users_collection.count_documents({'notify_minutes': None}),
"Отправлено уведомлений за сегодня": mongo.lessons_collection.count_documents({'notified': True}),
"Проиндексировано занятий из РУЗа": mongo.lessons_collection.count_documents({}),
"Пользователей из Москвы": mongo.users_collection.count_documents(
{'campus': 'Москва'}) + mongo.users_collection.count_documents({'campus': {'$exists': False}}),
"Пользователей из Москвы (регистрация)": mongo.users_collection.count_documents(
{'campus': 'Москва', 'email': {'$ne': None}}) + mongo.users_collection.count_documents(
{'campus': {'$exists': False}, 'email': {'$ne': None}}),
"Пользователей из Перми": mongo.users_collection.count_documents({'campus': 'Пермь'}),
"Пользователей из Перми (регистрация)": mongo.users_collection.count_documents(
{'campus': 'Пермь', 'email': {'$ne': None}}),
"Пользователей из Нижнего Новгорода": mongo.users_collection.count_documents({'campus': 'Нижний Новгород'}),
"Пользователей из Нижнего Новгорода (регистрация)": mongo.users_collection.count_documents(
{'campus': 'Нижний Новгород', 'email': {'$ne': None}}),
"Пользователей из Санкт-Петербурга": mongo.users_collection.count_documents({'campus': 'Санкт-Петербург'}),
"Пользователей из Санкт-Петербурга (регистрация)": mongo.users_collection.count_documents(
{'campus': 'Санкт-Петербург', 'email': {'$ne': None}})
}
@app.route('/stats', methods=['GET'])
def stats():
all_users = mongo.users_collection.count_documents({})
teachers = mongo.users_collection.count_documents({"is_teacher": True})
text = f"Всего пользователей: {all_users}<br>" \
f"Пользователей прошедших регистрацию: {mongo.users_collection.count_documents({'email': {'$ne': None}})}<br>" \
f"Студентов: {all_users - teachers}<br>" \
f"Преподавателей: {teachers}<br>" \
f"Отписались от уведомлений: {mongo.users_collection.count_documents({'notify_minutes': None})}<br>" \
f"Отправлено уведомлений за сегодня: {mongo.lessons_collection.count_documents({'notified': True})}<br>" \
f"Проиндексировано занятий из РУЗа: {mongo.lessons_collection.count_documents({})}<br>" \
f"<br>" \
f"<br>" \
f"Пользователей из Москвы: {mongo.users_collection.count_documents({'campus': 'Москва'}) + mongo.users_collection.count_documents({'campus': {'$exists': False}})}<br>" \
f"Пользователей из Москвы (регистрация): {mongo.users_collection.count_documents({'campus': 'Москва', 'email': {'$ne': None}}) + mongo.users_collection.count_documents({'campus': {'$exists': False}, 'email': {'$ne': None}})}<br>" \
f"Пользователей из Перми: {mongo.users_collection.count_documents({'campus': 'Пермь'})}<br>" \
f"Пользователей из Перми (регистрация): {mongo.users_collection.count_documents({'campus': 'Пермь', 'email': {'$ne': None}})}<br>" \
f"Пользователей из Нижнего Новгорода: {mongo.users_collection.count_documents({'campus': 'Нижний Новгород'})}<br>" \
f"Пользователей из Нижнего Новгорода (регистрация): {mongo.users_collection.count_documents({'campus': 'Нижний Новгород', 'email': {'$ne': None}})}<br>" \
f"Пользователей из Санкт-Петербурга: {mongo.users_collection.count_documents({'campus': 'Санкт-Петербург'})}<br>" \
f"Пользователей из Санкт-Петербурга (регистрация): {mongo.users_collection.count_documents({'campus': 'Санкт-Петербург', 'email': {'$ne': None}})}<br>"
return text
@app.route('/alice', methods=['POST'])
def alice():
req = request.json
processor = Processor(req)
response = {
"version": req['version'],
"session": req['session'],
"response": {
"end_session": False
@app.route('/stats/json', methods=['GET'])
def stats_json():
all_users = mongo.users_collection.count_documents({})
teachers = mongo.users_collection.count_documents({"is_teacher": True})
return {
"Всего пользователей": all_users,
"Пользователей прошедших регистрацию": mongo.users_collection.count_documents({'email': {'$ne': None}}),
"Преподавателей": teachers,
"Отписались от уведомлений": mongo.users_collection.count_documents({'notify_minutes': None}),
"Отправлено уведомлений за сегодня": mongo.lessons_collection.count_documents({'notified': True}),
"Проиндексировано занятий из РУЗа": mongo.lessons_collection.count_documents({}),
"Пользователей из Москвы": mongo.users_collection.count_documents(
{'campus': 'Москва'}) + mongo.users_collection.count_documents({'campus': {'$exists': False}}),
"Пользователей из Москвы (регистрация)": mongo.users_collection.count_documents(
{'campus': 'Москва', 'email': {'$ne': None}}) + mongo.users_collection.count_documents(
{'campus': {'$exists': False}, 'email': {'$ne': None}}),
"Пользователей из Перми": mongo.users_collection.count_documents({'campus': 'Пермь'}),
"Пользователей из Перми (регистрация)": mongo.users_collection.count_documents(
{'campus': 'Пермь', 'email': {'$ne': None}}),
"Пользователей из Нижнего Новгорода": mongo.users_collection.count_documents({'campus': 'Нижний Новгород'}),
"Пользователей из Нижнего Новгорода (регистрация)": mongo.users_collection.count_documents(
{'campus': 'Нижний Новгород', 'email': {'$ne': None}}),
"Пользователей из Санкт-Петербурга": mongo.users_collection.count_documents({'campus': 'Санкт-Петербург'}),
"Пользователей из Санкт-Петербурга (регистрация)": mongo.users_collection.count_documents(
{'campus': 'Санкт-Петербург', 'email': {'$ne': None}})
}
}
response['response'].update(processor.process())
return response
app.run(host="0.0.0.0", port=1238, debug=settings.DEBUG)
@app.route('/stats', methods=['GET'])
def stats():
all_users = mongo.users_collection.count_documents({})
teachers = mongo.users_collection.count_documents({"is_teacher": True})
text = f"Всего пользователей: {all_users}<br>" \
f"Пользователей прошедших регистрацию: {mongo.users_collection.count_documents({'email': {'$ne': None}})}<br>" \
f"Студентов: {all_users - teachers}<br>" \
f"Преподавателей: {teachers}<br>" \
f"Отписались от уведомлений: {mongo.users_collection.count_documents({'notify_minutes': None})}<br>" \
f"Отправлено уведомлений за сегодня: {mongo.lessons_collection.count_documents({'notified': True})}<br>" \
f"Проиндексировано занятий из РУЗа: {mongo.lessons_collection.count_documents({})}<br>" \
f"<br>" \
f"<br>" \
f"Пользователей из Москвы: {mongo.users_collection.count_documents({'campus': 'Москва'}) + mongo.users_collection.count_documents({'campus': {'$exists': False}})}<br>" \
f"Пользователей из Москвы (регистрация): {mongo.users_collection.count_documents({'campus': 'Москва', 'email': {'$ne': None}}) + mongo.users_collection.count_documents({'campus': {'$exists': False}, 'email': {'$ne': None}})}<br>" \
f"Пользователей из Перми: {mongo.users_collection.count_documents({'campus': 'Пермь'})}<br>" \
f"Пользователей из Перми (регистрация): {mongo.users_collection.count_documents({'campus': 'Пермь', 'email': {'$ne': None}})}<br>" \
f"Пользователей из Нижнего Новгорода: {mongo.users_collection.count_documents({'campus': 'Нижний Новгород'})}<br>" \
f"Пользователей из Нижнего Новгорода (регистрация): {mongo.users_collection.count_documents({'campus': 'Нижний Новгород', 'email': {'$ne': None}})}<br>" \
f"Пользователей из Санкт-Петербурга: {mongo.users_collection.count_documents({'campus': 'Санкт-Петербург'})}<br>" \
f"Пользователей из Санкт-Петербурга (регистрация): {mongo.users_collection.count_documents({'campus': 'Санкт-Петербург', 'email': {'$ne': None}})}<br>"
return text
@app.route('/alice', methods=['POST'])
def alice():
req = request.json
processor = Processor(req)
response = {
"version": req['version'],
"session": req['session'],
"response": {
"end_session": False
}
}
response['response'].update(processor.process())
return response
app.run(host="0.0.0.0", port=1238, debug=settings.DEBUG)

3
daemons/base.py Normal file
View File

@ -0,0 +1,3 @@
class Daemon:
def execute(self):
raise NotImplementedError

View File

@ -1,20 +0,0 @@
import os
import telebot
from telebot.types import Message
from helpers.mongo import mongo
bot = telebot.TeleBot(os.getenv("TELEGRAM_TOKEN"))
@bot.message_handler(commands=['start'])
def on_start(message: Message):
mongo.users_collection.delete_many({"chat_id": message.chat.id})
do_action(message)
@bot.message_handler()
def do_action(message: Message):
from helpers.answer import Answer
Answer(message).process()

View File

@ -6,6 +6,8 @@ from helpers import now, campus_timdelta
from helpers.mongo import mongo
from helpers.ruz import ruz
from daemons import base
def fetch_schedule_for_user(user: dict):
today = now(user)
@ -75,16 +77,17 @@ def delete_old():
mongo.lessons_collection.delete_many({"end": {"$lte": datetime.datetime.now() - datetime.timedelta(days=1)}})
def fetch():
while True:
logging.info("fetch start")
begin = datetime.datetime.now()
if begin.hour > 22 or begin.hour < 7:
logging.info("Too late, sleeping")
sleep(30 * 60)
continue
process()
end = datetime.datetime.now()
logging.info('fetch finished')
logging.info("time elapsed %s", (end - begin).total_seconds())
delete_old()
class Daemon(base.Daemon):
def execute(self):
while True:
logging.info("fetch start")
begin = datetime.datetime.now()
if begin.hour > 22 or begin.hour < 7:
logging.info("Too late, sleeping")
sleep(30 * 60)
continue
process()
end = datetime.datetime.now()
logging.info('fetch finished')
logging.info("time elapsed %s", (end - begin).total_seconds())
delete_old()

28
daemons/mailbox.py Normal file
View File

@ -0,0 +1,28 @@
import telebot
import os
from daemons import base
from utils import queues
class Daemon(base.Daemon, queues.TasksHandlerMixin):
def __init__(self):
super().__init__()
self.bot = telebot.TeleBot(os.getenv("TELEGRAM_TOKEN"))
@property
def queue_name(self):
return 'ruz_bot_mailbox'
def execute(self):
self.poll()
def process(self, payload):
body = {
'chat_id': payload['chat_id'],
'text': payload['text'],
}
reply_markup = payload.get('reply_markup')
if reply_markup:
body['reply_markup'] = reply_markup
self.bot.send_message(**body, parse_mode='Markdown')

View File

@ -2,12 +2,12 @@ import datetime
import logging
from time import sleep
from telebot.apihelper import ApiTelegramException
from daemons.bot import bot
from helpers import now
from helpers.mongo import mongo
from helpers.ruz import ruz
from daemons import base
from utils import queues
def process():
@ -25,13 +25,7 @@ def process():
ans += f"🧑‍🏫 {(lesson['lecturer'] or 'Неизвестно')}\n"
if lesson.get('link', None):
ans += f"🔗 {lesson['link']}"
try:
bot.send_message(
user["chat_id"],
f"Через {user['notify_minutes']} минут у тебя занятие!\n" + ans
)
except ApiTelegramException:
pass
queues.set_task('ruz_bot_mailbox', {'text': f"Через {user['notify_minutes']} минут у тебя занятие!\n" + ans, 'chat_id': user["chat_id"]}, 1)
mongo.lessons_collection.update_one({"_id": lesson['_id']}, {"$set": {"notified": True}})
time_now = datetime.datetime.now()
for user in mongo.users_collection.find({"next_daily_notify_time": {"$lte": time_now}}):
@ -46,11 +40,7 @@ def process():
else:
text = ruz.schedule_builder(lessons)
try:
bot.send_message(
user["chat_id"],
f"Уведомляю о занятиях! Твое расписание на {'сегодня' if user.get('daily_notify_today', True) else 'завтра'}:\n" + text,
parse_mode='Markdown'
)
queues.set_task('ruz_bot_mailbox', {'text': f"Уведомляю о занятиях! Твое расписание на {'сегодня' if user.get('daily_notify_today', True) else 'завтра'}:\n" + text, 'chat_id': user["chat_id"]}, 1)
except:
pass
mongo.users_collection.update_one(
@ -72,34 +62,29 @@ def process():
ans += f"🧑‍🏫 {(lesson['lecturer'] or 'Неизвестно')}\n"
if lesson.get('link', None):
ans += f"🔗 {lesson['link']}"
try:
mess = "Пары начутся через "
if user['first_lesson_notify'] == 30:
mess += "30 минут"
elif user['first_lesson_notify'] == 60:
mess += "1 час"
elif user['first_lesson_notify'] == 4 * 60:
mess += "4 часа"
else:
mess += "12 часов"
mess += "!\n\nТвоя первая пара:\n\n" + ans
bot.send_message(
user["chat_id"],
mess
)
except ApiTelegramException:
pass
mess = "Пары начутся через "
if user['first_lesson_notify'] == 30:
mess += "30 минут"
elif user['first_lesson_notify'] == 60:
mess += "1 час"
elif user['first_lesson_notify'] == 4 * 60:
mess += "4 часа"
else:
mess += "12 часов"
mess += "!\n\nТвоя первая пара:\n\n" + ans
queues.set_task('ruz_bot_mailbox', {'text': mess, 'chat_id': user["chat_id"]}, 1)
start_of_day = datetime.datetime(year=time_now.year, month=time_now.month, day=time_now.day)
mongo.lessons_collection.update_many({"begin": {"$gte": start_of_day, "$lt": (start_of_day + datetime.timedelta(days=1))}, "user_email": user["email"]}, {"$set": {"notified_today": True}})
break
def notify():
while True:
logging.info("notify start")
begin = datetime.datetime.now()
process()
end = datetime.datetime.now()
logging.info('notify finished')
logging.info("time elapsed %s", (end - begin).total_seconds())
sleep(63 - end.second)
class Daemon(base.Daemon):
def execute(self):
while True:
logging.info("notify start")
begin = datetime.datetime.now()
process()
end = datetime.datetime.now()
logging.info('notify finished')
logging.info("time elapsed %s", (end - begin).total_seconds())
sleep(63 - end.second)

15
daemons/poll.py Normal file
View File

@ -0,0 +1,15 @@
import os
import telebot
from daemons import base
from telebot import types
from utils import queues
class Daemon(base.Daemon):
def execute(self):
bot = telebot.TeleBot(os.getenv("TELEGRAM_TOKEN"))
@bot.message_handler()
def do_action(message: types.Message):
queues.set_task('ruz_bot_worker', message.json, 1)
bot.polling()

19
daemons/worker.py Normal file
View File

@ -0,0 +1,19 @@
from daemons import base
from utils import queues
import json
from telebot.types import Message
class Daemon(base.Daemon, queues.TasksHandlerMixin):
@property
def queue_name(self):
return 'ruz_bot_worker'
def execute(self):
self.poll()
def process(self, payload):
message: Message = Message.de_json(json.dumps(payload))
from helpers.answer import Answer
Answer(message).process()

View File

@ -2,10 +2,6 @@ import logging.config
import sys
import settings
from daemons.api import api
from daemons.bot import bot
from daemons.fetch import fetch
from daemons.notify import notify
import locale
@ -13,17 +9,25 @@ logging.config.dictConfig(settings.logging_config)
locale.setlocale(locale.LC_TIME, 'ru_RU.UTF-8')
arg = sys.argv[-1]
if arg == "bot":
logging.info("bot is starting")
bot.polling()
if arg == "poll":
logging.info("poll is starting")
from daemons.poll import Daemon
elif arg == 'worker':
logging.info("worker is starting")
from daemons.worker import Daemon
elif arg == 'mailbox':
logging.info("mailbox is starting")
from daemons.mailbox import Daemon
elif arg == "fetch":
logging.info("fetch is starting")
fetch()
from daemons.fetch import Daemon
elif arg == "notify":
logging.info("notify is starting")
notify()
from daemons.notify import Daemon
elif arg == "api":
logging.info("api is starting")
api()
from daemons.api import Daemon
else:
raise ValueError(f"Unknown param {arg}")
Daemon().execute()

View File

@ -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}})
bot.send_message(user['chat_id'], "Алиса успешно подключена!")
queues.set_task('ruz_bot_mailbox', {'text': "Алиса успешно подключена!", 'chat_id': user["chat_id"]}, 1)
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
}
}

View File

@ -4,7 +4,6 @@ from random import choice
from telebot.types import Message, ReplyKeyboardRemove
from daemons.bot import bot
from daemons.fetch import fetch_schedule_for_user
from helpers import get_next_daily_notify_time
from helpers.keyboards import main_keyboard, notify_keyboard, yes_no_keyboard, again_keyboard, no_daily_notify, \
@ -12,6 +11,7 @@ from helpers.keyboards import main_keyboard, notify_keyboard, yes_no_keyboard, a
from helpers.mongo import mongo
from helpers.sprint_platform import platform
from helpers.ruz import ruz
from utils import queues
class User:
@ -32,6 +32,8 @@ class Answer:
def __init__(self, message: Message):
self.message = message
self.message_text = message.text or message.caption or ""
if self.message_text.startswith('/start'):
mongo.users_collection.delete_many({"chat_id": message.chat.id})
user = mongo.users_collection.find_one({"chat_id": message.chat.id})
if user is None:
user = {
@ -72,7 +74,10 @@ class Answer:
def send_message(self, text, reply_markup=None, remove_keyboard=True, **kwargs):
if reply_markup is None and remove_keyboard:
reply_markup = ReplyKeyboardRemove()
bot.send_message(self.user['chat_id'], text, reply_markup=reply_markup, **kwargs)
body = {'text': text, 'chat_id': self.user['chat_id']}
if reply_markup:
body['reply_markup'] = reply_markup.to_json()
queues.set_task('ruz_bot_mailbox', body, 1)
def set_state(self, state: str):
self.user['state'] = state

View File

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

0
utils/__init__.py Normal file
View File

52
utils/queues.py Normal file
View File

@ -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})
if response.status_code == 404:
time.sleep(0.2)
continue
data = response.json()
try:
self.process(data['payload'])
except Exception as exc:
print(f'Error processing message id={data["id"]}, payload={data["payload"]}, exc={exc}')
continue
try:
resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': data['id']})
if resp.status_code != 202:
raise QueuesException
except:
print(f'Failed to finish task id={data["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