krosh/rc-tgbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

final_vers

Browse Source
This commit is contained in:
krosh
2025-02-28 22:23:13 +03:00
parent be24e124bb
commit 3392df2919
11 changed files with 343 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+5
View File
@@ -9,3 +9,8 @@
!/app/plugin
!app/plugin/*
app/bot.py
__pycache__/
.env
app/config.py
+13 -3
View File
@@ -7,8 +7,11 @@ load_dotenv()
class config:
DATE_START = datetime.strptime(os.environ.get("DATE_START"), "%Y-%m-%d").date()
BOT_TOKEN = os.environ.get("API_TOKEN")
DATE_START = datetime.strptime(
f'{os.environ.get("DATE_START")}-{os.environ.get("TIME_START")}',
"%Y-%m-%d-%H:%M",
)
API_TOKEN = os.environ.get("API_TOKEN")
ADMIN_ID = os.environ.get("ADMIN_ID")
ENV = os.getenv("ENV", "PROD")
@@ -30,7 +33,11 @@ class Messages:
WELCOME_LIST = [
"Привет! Это бот Quick dates (Random Coffee). Чтобы зарегистрироваться, пожалуйста, ответь на следующие вопросы." # несколько сообщений подряд
]
HELP = "Помощь"
HELPS = [
"Доступные команды:" "\n/start - начать регистрацию" "n/help - помощь",
"Начало мероприятия: {} \nЭто через целых {} минут!",
]
ERROR = "Произошла ошибка"
ENTER_NAME = "Как тебя зовут?"
ENTER_AGE = "Сколько тебе лет?"
@@ -47,3 +54,6 @@ class Messages:
\nГруппа: {}\
\n\nВсё верно?"
ALREADY_REGISTERED = "Ты уже зарегистрирован. Хочешь изменить данные?"
MATCHING_START = (
"Мы начали подбор пар. Пожалуйста, подтверди, что ты находишься в аудитории"
)
app/db.py
+3
View File
@@ -15,6 +15,9 @@ class DB:
def find(self, query):
return self.collection.find(query)
def find_one(self, query):
return self.collection.find_one(query)
def update(self, query, data):
self.collection.update_one(query, {"$set": data})
app/main.py
+22 -8
View File
@@ -8,6 +8,7 @@ from plugin.user import User
from plugin.admin import Admin, AdminMessages
from plugin.bot_instance import bot, BotCommands
from plugin.register import start_reg_name
from datetime import datetime, timedelta
import time
@@ -61,14 +62,11 @@ def start_message(message: types.Message):
@bot.message_handler(commands=BotCommands.Help)
def help_message(message: types.Message):
bot.send_message(
message.chat.id,
(
Messages.HELP
if not Admin.is_admin(message.chat.id)
else AdminMessages.ADMIN_HELP
),
)
if not Admin.is_admin(message.chat.id):
for msg in Messages.HELPS:
bot.send_message(message.chat.id, msg)
else:
bot.send_message(message.chat.id, AdminMessages.ADMIN_HELP, parse_mode="HTML")
return
@@ -76,7 +74,23 @@ def help_message(message: types.Message):
if __name__ == "__main__":
print("/t--- Bot started ---")
time_for_notif = config.DATE_START
while True:
if config.DATE_START - datetime.now() < timedelta(days=1):
users = User.get_all()
for user in users:
bot.send_message(
user["user_id"],
"Напоминание, что наше мероприятие пройдёт уже завтра! Следите за новостями)",
)
admins = Admin.get_all_admins()
for admin in admins:
bot.send_message(
admin["user_id"],
"Отправлено напоминание о скором начале мероприятия.",
)
time.sleep(10)
exit()
app/plugin/admin.py
+87 -13
View File
@@ -1,7 +1,10 @@
from db import DB
import config
import os
import threading
import time
from plugin.user import User
from plugin.shuffle import Shuffle
from plugin.shuffle import Shuffle, already_matched
from plugin.bot_instance import bot
@@ -29,9 +32,13 @@ class Admin:
return
@staticmethod
def get_all_admins():
return Admin._db.find({})
@staticmethod
def is_admin(user_id):
return Admin._db.find_one({"user_id": user_id}) is not None
return Admin._db.find({"user_id": user_id}) is not None
@staticmethod
def add_admin(new_user_id, name, unique_key) -> int:
@@ -206,8 +213,6 @@ class Admin:
with open("admin_stats.txt", "rb") as f:
bot.send_document(message.chat.id, f)
import os
os.remove("stats.txt")
os.remove("admin_stats.txt")
@@ -242,7 +247,7 @@ class Admin:
def clear_all_command(message):
bot.send_message(
message.chat.id,
"Отправьте <code>delete all</code> для подтверждения",
"Отправьте <code>delete all</code> для подтверждения. \n\nВнимание, вы вряд ли хотите это делать.",
parse_mode="HTML",
)
@@ -263,8 +268,76 @@ class Admin:
@bot.message_handler(commands=["random"])
@admin_only
def random_command(message):
users = User.get_all()
Shuffle(users)
already_matched.clear()
for admin in Admin.get_all_admins():
bot.send_message(
admin["user_id"],
"Отправлено приглашение пользователем. В данный момент подтвердило участие 0 человек. Запускайте команду /end_random для завершения регистрации",
)
User.start_shuffle_reg()
def monitor_participation():
while True:
participants = len(already_matched)
bot.send_message(
message.chat.id,
f"В данный момент подтвердило участие {participants} человек",
)
time.sleep(5)
thread = threading.Thread(
target=monitor_participation, name="monitor_participation"
)
thread.start()
return
@bot.message_handler(commands=["end_random", "random_again"])
@admin_only
def end_random_command(message):
for thread in threading.enumerate():
if thread.name == "monitor_participation":
thread.kill()
bot.send_message(message.chat.id, "Регистрация завершена. Ожидайте")
pairs = Shuffle()
for pair in pairs:
if pair[1]:
bot.send_message(
pair[0].user_id,
f"Ваша пара: {pair[1].name}. Парта находится под номером: {pairs.index(pair) + 1})",
)
bot.send_message(
pair[1].user_id,
f"Ваша пара: {pair[0].name}. Парта находится под номером: {pairs.index(pair) + 1})",
)
else:
bot.send_message(
pair[0].user_id,
"Подойди к организаторам мероприятия, они тебе подберут идеальную пару!",
)
with open("pairs.txt", "w") as f:
str_pair = ""
for pair in pairs:
str_pair += (
f"Парта {pairs.index(pair) + 1}: {pair[0].name} | {pair[1].name}\n"
)
f.write(str_pair)
with open("pairs.txt", "rb") as f:
for admin in Admin.get_all_admins():
bot.send_document(
admin["user_id"],
f,
caption="Распредение по парам.",
)
os.remove("pairs.txt")
return
@@ -273,10 +346,11 @@ class AdminMessages:
ADMIN_HELP = (
"Вы администратор! Вот что вы можете сделать:\n"
"/remove_admin - удалить админа\n"
"/generate_key - сгенерировать уникальный ключ"
"/delete_user - удалить пользователя"
"/stats - статистика"
"/send_message - отправить всем сообщение"
"/random - зарандомить людей для 1 тура"
"clear_all - удаляет ВСЕХ пользователей"
"/generate_key - сгенерировать уникальный ключ\n"
"/delete_user - удалить пользователя\n"
"/stats - статистика\n"
"/send_message - отправить всем сообщение\n"
"/random - зарандомить людей для 1 тура\n"
"/random_again - зарандомить людей без повторной регистрации\n"
"<code>clear_all</code> - удаляет ВСЕХ пользователей для их повторной регистрации\n"
)
app/plugin/bot_instance.py
+2 -2
View File
@@ -1,5 +1,5 @@
import telebot
import config
from config import config
import threading
@@ -9,7 +9,7 @@ class BotCommands:
bot = telebot.TeleBot(
config.BOT_TOKEN,
config.API_TOKEN,
colorful_logs=True,
)
app/plugin/register.py
+16 -1
View File
@@ -4,6 +4,8 @@ from config import Messages
from plugin.test import TestMessages
from plugin.bot_instance import bot
import os
import re
def start_reg_name(message: types.Message):
@@ -159,11 +161,24 @@ def show_result(message: types.Message):
TestMessages.TEST_RESULTS[user.type][0],
TestMessages.TEST_RESULTS[user.type][1],
)
directory = "pics/"
pattern = re.compile(f"{user.type}.*\.(jpg|jpeg|png|gif)$", re.IGNORECASE)
image_path = None
for filename in os.listdir(directory):
if pattern.match(filename):
image_path = os.path.join(directory, filename)
break
if image_path:
bot.send_photo(
message.chat.id,
open(f"pics/{user.type}.jpg", "rb"),
open(image_path, "rb"),
caption=text,
)
else:
print("Image not found")
bot.send_message(message.chat.id, TestMessages.TEST_FINISH)
app/plugin/shuffle.py
+152 -3
View File
@@ -1,6 +1,155 @@
from plugin.user import User
from plugin.user import User, users_reg
from typing import List, Tuple
import random
already_matched: List[Tuple[User, User]] = []
def Shuffle(users: list[User]):
def Shuffle() -> List[Tuple[User, User]]:
users = users_reg.copy()
random.shuffle(users)
pairs = []
return
if len(users) % 2 != 0:
pairs.append((users.pop(),))
# Step 1: Match men with women by compatibility and age
for user in users:
for potential_match in users:
if user == potential_match:
continue
if (
user.gender != potential_match.gender
and potential_match.type in compatibility[user.type]
and abs(user.age - potential_match.age) <= 5
and (user, potential_match) not in already_matched
and (potential_match, user) not in already_matched
):
pairs.append((user, potential_match))
users.remove(user)
users.remove(potential_match)
break
# Step 2: Match men with women with compatibility
for user in users:
for potential_match in users:
if user == potential_match:
continue
if (
user.gender != potential_match.gender
and potential_match.type in compatibility[user.type]
and (user, potential_match) not in already_matched
and (potential_match, user) not in already_matched
):
pairs.append((user, potential_match))
users.remove(user)
users.remove(potential_match)
break
# Step 3: Match men with women
for user in users:
for potential_match in users:
if user == potential_match:
continue
if (
user.gender != potential_match.gender
and (user, potential_match) not in already_matched
and (potential_match, user) not in already_matched
):
pairs.append((user, potential_match))
users.remove(user)
users.remove(potential_match)
break
# Step 4: Match people of the same age
for user in users:
for potential_match in users:
if user == potential_match:
continue
if (
user.age == potential_match.age
and (user, potential_match) not in already_matched
and (potential_match, user) not in already_matched
):
pairs.append((user, potential_match))
users.remove(user)
users.remove(potential_match)
break
# Step 5: Pair all remaining users
while len(users) > 1:
pairs.append((users.pop(), users.pop()))
already_matched.extend(pairs)
return pairs
compatibility = {
"INTJ": (
"ENFP",
"ENTP",
),
"INTP": (
"ENTJ",
"ESTJ",
),
"INFJ": (
"ENFP",
"ENTP",
),
"INFP": (
"ENFJ",
"ENTJ",
),
"ISTJ": (
"ESFP",
"ESTP",
),
"ISTP": (
"ESFJ",
"ESTJ",
),
"ISFJ": (
"ESFP",
"ESTP",
),
"ISFP": (
"ENFJ",
"ESFJ",
"ESTJ",
),
"ENTJ": (
"INFP",
"INTP",
),
"ENTP": (
"INTJ",
"ENTP",
),
"ENFJ": (
"INFP",
"ISFP",
"ENFJ",
),
"ENFP": (
"INFJ",
"INTJ",
),
"ESTJ": (
"ISFP",
"ISTP",
),
"ESTP": (
"ISFJ",
"ISTJ",
),
"ESFJ": (
"ISFP",
"ISTP",
),
"ESFP": (
"ISTJ",
"ISFJ",
),
}
app/plugin/user.py
+36
View File
@@ -1,4 +1,10 @@
from db import DB
from plugin.bot_instance import bot
from config import Messages
from telebot import types
from typing import List, Tuple
users_reg: List["User"] = []
class User:
@@ -119,12 +125,42 @@ class User:
data = self._db.find_one({"_id": self.user_id})
self._name = data["name"]
self._age = data["age"]
self._gender = data["gender"]
self._faculty = data["faculty"]
self._group = data["group"]
self._type = data["type"]
@staticmethod
def start_shuffle_reg():
users_reg.clear()
users = [user for user in User.get_all() if user.type]
markup = types.InlineKeyboardMarkup()
markup.add(
types.InlineKeyboardButton("Я тут!", callback_data="shuffle_agree"),
)
for user in users:
bot.send_message(user.user_id, Messages.MATCHING_START, reply_markup=markup)
@bot.callback_query_handler(func=lambda call: call.data == "shuffle_agree")
def add_user_to_reg(call: types.CallbackQuery):
user = User(call.message.chat.id)
users_reg.append(user)
bot.edit_message_text(
call.message.chat.id,
call.message.message_id,
"Вы добавлены в список участников! Пожалуйста, подождите, пока все зарегистрируются",
reply_markup=None,
)
return
return
def __str__(self):
text = [f"{var}: {vars(self)[var]}" for var in vars(self) if var]
text = "\n".join(text)
requirements.txt → app/requirements.txt
View File
docker-compose.yaml
+2 -2
View File
@@ -6,12 +6,12 @@ services:
container_name: python_bot
working_dir: /usr/src/app
volumes:
- .:/usr/src/app
- ./app:/usr/src/app
networks:
- rc_network
depends_on:
- mongodb
command: sh -c "pip install -r requirements.txt && python bot.py"
command: sh -c "pip install -r requirements.txt && python main.py"
mongodb:
image: mongo:latest