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

add stuckt

Browse Source
This commit is contained in:
jze9
2025-09-22 12:35:41 +05:00
parent 5b9c55a4dc
commit 046b610a34
28 changed files with 937 additions and 3576 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
DOCKER_COMMANDS.md
View File

@@ -1,22 +1,46 @@
# Docker команды для управления Telegram Bot API
# Docker команды для Telegram Bot Project
## Основные команды
## Новая архитектура: БОТ → API → БД
### 🚀 Быстрый старт
### Запуск и остановка
```bash
# Запуск всех сервисов
docker-compose up -d
# Запуск только API + БД (для разработки)
docker-compose --profile api up -d
# Запуск с пересборкой
docker-compose up --build -d
# Запуск всех компонентов (API + БОТ + ВЕБ)
docker-compose --profile all up -d
# Запуск с админкой
docker-compose --profile admin up -d
# Разработка с hot reload
docker-compose -f docker-compose.dev.yml up -d
```
# Остановка
### 📦 Отдельные компоненты
```bash
# Только база данных
docker-compose up postgres -d
# Только API сервер (обращается к БД)
docker-compose --profile api up api -d
# Только бот (обращается к API)
docker-compose --profile bot up bot -d
# Только веб-панель (обращается к API)
docker-compose --profile web up web -d
```
### 🔧 Управление
```bash
# Остановка всех сервисов
docker-compose down
# Остановка с удалением данных
# Пересборка и запуск
docker-compose --profile all up --build -d
# Удаление данных
docker-compose down -v
```

8
Dockerfile
View File

