add algam

This commit is contained in:
2025-09-22 11:54:42 +05:00
parent 3c418a97b4
commit 5b9c55a4dc
34 changed files with 4700 additions and 426 deletions

View File

@@ -5,7 +5,7 @@
from .database import Database, DatabaseConfig, db, get_db_session, init_database, test_db_connection
from .models import User, Message, BotSettings
from .repositories import UserRepository, MessageRepository, SettingsRepository
from .services import DatabaseService, db_service, setup_database
from .services import ImageService, TestService, QuestionService
__all__ = [
# Основные классы
@@ -14,13 +14,11 @@ __all__ = [
# Глобальные объекты
'db',
'db_service',
# Функции
'get_db_session',
'init_database',
'test_db_connection',
'setup_database',
# Модели
'User',
@@ -33,5 +31,7 @@ __all__ = [
'SettingsRepository',
# Сервисы
'DatabaseService',
'ImageService',
'TestService',
'QuestionService',
]

View File

@@ -110,9 +110,10 @@ class Database:
def test_connection(self) -> bool:
"""Проверка подключения к базе данных"""
from sqlalchemy import text
try:
with self.get_session() as session:
session.execute("SELECT 1")
session.execute(text("SELECT 1"))
logger.info("Подключение к базе данных работает корректно")
return True
except Exception as e:
@@ -137,9 +138,29 @@ def init_database():
"""Инициализация базы данных (создание таблиц)"""
db.create_tables()
def test_db_connection():
"""Тестирование подключения к базе данных"""
return db.test_connection()
def test_db_connection(max_retries: int = 5, delay: int = 2):
"""
Тестирование подключения к базе данных с повторными попытками
Args:
max_retries: Максимальное количество попыток
delay: Задержка между попытками в секундах
"""
import time
for attempt in range(max_retries):
try:
if db.test_connection():
return True
except Exception as e:
logger.warning(f"Попытка {attempt + 1}/{max_retries} подключения не удалась: {e}")
if attempt < max_retries - 1:
logger.info(f"Ожидание {delay} секунд перед следующей попыткой...")
time.sleep(delay)
logger.error(f"Не удалось подключиться к базе данных после {max_retries} попыток")
return False
def load_image_from_file(file_path: str, alt_text: str = None):
"""

View File

@@ -1,145 +1,284 @@
"""
Пример использования базы данных
Скрипт для создания примеров данных для системы тестирования
Создает тестовые категории, тесты, вопросы и варианты ответов
"""
import os
import sys
from pathlib import Path
import os
from datetime import datetime
# Добавляем путь к проекту в PYTHONPATH
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
# Добавляем путь к корневой директории проекта
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from database.database import db, test_db_connection
from database.services import db_service, setup_database
import logging
from database.database import db, init_database
from database.models import TestCategory, Test, Question, AnswerOption, Image
from database.repositories import TestRepository, QuestionRepository, ImageRepository
from sqlalchemy.orm import Session
# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def create_sample_data():
"""Создание примеров данных для тестирования"""
print("Создание примеров данных для системы тестирования...")
# Инициализируем базу данных
init_database()
with db.get_session() as session:
# Создаем категории тестов
print("\n1. Создание категорий тестов...")
categories_data = [
{"name": "Программирование", "description": "Тесты по программированию и разработке"},
{"name": "Математика", "description": "Математические тесты и задачи"},
{"name": "Общие знания", "description": "Тесты на общую эрудицию"},
{"name": "История", "description": "Исторические факты и события"},
]
categories = {}
for cat_data in categories_data:
# Проверяем, есть ли уже такая категория
existing = session.query(TestCategory).filter(
TestCategory.name == cat_data["name"]
).first()
if not existing:
category = TestCategory(
name=cat_data["name"],
description=cat_data["description"]
)
session.add(category)
session.flush() # Получаем ID
categories[cat_data["name"]] = category
print(f" ✓ Создана категория: {cat_data['name']}")
else:
categories[cat_data["name"]] = existing
print(f" - Категория уже существует: {cat_data['name']}")
session.commit()
# Создаем тесты
print("\n2. Создание тестов...")
test_repo = TestRepository(session)
tests_data = [
{
"title": "Python для начинающих",
"description": "Основы программирования на Python",
"category": "Программирование",
"time_limit": 30
},
{
"title": "Базовая алгебра",
"description": "Простые алгебраические задачи",
"category": "Математика",
"time_limit": 45
},
{
"title": "Столицы мира",
"description": "Знание столиц различных стран",
"category": "Общие знания",
"time_limit": 15
},
]
tests = {}
for test_data in tests_data:
# Проверяем, есть ли уже такой тест
existing = session.query(Test).filter(Test.title == test_data["title"]).first()
if not existing:
test = test_repo.create_test(
title=test_data["title"],
description=test_data["description"],
category_id=categories[test_data["category"]].id,
time_limit_minutes=test_data["time_limit"]
)
tests[test_data["title"]] = test
print(f" ✓ Создан тест: {test_data['title']}")
else:
tests[test_data["title"]] = existing
print(f" - Тест уже существует: {test_data['title']}")
# Создаем вопросы и варианты ответов
print("\n3. Создание вопросов и вариантов ответов...")
question_repo = QuestionRepository(session)
# Вопросы для теста "Python для начинающих"
python_test = tests["Python для начинающих"]
python_questions = [
{
"text": "Какой оператор используется для присваивания в Python?",
"type": "single_choice",
"order": 1,
"points": 1,
"answers": [
{"text": "=", "correct": True, "order": 1},
{"text": "==", "correct": False, "order": 2},
{"text": ":=", "correct": False, "order": 3},
{"text": "->", "correct": False, "order": 4},
]
},
{
"text": "Как создать список в Python?",
"type": "single_choice",
"order": 2,
"points": 1,
"answers": [
{"text": "list = {}", "correct": False, "order": 1},
{"text": "list = []", "correct": True, "order": 2},
{"text": "list = ()", "correct": False, "order": 3},
{"text": "list = <>", "correct": False, "order": 4},
]
},
{
"text": "Какая функция используется для вывода текста в Python?",
"type": "single_choice",
"order": 3,
"points": 1,
"answers": [
{"text": "echo()", "correct": False, "order": 1},
{"text": "printf()", "correct": False, "order": 2},
{"text": "print()", "correct": True, "order": 3},
{"text": "write()", "correct": False, "order": 4},
]
}
]
_create_questions_for_test(session, question_repo, python_test.id, python_questions)
# Вопросы для теста "Базовая алгебра"
math_test = tests["Базовая алгебра"]
math_questions = [
{
"text": "Чему равно 2 + 2 * 3?",
"type": "single_choice",
"order": 1,
"points": 2,
"answers": [
{"text": "12", "correct": False, "order": 1},
{"text": "8", "correct": True, "order": 2},
{"text": "10", "correct": False, "order": 3},
{"text": "6", "correct": False, "order": 4},
]
},
{
"text": "Решите уравнение: x + 5 = 12",
"type": "single_choice",
"order": 2,
"points": 2,
"answers": [
{"text": "x = 5", "correct": False, "order": 1},
{"text": "x = 7", "correct": True, "order": 2},
{"text": "x = 17", "correct": False, "order": 3},
{"text": "x = 12", "correct": False, "order": 4},
]
}
]
_create_questions_for_test(session, question_repo, math_test.id, math_questions)
# Вопросы для теста "Столицы мира"
capitals_test = tests["Столицы мира"]
capitals_questions = [
{
"text": "Какая столица Франции?",
"type": "single_choice",
"order": 1,
"points": 1,
"answers": [
{"text": "Лондон", "correct": False, "order": 1},
{"text": "Париж", "correct": True, "order": 2},
{"text": "Берлин", "correct": False, "order": 3},
{"text": "Рим", "correct": False, "order": 4},
]
},
{
"text": "Столица Японии?",
"type": "single_choice",
"order": 2,
"points": 1,
"answers": [
{"text": "Осака", "correct": False, "order": 1},
{"text": "Киото", "correct": False, "order": 2},
{"text": "Токио", "correct": True, "order": 3},
{"text": "Хиросима", "correct": False, "order": 4},
]
}
]
_create_questions_for_test(session, question_repo, capitals_test.id, capitals_questions)
print(f"\n✅ Создание примеров данных завершено!")
print(f"Создано категорий: {len(categories)}")
print(f"Создано тестов: {len(tests)}")
# Выводим статистику
print("\n📊 Статистика:")
for test_name, test in tests.items():
question_count = session.query(Question).filter(Question.test_id == test.id).count()
print(f" - {test_name}: {question_count} вопросов")
def main():
"""Основная функция для демонстрации работы с базой данных"""
print("🚀 Пример работы с PostgreSQL через SQLAlchemy")
print("=" * 50)
# 1. Тестирование подключения
print("1. Тестирование подключения к базе данных...")
if test_db_connection():
print("✅ Подключение успешно!")
else:
print("❌ Ошибка подключения!")
return
# 2. Настройка базы данных (создание таблиц)
print("\n2. Настройка базы данных...")
if setup_database():
print("✅ База данных настроена!")
else:
print("❌ Ошибка настройки базы данных!")
return
# 3. Работа с пользователями
print("\n3. Работа с пользователями...")
# Регистрация пользователя
user, created = db_service.register_user(
telegram_id=123456789,
username="test_user",
first_name="Иван",
last_name="Иванов"
)
print(f"Пользователь: {user.first_name} {user.last_name} (@{user.username})")
print(f"Создан новый: {'Да' if created else 'Нет'}")
# Повторная регистрация (должен вернуть существующего)
user2, created2 = db_service.register_user(
telegram_id=123456789,
username="test_user_updated"
)
print(f"Повторная регистрация - создан новый: {'Да' if created2 else 'Нет'}")
# 4. Работа с сообщениями
print("\n4. Работа с сообщениями...")
# Сохранение сообщения
message = db_service.save_message(
telegram_message_id=1001,
telegram_user_id=123456789,
text="Привет! Это тестовое сообщение",
message_type="text"
)
print(f"Сообщение сохранено: ID {message.id if message else 'Ошибка'}")
# Получение сообщений пользователя
messages = db_service.get_user_messages(123456789, limit=10)
print(f"Количество сообщений пользователя: {len(messages)}")
# 5. Работа с настройками
print("\n5. Работа с настройками...")
# Установка настройки
db_service.set_setting("welcome_message", "Добро пожаловать в бот!", "Приветственное сообщение")
# Получение настройки
welcome_msg = db_service.get_setting("welcome_message", "Привет!")
print(f"Приветственное сообщение: {welcome_msg}")
# 6. Статистика
print("\n6. Статистика...")
user_count = db_service.get_user_count()
print(f"Всего пользователей: {user_count}")
print("\n✅ Демонстрация завершена успешно!")
def example_usage_in_telegram_bot():
"""Пример использования в телеграм боте"""
print("\n" + "=" * 50)
print("📱 Пример использования в телеграм боте:")
print("=" * 50)
def _create_questions_for_test(session: Session, question_repo: QuestionRepository, test_id: int, questions_data: list):
"""Вспомогательная функция для создания вопросов теста"""
# Имитация обработки сообщения от пользователя
telegram_user_id = 987654321
message_text = "Какая сегодня погода?"
for q_data in questions_data:
# Проверяем, есть ли уже такой вопрос
existing_question = session.query(Question).filter(
Question.test_id == test_id,
Question.order_number == q_data["order"]
).first()
if not existing_question:
question = question_repo.create_question(
test_id=test_id,
question_text=q_data["text"],
question_type=q_data["type"],
order_number=q_data["order"],
points=q_data["points"]
)
# Создаем варианты ответов
for answer_data in q_data["answers"]:
answer_option = AnswerOption(
question_id=question.id,
option_text=answer_data["text"],
is_correct=answer_data["correct"],
order_number=answer_data["order"]
)
session.add(answer_option)
session.commit()
print(f" ✓ Создан вопрос: {q_data['text'][:50]}...")
else:
print(f" - Вопрос уже существует: {q_data['text'][:50]}...")
def clear_test_data():
"""Очистка тестовых данных"""
print("\n🗑️ Очистка тестовых данных...")
# 1. Регистрируем/получаем пользователя
user, created = db_service.register_user(
telegram_id=telegram_user_id,
username="weather_user",
first_name="Мария"
)
if created:
print(f"🆕 Новый пользователь: {user.first_name}")
# Отправляем приветственное сообщение
welcome_msg = db_service.get_setting("welcome_message", "Привет!")
print(f"Отправляем: {welcome_msg}")
# 2. Сохраняем сообщение
message = db_service.save_message(
telegram_message_id=2001,
telegram_user_id=telegram_user_id,
text=message_text
)
print(f"💬 Сохранено сообщение: '{message_text[:30]}...'")
# 3. Обновляем активность пользователя
db_service.update_user_info(telegram_user_id, is_active=True)
print("✅ Обработка сообщения завершена")
with db.get_session() as session:
# Удаляем в правильном порядке (от дочерних к родительским)
session.query(AnswerOption).delete()
session.query(Question).delete()
session.query(Test).delete()
session.query(TestCategory).delete()
session.commit()
print("✅ Тестовые данные очищены")
if __name__ == "__main__":
try:
main()
example_usage_in_telegram_bot()
except Exception as e:
logger.error(f"Ошибка в примере: {e}")
print(f"❌ Ошибка: {e}")
finally:
# Закрываем подключение
db.close()
print("\n🔒 Подключение к базе данных закрыто")
import argparse
parser = argparse.ArgumentParser(description="Управление примерами данных для системы тестирования")
parser.add_argument("--clear", action="store_true", help="Очистить все тестовые данные")
args = parser.parse_args()
if args.clear:
clear_test_data()
else:
create_sample_data()

View File

@@ -182,4 +182,153 @@ class ImageRepository:
if image:
session.delete(image)
return True
return False
class TestRepository:
"""Репозиторий для работы с тестами"""
def __init__(self, session: Session):
self.session = session
def create_test(
self,
title: str,
description: str = None,
category_id: int = None,
is_active: bool = True,
time_limit_minutes: int = None
) -> 'Test':
"""Создание нового теста"""
from database.models import Test
test = Test(
title=title,
description=description,
category_id=category_id,
is_active=is_active,
time_limit_minutes=time_limit_minutes
)
self.session.add(test)
self.session.commit()
self.session.refresh(test)
return test
def get_test_by_id(self, test_id: int) -> Optional['Test']:
"""Получение теста по ID"""
from database.models import Test
return self.session.query(Test).filter(Test.id == test_id).first()
def get_tests(
self,
is_active: bool = None,
category_id: int = None,
limit: int = None
) -> List['Test']:
"""Получение списка тестов с фильтрацией"""
from database.models import Test
query = self.session.query(Test)
if is_active is not None:
query = query.filter(Test.is_active == is_active)
if category_id is not None:
query = query.filter(Test.category_id == category_id)
if limit:
query = query.limit(limit)
return query.all()
def update_test(self, test_id: int, **kwargs) -> Optional['Test']:
"""Обновление теста"""
from database.models import Test
test = self.session.query(Test).filter(Test.id == test_id).first()
if test:
for key, value in kwargs.items():
if hasattr(test, key):
setattr(test, key, value)
self.session.commit()
self.session.refresh(test)
return test
def delete_test(self, test_id: int) -> bool:
"""Удаление теста"""
from database.models import Test
test = self.session.query(Test).filter(Test.id == test_id).first()
if test:
self.session.delete(test)
self.session.commit()
return True
return False
class QuestionRepository:
"""Репозиторий для работы с вопросами"""
def __init__(self, session: Session):
self.session = session
def create_question(
self,
test_id: int,
question_text: str,
question_type: str = "single_choice",
order_number: int = 1,
points: int = 1,
image_id: int = None
) -> 'Question':
"""Создание нового вопроса"""
from database.models import Question
question = Question(
test_id=test_id,
question_text=question_text,
question_type=question_type,
order_number=order_number,
points=points,
image_id=image_id
)
self.session.add(question)
self.session.commit()
self.session.refresh(question)
return question
def get_question_by_id(self, question_id: int) -> Optional['Question']:
"""Получение вопроса по ID"""
from database.models import Question
return self.session.query(Question).filter(Question.id == question_id).first()
def get_questions_by_test_id(self, test_id: int) -> List['Question']:
"""Получение всех вопросов для теста"""
from database.models import Question
return self.session.query(Question).filter(
Question.test_id == test_id
).order_by(Question.order_number).all()
def update_question(self, question_id: int, **kwargs) -> Optional['Question']:
"""Обновление вопроса"""
from database.models import Question
question = self.session.query(Question).filter(Question.id == question_id).first()
if question:
for key, value in kwargs.items():
if hasattr(question, key):
setattr(question, key, value)
self.session.commit()
self.session.refresh(question)
return question
def delete_question(self, question_id: int) -> bool:
"""Удаление вопроса"""
from database.models import Question
question = self.session.query(Question).filter(Question.id == question_id).first()
if question:
self.session.delete(question)
self.session.commit()
return True
return False