self/ruz-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: self:master
Branches Tags
self:master
self:prod
self:dev
self:queues
..
compare: self:queues
Branches Tags
self:prod
self:master
self:dev
self:queues

No commits in common. "master" and "queues" have entirely different histories.
master ... queues

9 changed files with 146 additions and 54 deletions
Show Stats Expand all files Collapse all files

38
.deploy/deploy-dev.yaml
View File

@ -3,12 +3,29 @@ version: "3.4"
services:
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
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
networks:
- queues-development
- configurator
@ -21,6 +38,22 @@ services:
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:
condition: any
update_config:
parallelism: 1
order: start-first
fetch:
image: mathwave/sprint-repo:ruz-bot
environment:
@ -60,13 +93,14 @@ services:
ruz-bot-nginx:
image: mathwave/sprint-repo:ruz-bot
networks:
- common-infra-nginx-development
- queues-development
- common-infra-nginx
environment:
MONGO_HOST: "mongo.develop.sprinthub.ru"
STAGE: "development"
MONGO_PASSWORD: $MONGO_PASSWORD_DEV
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
networks:
- queues-development
command: api
deploy:
mode: replicated

51
.deploy/deploy-prod.yaml
View File

@ -3,12 +3,29 @@ version: "3.4"
services:
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
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
networks:
- queues
- configurator
@ -21,21 +38,41 @@ services:
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
update_config:
parallelism: 1
order: start-first
fetch:
image: mathwave/sprint-repo:ruz-bot
environment:
MONGO_HOST: "mongo.sprinthub.ru"
STAGE: "production"
MONGO_PASSWORD: $MONGO_PASSWORD_PROD
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
DEBUG: "false"
networks:
- queues
- configurator
command: fetch
deploy:
mode: replicated
restart_policy:
condition: any
placement:
constraints:
- node.role == worker
- node.labels.zone == ru
update_config:
parallelism: 1
order: start-first
@ -46,15 +83,19 @@ services:
MONGO_HOST: "mongo.sprinthub.ru"
STAGE: "production"
MONGO_PASSWORD: $MONGO_PASSWORD_PROD
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
DEBUG: "false"
networks:
- queues
- configurator
command: notify
deploy:
mode: replicated
restart_policy:
condition: any
placement:
constraints:
- node.role == worker
- node.labels.zone == ru
update_config:
parallelism: 1
order: start-first
@ -63,17 +104,21 @@ services:
image: mathwave/sprint-repo:ruz-bot
networks:
- common-infra-nginx
- configurator
environment:
MONGO_HOST: "mongo.sprinthub.ru"
STAGE: "production"
MONGO_PASSWORD: $MONGO_PASSWORD_PROD
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
DEBUG: "false"
command: api
deploy:
mode: replicated
restart_policy:
condition: any
placement:
constraints:
- node.role == worker
- node.labels.zone == ru
update_config:
parallelism: 1
order: start-first

1
.gitea/workflows/deploy-dev.yaml
View File

@ -41,4 +41,5 @@ jobs:
env:
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

31
daemons/mailbox.py Normal file
View File

@ -0,0 +1,31 @@
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
try:
self.bot.send_message(**body, parse_mode='Markdown')
except Exception as exc:
print('Error', str(exc))

6
daemons/notify.py
View File

@ -25,7 +25,7 @@ def process():
ans += f"🧑‍🏫 {(lesson['lecturer'] or 'Неизвестно')}\n"
if lesson.get('link', None):
ans += f"🔗 {lesson['link']}"
queues.set_task('botalka_mailbox', {'project': 'ruz-bot', 'name': 'telegram-bot', 'body': {'text': f"Через {user['notify_minutes']} минут у тебя занятие!\n" + ans, 'chat_id': user["chat_id"], 'parse_mode': 'Markdown'}}, 1)
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}}):
@ -40,7 +40,7 @@ def process():
else:
text = ruz.schedule_builder(lessons)
try:
queues.set_task('botalka_mailbox', {'project': 'ruz-bot', 'name': 'telegram-bot', 'body': {'text': f"Уведомляю о занятиях! Твое расписание на {'сегодня' if user.get('daily_notify_today', True) else 'завтра'}:\n" + text, 'chat_id': user["chat_id"], 'parse_mode': 'Markdown'}}, 1)
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,7 +72,7 @@ def process():
else:
mess += "12 часов"
mess += "!\n\nТвоя первая пара:\n\n" + ans
queues.set_task('botalka_mailbox', {'project': 'ruz-bot', 'name': 'telegram-bot', 'body': {'text': mess, 'chat_id': user["chat_id"], 'parse_mode': 'Markdown'}}, 1)
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

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()

4
helpers/answer.py
View File

@ -74,10 +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()
body = {'text': text, 'chat_id': self.user['chat_id'], 'parse_mode': 'Markdown'}
body = {'text': text, 'chat_id': self.user['chat_id']}
if reply_markup:
body['reply_markup'] = reply_markup.to_json()
queues.set_task('botalka_mailbox', {'project': 'ruz-bot', 'name': 'telegram-bot', 'body': body}, 5)
queues.set_task('ruz_bot_mailbox', body, 1)
def set_state(self, state: str):
self.user['state'] = state

3
helpers/ruz.py
View File

@ -12,8 +12,7 @@ fields = [
'date_start',
'date_end',
'lecturer_profiles',
'stream_links',
'type',
'stream_links'
]

51
utils/queues.py
View File

@ -1,7 +1,4 @@
from concurrent.futures import ThreadPoolExecutor
import datetime
import os
import zoneinfo
import requests
import time
@ -18,54 +15,24 @@ class QueuesException(Exception):
class TasksHandlerMixin:
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.executor = ThreadPoolExecutor(max_workers=1)
def _send_metric(self, start: datetime.datetime, end: datetime.datetime, success: bool):
def send():
requests.post(f'{QUEUES_URL}/api/v1/metric', json={
'service': 'ruz-bot',
'queue': self.queue_name,
'success': success,
'timestamp': start.strftime("%Y-%m-%dT%H:%M:%S") + "Z",
"success": success,
"execution_time_ms": (end - start).microseconds // 1000,
"environment": stage,
})
self.executor.submit(send)
def poll(self):
while True:
try:
response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name}).json()
except requests.JSONDecodeError:
print('Unable to decode json')
time.sleep(3)
continue
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
start = datetime.datetime.now(zoneinfo.ZoneInfo("Europe/Moscow"))
try:
print(f'process task with id {task["id"]}, attempt {task["attempt"]}')
self.process(task['payload'])
success = True
except Exception as exc:
print(f'Error processing message id={task["id"]}, payload={task["payload"]}, exc={exc}')
success = False
end = datetime.datetime.now(zoneinfo.ZoneInfo("Europe/Moscow"))
if success:
try:
resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': task['id']})
if resp.status_code != 202:
raise QueuesException
print(f'finish task with id {task["id"]}')
except:
print(f'Failed to finish task id={task["id"]}')
self._send_metric(start, end, success)
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):
@ -82,4 +49,4 @@ def set_task(queue_name: str, payload: dict, seconds_to_execute: int, delay: int
'delay': delay,
})
if resp.status_code != 202:
raise QueuesException
raise QueuesException