@@ -19,14 +19,18 @@ RUN uv sync --frozen
# Этап 2: Производственный образ
FROM python:3.12-slim AS production
# Устанавливаем curl для healthcheck
RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/*
WORKDIR /app
# Копируем виртуальное окружение с установленными пакетами
COPY --from=builder /app/.venv /app/.venv
# Копируем исходный код приложения
COPY api/ ./api/
COPY database/ ./database/
COPY app.api/ ./app.api/
COPY app.bot/ ./app.bot/
COPY app.control-bot/ ./app.control-bot/
COPY *.py ./
COPY pyproject.toml ./

43
api/Dockerfile
View File

@@ -1,43 +0,0 @@
# Dockerfile для API сервера
FROM python:3.12-slim
# Устанавливаем рабочую директорию
WORKDIR /app
# Устанавливаем системные зависимости
RUN apt-get update && apt-get install -y \
curl \
&& rm -rf /var/lib/apt/lists/*
# Копируем файлы зависимостей
COPY requirements.txt* ./
COPY pyproject.toml* ./
# Устанавливаем Python зависимости
RUN pip install --no-cache-dir --upgrade pip
RUN if [ -f requirements.txt ]; then pip install --no-cache-dir -r requirements.txt; fi
RUN if [ -f pyproject.toml ]; then pip install --no-cache-dir -e .; fi
# Устанавливаем основные зависимости для FastAPI
RUN pip install --no-cache-dir \
fastapi \
uvicorn[standard] \
sqlalchemy \
python-multipart \
python-dotenv
# Копируем исходный код
COPY . .
# Создаем директорию для данных
RUN mkdir -p /app/data
# Экспонируем порт
EXPOSE 8000
# Проверка здоровья
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# Команда запуска
CMD ["python", "run_api.py"]

27
api/__init__.py
View File

@@ -1,27 +0,0 @@
"""
API пакет для Telegram Bot
"""
from .main import app
from .routers import users_router, messages_router, settings_router
from .schemas import (
UserCreate, UserUpdate, UserResponse,
MessageCreate, MessageResponse,
BotSettingCreate, BotSettingUpdate, BotSettingResponse
)
__version__ = "1.0.0"
__all__ = [
"app",
"users_router",
"messages_router",
"settings_router",
"UserCreate",
"UserUpdate",
"UserResponse",
"MessageCreate",
"MessageResponse",
"BotSettingCreate",
"BotSettingUpdate",
"BotSettingResponse"
]

37
api/config.py
View File

@@ -1,37 +0,0 @@
"""
Конфигурация для API приложения
"""
import os
from dotenv import load_dotenv
# Загрузка переменных окружения из .env файла
load_dotenv()
class APIConfig:
"""Конфигурация API приложения"""
# Основные настройки
TITLE = "Telegram Bot API"
DESCRIPTION = "API для взаимодействия с Telegram ботом и базой данных"
VERSION = "1.0.0"
# Настройки сервера
HOST = os.getenv("API_HOST", "0.0.0.0")
PORT = int(os.getenv("API_PORT", "8000"))
RELOAD = os.getenv("API_RELOAD", "true").lower() == "true"
# Настройки CORS
CORS_ORIGINS = ["*"] # В продакшене стоит ограничить
CORS_CREDENTIALS = True
CORS_METHODS = ["*"]
CORS_HEADERS = ["*"]
# Настройки логирования
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
# API префикс
API_V1_PREFIX = "/api/v1"
# Глобальный экземпляр конфигурации
config = APIConfig()

106
api/main.py
View File

@@ -1,106 +0,0 @@
"""
FastAPI приложение для взаимодействия с Telegram ботом
"""
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from contextlib import asynccontextmanager
import logging
import sys
import os
# Добавляем путь к корневой директории проекта
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from database.database import db, get_db_session, init_database, test_db_connection
from database.models import User, Message, BotSettings
from api.routers import (
users_router, messages_router, settings_router,
test_categories_router, tests_router, questions_router,
answer_options_router, images_router
)
from api.schemas import HealthCheckResponse, RootResponse
from api.config import config
# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Управление жизненным циклом приложения"""
# Инициализация при запуске
logger.info("Запуск FastAPI приложения...")
# Проверка подключения к БД
if not test_db_connection():
logger.error("Не удалось подключиться к базе данных!")
raise Exception("Database connection failed")
# Создание таблиц если они не существуют
try:
init_database()
logger.info("База данных инициализирована")
except Exception as e:
logger.error(f"Ошибка инициализации базы данных: {e}")
raise
yield
# Очистка при завершении
logger.info("Завершение работы FastAPI приложения...")
db.close()
# Создание FastAPI приложения
app = FastAPI(
title=config.TITLE,
description=config.DESCRIPTION,
version=config.VERSION,
lifespan=lifespan
)
# Настройка CORS
app.add_middleware(
CORSMiddleware,
allow_origins=config.CORS_ORIGINS,
allow_credentials=config.CORS_CREDENTIALS,
allow_methods=config.CORS_METHODS,
allow_headers=config.CORS_HEADERS,
)
# Подключение роутеров
app.include_router(users_router, prefix=config.API_V1_PREFIX)
app.include_router(messages_router, prefix=config.API_V1_PREFIX)
app.include_router(settings_router, prefix=config.API_V1_PREFIX)
# Подключение роутеров для системы тестирования
app.include_router(test_categories_router, prefix=config.API_V1_PREFIX)
app.include_router(tests_router, prefix=config.API_V1_PREFIX)
app.include_router(questions_router, prefix=config.API_V1_PREFIX)
app.include_router(answer_options_router, prefix=config.API_V1_PREFIX)
app.include_router(images_router, prefix=config.API_V1_PREFIX)
@app.get("/", response_model=RootResponse)
async def root():
"""Корневой эндпоинт"""
return RootResponse(
message=config.TITLE,
version=config.VERSION,
status="running"
)
@app.get("/health", response_model=HealthCheckResponse)
async def health_check():
"""Проверка состояния приложения"""
db_status = test_db_connection()
return HealthCheckResponse(
status="healthy" if db_status else "unhealthy",
database="connected" if db_status else "disconnected"
)
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host=config.HOST, port=config.PORT, reload=config.RELOAD)

460
api/routers.py
View File

@@ -1,460 +0,0 @@
"""
API маршруты для работы с пользователями, сообщениями и настройками
"""
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from typing import List, Optional
import sys
import os
# Добавляем путь к корневой директории проекта
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from database.database import db
from database.models import User, Message, BotSettings
from api.schemas import (
UserCreate, UserUpdate, UserResponse,
MessageCreate, MessageResponse,
BotSettingCreate, BotSettingUpdate, BotSettingResponse,
MessageResponseModel, ErrorResponse
)
# Зависимость для получения сессии БД
def get_db():
"""Получение сессии базы данных"""
session = db.get_session_sync()
try:
yield session
finally:
session.close()
# Роутер для работы с пользователями
users_router = APIRouter(prefix="/users", tags=["Users"])
@users_router.get("/{telegram_id}", response_model=UserResponse)
async def get_user_by_telegram_id(telegram_id: int, db_session: Session = Depends(get_db)):
"""Получение пользователя по Telegram ID"""
user = db_session.query(User).filter(User.telegram_id == telegram_id).first()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Пользователь не найден"
)
return user
@users_router.get("/", response_model=List[UserResponse])
async def get_all_users(
limit: int = Query(default=100, ge=1, le=1000),
offset: int = Query(default=0, ge=0),
is_active: Optional[bool] = Query(default=None),
db_session: Session = Depends(get_db)
):
"""Получение списка всех пользователей с фильтрацией"""
query = db_session.query(User)
if is_active is not None:
query = query.filter(User.is_active == is_active)
users = query.offset(offset).limit(limit).all()
return users
@users_router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(user_data: UserCreate, db_session: Session = Depends(get_db)):
"""Создание нового пользователя"""
# Проверяем, существует ли уже пользователь
existing_user = db_session.query(User).filter(User.telegram_id == user_data.telegram_id).first()
if existing_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Пользователь уже существует"
)
# Создаем нового пользователя
new_user = User(**user_data.model_dump())
db_session.add(new_user)
db_session.commit()
db_session.refresh(new_user)
return new_user
@users_router.put("/{telegram_id}", response_model=UserResponse)
async def update_user(
telegram_id: int,
user_data: UserUpdate,
db_session: Session = Depends(get_db)
):
"""Обновление данных пользователя"""
user = db_session.query(User).filter(User.telegram_id == telegram_id).first()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Пользователь не найден"
)
# Обновляем только переданные поля
update_data = user_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(user, field, value)
db_session.commit()
db_session.refresh(user)
return user
@users_router.delete("/{telegram_id}", response_model=MessageResponseModel)
async def delete_user(telegram_id: int, db_session: Session = Depends(get_db)):
"""Деактивация пользователя (мягкое удаление)"""
user = db_session.query(User).filter(User.telegram_id == telegram_id).first()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Пользователь не найден"
)
user.is_active = False
db_session.commit()
return MessageResponseModel(message="Пользователь деактивирован")
# Роутер для работы с сообщениями
messages_router = APIRouter(prefix="/messages", tags=["Messages"])
@messages_router.get("/", response_model=List[MessageResponse])
async def get_messages(
user_id: Optional[int] = Query(default=None),
message_type: Optional[str] = Query(default=None),
limit: int = Query(default=100, ge=1, le=1000),
offset: int = Query(default=0, ge=0),
db_session: Session = Depends(get_db)
):
"""Получение сообщений с фильтрацией"""
query = db_session.query(Message)
if user_id:
query = query.filter(Message.user_id == user_id)
if message_type:
query = query.filter(Message.message_type == message_type)
messages = query.order_by(Message.created_at.desc()).offset(offset).limit(limit).all()
return messages
@messages_router.get("/{message_id}", response_model=MessageResponse)
async def get_message(message_id: int, db_session: Session = Depends(get_db)):
"""Получение конкретного сообщения по ID"""
message = db_session.query(Message).filter(Message.id == message_id).first()
if not message:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Сообщение не найдено"
)
return message
@messages_router.post("/", response_model=MessageResponse, status_code=status.HTTP_201_CREATED)
async def create_message(message_data: MessageCreate, db_session: Session = Depends(get_db)):
"""Создание нового сообщения"""
# Проверяем, существует ли пользователь
user = db_session.query(User).filter(User.id == message_data.user_id).first()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Пользователь не найден"
)
new_message = Message(**message_data.model_dump())
db_session.add(new_message)
db_session.commit()
db_session.refresh(new_message)
return new_message
# Роутер для работы с настройками бота
settings_router = APIRouter(prefix="/settings", tags=["Bot Settings"])
@settings_router.get("/", response_model=dict)
async def get_all_settings(db_session: Session = Depends(get_db)):
"""Получение всех настроек бота в виде словаря"""
settings = db_session.query(BotSettings).all()
return {setting.key: setting.value for setting in settings}
@settings_router.get("/list", response_model=List[BotSettingResponse])
async def get_all_settings_list(db_session: Session = Depends(get_db)):
"""Получение всех настроек бота в виде списка объектов"""
settings = db_session.query(BotSettings).all()
return settings
@settings_router.get("/{key}", response_model=BotSettingResponse)
async def get_setting(key: str, db_session: Session = Depends(get_db)):
"""Получение конкретной настройки по ключу"""
setting = db_session.query(BotSettings).filter(BotSettings.key == key).first()
if not setting:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Настройка не найдена"
)
return setting
@settings_router.post("/", response_model=BotSettingResponse, status_code=status.HTTP_201_CREATED)
async def create_or_update_setting(
setting_data: BotSettingCreate,
db_session: Session = Depends(get_db)
):
"""Создание или обновление настройки"""
existing_setting = db_session.query(BotSettings).filter(BotSettings.key == setting_data.key).first()
if existing_setting:
# Обновляем существующую настройку
update_data = setting_data.model_dump(exclude={"key"})
for field, value in update_data.items():
if value is not None:
setattr(existing_setting, field, value)
db_session.commit()
db_session.refresh(existing_setting)
return existing_setting
else:
# Создаем новую настройку
new_setting = BotSettings(**setting_data.model_dump())
db_session.add(new_setting)
db_session.commit()
db_session.refresh(new_setting)
return new_setting
@settings_router.put("/{key}", response_model=BotSettingResponse)
async def update_setting(
key: str,
setting_data: BotSettingUpdate,
db_session: Session = Depends(get_db)
):
"""Обновление существующей настройки"""
setting = db_session.query(BotSettings).filter(BotSettings.key == key).first()
if not setting:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Настройка не найдена"
)
update_data = setting_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(setting, field, value)
db_session.commit()
db_session.refresh(setting)
return setting
@settings_router.delete("/{key}", response_model=MessageResponseModel)
async def delete_setting(key: str, db_session: Session = Depends(get_db)):
"""Удаление настройки"""
setting = db_session.query(BotSettings).filter(BotSettings.key == key).first()
if not setting:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Настройка не найдена"
)
db_session.delete(setting)
db_session.commit()
return MessageResponseModel(message=f"Настройка '{key}' удалена")
# Добавляем импорты для работы с тестами
from database.models import Test, TestCategory, Question, AnswerOption, Image
from database.services import ImageService, TestService, QuestionService
from database.repositories import TestRepository, QuestionRepository, ImageRepository
from api.schemas import (
TestCategoryCreate, TestCategoryResponse, TestCreate, TestResponse,
QuestionCreate, QuestionResponse, AnswerOptionCreate, AnswerOptionResponse,
ImageUploadResponse
)
from fastapi import UploadFile, File
# Роутер для работы с категориями тестов
test_categories_router = APIRouter(prefix="/test-categories", tags=["Test Categories"])
@test_categories_router.post("/", response_model=TestCategoryResponse)
async def create_test_category(
category_data: TestCategoryCreate,
db_session: Session = Depends(get_db)
):
"""Создание новой категории тестов"""
category = TestCategory(
name=category_data.name,
description=category_data.description
)
db_session.add(category)
db_session.commit()
db_session.refresh(category)
return category
@test_categories_router.get("/", response_model=List[TestCategoryResponse])
async def get_test_categories(db_session: Session = Depends(get_db)):
"""Получение всех категорий тестов"""
categories = db_session.query(TestCategory).all()
return categories
# Роутер для работы с тестами
tests_router = APIRouter(prefix="/tests", tags=["Tests"])
@tests_router.post("/", response_model=TestResponse)
async def create_test(
test_data: TestCreate,
db_session: Session = Depends(get_db)
):
"""Создание нового теста"""
test_repo = TestRepository(db_session)
test = test_repo.create_test(
title=test_data.title,
description=test_data.description,
category_id=test_data.category_id,
is_active=test_data.is_active,
time_limit_minutes=test_data.time_limit_minutes
)
return test
@tests_router.get("/", response_model=List[TestResponse])
async def get_tests(
is_active: Optional[bool] = Query(default=None),
category_id: Optional[int] = Query(default=None),
db_session: Session = Depends(get_db)
):
"""Получение списка тестов с фильтрацией"""
test_repo = TestRepository(db_session)
tests = test_repo.get_tests(is_active=is_active, category_id=category_id)
return tests
@tests_router.get("/{test_id}", response_model=TestResponse)
async def get_test_by_id(test_id: int, db_session: Session = Depends(get_db)):
"""Получение теста по ID"""
test_repo = TestRepository(db_session)
test = test_repo.get_test_by_id(test_id)
if not test:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Тест не найден"
)
return test
# Роутер для работы с вопросами
questions_router = APIRouter(prefix="/questions", tags=["Questions"])
@questions_router.post("/", response_model=QuestionResponse)
async def create_question(
question_data: QuestionCreate,
db_session: Session = Depends(get_db)
):
"""Создание нового вопроса"""
question_repo = QuestionRepository(db_session)
question = question_repo.create_question(
test_id=question_data.test_id,
question_text=question_data.question_text,
question_type=question_data.question_type,
order_number=question_data.order_number,
points=question_data.points,
image_id=question_data.image_id
)
return question
@questions_router.get("/test/{test_id}", response_model=List[QuestionResponse])
async def get_questions_by_test(test_id: int, db_session: Session = Depends(get_db)):
"""Получение всех вопросов для теста"""
question_repo = QuestionRepository(db_session)
questions = question_repo.get_questions_by_test_id(test_id)
return questions
@questions_router.get("/{question_id}", response_model=QuestionResponse)
async def get_question_by_id(question_id: int, db_session: Session = Depends(get_db)):
"""Получение вопроса по ID"""
question_repo = QuestionRepository(db_session)
question = question_repo.get_question_by_id(question_id)
if not question:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Вопрос не найден"
)
return question
# Роутер для работы с вариантами ответов
answer_options_router = APIRouter(prefix="/answer-options", tags=["Answer Options"])
@answer_options_router.post("/", response_model=AnswerOptionResponse)
async def create_answer_option(
option_data: AnswerOptionCreate,
db_session: Session = Depends(get_db)
):
"""Создание нового варианта ответа"""
option = AnswerOption(
question_id=option_data.question_id,
option_text=option_data.option_text,
is_correct=option_data.is_correct,
order_number=option_data.order_number
)
db_session.add(option)
db_session.commit()
db_session.refresh(option)
return option
@answer_options_router.get("/question/{question_id}", response_model=List[AnswerOptionResponse])
async def get_answer_options_by_question(question_id: int, db_session: Session = Depends(get_db)):
"""Получение всех вариантов ответов для вопроса"""
options = db_session.query(AnswerOption).filter(
AnswerOption.question_id == question_id
).order_by(AnswerOption.order_number).all()
return options
# Роутер для работы с изображениями
images_router = APIRouter(prefix="/images", tags=["Images"])
@images_router.post("/upload", response_model=ImageUploadResponse)
async def upload_image(
file: UploadFile = File(...),
alt_text: Optional[str] = None,
db_session: Session = Depends(get_db)
):
"""Загрузка изображения в базу данных"""
# Проверяем тип файла
if not file.content_type.startswith('image/'):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Загружаемый файл должен быть изображением"
)
# Читаем содержимое файла
image_data = await file.read()
# Сохраняем в БД через сервис
image = ImageService.save_image_to_db(
db=db_session,
filename=file.filename,
content_type=file.content_type,
image_data=image_data,
alt_text=alt_text
)
return ImageUploadResponse(
id=image.id,
filename=image.filename,
content_type=image.content_type,
file_size=image.file_size
)
@images_router.get("/{image_id}")
async def get_image(image_id: int, db_session: Session = Depends(get_db)):
"""Получение изображения по ID"""
from fastapi.responses import Response
image_repo = ImageRepository(db_session)
image = image_repo.get_image_by_id(image_id)
if not image:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Изображение не найдено"
)
return Response(
content=image.image_data,
media_type=image.content_type,
headers={"Content-Disposition": f"inline; filename={image.filename}"}
)

18
api/run_api.py
View File

@@ -1,18 +0,0 @@
"""
Скрипт для запуска API сервера
"""
import uvicorn
from api.config import config
if __name__ == "__main__":
print(f"Запуск {config.TITLE} v{config.VERSION}")
print(f"Документация будет доступна по адресу: http://{config.HOST}:{config.PORT}/docs")
uvicorn.run(
"api.main:app",
host=config.HOST,
port=config.PORT,
reload=config.RELOAD,
log_level=config.LOG_LEVEL.lower()
)

208
api/schemas.py
View File

@@ -1,208 +0,0 @@
"""
Pydantic схемы для валидации данных API
"""
from pydantic import BaseModel, Field, ConfigDict
from typing import Optional
from datetime import datetime
# Базовые схемы для пользователей
class UserBase(BaseModel):
"""Базовая схема пользователя"""
telegram_id: int = Field(..., description="Telegram ID пользователя")
username: Optional[str] = Field(None, max_length=255, description="Username в Telegram")
first_name: Optional[str] = Field(None, max_length=255, description="Имя пользователя")
last_name: Optional[str] = Field(None, max_length=255, description="Фамилия пользователя")
language_code: str = Field(default="ru", max_length=10, description="Код языка")
class UserCreate(UserBase):
"""Схема для создания пользователя"""
pass
class UserUpdate(BaseModel):
"""Схема для обновления пользователя"""
username: Optional[str] = Field(None, max_length=255)
first_name: Optional[str] = Field(None, max_length=255)
last_name: Optional[str] = Field(None, max_length=255)
language_code: Optional[str] = Field(None, max_length=10)
is_active: Optional[bool] = None
class UserResponse(UserBase):
"""Схема ответа с данными пользователя"""
model_config = ConfigDict(from_attributes=True)
id: int
is_active: bool
is_admin: bool
created_at: datetime
updated_at: Optional[datetime] = None
# Базовые схемы для сообщений
class MessageBase(BaseModel):
"""Базовая схема сообщения"""
telegram_message_id: int = Field(..., description="ID сообщения в Telegram")
user_id: int = Field(..., description="ID пользователя в нашей БД")
text: Optional[str] = Field(None, description="Текст сообщения")
message_type: str = Field(default="text", max_length=50, description="Тип сообщения")
class MessageCreate(MessageBase):
"""Схема для создания сообщения"""
pass
class MessageResponse(MessageBase):
"""Схема ответа с данными сообщения"""
model_config = ConfigDict(from_attributes=True)
id: int
created_at: datetime
# Базовые схемы для настроек бота
class BotSettingBase(BaseModel):
"""Базовая схема настройки бота"""
key: str = Field(..., max_length=255, description="Ключ настройки")
value: Optional[str] = Field(None, description="Значение настройки")
description: Optional[str] = Field(None, description="Описание настройки")
class BotSettingCreate(BotSettingBase):
"""Схема для создания настройки"""
pass
class BotSettingUpdate(BaseModel):
"""Схема для обновления настройки"""
value: Optional[str] = None
description: Optional[str] = None
class BotSettingResponse(BotSettingBase):
"""Схема ответа с данными настройки"""
model_config = ConfigDict(from_attributes=True)
id: int
created_at: datetime
updated_at: Optional[datetime] = None
# Схемы для общих ответов API
class MessageResponseModel(BaseModel):
"""Общая схема ответа с сообщением"""
message: str
status: str = "success"
class ErrorResponse(BaseModel):
"""Схема ответа с ошибкой"""
detail: str
status: str = "error"
class HealthCheckResponse(BaseModel):
"""Схема ответа проверки здоровья"""
status: str
database: str
class RootResponse(BaseModel):
"""Схема корневого ответа"""
message: str
version: str
status: str
# Схемы для пагинации
class PaginationParams(BaseModel):
"""Параметры пагинации"""
limit: int = Field(default=100, ge=1, le=1000, description="Количество элементов на странице")
offset: int = Field(default=0, ge=0, description="Смещение")
class PaginatedResponse(BaseModel):
"""Схема ответа с пагинацией"""
items: list
total: int
limit: int
offset: int
has_next: bool
has_prev: bool
# Схемы для работы с тестами
class TestCategoryCreate(BaseModel):
"""Схема для создания категории тестов"""
name: str = Field(..., max_length=100, description="Название категории")
description: Optional[str] = Field(None, description="Описание категории")
class TestCategoryResponse(BaseModel):
"""Схема ответа с категорией тестов"""
model_config = ConfigDict(from_attributes=True)
id: int
name: str
description: Optional[str]
created_at: datetime
class TestCreate(BaseModel):
"""Схема для создания теста"""
title: str = Field(..., max_length=200, description="Название теста")
description: Optional[str] = Field(None, description="Описание теста")
category_id: Optional[int] = Field(None, description="ID категории")
is_active: bool = Field(default=True, description="Активен ли тест")
time_limit_minutes: Optional[int] = Field(None, ge=1, description="Лимит времени в минутах")
class TestResponse(BaseModel):
"""Схема ответа с тестом"""
model_config = ConfigDict(from_attributes=True)
id: int
title: str
description: Optional[str]
category_id: Optional[int]
is_active: bool
time_limit_minutes: Optional[int]
created_at: datetime
updated_at: Optional[datetime]
class QuestionCreate(BaseModel):
"""Схема для создания вопроса"""
test_id: int = Field(..., description="ID теста")
question_text: str = Field(..., description="Текст вопроса")
question_type: str = Field(default="single_choice", description="Тип вопроса")
order_number: int = Field(..., ge=1, description="Порядковый номер вопроса")
points: int = Field(default=1, ge=0, description="Количество баллов за правильный ответ")
image_id: Optional[int] = Field(None, description="ID изображения к вопросу")
class QuestionResponse(BaseModel):
"""Схема ответа с вопросом"""
model_config = ConfigDict(from_attributes=True)
id: int
test_id: int
question_text: str
question_type: str
order_number: int
points: int
image_id: Optional[int]
created_at: datetime
class AnswerOptionCreate(BaseModel):
"""Схема для создания варианта ответа"""
question_id: int = Field(..., description="ID вопроса")
option_text: str = Field(..., description="Текст варианта ответа")
is_correct: bool = Field(..., description="Правильный ли этот вариант")
order_number: int = Field(..., ge=1, description="Порядковый номер варианта")
class AnswerOptionResponse(BaseModel):
"""Схема ответа с вариантом ответа"""
model_config = ConfigDict(from_attributes=True)
id: int
question_id: int
option_text: str
is_correct: bool
order_number: int
created_at: datetime
class ImageUploadResponse(BaseModel):
"""Схема ответа при загрузке изображения"""
id: int
filename: str
content_type: str
file_size: int
message: str = "Изображение успешно загружено"

10
app.bot/__init__.py Normal file
View File

@@ -0,0 +1,10 @@
"""
Telegram Bot Module
Модуль содержит все компоненты Telegram бота:
- telegram_bot.py - основной файл бота
- bot_config.py - конфигурация
- api_client.py - HTTP клиент для API
"""
__version__ = "1.0.0"

307
app.bot/api_client.py Normal file
View File

@@ -0,0 +1,307 @@
"""
HTTP клиент для взаимодействия с API
"""
import aiohttp
import asyncio
import logging
from typing import Optional, List, Dict, Any, Union
from dataclasses import dataclass
import json
from bot_config import config
logger = logging.getLogger(__name__)
@dataclass
class APIResponse:
"""Стандартный ответ API"""
success: bool
data: Any = None
error: Optional[str] = None
status_code: int = 200
class APIClient:
"""HTTP клиент для работы с FastAPI"""
def __init__(self, base_url: str = None, timeout: int = 30):
self.base_url = base_url or config.API_BASE_URL
self.timeout = timeout
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
"""Асинхронный контекстный менеджер"""
await self.create_session()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Закрытие сессии"""
await self.close_session()
async def create_session(self):
"""Создание HTTP сессии"""
if not self.session or self.session.closed:
timeout = aiohttp.ClientTimeout(total=self.timeout)
self.session = aiohttp.ClientSession(
timeout=timeout,
headers={'Content-Type': 'application/json'}
)
async def close_session(self):
"""Закрытие HTTP сессии"""
if self.session and not self.session.closed:
await self.session.close()
async def _make_request(
self,
method: str,
endpoint: str,
data: Optional[Dict] = None,
params: Optional[Dict] = None,
files: Optional[Dict] = None
) -> APIResponse:
"""Базовый метод для HTTP запросов"""
await self.create_session()
url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
try:
# Если есть файлы, не устанавливаем Content-Type
headers = {}
if not files:
headers['Content-Type'] = 'application/json'
kwargs = {
'params': params,
'headers': headers
}
# Данные для запроса
if files:
# Для файлов используем FormData
kwargs['data'] = aiohttp.FormData()
if data:
for key, value in data.items():
kwargs['data'].add_field(key, str(value))
for key, file_data in files.items():
kwargs['data'].add_field(key, file_data)
elif data:
kwargs['data'] = json.dumps(data)
async with self.session.request(method, url, **kwargs) as response:
status_code = response.status
# Обработка ответа
if response.content_type == 'application/json':
response_data = await response.json()
else:
response_data = await response.text()
if 200 <= status_code < 300:
return APIResponse(
success=True,
data=response_data,
status_code=status_code
)
else:
error_msg = response_data.get('detail', f'HTTP {status_code}') if isinstance(response_data, dict) else str(response_data)
logger.error(f"API Error {status_code}: {error_msg}")
return APIResponse(
success=False,
error=error_msg,
status_code=status_code
)
except aiohttp.ClientError as e:
logger.error(f"HTTP Client Error: {e}")
return APIResponse(
success=False,
error=f"Connection error: {str(e)}",
status_code=500
)
except Exception as e:
logger.error(f"Unexpected error: {e}")
return APIResponse(
success=False,
error=f"Unexpected error: {str(e)}",
status_code=500
)
# ===== ПОЛЬЗОВАТЕЛИ =====
async def get_user_by_telegram_id(self, telegram_id: int) -> APIResponse:
"""Получение пользователя по Telegram ID"""
return await self._make_request('GET', f'/users/{telegram_id}')
async def create_user(self, user_data: Dict) -> APIResponse:
"""Создание нового пользователя"""
return await self._make_request('POST', '/users/', data=user_data)
async def update_user(self, telegram_id: int, user_data: Dict) -> APIResponse:
"""Обновление данных пользователя"""
return await self._make_request('PUT', f'/users/{telegram_id}', data=user_data)
async def get_all_users(self, limit: int = 100, offset: int = 0, is_active: Optional[bool] = None) -> APIResponse:
"""Получение списка пользователей"""
params = {'limit': limit, 'offset': offset}
if is_active is not None:
params['is_active'] = is_active
return await self._make_request('GET', '/users/', params=params)
# ===== ТЕСТЫ =====
async def get_tests(self, is_active: Optional[bool] = None, category_id: Optional[int] = None) -> APIResponse:
"""Получение списка тестов"""
params = {}
if is_active is not None:
params['is_active'] = is_active
if category_id is not None:
params['category_id'] = category_id
return await self._make_request('GET', '/tests/', params=params)
async def get_test_by_id(self, test_id: int) -> APIResponse:
"""Получение теста по ID"""
return await self._make_request('GET', f'/tests/{test_id}')
async def create_test(self, test_data: Dict) -> APIResponse:
"""Создание нового теста"""
return await self._make_request('POST', '/tests/', data=test_data)
# ===== ВОПРОСЫ =====
async def get_questions_by_test(self, test_id: int) -> APIResponse:
"""Получение всех вопросов для теста"""
return await self._make_request('GET', f'/questions/test/{test_id}')
async def get_question_by_id(self, question_id: int) -> APIResponse:
"""Получение вопроса по ID"""
return await self._make_request('GET', f'/questions/{question_id}')
async def create_question(self, question_data: Dict) -> APIResponse:
"""Создание нового вопроса"""
return await self._make_request('POST', '/questions/', data=question_data)
# ===== ВАРИАНТЫ ОТВЕТОВ =====
async def get_answer_options_by_question(self, question_id: int) -> APIResponse:
"""Получение вариантов ответов для вопроса"""
return await self._make_request('GET', f'/answer-options/question/{question_id}')
async def create_answer_option(self, option_data: Dict) -> APIResponse:
"""Создание варианта ответа"""
return await self._make_request('POST', '/answer-options/', data=option_data)
# ===== ИЗОБРАЖЕНИЯ =====
async def upload_image(self, file_content: bytes, filename: str, content_type: str, alt_text: Optional[str] = None) -> APIResponse:
"""Загрузка изображения"""
files = {'file': (filename, file_content, content_type)}
data = {}
if alt_text:
data['alt_text'] = alt_text
return await self._make_request('POST', '/images/upload', data=data, files=files)
async def get_image(self, image_id: int) -> APIResponse:
"""Получение изображения по ID"""
return await self._make_request('GET', f'/images/{image_id}')
# ===== НАСТРОЙКИ =====
async def get_all_settings(self) -> APIResponse:
"""Получение всех настроек"""
return await self._make_request('GET', '/settings/')
async def get_setting(self, key: str) -> APIResponse:
"""Получение настройки по ключу"""
return await self._make_request('GET', f'/settings/{key}')
async def create_setting(self, setting_data: Dict) -> APIResponse:
"""Создание новой настройки"""
return await self._make_request('POST', '/settings/', data=setting_data)
async def update_setting(self, key: str, setting_data: Dict) -> APIResponse:
"""Обновление настройки"""
return await self._make_request('PUT', f'/settings/{key}', data=setting_data)
# ===== СООБЩЕНИЯ =====
async def create_message(self, message_data: Dict) -> APIResponse:
"""Создание записи о сообщении"""
return await self._make_request('POST', '/messages/', data=message_data)
async def get_messages(self, user_id: Optional[int] = None, message_type: Optional[str] = None, limit: int = 100) -> APIResponse:
"""Получение списка сообщений"""
params = {'limit': limit}
if user_id:
params['user_id'] = user_id
if message_type:
params['message_type'] = message_type
return await self._make_request('GET', '/messages/', params=params)
# ===== УТИЛИТЫ =====
async def health_check(self) -> bool:
"""Проверка доступности API"""
try:
response = await self._make_request('GET', '/health')
return response.success
except Exception as e:
logger.error(f"Health check failed: {e}")
return False
# Глобальный экземпляр клиента
api_client = APIClient()
# Вспомогательные функции для удобства использования
async def ensure_user_exists(telegram_id: int, user_data: Dict) -> Optional[Dict]:
"""Убедиться, что пользователь существует, создать если нет"""
async with api_client:
# Попробуем получить пользователя
response = await api_client.get_user_by_telegram_id(telegram_id)
if response.success:
return response.data
# Если пользователь не найден, создаем нового
if response.status_code == 404:
create_response = await api_client.create_user(user_data)
if create_response.success:
logger.info(f"Создан новый пользователь: {telegram_id}")
return create_response.data
else:
logger.error(f"Ошибка создания пользователя: {create_response.error}")
return None
else:
logger.error(f"Ошибка получения пользователя: {response.error}")
return None
async def get_available_tests() -> List[Dict]:
"""Получить доступные тесты"""
async with api_client:
response = await api_client.get_tests(is_active=True)
if response.success:
return response.data
else:
logger.error(f"Ошибка получения тестов: {response.error}")
return []
async def get_test_questions_with_options(test_id: int) -> List[Dict]:
"""Получить вопросы теста с вариантами ответов"""
async with api_client:
# Получаем вопросы
questions_response = await api_client.get_questions_by_test(test_id)
if not questions_response.success:
logger.error(f"Ошибка получения вопросов: {questions_response.error}")
return []
questions = questions_response.data
# Для каждого вопроса получаем варианты ответов
for question in questions:
options_response = await api_client.get_answer_options_by_question(question['id'])
if options_response.success:
question['options'] = options_response.data
else:
question['options'] = []
logger.error(f"Ошибка получения вариантов для вопроса {question['id']}: {options_response.error}")
return questions

6
bot_config.py → app.bot/bot_config.py
View File

@@ -25,6 +25,12 @@ class BotConfig:
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
DEBUG = os.getenv("DEBUG", "False").lower() == "true"
# API Settings
API_HOST = os.getenv("API_HOST", "localhost")
API_PORT = int(os.getenv("API_PORT", "8000"))
API_BASE_URL = os.getenv("API_BASE_URL", f"http://{API_HOST}:{API_PORT}")
API_TIMEOUT = int(os.getenv("API_TIMEOUT", "30"))
@classmethod
def validate(cls):
"""Проверка обязательных конфигурационных параметров"""

60
app.bot/main.py Normal file
View File

@@ -0,0 +1,60 @@
"""
Точка входа для Telegram бота
Бот работает только через API, без прямого доступа к БД
"""
import asyncio
import logging
import sys
import os
# Добавляем текущую директорию в PYTHONPATH
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from bot_config import config
def setup_logging():
"""Настройка системы логирования"""
logging.basicConfig(
level=getattr(logging, config.LOG_LEVEL),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout),
logging.FileHandler('bot.log') if config.APP_ENV == 'production' else logging.NullHandler()
]
)
async def main():
"""Основная функция запуска бота"""
# Настраиваем логирование
setup_logging()
logger = logging.getLogger(__name__)
try:
# Проверяем конфигурацию
config.validate()
logger.info(f"Конфигурация проверена. Режим: {config.APP_ENV}")
logger.info(f"API URL: {config.API_BASE_URL}")
# Проверяем доступность API
from api_client import api_client
async with api_client:
api_available = await api_client.health_check()
if not api_available:
logger.error("API недоступен! Проверьте, что API сервер запущен.")
return
logger.info("API доступен, запускаем бота...")
# Запускаем бота
import telegram_bot
await telegram_bot.main()
except KeyboardInterrupt:
logger.info("Получен сигнал остановки")
except Exception as e:
logger.error(f"Критическая ошибка: {e}")
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())

469
app.bot/telegram_bot.py Normal file
View File

@@ -0,0 +1,469 @@
"""
Современный Telegram бот, работающий только через API
БЕЗ прямого доступа к базе данных
"""
import asyncio
import io
from datetime import datetime
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.types import KeyboardButton, ReplyKeyboardMarkup, ReplyKeyboardRemove, BufferedInputFile
from aiogram.filters.state import StateFilter
import logging
from bot_config import config
from api_client import api_client
# Настройка логирования
logging.basicConfig(
level=getattr(logging, config.LOG_LEVEL),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Инициализация бота
bot = Bot(token=config.TELEGRAM_BOT_TOKEN)
dp = Dispatcher(storage=MemoryStorage())
# ===== FSM States =====
class Form(StatesGroup):
consent = State()
fio = State()
dob = State()
phone = State()
city = State()
test_in_progress = State()
# ===== /start command =====
@dp.message(Command("start"))
async def start(message: types.Message, state: FSMContext):
"""Обработка команды /start"""
try:
# Создаем или получаем пользователя через API
user_data = {
"telegram_id": message.from_user.id,
"username": message.from_user.username,
"first_name": message.from_user.first_name,
"last_name": message.from_user.last_name,
"language_code": message.from_user.language_code or 'ru'
}
async with api_client:
# Попробуем получить пользователя
response = await api_client.get_user_by_telegram_id(message.from_user.id)
if response.success:
user = response.data
logger.info(f"Найден существующий пользователь: {user['telegram_id']}")
else:
# Создаем нового пользователя
create_response = await api_client.create_user(user_data)
if create_response.success:
user = create_response.data
logger.info(f"Создан новый пользователь: {user['telegram_id']}")
else:
logger.error(f"Ошибка создания пользователя: {create_response.error}")
await message.answer("Произошла ошибка при регистрации. Попробуйте позже.")
return
# Сохраняем данные пользователя в состояние
await state.update_data(user_id=user['id'], telegram_id=user['telegram_id'])
welcome_text = f"Добро пожаловать, {user.get('first_name', 'пользователь')}!\n\n"
welcome_text += "Этот бот поможет вам пройти тестирование.\n"
welcome_text += "Для продолжения необходимо дать согласие на обработку персональных данных.\n\n"
welcome_text += "Даете ли вы согласие на обработку персональных данных?"
keyboard = ReplyKeyboardMarkup(
keyboard=[
[KeyboardButton(text="Да, согласен")],
[KeyboardButton(text="Нет, не согласен")]
],
resize_keyboard=True,
one_time_keyboard=True
)
await message.answer(welcome_text, reply_markup=keyboard)
await state.set_state(Form.consent)
except Exception as e:
logger.error(f"Ошибка в start handler: {e}")
await message.answer("Произошла ошибка. Попробуйте позже.")
# ===== Согласие на обработку данных =====
@dp.message(Form.consent)
async def process_consent(message: types.Message, state: FSMContext):
"""Обработка согласия на обработку данных"""
if message.text == "Да, согласен":
await message.answer(
"Спасибо! Теперь введите ваше ФИО:",
reply_markup=ReplyKeyboardRemove()
)
await state.set_state(Form.fio)
elif message.text == "Нет, не согласен":
await message.answer(
"К сожалению, без согласия на обработку персональных данных использование бота невозможно.",
reply_markup=ReplyKeyboardRemove()
)
await state.clear()
else:
await message.answer("Пожалуйста, выберите один из предложенных вариантов.")
# ===== Сбор ФИО =====
@dp.message(Form.fio)
async def process_fio(message: types.Message, state: FSMContext):
"""Обработка ввода ФИО"""
if len(message.text.strip()) < 5:
await message.answer("Пожалуйста, введите полное ФИО (минимум 5 символов).")
return
await state.update_data(fio=message.text.strip())
await message.answer("Введите вашу дату рождения в формате ДД.ММ.ГГГГ:")
await state.set_state(Form.dob)
# ===== Дата рождения =====
@dp.message(Form.dob)
async def process_dob(message: types.Message, state: FSMContext):
"""Обработка ввода даты рождения"""
try:
# Проверяем формат даты
dob = datetime.strptime(message.text.strip(), "%d.%m.%Y")
# Проверяем возраст (должен быть от 16 до 80 лет)
age = (datetime.now() - dob).days // 365
if age < 16 or age > 80:
await message.answer("Возраст должен быть от 16 до 80 лет. Проверьте правильность даты.")
return
await state.update_data(date_of_birth=message.text.strip())
await message.answer("Введите ваш номер телефона:")
await state.set_state(Form.phone)
except ValueError:
await message.answer("Неверный формат даты. Используйте формат ДД.ММ.ГГГГ (например, 15.05.1990):")
# ===== Телефон =====
@dp.message(Form.phone)
async def process_phone(message: types.Message, state: FSMContext):
"""Обработка ввода телефона"""
phone = message.text.strip()
# Простая проверка телефона
if len(phone) < 10 or not any(char.isdigit() for char in phone):
await message.answer("Пожалуйста, введите корректный номер телефона.")
return
await state.update_data(phone=phone)
await message.answer("Введите ваш город:")
await state.set_state(Form.city)
# ===== Город =====
@dp.message(Form.city)
async def process_city(message: types.Message, state: FSMContext):
"""Обработка ввода города"""
city = message.text.strip()
if len(city) < 2:
await message.answer("Пожалуйста, введите название города (минимум 2 символа).")
return
# Получаем все собранные данные
data = await state.get_data()
telegram_id = data['telegram_id']
# Обновляем данные пользователя через API
update_data = {
"first_name": data.get('fio', '').split()[0] if data.get('fio') else None,
"last_name": ' '.join(data.get('fio', '').split()[1:]) if len(data.get('fio', '').split()) > 1 else None,
# Можно добавить дополнительные поля в модель User для хранения ФИО, телефона, города и даты рождения
}
async with api_client:
response = await api_client.update_user(telegram_id, update_data)
if not response.success:
logger.error(f"Ошибка обновления пользователя: {response.error}")
await message.answer(
f"Спасибо! Ваши данные сохранены:\n"
f"ФИО: {data['fio']}\n"
f"Дата рождения: {data['date_of_birth']}\n"
f"Телефон: {data['phone']}\n"
f"Город: {city}\n\n"
f"Теперь вы можете приступить к тестированию!"
)
# Показываем доступные тесты
await show_available_tests(message, state)
# ===== Показать доступные тесты =====
async def show_available_tests(message: types.Message, state: FSMContext):
"""Показать список доступных тестов"""
try:
async with api_client:
response = await api_client.get_tests(is_active=True)
if not response.success:
logger.error(f"Ошибка получения тестов: {response.error}")
await message.answer("Ошибка загрузки тестов. Попробуйте позже.")
return
tests = response.data
if not tests:
await message.answer("На данный момент нет доступных тестов.")
return
# Создаем клавиатуру с тестами
keyboard_buttons = []
for test in tests:
keyboard_buttons.append([KeyboardButton(text=test['title'])])
keyboard = ReplyKeyboardMarkup(
keyboard=keyboard_buttons,
resize_keyboard=True,
one_time_keyboard=True
)
await message.answer(
"Выберите тест для прохождения:",
reply_markup=keyboard
)
await state.clear() # Очищаем предыдущее состояние
except Exception as e:
logger.error(f"Ошибка в show_available_tests: {e}")
await message.answer("Произошла ошибка при загрузке тестов.")
# ===== Выбор теста =====
@dp.message()
async def handle_test_selection(message: types.Message, state: FSMContext):
"""Обработка выбора теста"""
try:
async with api_client:
# Получаем все тесты и ищем по названию
response = await api_client.get_tests(is_active=True)
if not response.success:
await message.answer("Ошибка загрузки тестов.")
return
selected_test = None
for test in response.data:
if test['title'] == message.text:
selected_test = test
break
if not selected_test:
await message.answer(
"Тест не найден. Используйте /start для просмотра доступных тестов.",
reply_markup=ReplyKeyboardRemove()
)
return
# Получаем вопросы теста
questions_response = await api_client.get_questions_by_test(selected_test['id'])
if not questions_response.success:
await message.answer("Ошибка загрузки вопросов теста.")
return
questions = questions_response.data
if not questions:
await message.answer("В этом тесте пока нет вопросов.")
return
# Загружаем варианты ответов для всех вопросов
for question in questions:
options_response = await api_client.get_answer_options_by_question(question['id'])
if options_response.success:
question['options'] = options_response.data
else:
question['options'] = []
# Сохраняем данные теста в состояние
await state.update_data(
test=selected_test,
questions=questions,
current_question=0,
answers=[]
)
await message.answer(
f"Вы выбрали тест: {selected_test['title']}\n"
f"Описание: {selected_test.get('description', 'Описание отсутствует')}\n"
f"Количество вопросов: {len(questions)}\n\n"
f"Готовы начать?",
reply_markup=ReplyKeyboardMarkup(
keyboard=[[KeyboardButton(text="Начать тест")]],
resize_keyboard=True,
one_time_keyboard=True
)
)
await state.set_state(Form.test_in_progress)
except Exception as e:
logger.error(f"Ошибка в handle_test_selection: {e}")
await message.answer("Произошла ошибка при выборе теста.")
# ===== Прохождение теста =====
@dp.message(Form.test_in_progress)
async def handle_test_progress(message: types.Message, state: FSMContext):
"""Обработка прохождения теста"""
try:
data = await state.get_data()
if message.text == "Начать тест":
# Показываем первый вопрос
await show_question(message, state, 0)
return
# Обрабатываем ответ на вопрос
questions = data.get('questions', [])
current_question_idx = data.get('current_question', 0)
answers = data.get('answers', [])
if current_question_idx >= len(questions):
await message.answer("Тест уже завершен.")
return
current_question = questions[current_question_idx]
# Находим выбранный вариант ответа
selected_option = None
for option in current_question.get('options', []):
if option['option_text'] == message.text:
selected_option = option
break
if not selected_option:
await message.answer("Пожалуйста, выберите один из предложенных вариантов ответа.")
return
# Сохраняем ответ
answers.append({
'question_id': current_question['id'],
'option_id': selected_option['id'],
'is_correct': selected_option.get('is_correct', False)
})
# Переходим к следующему вопросу
next_question_idx = current_question_idx + 1
await state.update_data(
answers=answers,
current_question=next_question_idx
)
if next_question_idx >= len(questions):
# Тест завершен
await finish_test(message, state)
else:
# Показываем следующий вопрос
await show_question(message, state, next_question_idx)
except Exception as e:
logger.error(f"Ошибка в handle_test_progress: {e}")
await message.answer("Произошла ошибка при обработке ответа.")
async def show_question(message: types.Message, state: FSMContext, question_idx: int):
"""Показать вопрос"""
try:
data = await state.get_data()
questions = data.get('questions', [])
if question_idx >= len(questions):
return
question = questions[question_idx]
# Формируем текст вопроса
question_text = f"Вопрос {question_idx + 1} из {len(questions)}:\n\n"
question_text += question['question_text']
# Создаем клавиатуру с вариантами ответов
keyboard_buttons = []
for option in question.get('options', []):
keyboard_buttons.append([KeyboardButton(text=option['option_text'])])
keyboard = ReplyKeyboardMarkup(
keyboard=keyboard_buttons,
resize_keyboard=True,
one_time_keyboard=True
)
await message.answer(question_text, reply_markup=keyboard)
except Exception as e:
logger.error(f"Ошибка в show_question: {e}")
await message.answer("Ошибка при показе вопроса.")
async def finish_test(message: types.Message, state: FSMContext):
"""Завершить тест и показать результаты"""
try:
data = await state.get_data()
answers = data.get('answers', [])
test = data.get('test', {})
# Подсчитываем результаты
correct_answers = sum(1 for answer in answers if answer.get('is_correct', False))
total_questions = len(answers)
score_percentage = (correct_answers / total_questions * 100) if total_questions > 0 else 0
result_text = f"🎉 Тест завершен!\n\n"
result_text += f"Тест: {test.get('title', 'Неизвестный тест')}\n"
result_text += f"Правильных ответов: {correct_answers} из {total_questions}\n"
result_text += f"Результат: {score_percentage:.1f}%\n\n"
if score_percentage >= 80:
result_text += "Отличный результат! 🏆"
elif score_percentage >= 60:
result_text += "Хороший результат! 👍"
elif score_percentage >= 40:
result_text += "Удовлетворительный результат. 👌"
else:
result_text += "Рекомендуем повторить материал. 📚"
await message.answer(
result_text,
reply_markup=ReplyKeyboardMarkup(
keyboard=[[KeyboardButton(text="/start")]],
resize_keyboard=True,
one_time_keyboard=True
)
)
await state.clear()
except Exception as e:
logger.error(f"Ошибка в finish_test: {e}")
await message.answer("Ошибка при завершении теста.")
# ===== Основная функция =====
async def main():
"""Основная функция запуска бота"""
logger.info("Запуск Telegram бота...")
try:
# Проверяем доступность API
async with api_client:
if not await api_client.health_check():
logger.error("API недоступен! Убедитесь, что API сервер запущен.")
return
logger.info("API доступен, бот готов к работе")
# Запускаем бота
await dp.start_polling(bot)
except Exception as e:
logger.error(f"Ошибка при запуске бота: {e}")
finally:
await bot.session.close()
if __name__ == "__main__":
asyncio.run(main())

337
bot_old.py
View File

@@ -1,337 +0,0 @@
import os
from io import BytesIO
from aiogram.types import KeyboardButton, ReplyKeyboardMarkup
import asyncio
from datetime import datetime
import psycopg2
import psycopg2.extras
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.types import KeyboardButton, ReplyKeyboardMarkup, ReplyKeyboardRemove
from aiogram.filters.state import StateFilter
from dotenv import load_dotenv
import logging
# Загрузка переменных окружения
load_dotenv()
# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Конфигурация из переменных окружения
TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
if not TOKEN:
raise ValueError("TELEGRAM_BOT_TOKEN не найден в переменных окружения")
# ===== PostgreSQL =====
try:
conn = psycopg2.connect(
dbname=os.getenv("DB_NAME", "botdb"),
user=os.getenv("DB_USER", "botuser"),
password=os.getenv("DB_PASSWORD"),
host=os.getenv("DB_HOST", "localhost"),
port=os.getenv("DB_PORT", "5432")
)
cur = conn.cursor()
logger.info("Подключение к базе данных установлено")
except Exception as e:
logger.error(f"Ошибка подключения к базе данных: {e}")
raise
bot = Bot(token=TOKEN)
dp = Dispatcher(storage=MemoryStorage())
# ===== FSM =====
class Form(StatesGroup):
consent = State()
fio = State()
dob = State()
phone = State()
city = State()
klimov = State()
# ===== /start =====
@dp.message(Command("start"))
async def start(message: types.Message, state: FSMContext):
kb = [
[KeyboardButton(text="✅ Да, согласен"), KeyboardButton(text="❌ Нет")]
]
keyboard = ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True)
await message.answer(
"Привет! Я помогу пройти тест Климова.\n\n"
"Сначала — согласие на обработку персональных данных.\n"
"Ты согласен?",
reply_markup=keyboard
)
await state.set_state(Form.consent)
# ===== Согласие =====
@dp.message(Form.consent)
async def consent(message: types.Message, state: FSMContext):
if message.text.startswith("✅"):
await state.set_state(Form.fio)
await message.answer("Введите своё ФИО:", reply_markup=ReplyKeyboardRemove())
else:
await message.answer("Без согласия мы не можем продолжить 🙏", reply_markup=ReplyKeyboardRemove())
await state.clear()
# ===== ФИО =====
@dp.message(Form.fio)
async def fio(message: types.Message, state: FSMContext):
await state.update_data(fio=message.text)
await state.set_state(Form.dob)
await message.answer("Введите дату рождения (ДД.MM.ГГГГ):")
# ===== Дата рождения =====
@dp.message(Form.dob)
async def dob(message: types.Message, state: FSMContext):
await state.update_data(dob=message.text)
await state.set_state(Form.phone)
await message.answer("Введите номер телефона:")
# ===== Телефон =====
@dp.message(Form.phone)
async def phone(message: types.Message, state: FSMContext):
await state.update_data(phone=message.text)
await state.set_state(Form.city)
await message.answer("Из какого ты населённого пункта?")
# ===== Город + меню тестов =====
@dp.message(Form.city)
async def city(message: types.Message, state: FSMContext):
await state.update_data(city=message.text)
data = await state.get_data()
# Преобразуем дату в формат PostgreSQL
try:
dob_parsed = datetime.strptime(data["dob"], "%d.%m.%Y").date()
except ValueError:
await message.answer("Неверный формат даты. Введите дату в формате ДД.MM.ГГГГ")
await state.set_state(Form.dob)
return
try:
cur.execute(
"""
INSERT INTO users (telegram_id, fio, dob, phone, city)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (telegram_id) DO UPDATE
SET fio=EXCLUDED.fio, dob=EXCLUDED.dob, phone=EXCLUDED.phone, city=EXCLUDED.city
""",
(message.from_user.id, data["fio"], dob_parsed, data["phone"], data["city"])
)
conn.commit()
await message.answer("Анкета сохранена ✅")
logger.info(f"Пользователь {message.from_user.id} сохранил анкету")
except Exception as e:
conn.rollback()
logger.error(f"Ошибка при сохранении анкеты пользователя {message.from_user.id}: {e}")
await message.answer("Произошла ошибка при сохранении анкеты. Попробуйте позже.")
return
kb = [
[KeyboardButton(text="📘 Тест Климова")],
[KeyboardButton(text="📗 Тест №2"), KeyboardButton(text="📙 Тест №3")]
]
keyboard = ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True)
await state.clear()
await message.answer("Теперь выбери тест:", reply_markup=keyboard)
# ===== Выбор теста после анкеты =====
# Выбор теста — только когда FSM пустой (после анкеты)
@dp.message(StateFilter(None))
async def select_test(message: types.Message, state: FSMContext):
if message.text == "📘 Тест Климова":
try:
cur.execute("SELECT id, text1, category1, text2, category2, image FROM klimov_questions ORDER BY id")
questions = cur.fetchall()
if not questions:
await message.answer("Вопросы теста не найдены. Обратитесь к администратору.")
logger.warning("Вопросы теста Климова не найдены в базе данных")
return
await state.update_data(klimov_questions=questions, klimov_index=0, klimov_result={})
await state.set_state(Form.klimov)
await send_klimov_question(message.from_user.id, state)
logger.info(f"Пользователь {message.from_user.id} начал тест Климова")
except Exception as e:
logger.error(f"Ошибка при загрузке теста Климова: {e}")
await message.answer("Произошла ошибка при загрузке теста. Попробуйте позже.")
else:
await message.answer("Пока этот тест не реализован.")
import os
from aiogram.types import KeyboardButton, ReplyKeyboardMarkup, FSInputFile
from aiogram.fsm.context import FSMContext
async def send_klimov_question(user_id, state: FSMContext):
data = await state.get_data()
index = data["klimov_index"]
questions = data["klimov_questions"]
if index >= len(questions):
await send_klimov_result(user_id, state)
await state.clear()
return
q = questions[index] # (id, text1, category1, text2, category2, image)
# Клавиатура с вариантами
kb = [
[KeyboardButton(text=q[1])],
[KeyboardButton(text=q[3])]
]
keyboard = ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True)
# Отправка картинки, если есть
image_filename = os.path.basename(q[5].strip()) if q[5] else None
if image_filename:
image_path = os.path.join(os.getcwd(), "images", image_filename)
logger.debug(f"Путь к изображению: {image_path}, существует: {os.path.isfile(image_path)}")
if os.path.isfile(image_path):
try:
photo = FSInputFile(image_path)
await bot.send_photo(
chat_id=user_id,
photo=photo,
caption=f"Вопрос {index+1} из {len(questions)}:\nВыберите вариант",
reply_markup=keyboard
)
return
except Exception as e:
logger.error(f"Ошибка отправки изображения {image_path}: {e}")
# Продолжаем с текстовым сообщением
# Если картинки нет, просто отправляем текст
await bot.send_message(
chat_id=user_id,
text=f"Вопрос {index+1} из {len(questions)}:\nВыберите вариант",
reply_markup=keyboard
)
# ===== Ответ на вопросы =====
@dp.message(StateFilter(Form.klimov))
async def klimov_answer(message: types.Message, state: FSMContext):
data = await state.get_data()
index = data["klimov_index"]
questions = data["klimov_questions"]
result = data["klimov_result"]
q = questions[index]
# Проверяем какой вариант выбрал пользователь
if message.text == q[1]:
result[q[2]] = result.get(q[2], 0) + 1
elif message.text == q[3]:
result[q[4]] = result.get(q[4], 0) + 1
index += 1
await state.update_data(klimov_index=index, klimov_result=result)
# Переходим к следующему вопросу
await send_klimov_question(message.from_user.id, state)
# ===== Функция показа вопроса =====
# Функция показа вопроса
# ===== Функция для сохранения результата в БД =====
async def save_klimov_result(user_id, state: FSMContext):
data = await state.get_data()
result = data.get("klimov_result", {})
try:
conn = psycopg2.connect(
dbname=os.getenv("DB_NAME", "botdb"),
user=os.getenv("DB_USER", "botuser"),
password=os.getenv("DB_PASSWORD"),
host=os.getenv("DB_HOST", "localhost"),
port=os.getenv("DB_PORT", "5432")
)
cur = conn.cursor()
# Получаем user_id из таблицы users
cur.execute("SELECT id FROM users WHERE telegram_id = %s", (user_id,))
user_row = cur.fetchone()
if user_row:
db_user_id = user_row[0]
cur.execute(
"INSERT INTO results (user_id, test_name, result) VALUES (%s, %s, %s)",
(db_user_id, "Тест Климова", psycopg2.extras.Json(result))
)
conn.commit()
logger.info(f"Результат теста сохранен для пользователя {user_id}")
cur.close()
conn.close()
except Exception as e:
logger.error(f"Ошибка сохранения результата теста: {e}")
# Не прерываем работу бота из-за ошибки сохранения
# Словарь с описаниями профессий по категориям
professions_desc = {
"Человек-природа": "К типу 'человек-природа' можно отнести профессии, связанные с изучением живой и неживой природы (микробиолог, агрохимик, геолог), с уходом за растениями и животными (лесовод, овощевод, зоотехник).",
"Человек-техника": "Тип 'человек-техника' включает в себя профессии, связанные с созданием, монтажом, сборкой и наладкой технических устройств (каменщик, монтажник, сварщик, инженер-конструктор), эксплуатацией технических средств (водитель, крановщик, токарь, швея), ремонтом техники.",
"Человек-человек": "К профессиям типа 'человек-человек' относятся профессии, связанные с медицинским обслуживанием (врач, медсестра, санитарка), обучением и воспитанием (воспитатель, няня, учитель, преподаватель, тренер), бытовым обслуживанием (продавец, проводник, официант).",
"Человек-знаковая система": "Тип 'человек - знаковая система' объединяет профессии, связанные с текстами (корректор, оператор ПК, переводчик, библиотекарь), с цифрами, формулами и таблицами (программист, экономист, бухгалтер, кассир), с чертежами, картами, схемами (штурман, чертежник).",
"Человек-художественный образ": "К типу 'человек - художественный образ' можно отнести профессии, связанные с созданием, проектированием, моделированием художественных произведений (художник, журналист, модельер, композитор), с воспроизведением, изготовлением различных изделий по эскизу, образцу."
}
# ===== Функция для форматированного вывода результатов =====
# ===== Форматированный вывод результатов с процентами по максимуму =====
async def send_klimov_result(user_id, state: FSMContext):
data = await state.get_data()
result = data.get("klimov_result", {})
questions = data.get("klimov_questions", [])
if not questions:
await bot.send_message(user_id, "Вы не ответили ни на один вопрос.")
return
# Подсчёт максимального количества очков для каждой категории
max_points = {}
for q in questions:
# q[2] и q[4] — категории для вариантов
max_points[q[2]] = max_points.get(q[2], 0) + 1
max_points[q[4]] = max_points.get(q[4], 0) + 1
output = "Ваши результаты:\n"
for category, max_pt in max_points.items():
pts = result.get(category, 0)
percent = (pts / max_pt) * 100
output += f"- {category} - {percent:.1f}%\n"
# Определяем категорию с наибольшим результатом
if result:
top_category = max(result, key=result.get)
output += f"\nТип профессии: {top_category}"
await bot.send_message(user_id, output, reply_markup=types.ReplyKeyboardRemove())
# Затем по очереди отправляем описания
for category in ["Человек-природа", "Человек-техника", "Человек-человек",
"Человек-знаковая система", "Человек-художественный образ"]:
await bot.send_message(user_id, f"🟢 {category}\n\n{professions_desc[category]}")
# Последнее сообщение с контактами / ссылками
await bot.send_message(
user_id,
" Задать вопрос или записаться на индивидуальную консультацию по построению образовательного маршрута можно здесь: @chrtakova11\n\n"
"Узнать подробнее о профессиях: https://postupi.online/professii/so/"
)
# ===== Запуск бота =====
async def main():
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())

11
database/.env.example
View File

@@ -1,11 +0,0 @@
# Переменные окружения для подключения к PostgreSQL
# Параметры подключения к базе данных
DB_USER=postgres
DB_PASSWORD=your_password_here
DB_HOST=localhost
DB_PORT=5432
DB_NAME=telegram_bot
# Для продакшена можно также использовать полную строку подключения:
# DATABASE_URL=postgresql://username:password@localhost:5432/database_name

102
database/README.md
View File

@@ -1,102 +0,0 @@
# База данных PostgreSQL для Telegram бота
Этот модуль содержит полную настройку для работы с PostgreSQL через SQLAlchemy.
## Структура файлов
- `database.py` - Основные классы для подключения к базе данных
- `models.py` - Модели SQLAlchemy (таблицы)
- `repositories.py` - Репозитории для работы с данными
- `services.py` - Высокоуровневые сервисы
- `example.py` - Примеры использования
- `requirements.txt` - Зависимости
- `.env.example` - Пример файла с переменными окружения
## Быстрый старт
### 1. Установка зависимостей
```bash
pip install -r requirements.txt
```
### 2. Настройка базы данных
Создайте файл `.env` на основе `.env.example`:
```bash
DB_USER=postgres
DB_PASSWORD=your_password
DB_HOST=localhost
DB_PORT=5432
DB_NAME=telegram_bot
```
### 3. Основное использование
```python
from database import db_service, setup_database
# Настройка базы данных (создание таблиц)
setup_database()
# Регистрация пользователя
user, created = db_service.register_user(
telegram_id=123456789,
username="example_user",
first_name="Иван"
)
# Сохранение сообщения
message = db_service.save_message(
telegram_message_id=1001,
telegram_user_id=123456789,
text="Привет!"
)
# Работа с настройками
db_service.set_setting("bot_name", "Мой Бот")
bot_name = db_service.get_setting("bot_name")
```
## Основные возможности
### Модели данных
- **User** - Пользователи телеграм бота
- **Message** - Сообщения от пользователей
- **BotSettings** - Настройки бота
### Репозитории
- **UserRepository** - CRUD операции с пользователями
- **MessageRepository** - CRUD операции с сообщениями
- **SettingsRepository** - CRUD операции с настройками
### Сервисы
- **DatabaseService** - Высокоуровневый API для работы с данными
## Примеры использования
Запустите `example.py` для демонстрации основных возможностей:
```bash
python example.py
```
## Настройки
Все настройки подключения к базе данных можно задать через переменные окружения:
- `DB_USER` - имя пользователя PostgreSQL
- `DB_PASSWORD` - пароль
- `DB_HOST` - хост (по умолчанию localhost)
- `DB_PORT` - порт (по умолчанию 5432)
- `DB_NAME` - имя базы данных
## Требования
- Python 3.8+
- PostgreSQL 10+
- SQLAlchemy 2.0+

37
database/__init__.py
View File

@@ -1,37 +0,0 @@
"""
Инициализация пакета database
"""
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 ImageService, TestService, QuestionService
__all__ = [
# Основные классы
'Database',
'DatabaseConfig',
# Глобальные объекты
'db',
# Функции
'get_db_session',
'init_database',
'test_db_connection',
# Модели
'User',
'Message',
'BotSettings',
# Репозитории
'UserRepository',
'MessageRepository',
'SettingsRepository',
# Сервисы
'ImageService',
'TestService',
'QuestionService',
]

201
database/database.py
View File

@@ -1,201 +0,0 @@
"""
Модуль для подключения к базе данных PostgreSQL через SQLAlchemy
"""
from sqlalchemy import create_engine, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from contextlib import contextmanager
import os
from typing import Generator
import logging
# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Базовый класс для всех моделей
Base = declarative_base()
class DatabaseConfig:
"""Конфигурация базы данных"""
def __init__(self):
# Параметры подключения к PostgreSQL
self.DB_USER = os.getenv('DB_USER', 'postgres')
self.DB_PASSWORD = os.getenv('DB_PASSWORD', 'password')
self.DB_HOST = os.getenv('DB_HOST', 'localhost')
self.DB_PORT = os.getenv('DB_PORT', '5432')
self.DB_NAME = os.getenv('DB_NAME', 'telegram_bot')
# Строка подключения
self.DATABASE_URL = f"postgresql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
# Настройки SQLAlchemy
self.SQLALCHEMY_ENGINE_OPTIONS = {
'pool_pre_ping': True, # Проверка соединения перед использованием
'pool_recycle': 300, # Переподключение каждые 5 минут
'pool_size': 10, # Размер пула соединений
'max_overflow': 20, # Максимальное количество дополнительных соединений
'echo': False, # Логирование SQL запросов (для отладки)
}
class Database:
"""Класс для управления подключением к базе данных"""
def __init__(self, config: DatabaseConfig = None):
self.config = config or DatabaseConfig()
self.engine = None
self.SessionLocal = None
self._initialize()
def _initialize(self):
"""Инициализация подключения к базе данных"""
try:
# Создание движка SQLAlchemy
self.engine = create_engine(
self.config.DATABASE_URL,
**self.config.SQLALCHEMY_ENGINE_OPTIONS
)
# Создание фабрики сессий
self.SessionLocal = sessionmaker(
autocommit=False,
autoflush=False,
bind=self.engine
)
logger.info("Подключение к базе данных успешно инициализировано")
except Exception as e:
logger.error(f"Ошибка при инициализации базы данных: {e}")
raise
def create_tables(self):
"""Создание всех таблиц в базе данных"""
try:
Base.metadata.create_all(bind=self.engine)
logger.info("Таблицы успешно созданы")
except Exception as e:
logger.error(f"Ошибка при создании таблиц: {e}")
raise
def drop_tables(self):
"""Удаление всех таблиц из базы данных"""
try:
Base.metadata.drop_all(bind=self.engine)
logger.info("Таблицы успешно удалены")
except Exception as e:
logger.error(f"Ошибка при удалении таблиц: {e}")
raise
@contextmanager
def get_session(self) -> Generator:
"""Контекстный менеджер для получения сессии базы данных"""
session = self.SessionLocal()
try:
yield session
session.commit()
except Exception as e:
session.rollback()
logger.error(f"Ошибка в сессии базы данных: {e}")
raise
finally:
session.close()
def get_session_sync(self):
"""Получение синхронной сессии (не забудьте закрыть!)"""
return self.SessionLocal()
def test_connection(self) -> bool:
"""Проверка подключения к базе данных"""
from sqlalchemy import text
try:
with self.get_session() as session:
session.execute(text("SELECT 1"))
logger.info("Подключение к базе данных работает корректно")
return True
except Exception as e:
logger.error(f"Ошибка подключения к базе данных: {e}")
return False
def close(self):
"""Закрытие подключения к базе данных"""
if self.engine:
self.engine.dispose()
logger.info("Подключение к базе данных закрыто")
# Глобальный экземпляр базы данных
db = Database()
# Удобные функции для использования в приложении
def get_db_session():
"""Получение сессии базы данных для использования в приложении"""
return db.get_session()
def init_database():
"""Инициализация базы данных (создание таблиц)"""
db.create_tables()
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):
"""
Быстрая загрузка изображения из файла в базу данных
Args:
file_path: Путь к файлу изображения
alt_text: Альтернативный текст
Returns:
Image: Объект изображения или None при ошибке
"""
import os
import mimetypes
from database.repositories import ImageRepository
try:
if not os.path.isfile(file_path):
return None
filename = os.path.basename(file_path)
mime_type, _ = mimetypes.guess_type(filename)
if not mime_type:
mime_type = 'image/jpeg'
with open(file_path, 'rb') as f:
image_data = f.read()
with get_db_session() as session:
image = ImageRepository.create_image(
session, filename, mime_type, image_data, alt_text
)
session.commit()
return image
except Exception as e:
logger.error(f"Ошибка загрузки изображения {file_path}: {e}")
return None

284
database/example.py
View File

@@ -1,284 +0,0 @@
"""
Скрипт для создания примеров данных для системы тестирования
Создает тестовые категории, тесты, вопросы и варианты ответов
"""
import sys
import os
from datetime import datetime
# Добавляем путь к корневой директории проекта
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
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
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 _create_questions_for_test(session: Session, question_repo: QuestionRepository, test_id: int, questions_data: list):
"""Вспомогательная функция для создания вопросов теста"""
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🗑️ Очистка тестовых данных...")
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__":
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()

271
database/load_images.py
View File

@@ -1,271 +0,0 @@
"""
Скрипт для загрузки изображений из папки images/ в базу данных
"""
import os
import mimetypes
from pathlib import Path
from database.database import db, get_db_session
from database.services import ImageService
from database.repositories import ImageRepository
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def get_image_content_type(filename: str) -> str:
"""Определение MIME типа изображения по расширению файла"""
mime_type, _ = mimetypes.guess_type(filename)
# Если mimetypes не смог определить, делаем по расширению
if not mime_type:
ext = filename.lower().split('.')[-1]
mime_types = {
'jpg': 'image/jpeg',
'jpeg': 'image/jpeg',
'png': 'image/png',
'gif': 'image/gif',
'bmp': 'image/bmp',
'webp': 'image/webp'
}
mime_type = mime_types.get(ext, 'image/jpeg')
return mime_type
def load_image_to_db(image_path: str, alt_text: str = None) -> bool:
"""
Загрузка одного изображения в базу данных
Args:
image_path: Путь к файлу изображения
alt_text: Альтернативный текст для изображения
Returns:
bool: True если загрузка успешна, False если ошибка
"""
try:
# Проверяем, существует ли файл
if not os.path.isfile(image_path):
logger.error(f"Файл не найден: {image_path}")
return False
# Получаем информацию о файле
filename = os.path.basename(image_path)
content_type = get_image_content_type(filename)
# Читаем бинарные данные
with open(image_path, 'rb') as f:
image_data = f.read()
# Проверяем размер файла (ограничение PostgreSQL ~1GB, но лучше меньше)
file_size = len(image_data)
if file_size > 10 * 1024 * 1024: # 10MB лимит
logger.warning(f"Файл слишком большой ({file_size} байт): {filename}")
return False
# Сохраняем в базу данных
with get_db_session() as session:
# Проверяем, не существует ли уже такой файл
existing_image = ImageRepository.get_image_by_filename(session, filename)
if existing_image:
logger.info(f"Изображение уже существует в БД: {filename}")
return True
# Создаем новую запись
image = ImageRepository.create_image(
session=session,
filename=filename,
content_type=content_type,
image_data=image_data,
alt_text=alt_text or f"Изображение {filename}"
)
session.commit()
logger.info(f"Загружено изображение: {filename} (ID: {image.id}, размер: {file_size} байт)")
return True
except Exception as e:
logger.error(f"Ошибка при загрузке изображения {image_path}: {e}")
return False
def load_images_from_directory(images_dir: str = "images") -> dict:
"""
Загрузка всех изображений из указанной директории
Args:
images_dir: Путь к директории с изображениями
Returns:
dict: Статистика загрузки
"""
if not os.path.isdir(images_dir):
logger.error(f"Директория не найдена: {images_dir}")
return {"success": 0, "failed": 0, "skipped": 0}
# Поддерживаемые форматы изображений
supported_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}
stats = {"success": 0, "failed": 0, "skipped": 0}
logger.info(f"Начинаем загрузку изображений из директории: {images_dir}")
# Проходим по всем файлам в директории
for filename in os.listdir(images_dir):
file_path = os.path.join(images_dir, filename)
# Пропускаем директории
if os.path.isdir(file_path):
continue
# Проверяем расширение файла
file_ext = Path(filename).suffix.lower()
if file_ext not in supported_extensions:
logger.info(f"Пропускаем файл неподдерживаемого формата: {filename}")
stats["skipped"] += 1
continue
# Загружаем изображение
if load_image_to_db(file_path):
stats["success"] += 1
else:
stats["failed"] += 1
logger.info(f"Загрузка завершена. Успешно: {stats['success']}, "
f"Ошибок: {stats['failed']}, Пропущено: {stats['skipped']}")
return stats
def list_images_in_db():
"""Показать все изображения в базе данных"""
try:
with get_db_session() as session:
images = ImageRepository.get_all_images(session)
if not images:
logger.info("В базе данных нет изображений")
return
logger.info(f"Изображения в базе данных ({len(images)} шт.):")
for image in images:
size_kb = image.file_size / 1024
logger.info(f" ID: {image.id}, Файл: {image.filename}, "
f"Тип: {image.content_type}, Размер: {size_kb:.1f} KB")
except Exception as e:
logger.error(f"Ошибка при получении списка изображений: {e}")
def delete_image_from_db(filename: str = None, image_id: int = None) -> bool:
"""
Удаление изображения из базы данных
Args:
filename: Имя файла для удаления
image_id: ID изображения для удаления
Returns:
bool: True если удаление успешно
"""
try:
with get_db_session() as session:
if filename:
image = ImageRepository.get_image_by_filename(session, filename)
elif image_id:
image = ImageRepository.get_image_by_id(session, image_id)
else:
logger.error("Необходимо указать filename или image_id")
return False
if not image:
logger.error("Изображение не найдено")
return False
if ImageRepository.delete_image(session, image.id):
session.commit()
logger.info(f"Изображение удалено: {image.filename}")
return True
else:
logger.error("Ошибка при удалении изображения")
return False
except Exception as e:
logger.error(f"Ошибка при удалении изображения: {e}")
return False
if __name__ == "__main__":
"""Главная функция для запуска скрипта"""
print("🖼️ Скрипт загрузки изображений в базу данных")
print("=" * 50)
# Проверяем подключение к БД
try:
from database.database import test_db_connection
if not test_db_connection():
print("❌ Ошибка подключения к базе данных!")
exit(1)
print("✅ Подключение к базе данных успешно")
except Exception as e:
print(f"❌ Ошибка: {e}")
exit(1)
# Меню
while True:
print("\nВыберите действие:")
print("1. Загрузить все изображения из папки images/")
print("2. Загрузить конкретное изображение")
print("3. Показать все изображения в БД")
print("4. Удалить изображение из БД")
print("5. Выход")
choice = input("\nВведите номер (1-5): ").strip()
if choice == "1":
stats = load_images_from_directory("images")
print(f"\n📊 Результат: загружено {stats['success']}, "
f"ошибок {stats['failed']}, пропущено {stats['skipped']}")
elif choice == "2":
image_path = input("Введите путь к изображению: ").strip()
alt_text = input("Введите описание (необязательно): ").strip() or None
if load_image_to_db(image_path, alt_text):
print("✅ Изображение загружено!")
else:
print("❌ Ошибка загрузки")
elif choice == "3":
list_images_in_db()
elif choice == "4":
print("Удаление по:")
print("1. Имени файла")
print("2. ID")
sub_choice = input("Выберите (1-2): ").strip()
if sub_choice == "1":
filename = input("Введите имя файла: ").strip()
if delete_image_from_db(filename=filename):
print("✅ Изображение удалено!")
else:
print("❌ Ошибка удаления")
elif sub_choice == "2":
try:
image_id = int(input("Введите ID: ").strip())
if delete_image_from_db(image_id=image_id):
print("✅ Изображение удалено!")
else:
print("❌ Ошибка удаления")
except ValueError:
print("❌ Неверный ID")
elif choice == "5":
print("👋 До свидания!")
break
else:
print("❌ Неверный выбор")

208
database/models.py
View File

@@ -1,208 +0,0 @@
"""
Базовые модели для работы с базой данных
"""
from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text, BigInteger, ForeignKey, LargeBinary
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from database.database import Base
class User(Base):
"""Модель пользователя телеграм бота"""
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
telegram_id = Column(BigInteger, unique=True, index=True, nullable=False)
username = Column(String(255), nullable=True)
first_name = Column(String(255), nullable=True)
last_name = Column(String(255), nullable=True)
language_code = Column(String(10), default='ru')
is_active = Column(Boolean, default=True)
is_admin = Column(Boolean, default=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Связь с результатами тестов
test_results = relationship("TestResult", back_populates="user")
def __repr__(self):
return f"<User(id={self.id}, telegram_id={self.telegram_id}, username={self.username})>"
class Message(Base):
"""Модель сообщений"""
__tablename__ = "messages"
id = Column(Integer, primary_key=True, index=True)
telegram_message_id = Column(BigInteger, nullable=False)
user_id = Column(Integer, nullable=False) # Внешний ключ на users.id
text = Column(Text, nullable=True)
message_type = Column(String(50), default='text') # text, photo, document, etc.
created_at = Column(DateTime(timezone=True), server_default=func.now())
def __repr__(self):
return f"<Message(id={self.id}, user_id={self.user_id}, type={self.message_type})>"
class BotSettings(Base):
"""Модель настроек бота"""
__tablename__ = "bot_settings"
id = Column(Integer, primary_key=True, index=True)
key = Column(String(255), unique=True, nullable=False)
value = Column(Text, nullable=True)
description = Column(Text, nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
def __repr__(self):
return f"<BotSettings(key={self.key}, value={self.value})>"
# ===== МОДЕЛИ ДЛЯ СИСТЕМЫ ТЕСТИРОВАНИЯ =====
class TestCategory(Base):
"""Категории тестов"""
__tablename__ = "test_categories"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(255), unique=True, nullable=False) # "Тест Климова", "Профориентация" и т.д.
description = Column(Text, nullable=True)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
# Связь с тестами
tests = relationship("Test", back_populates="category")
def __repr__(self):
return f"<TestCategory(id={self.id}, name={self.name})>"
class Test(Base):
"""Конкретные тесты"""
__tablename__ = "tests"
id = Column(Integer, primary_key=True, index=True)
category_id = Column(Integer, ForeignKey("test_categories.id"), nullable=False)
title = Column(String(255), nullable=False) # "📘 Тест Климова"
description = Column(Text, nullable=True)
instructions = Column(Text, nullable=True) # Инструкции перед тестом
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
# Связи
category = relationship("TestCategory", back_populates="tests")
questions = relationship("Question", back_populates="test", cascade="all, delete-orphan")
user_results = relationship("TestResult", back_populates="test")
def __repr__(self):
return f"<Test(id={self.id}, title={self.title})>"
class Image(Base):
"""Хранение изображений в базе данных"""
__tablename__ = "images"
id = Column(Integer, primary_key=True, index=True)
filename = Column(String(255), nullable=False) # Оригинальное имя файла
content_type = Column(String(100), nullable=False) # image/jpeg, image/png и т.д.
file_size = Column(Integer, nullable=False) # Размер в байтах
image_data = Column(LargeBinary, nullable=False) # Бинарные данные изображения
alt_text = Column(String(255), nullable=True) # Альтернативный текст для доступности
created_at = Column(DateTime(timezone=True), server_default=func.now())
# Связь с вопросами
questions = relationship("Question", back_populates="image")
def __repr__(self):
return f"<Image(id={self.id}, filename={self.filename}, size={self.file_size})>"
class Question(Base):
"""Вопросы тестов"""
__tablename__ = "questions"
id = Column(Integer, primary_key=True, index=True)
test_id = Column(Integer, ForeignKey("tests.id"), nullable=False)
order_index = Column(Integer, nullable=False) # Порядок вопроса в тесте
question_text = Column(Text, nullable=True) # Текст вопроса (может быть пустым если только картинка)
image_id = Column(Integer, ForeignKey("images.id"), nullable=True) # Привязка к изображению
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
# Связи
test = relationship("Test", back_populates="questions")
image = relationship("Image", back_populates="questions")
answer_options = relationship("AnswerOption", back_populates="question", cascade="all, delete-orphan")
user_answers = relationship("UserAnswer", back_populates="question")
def __repr__(self):
return f"<Question(id={self.id}, test_id={self.test_id}, order={self.order_index})>"
class AnswerOption(Base):
"""Варианты ответов на вопросы"""
__tablename__ = "answer_options"
id = Column(Integer, primary_key=True, index=True)
question_id = Column(Integer, ForeignKey("questions.id"), nullable=False)
option_text = Column(Text, nullable=False) # Текст варианта ответа
category = Column(String(255), nullable=True) # Категория для подсчета результатов (например, "Человек-природа")
order_index = Column(Integer, nullable=False) # Порядок варианта (1, 2, 3...)
is_correct = Column(Boolean, default=False) # Для тестов с правильными ответами
created_at = Column(DateTime(timezone=True), server_default=func.now())
# Связи
question = relationship("Question", back_populates="answer_options")
user_answers = relationship("UserAnswer", back_populates="selected_option")
def __repr__(self):
return f"<AnswerOption(id={self.id}, question_id={self.question_id}, text='{self.option_text[:50]}')>"
class TestResult(Base):
"""Результаты прохождения тестов пользователями"""
__tablename__ = "test_results"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
test_id = Column(Integer, ForeignKey("tests.id"), nullable=False)
started_at = Column(DateTime(timezone=True), server_default=func.now())
completed_at = Column(DateTime(timezone=True), nullable=True)
is_completed = Column(Boolean, default=False)
result_data = Column(Text, nullable=True) # JSON с результатами по категориям
# Связи
user = relationship("User", back_populates="test_results")
test = relationship("Test", back_populates="user_results")
answers = relationship("UserAnswer", back_populates="test_result", cascade="all, delete-orphan")
def __repr__(self):
return f"<TestResult(id={self.id}, user_id={self.user_id}, test_id={self.test_id}, completed={self.is_completed})>"
class UserAnswer(Base):
"""Ответы пользователей на конкретные вопросы"""
__tablename__ = "user_answers"
id = Column(Integer, primary_key=True, index=True)
test_result_id = Column(Integer, ForeignKey("test_results.id"), nullable=False)
question_id = Column(Integer, ForeignKey("questions.id"), nullable=False)
selected_option_id = Column(Integer, ForeignKey("answer_options.id"), nullable=False)
answered_at = Column(DateTime(timezone=True), server_default=func.now())
# Связи
test_result = relationship("TestResult", back_populates="answers")
question = relationship("Question", back_populates="user_answers")
selected_option = relationship("AnswerOption", back_populates="user_answers")
def __repr__(self):
return f"<UserAnswer(id={self.id}, question_id={self.question_id}, option_id={self.selected_option_id})>"

334
database/repositories.py
View File

@@ -1,334 +0,0 @@
"""
Репозитории для работы с моделями базы данных системы тестирования
"""
from typing import Optional, List
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_
from database.models import (
User, Message, BotSettings, Test, TestCategory, Question,
AnswerOption, TestResult, UserAnswer, Image
)
import logging
logger = logging.getLogger(__name__)
class UserRepository:
"""Репозиторий для работы с пользователями"""
@staticmethod
def create_user(session: Session, telegram_id: int, username: str = None,
first_name: str = None, last_name: str = None) -> User:
"""Создание нового пользователя"""
user = User(
telegram_id=telegram_id,
username=username,
first_name=first_name,
last_name=last_name
)
session.add(user)
session.flush() # Получаем ID без коммита
return user
@staticmethod
def get_user_by_telegram_id(session: Session, telegram_id: int) -> Optional[User]:
"""Получение пользователя по Telegram ID"""
return session.query(User).filter(User.telegram_id == telegram_id).first()
@staticmethod
def get_user_by_id(session: Session, user_id: int) -> Optional[User]:
"""Получение пользователя по ID"""
return session.query(User).filter(User.id == user_id).first()
@staticmethod
def get_all_users(session: Session, active_only: bool = True) -> List[User]:
"""Получение всех пользователей"""
query = session.query(User)
if active_only:
query = query.filter(User.is_active == True)
return query.all()
@staticmethod
def update_user(session: Session, telegram_id: int, **kwargs) -> Optional[User]:
"""Обновление данных пользователя"""
user = UserRepository.get_user_by_telegram_id(session, telegram_id)
if user:
for key, value in kwargs.items():
if hasattr(user, key):
setattr(user, key, value)
session.flush()
return user
@staticmethod
def get_or_create_user(session: Session, telegram_id: int, **kwargs) -> tuple[User, bool]:
"""Получение или создание пользователя. Возвращает (user, created)"""
user = UserRepository.get_user_by_telegram_id(session, telegram_id)
if user:
return user, False
else:
user = UserRepository.create_user(session, telegram_id, **kwargs)
return user, True
class MessageRepository:
"""Репозиторий для работы с сообщениями"""
@staticmethod
def create_message(session: Session, telegram_message_id: int, user_id: int,
text: str = None, message_type: str = 'text') -> Message:
"""Создание нового сообщения"""
message = Message(
telegram_message_id=telegram_message_id,
user_id=user_id,
text=text,
message_type=message_type
)
session.add(message)
session.flush()
return message
@staticmethod
def get_messages_by_user(session: Session, user_id: int, limit: int = 100) -> List[Message]:
"""Получение сообщений пользователя"""
return (session.query(Message)
.filter(Message.user_id == user_id)
.order_by(Message.created_at.desc())
.limit(limit)
.all())
@staticmethod
def get_recent_messages(session: Session, limit: int = 100) -> List[Message]:
"""Получение последних сообщений"""
return (session.query(Message)
.order_by(Message.created_at.desc())
.limit(limit)
.all())
class SettingsRepository:
"""Репозиторий для работы с настройками бота"""
@staticmethod
def get_setting(session: Session, key: str) -> Optional[str]:
"""Получение значения настройки"""
setting = session.query(BotSettings).filter(BotSettings.key == key).first()
return setting.value if setting else None
@staticmethod
def set_setting(session: Session, key: str, value: str, description: str = None) -> BotSettings:
"""Установка значения настройки"""
setting = session.query(BotSettings).filter(BotSettings.key == key).first()
if setting:
setting.value = value
if description:
setting.description = description
else:
setting = BotSettings(key=key, value=value, description=description)
session.add(setting)
session.flush()
return setting
@staticmethod
def get_all_settings(session: Session) -> List[BotSettings]:
"""Получение всех настроек"""
return session.query(BotSettings).all()
@staticmethod
def delete_setting(session: Session, key: str) -> bool:
"""Удаление настройки"""
setting = session.query(BotSettings).filter(BotSettings.key == key).first()
if setting:
session.delete(setting)
session.flush()
return True
return False
class ImageRepository:
"""Репозиторий для работы с изображениями"""
@staticmethod
def create_image(session: Session, filename: str, content_type: str,
image_data: bytes, alt_text: str = None) -> Image:
"""Создание записи изображения"""
image = Image(
filename=filename,
content_type=content_type,
file_size=len(image_data),
image_data=image_data,
alt_text=alt_text
)
session.add(image)
session.flush()
return image
@staticmethod
def get_image_by_id(session: Session, image_id: int) -> Optional[Image]:
"""Получение изображения по ID"""
return session.query(Image).filter(Image.id == image_id).first()
@staticmethod
def get_image_by_filename(session: Session, filename: str) -> Optional[Image]:
"""Получение изображения по имени файла"""
return session.query(Image).filter(Image.filename == filename).first()
@staticmethod
def get_all_images(session: Session) -> List[Image]:
"""Получение всех изображений"""
return session.query(Image).all()
@staticmethod
def delete_image(session: Session, image_id: int) -> bool:
"""Удаление изображения"""
image = session.query(Image).filter(Image.id == image_id).first()
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

6
database/requirements.txt
View File

@@ -1,6 +0,0 @@
# Необходимые зависимости для работы с PostgreSQL и SQLAlchemy
sqlalchemy>=2.0.0
psycopg2-binary>=2.9.0 # PostgreSQL адаптер для Python
python-dotenv>=1.0.0 # Для работы с переменными окружения
alembic>=1.12.0 # Для миграций базы данных (опционально)

358
database/services.py
View File

@@ -1,358 +0,0 @@
"""
Сервисы для работы с системой тестирования
Основная бизнес-логика для получения вопросов, сохранения ответов и работы с изображениями
"""
import json
import io
from typing import List, Optional, Dict, Any, Tuple
from sqlalchemy.orm import Session
from sqlalchemy import and_
from database.models import (
User, Test, TestCategory, Question, AnswerOption,
TestResult, UserAnswer, Image
)
import logging
logger = logging.getLogger(__name__)
class ImageService:
"""Сервис для работы с изображениями в базе данных"""
@staticmethod
def save_image_to_db(
db: Session,
filename: str,
content_type: str,
image_data: bytes,
alt_text: str = None
) -> Image:
"""
Сохранение изображения в базу данных
Args:
db: Сессия базы данных
filename: Имя файла
content_type: MIME тип (image/jpeg, image/png и т.д.)
image_data: Бинарные данные изображения
alt_text: Альтернативный текст
Returns:
Image: Созданная запись изображения
"""
image = Image(
filename=filename,
content_type=content_type,
file_size=len(image_data),
image_data=image_data,
alt_text=alt_text
)
db.add(image)
db.commit()
db.refresh(image)
return image
@staticmethod
def get_image_data(db: Session, image_id: int) -> Optional[Tuple[bytes, str, str]]:
"""
Получение данных изображения из БД
Args:
db: Сессия базы данных
image_id: ID изображения
Returns:
Tuple[bytes, str, str]: (image_data, content_type, filename) или None
"""
image = db.query(Image).filter(Image.id == image_id).first()
if image:
return image.image_data, image.content_type, image.filename
return None
@staticmethod
def get_image_by_filename(db: Session, filename: str) -> Optional[Image]:
"""Получение изображения по имени файла"""
return db.query(Image).filter(Image.filename == filename).first()
class TestService:
"""Сервис для работы с тестами"""
@staticmethod
def get_available_tests(db: Session) -> List[Test]:
"""Получение всех активных тестов"""
return db.query(Test).filter(Test.is_active == True).all()
@staticmethod
def get_test_by_title(db: Session, title: str) -> Optional[Test]:
"""Получение теста по названию (например, '📘 Тест Климова')"""
return db.query(Test).filter(
and_(Test.title == title, Test.is_active == True)
).first()
@staticmethod
def get_test_questions(db: Session, test_id: int) -> List[Question]:
"""
Получение всех вопросов теста в правильном порядке
Args:
db: Сессия базы данных
test_id: ID теста
Returns:
List[Question]: Список вопросов с загруженными связями
"""
return db.query(Question).filter(
and_(Question.test_id == test_id, Question.is_active == True)
).order_by(Question.order_index).all()
@staticmethod
def start_test(db: Session, user_id: int, test_id: int) -> TestResult:
"""
Начало прохождения теста пользователем
Args:
db: Сессия базы данных
user_id: ID пользователя
test_id: ID теста
Returns:
TestResult: Созданная запись результата теста
"""
# Проверяем, есть ли незавершенный тест
existing_result = db.query(TestResult).filter(
and_(
TestResult.user_id == user_id,
TestResult.test_id == test_id,
TestResult.is_completed == False
)
).first()
if existing_result:
return existing_result
# Создаем новый результат теста
test_result = TestResult(
user_id=user_id,
test_id=test_id,
is_completed=False
)
db.add(test_result)
db.commit()
db.refresh(test_result)
return test_result
@staticmethod
def save_answer(
db: Session,
test_result_id: int,
question_id: int,
selected_option_id: int
) -> UserAnswer:
"""
Сохранение ответа пользователя на вопрос
Args:
db: Сессия базы данных
test_result_id: ID результата теста
question_id: ID вопроса
selected_option_id: ID выбранного варианта ответа
Returns:
UserAnswer: Созданная запись ответа
"""
# Проверяем, есть ли уже ответ на этот вопрос
existing_answer = db.query(UserAnswer).filter(
and_(
UserAnswer.test_result_id == test_result_id,
UserAnswer.question_id == question_id
)
).first()
if existing_answer:
# Обновляем существующий ответ
existing_answer.selected_option_id = selected_option_id
db.commit()
return existing_answer
# Создаем новый ответ
user_answer = UserAnswer(
test_result_id=test_result_id,
question_id=question_id,
selected_option_id=selected_option_id
)
db.add(user_answer)
db.commit()
db.refresh(user_answer)
return user_answer
@staticmethod
def complete_test(db: Session, test_result_id: int) -> TestResult:
"""
Завершение теста и подсчет результатов
Args:
db: Сессия базы данных
test_result_id: ID результата теста
Returns:
TestResult: Обновленная запись результата с подсчитанными данными
"""
from datetime import datetime
test_result = db.query(TestResult).filter(
TestResult.id == test_result_id
).first()
if not test_result:
raise ValueError(f"TestResult с ID {test_result_id} не найден")
# Получаем все ответы пользователя для этого теста
answers = db.query(UserAnswer).filter(
UserAnswer.test_result_id == test_result_id
).all()
# Подсчитываем результаты по категориям
category_scores = {}
for answer in answers:
option = db.query(AnswerOption).filter(
AnswerOption.id == answer.selected_option_id
).first()
if option and option.category:
category = option.category
category_scores[category] = category_scores.get(category, 0) + 1
# Сохраняем результаты в JSON формате
test_result.result_data = json.dumps(category_scores, ensure_ascii=False)
test_result.completed_at = datetime.now()
test_result.is_completed = True
db.commit()
db.refresh(test_result)
return test_result
@staticmethod
def get_test_results(db: Session, test_result_id: int) -> Dict[str, Any]:
"""
Получение и форматирование результатов теста
Args:
db: Сессия базы данных
test_result_id: ID результата теста
Returns:
Dict: Словарь с результатами и процентами
"""
test_result = db.query(TestResult).filter(
TestResult.id == test_result_id
).first()
if not test_result or not test_result.result_data:
return {}
# Парсим результаты
category_scores = json.loads(test_result.result_data)
# Получаем максимальные возможные баллы для каждой категории
questions = TestService.get_test_questions(db, test_result.test_id)
max_scores = {}
for question in questions:
for option in question.answer_options:
if option.category:
max_scores[option.category] = max_scores.get(option.category, 0) + 1
# Вычисляем проценты
results = {}
for category, score in category_scores.items():
max_score = max_scores.get(category, 1)
percentage = (score / max_score) * 100
results[category] = {
'score': score,
'max_score': max_score,
'percentage': round(percentage, 1)
}
return results
@staticmethod
def get_user_test_history(db: Session, user_id: int) -> List[TestResult]:
"""Получение истории тестов пользователя"""
return db.query(TestResult).filter(
and_(TestResult.user_id == user_id, TestResult.is_completed == True)
).order_by(TestResult.completed_at.desc()).all()
class QuestionService:
"""Сервис для работы с вопросами"""
@staticmethod
def get_question_with_options(db: Session, question_id: int) -> Optional[Question]:
"""
Получение вопроса со всеми вариантами ответов и изображением
Args:
db: Сессия базы данных
question_id: ID вопроса
Returns:
Question: Вопрос с загруженными связями или None
"""
return db.query(Question).filter(Question.id == question_id).first()
@staticmethod
def format_question_for_bot(db: Session, question: Question) -> Dict[str, Any]:
"""
Форматирование вопроса для отправки в бот
Args:
db: Сессия базы данных
question: Объект вопроса
Returns:
Dict: Сформатированные данные вопроса
"""
# Получаем варианты ответов
options = db.query(AnswerOption).filter(
AnswerOption.question_id == question.id
).order_by(AnswerOption.order_index).all()
# Получаем данные изображения, если есть
image_data = None
if question.image_id:
image_info = ImageService.get_image_data(db, question.image_id)
if image_info:
image_data = {
'data': image_info[0],
'content_type': image_info[1],
'filename': image_info[2]
}
return {
'id': question.id,
'text': question.question_text,
'order_index': question.order_index,
'image': image_data,
'options': [
{
'id': option.id,
'text': option.option_text,
'category': option.category,
'order_index': option.order_index
}
for option in options
]
}

58
docker-compose.yml
View File

@@ -20,32 +20,34 @@ services:
timeout: 5s
retries: 5
# Telegram Bot приложение
# Telegram Bot приложение (работает ТОЛЬКО через API)
bot:
build: .
container_name: telegram_bot
environment:
# Загружаем переменные из .env файла
# Настройки бота
TELEGRAM_BOT_TOKEN: ${TELEGRAM_BOT_TOKEN}
DB_HOST: postgres
DB_PORT: ${DB_PORT:-5432}
DB_NAME: ${DB_NAME:-botdb}
DB_USER: ${DB_USER:-botuser}
DB_PASSWORD: ${DB_PASSWORD}
APP_ENV: ${APP_ENV:-production}
LOG_LEVEL: ${LOG_LEVEL:-INFO}
DEBUG: ${DEBUG:-False}
command: python main.py --mode bot
# Настройки подключения к API (НЕ к БД!)
API_HOST: api
API_PORT: 8000
API_BASE_URL: http://api:8000
API_TIMEOUT: 30
command: python app.bot/main.py
depends_on:
postgres:
api:
condition: service_healthy
networks:
- bot_network
restart: unless-stopped
volumes:
- ./images:/app/images:ro
profiles:
- bot
- all
# API сервер (опционально)
# API сервер (единственный доступ к БД)
api:
build: .
container_name: telegram_bot_api
@@ -60,10 +62,10 @@ services:
# Настройки API сервера
API_HOST: 0.0.0.0
API_PORT: ${API_PORT:-8000}
API_RELOAD: "false"
LOG_LEVEL: ${LOG_LEVEL:-INFO}
APP_ENV: ${APP_ENV:-production}
command: python main.py --mode api
DEBUG: ${DEBUG:-False}
command: python app.api/run_server.py
ports:
- "${API_PORT:-8000}:${API_PORT:-8000}"
depends_on:
@@ -72,8 +74,36 @@ services:
networks:
- bot_network
restart: unless-stopped
healthcheck:
test: ["CMD-SHELL", "curl -f http://localhost:8000/health || exit 1"]
interval: 30s
timeout: 10s
retries: 3
profiles:
- api
- all
# Веб-интерфейс для управления (работает через API)
web:
build: .
container_name: telegram_bot_web
environment:
# Настройки подключения к API
API_HOST: api
API_PORT: 8000
API_BASE_URL: http://api:8000
command: python app.control-bot/main.py
ports:
- "${WEB_PORT:-3000}:3000"
depends_on:
api:
condition: service_healthy
networks:
- bot_network
restart: unless-stopped
profiles:
- web
- all
# Сети для изоляции сервисов
networks:

111
main.py
View File

@@ -1,111 +0,0 @@
"""
Основная точка входа для Telegram бота
Поддерживает запуск бота и API сервера
"""
import asyncio
import argparse
import logging
import sys
import os
# Добавляем текущую директорию в PYTHONPATH
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from bot_config import config
def setup_logging():
"""Настройка системы логирования"""
logging.basicConfig(
level=getattr(logging, config.LOG_LEVEL),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout),
logging.FileHandler('bot.log') if config.APP_ENV == 'production' else logging.NullHandler()
]
)
async def run_bot():
"""Запуск Telegram бота"""
try:
import telegram_bot
logging.info("Запуск Telegram бота...")
await telegram_bot.main()
except Exception as e:
logging.error(f"Ошибка при запуске бота: {e}")
raise
def run_api():
"""Запуск API сервера"""
import uvicorn
from api.main import app
logging.info("Запуск API сервера...")
uvicorn.run(
app,
host=config.DB_HOST if hasattr(config, 'API_HOST') else "0.0.0.0",
port=int(os.getenv('API_PORT', 8000)),
reload=config.DEBUG,
log_level=config.LOG_LEVEL.lower()
)
async def run_both():
"""Запуск бота и API одновременно"""
import uvicorn
from api.main import app
import telegram_bot
logging.info("Запуск бота и API сервера...")
# Создаем задачи для параллельного запуска
api_config = uvicorn.Config(
app,
host="0.0.0.0",
port=int(os.getenv('API_PORT', 8000)),
log_level=config.LOG_LEVEL.lower()
)
api_server = uvicorn.Server(api_config)
# Запускаем параллельно
await asyncio.gather(
telegram_bot.main(),
api_server.serve()
)
def main():
"""Основная функция"""
parser = argparse.ArgumentParser(description='Telegram Bot Application')
parser.add_argument(
'--mode',
choices=['bot', 'api', 'both'],
default='bot',
help='Режим запуска: bot (только бот), api (только API), both (бот + API)'
)
args = parser.parse_args()
# Настраиваем логирование
setup_logging()
logger = logging.getLogger(__name__)
try:
# Проверяем конфигурацию
config.validate()
logger.info(f"Конфигурация проверена. Режим: {config.APP_ENV}")
# Запускаем в зависимости от режима
if args.mode == 'bot':
asyncio.run(run_bot())
elif args.mode == 'api':
run_api()
elif args.mode == 'both':
asyncio.run(run_both())
except KeyboardInterrupt:
logger.info("Получен сигнал остановки")
except Exception as e:
logger.error(f"Критическая ошибка: {e}")
sys.exit(1)
if __name__ == "__main__":
main()

390
telegram_bot.py
View File

@@ -1,390 +0,0 @@
"""
Современный Telegram бот с использованием SQLAlchemy и сервисов
"""
import asyncio
import io
from datetime import datetime
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.types import KeyboardButton, ReplyKeyboardMarkup, ReplyKeyboardRemove, BufferedInputFile
from aiogram.filters.state import StateFilter
import logging
from bot_config import config
from database.database import get_db_session
from database.repositories import UserRepository, SettingsRepository
from database.services import TestService, QuestionService, ImageService
# Настройка логирования
logging.basicConfig(
level=getattr(logging, config.LOG_LEVEL),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Инициализация бота
bot = Bot(token=config.TELEGRAM_BOT_TOKEN)
dp = Dispatcher(storage=MemoryStorage())
# ===== FSM States =====
class Form(StatesGroup):
consent = State()
fio = State()
dob = State()
phone = State()
city = State()
test_in_progress = State()
# ===== /start command =====
@dp.message(Command("start"))
async def start(message: types.Message, state: FSMContext):
"""Обработка команды /start"""
try:
# Создаем или получаем пользователя
with get_db_session() as session:
user, created = UserRepository.get_or_create_user(
session,
telegram_id=message.from_user.id,
username=message.from_user.username,
first_name=message.from_user.first_name,
last_name=message.from_user.last_name
)
session.commit()
if created:
logger.info(f"Создан новый пользователь: {message.from_user.id}")
kb = [
[KeyboardButton(text="✅ Да, согласен"), KeyboardButton(text="❌ Нет")]
]
keyboard = ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True)
await message.answer(
"Привет! Я помогу пройти тест Климова.\\n\\n"
"Сначала — согласие на обработку персональных данных.\\n"
"Ты согласен?",
reply_markup=keyboard
)
await state.set_state(Form.consent)
except Exception as e:
logger.error(f"Ошибка в команде start: {e}")
await message.answer("Произошла ошибка. Попробуйте позже.")
# ===== Согласие на обработку данных =====
@dp.message(Form.consent)
async def consent(message: types.Message, state: FSMContext):
"""Обработка согласия на обработку персональных данных"""
if message.text.startswith("✅"):
await state.set_state(Form.fio)
await message.answer("Введите своё ФИО:", reply_markup=ReplyKeyboardRemove())
else:
await message.answer("Без согласия мы не можем продолжить 🙏", reply_markup=ReplyKeyboardRemove())
await state.clear()
# ===== Сбор ФИО =====
@dp.message(Form.fio)
async def collect_fio(message: types.Message, state: FSMContext):
"""Сбор ФИО пользователя"""
await state.update_data(fio=message.text)
await state.set_state(Form.dob)
await message.answer("Введите дату рождения (ДД.MM.ГГГГ):")
# ===== Сбор даты рождения =====
@dp.message(Form.dob)
async def collect_dob(message: types.Message, state: FSMContext):
"""Сбор даты рождения"""
try:
# Проверяем формат даты
datetime.strptime(message.text, "%d.%m.%Y")
await state.update_data(dob=message.text)
await state.set_state(Form.phone)
await message.answer("Введите номер телефона:")
except ValueError:
await message.answer("Неверный формат даты. Введите дату в формате ДД.MM.ГГГГ")
# ===== Сбор телефона =====
@dp.message(Form.phone)
async def collect_phone(message: types.Message, state: FSMContext):
"""Сбор номера телефона"""
await state.update_data(phone=message.text)
await state.set_state(Form.city)
await message.answer("Из какого ты населённого пункта?")
# ===== Сбор города и сохранение анкеты =====
@dp.message(Form.city)
async def collect_city_and_save(message: types.Message, state: FSMContext):
"""Сбор города и сохранение анкеты пользователя"""
try:
await state.update_data(city=message.text)
data = await state.get_data()
# Преобразуем дату
dob_parsed = datetime.strptime(data["dob"], "%d.%m.%Y").date()
# Сохраняем данные пользователя
with get_db_session() as session:
user = UserRepository.get_user_by_telegram_id(session, message.from_user.id)
if user:
# Обновляем дополнительные данные пользователя
# Здесь нужно будет расширить модель User для хранения ФИО, даты рождения и т.д.
# Или создать отдельную таблицу для профилей пользователей
logger.info(f"Пользователь {message.from_user.id} заполнил анкету: "
f"ФИО: {data['fio']}, ДР: {dob_parsed}, Телефон: {data['phone']}, Город: {data['city']}")
session.commit()
await message.answer("Анкета сохранена ✅")
# Показываем доступные тесты
await show_available_tests(message, state)
except ValueError:
await message.answer("Неверный формат даты. Введите дату в формате ДД.MM.ГГГГ")
await state.set_state(Form.dob)
except Exception as e:
logger.error(f"Ошибка при сохранении анкеты пользователя {message.from_user.id}: {e}")
await message.answer("Произошла ошибка при сохранении анкеты. Попробуйте позже.")
async def show_available_tests(message: types.Message, state: FSMContext):
"""Показать доступные тесты"""
try:
with get_db_session() as session:
tests = TestService.get_available_tests(session)
if not tests:
await message.answer("К сожалению, тесты пока недоступны.")
return
# Создаем клавиатуру с тестами
kb = []
for test in tests:
kb.append([KeyboardButton(text=test.title)])
keyboard = ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True)
await state.clear()
await message.answer("Выберите тест:", reply_markup=keyboard)
except Exception as e:
logger.error(f"Ошибка при получении списка тестов: {e}")
await message.answer("Произошла ошибка при загрузке тестов.")
# ===== Выбор теста =====
@dp.message(StateFilter(None))
async def select_test(message: types.Message, state: FSMContext):
"""Обработка выбора теста"""
try:
with get_db_session() as session:
# Получаем пользователя
user = UserRepository.get_user_by_telegram_id(session, message.from_user.id)
if not user:
await message.answer("Пользователь не найден. Используйте /start")
return
# Ищем тест по названию
test = TestService.get_test_by_title(session, message.text)
if not test:
await message.answer("Тест не найден или недоступен.")
return
# Начинаем тест
test_result = TestService.start_test(session, user.id, test.id)
session.commit()
# Получаем вопросы теста
questions = TestService.get_test_questions(session, test.id)
if not questions:
await message.answer("В тесте нет вопросов. Обратитесь к администратору.")
return
# Сохраняем данные теста в состоянии
await state.update_data(
test_id=test.id,
test_result_id=test_result.id,
questions=[(q.id, q.order_index) for q in questions],
current_question_index=0
)
await state.set_state(Form.test_in_progress)
# Отправляем первый вопрос
await send_question(message.from_user.id, state, session)
logger.info(f"Пользователь {message.from_user.id} начал тест: {test.title}")
except Exception as e:
logger.error(f"Ошибка при выборе теста: {e}")
await message.answer("Произошла ошибка при запуске теста. Попробуйте позже.")
async def send_question(user_id: int, state: FSMContext, session):
"""Отправка вопроса пользователю"""
try:
data = await state.get_data()
questions = data.get("questions", [])
current_index = data.get("current_question_index", 0)
if current_index >= len(questions):
# Тест завершен
await complete_test(user_id, state, session)
return
question_id, _ = questions[current_index]
# Получаем вопрос с вариантами ответов
question = QuestionService.get_question_with_options(session, question_id)
if not question:
await bot.send_message(user_id, "Ошибка при загрузке вопроса.")
return
# Форматируем вопрос для бота
question_data = QuestionService.format_question_for_bot(session, question)
# Создаем клавиатуру с вариантами ответов
kb = []
for option in question_data['options']:
kb.append([KeyboardButton(text=option['text'])])
keyboard = ReplyKeyboardMarkup(keyboard=kb, resize_keyboard=True)
# Отправляем изображение, если есть
if question_data.get('image'):
image_data = question_data['image']
try:
image_file = BufferedInputFile(
image_data['data'],
filename=image_data['filename']
)
await bot.send_photo(
chat_id=user_id,
photo=image_file,
caption=f"Вопрос {current_index + 1} из {len(questions)}:\\n{question_data['text'] or 'Выберите вариант'}",
reply_markup=keyboard
)
return
except Exception as e:
logger.error(f"Ошибка отправки изображения: {e}")
# Отправляем текстовый вопрос
question_text = question_data['text'] or "Выберите вариант"
await bot.send_message(
chat_id=user_id,
text=f"Вопрос {current_index + 1} из {len(questions)}:\\n{question_text}",
reply_markup=keyboard
)
except Exception as e:
logger.error(f"Ошибка при отправке вопроса: {e}")
await bot.send_message(user_id, "Произошла ошибка. Попробуйте позже.")
# ===== Обработка ответов на вопросы =====
@dp.message(Form.test_in_progress)
async def handle_answer(message: types.Message, state: FSMContext):
"""Обработка ответа на вопрос теста"""
try:
data = await state.get_data()
questions = data.get("questions", [])
current_index = data.get("current_question_index", 0)
test_result_id = data.get("test_result_id")
if current_index >= len(questions):
return
question_id, _ = questions[current_index]
with get_db_session() as session:
# Получаем вопрос с вариантами ответов
question = QuestionService.get_question_with_options(session, question_id)
if not question:
await message.answer("Ошибка при обработке ответа.")
return
# Ищем выбранный вариант ответа
selected_option = None
for option in question.answer_options:
if option.option_text == message.text:
selected_option = option
break
if not selected_option:
await message.answer("Пожалуйста, выберите один из предложенных вариантов.")
return
# Сохраняем ответ
TestService.save_answer(session, test_result_id, question_id, selected_option.id)
session.commit()
# Переходим к следующему вопросу
next_index = current_index + 1
await state.update_data(current_question_index=next_index)
# Отправляем следующий вопрос или завершаем тест
await send_question(message.from_user.id, state, session)
except Exception as e:
logger.error(f"Ошибка при обработке ответа: {e}")
await message.answer("Произошла ошибка при обработке ответа.")
async def complete_test(user_id: int, state: FSMContext, session):
"""Завершение теста и показ результатов"""
try:
data = await state.get_data()
test_result_id = data.get("test_result_id")
# Завершаем тест и подсчитываем результаты
test_result = TestService.complete_test(session, test_result_id)
session.commit()
# Получаем форматированные результаты
results = TestService.get_test_results(session, test_result_id)
# Формируем сообщение с результатами
result_text = "🎉 Тест завершен!\\n\\nВаши результаты:\\n"
for category, data in results.items():
result_text += f"• {category}: {data['percentage']}%\\n"
# Определяем ведущий тип
if results:
top_category = max(results.keys(), key=lambda x: results[x]['percentage'])
result_text += f"\\n🏆 Ваш ведущий тип: {top_category}"
await bot.send_message(user_id, result_text, reply_markup=ReplyKeyboardRemove())
# Отправляем описания профессий
profession_descriptions = {
"Человек-природа": "К типу 'человек-природа' можно отнести профессии, связанные с изучением живой и неживой природы (микробиолог, агрохимик, геолог), с уходом за растениями и животными (лесовод, овощевод, зоотехник).",
"Человек-техника": "Тип 'человек-техника' включает в себя профессии, связанные с созданием, монтажом, сборкой и наладкой технических устройств (каменщик, монтажник, сварщик, инженер-конструктор), эксплуатацией технических средств (водитель, крановщик, токарь, швея), ремонтом техники.",
"Человек-человек": "К профессиям типа 'человек-человек' относятся профессии, связанные с медицинским обслуживанием (врач, медсестра, санитарка), обучением и воспитанием (воспитатель, няня, учитель, преподаватель, тренер), бытовым обслуживанием (продавец, проводник, официант).",
"Человек-знаковая система": "Тип 'человек - знаковая система' объединяет профессии, связанные с текстами (корректор, оператор ПК, переводчик, библиотекарь), с цифрами, формулами и таблицами (программист, экономист, бухгалтер, кассир), с чертежами, картами, схемами (штурман, чертежник).",
"Человек-художественный образ": "К типу 'человек - художественный образ' можно отнести профессии, связанные с созданием, проектированием, моделированием художественных произведений (художник, журналист, модельер, композитор), с воспроизведением, изготовлением различных изделий по эскизу, образцу."
}
for category, description in profession_descriptions.items():
await bot.send_message(user_id, f"🟢 {category}\\n\\n{description}")
# Финальное сообщение
await bot.send_message(user_id,
"Задать вопрос или записаться на индивидуальную консультацию по построению образовательного маршрута можно здесь: @chrtakova11\\n\\n"
"Узнать подробнее о профессиях: https://postupi.online/professii/so/")
await state.clear()
logger.info(f"Пользователь {user_id} завершил тест")
except Exception as e:
logger.error(f"Ошибка при завершении теста: {e}")
await bot.send_message(user_id, "Произошла ошибка при подсчете результатов.")
# ===== Запуск бота =====
async def main():
"""Основная функция запуска бота"""
logger.info("Запуск Telegram бота...")
try:
await dp.start_polling(bot)
except Exception as e:
logger.error(f"Ошибка при запуске бота: {e}")
finally:
logger.info("Бот остановлен")
if __name__ == "__main__":
asyncio.run(main())