import os import pandas as pd from dotenv import load_dotenv from aiogram import Bot, Dispatcher, F, Router from aiogram.filters import Command from aiogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton, CallbackQuery, ContentType, InputFile import asyncio from database.db import BotDB from utils.response import get_days_of_week, get_json from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from aiogram.fsm.storage.memory import MemoryStorage from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from aiogram.fsm.storage.memory import MemoryStorage from aiogram.filters.state import StateFilter from datetime import datetime from aiogram.types import ContentType, FSInputFile from utils.pdf_generator import generate_student_dossier import tempfile load_dotenv() BOT_TOKEN = os.getenv("BOT_TOKEN") if not BOT_TOKEN: raise ValueError("Токен бота не найден. Убедитесь, что BOT_TOKEN задан в .env файле.") # Создаем объект бота и диспетчера bot = Bot(token=BOT_TOKEN) dp = Dispatcher() # Подключение к базе данных db = BotDB('database/database.db') # Обработчик команды /start @dp.message(Command(commands=["start"])) async def send_welcome(message: Message): db.add_user(user_id=message.from_user.id, username=message.from_user.username, group_id="") await message.answer("Привет! Я твой телеграм-бот, используй команду /raspisanye для просмотра расписания.") # Обработчик команды /raspisanye @dp.message(Command(commands=["raspisanye"])) async def handle_raspisanye(message: Message): user = db.get_user(message.from_user.id) if not user: await message.answer("Ваши данные не найдены в базе. Пожалуйста, начните с команды /start.") return # Получаем список дней недели для расписания (из response.py) try: days_of_week = get_days_of_week(user['group_id']) except Exception as e: await message.answer(f"Ошибка при получении расписания: {str(e)}") return # Создаем кнопки для выбора дня недели keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=day, callback_data=f"schedule_{day}")] for day in days_of_week ]) await message.answer("Выберите день недели, чтобы увидеть расписание:", reply_markup=keyboard) # Обработчик callback'ов для выбора дня недели и отображения расписания @dp.callback_query(lambda call: call.data.startswith("schedule_")) async def handle_schedule_selection(callback: CallbackQuery): callback_data = callback.data.split("_") selected_day = callback_data[1] user = db.get_user(callback.from_user.id) if not user: await callback.message.answer("Ваши данные не найдены в базе. Пожалуйста, начните с команды /start.") await callback.answer() return # Получаем данные о расписании для конкретного дня недели try: schedule_data = get_json(user['group_id']) except Exception as e: await callback.message.answer(f"Ошибка при получении расписания: {str(e)}") await callback.answer() return if selected_day not in schedule_data: await callback.message.answer(f"Расписание на {selected_day} не найдено.") await callback.answer() return # Формируем сообщение с расписанием на выбранный день day_schedule = schedule_data[selected_day] response_text = f"Расписание на {selected_day}:\n\n" for pair_number, pair_info in day_schedule.items(): response_text += f"Пара {pair_number}: {pair_info['time']}\n" response_text += f"Предмет: {pair_info['name']}\n" # Проверяем наличие преподавателя и добавляем его, если доступен if 'teacher' in pair_info and pair_info['teacher']: response_text += f"{pair_info['teacher']}\n" # Проверяем наличие метода и добавляем его, если доступен if 'method' in pair_info and pair_info['method']: response_text += f"Метод: {pair_info['method']}\n" response_text += "\n" await callback.message.answer(response_text) await callback.answer() # Обработчик команды /propuski @dp.message(Command(commands=["propuski"])) async def start_absence_tracking(message: Message): user = db.get_user(message.from_user.id) if not user: await message.answer("Ваши данные не найдены в базе. Пожалуйста, начните с команды /start.") return # Получаем дни недели try: days_of_week = get_days_of_week(user['group_id']) except Exception as e: await message.answer(f"Ошибка при получении дней недели: {str(e)}") return # Создаем кнопки для выбора дня недели, используем точное значение, как оно в расписании keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=day, callback_data=f"absence_day_{day}")] for day in days_of_week ]) await message.answer("Выберите день недели, чтобы отметить пропуски:", reply_markup=keyboard) # Обработчик выбора дня для отметки пропусков @dp.callback_query(lambda call: call.data.startswith("absence_day_")) async def handle_absence_day(callback: CallbackQuery): callback_data = callback.data.split("_", 2) selected_day = callback_data[2] # Сохраняем точное название дня, чтобы оно совпадало с ключами в расписании user = db.get_user(callback.from_user.id) if not user: await callback.message.answer("Ваши данные не найдены в базе. Пожалуйста, начните с команды /start.") await callback.answer() return # Получаем данные о расписании для всей недели try: schedule_data = get_json(user['group_id']) except Exception as e: await callback.message.answer(f"Ошибка при получении расписания: {str(e)}") await callback.answer() return # Проверяем наличие расписания для выбранного дня if selected_day not in schedule_data: await callback.message.answer(f"Расписание на {selected_day} не найдено.") await callback.answer() return # Формируем кнопки с номерами пар day_schedule = schedule_data[selected_day] pairs_keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=f"Пара {pair_number}: {pair_info['name']}", callback_data=f"absence_pair_{selected_day}_{pair_number}")] for pair_number, pair_info in day_schedule.items() ]) await callback.message.answer("Выберите номер пары, для которой хотите отметить пропуски:", reply_markup=pairs_keyboard) await callback.answer() # Обработчик выбора пары для отметки пропусков @dp.callback_query(lambda call: call.data.startswith("absence_pair_")) async def handle_absence_pair(callback: CallbackQuery): callback_data = callback.data.split("_") selected_day = callback_data[2] pair_number = callback_data[3] # Получаем список всех студентов из базы данных students = db.get_all_students() if not students: await callback.message.answer("Студенты не найдены в базе данных.") await callback.answer() return # Формируем кнопки с именами студентов buttons = [ [InlineKeyboardButton(text=f"{student[1][:20]}", callback_data=f"abs_{selected_day[:3]}_{pair_number}_{student[0]}")] for student in students ] students_keyboard = InlineKeyboardMarkup(inline_keyboard=buttons) await callback.message.answer("Выберите студента, чтобы отметить пропуск:", reply_markup=students_keyboard) await callback.answer() # Обработчик выбора студента для отметки пропуска @dp.callback_query(lambda call: call.data.startswith("abs_")) async def handle_student_selection(callback: CallbackQuery): callback_data = callback.data.split("_") selected_day = callback_data[1] pair_number = callback_data[2] student_id = callback_data[3] # Спрашиваем о причине пропуска reason_markup = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="Опоздал (+1 час)", callback_data=f"reason_late_{student_id}_{selected_day}_{pair_number}")], [InlineKeyboardButton(text="Отсутствовал (+2 часа)", callback_data=f"reason_absent_{student_id}_{selected_day}_{pair_number}")] ]) await callback.message.answer("Укажите причину пропуска:", reply_markup=reason_markup) await callback.answer() # Обработчик выбора причины пропуска @dp.callback_query(lambda call: call.data.startswith("reason_")) async def handle_absence_reason(callback: CallbackQuery): callback_data = callback.data.split("_") reason_type = callback_data[1] # late или absent student_id = callback_data[2] selected_day = callback_data[3] pair_number = callback_data[4] # Спросим, была ли причина уважительной или нет respect_markup = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="Уважительная", callback_data=f"final_{reason_type}_respect_{student_id}_{selected_day}_{pair_number}")], [InlineKeyboardButton(text="Неуважительная", callback_data=f"final_{reason_type}_norespect_{student_id}_{selected_day}_{pair_number}")] ]) reason_text = "опоздал" if reason_type == "late" else "отсутствовал" await callback.message.answer(f"Вы выбрали, что студент {reason_text}. Укажите, была ли причина уважительной:", reply_markup=respect_markup) await callback.answer() # Обработчик финального выбора для записи в базу данных @dp.callback_query(lambda call: call.data.startswith("final_")) async def handle_final_selection(callback: CallbackQuery): callback_data = callback.data.split("_") reason_type = callback_data[1] # late или absent respect_type = callback_data[2] # respect или norespect student_id = callback_data[3] selected_day = callback_data[4] pair_number = callback_data[5] # Записываем данные в базу данных status = "Опоздал" if reason_type == "late" else "Отсутствовал" respect = "Уважительно" if respect_type == "respect" else "Неуважительно" db.add_attendance_record(student_id, selected_day, pair_number, status, respect) await callback.message.answer(f"Пропуск студента успешно отмечен: {status}, причина: {respect}.") await callback.answer() # Обработчик команды /proguli @dp.message(Command(commands=["proguli"])) async def handle_proguli(message: Message): students = db.get_all_students() if not students: await message.answer("Студенты не найдены в базе данных.") return # Создаем кнопки для каждого студента students_keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=student[1], callback_data=f"proguli_student_{student[0]}")] for student in students ]) await message.answer("Выберите студента для просмотра досье:", reply_markup=students_keyboard) # Обработчик выбора студента для просмотра досье @dp.callback_query(lambda call: call.data.startswith("proguli_student_")) async def handle_student_profile(callback: CallbackQuery, state: FSMContext): callback_data = callback.data.split("_") student_id = callback_data[2] # Получаем данные о студенте, включая прогулы, характеристику и фото student = db.get_student_by_id(student_id) if not student: await callback.message.answer("Студент не найден в базе данных.") await callback.answer() return # Формируем текст для отображения досье студента profile_text = ( " <b>Досье студента:</b>\n" f" <b>ФИО:</b> {student['first_name']} {student['last_name']}\n" f" <b>Номер телефона:</b> {student['phone_number'] if student['phone_number'] else 'Не указано'}\n" f" <b>Адрес прописки:</b> {student['registration_address'] if student['registration_address'] else 'Не указано'}\n" f" <b>Фактический адрес проживания:</b> {student['actual_address'] if student['actual_address'] else 'Не указано'}\n" f" <b>Telegram тег:</b> {student['telegram_tag'] if student['telegram_tag'] else 'Не указан'}\n" f"⚕ <b>Хронические заболевания:</b> {student['chronic_diseases'] if student['chronic_diseases'] else 'Не указано'}\n" f" <b>Характеристика:</b> {student['characteristic']}\n\n" ) profile_text += f"✅ <b>Положительные черты:</b> {student.get('positive_traits', 'Не указаны')}\n" profile_text += f"❌ <b>Негативные черты:</b> {student.get('negative_traits', 'Не указаны')}\n" # Подсчет часов пропусков respectful_hours = sum(1 for absence in student["absences"] if absence["reason"].lower() == "уважительно") disrespectful_hours = sum(1 for absence in student["absences"] if absence["reason"].lower() != "уважительно") profile_text += ( f"⏳ <b>Часы пропусков (уважительно):</b> {respectful_hours} часов\n" f"⏱ <b>Часы пропусков (неуважительно):</b> {disrespectful_hours} часов\n\n" ) # Добавляем записи о прогулах, если они есть if student["absences"]: profile_text += " <b>Прогулы:</b>\n" for absence in student["absences"]: profile_text += f"- Дата: {absence['date']}, Пара: {absence['pair_number']}, Причина: {absence['reason']}\n" else: profile_text += "Прогулов не зафиксировано." # Формируем кнопки для редактирования данных edit_keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="Изменить номер телефона", callback_data=f"edit_phone_{student_id}")], [InlineKeyboardButton(text="Изменить адрес прописки", callback_data=f"edit_registration_{student_id}")], [InlineKeyboardButton(text="Изменить фактический адрес", callback_data=f"edit_actual_{student_id}")], [InlineKeyboardButton(text="Изменить телеграмм тег", callback_data=f"edit_tag_{student_id}")], [InlineKeyboardButton(text="Изменить хронические заболевания", callback_data=f"edit_diseases_{student_id}")], [InlineKeyboardButton(text="Изменить характеристику", callback_data=f"edit_characteristic_{student_id}")], [InlineKeyboardButton(text="Изменить фотографию", callback_data=f"edit_photo_{student_id}")], [InlineKeyboardButton(text="Изменить положительные черты", callback_data=f"edit_positive_traits_{student_id}")], [InlineKeyboardButton(text="Изменить негативные черты", callback_data=f"edit_negative_traits_{student_id}")], ]) # Отправляем фотографию, если она существует if student["photo_id"]: await callback.message.answer_photo(student["photo_id"], caption=profile_text, reply_markup=edit_keyboard, parse_mode="HTML") else: await callback.message.answer(profile_text, reply_markup=edit_keyboard, parse_mode="HTML") # Подтверждаем обработку колбэка await callback.answer() class EditStudentState(StatesGroup): edit_phone = State() edit_registration = State() edit_actual = State() edit_tag = State() edit_diseases = State() edit_characteristic = State() edit_photo = State() edit_positive_traits = State() edit_negative_traits = State() @dp.callback_query(lambda call: call.data.startswith("edit_")) async def handle_edit_request(callback: CallbackQuery, state: FSMContext): callback_data = callback.data.split("_") edit_field = callback_data[1] # Поле для редактирования (phone, registration, actual, tag, diseases) student_id = callback_data[2] # Устанавливаем prompt для каждой редактируемой информации edit_prompts = { "phone": "Введите новый номер телефона:", "registration": "Введите новый адрес прописки:", "actual": "Введите новый фактический адрес проживания:", "tag": "Введите новый Telegram тег:", "diseases": "Введите новую информацию о хронических заболеваниях:", "characteristic": "Введите новую характеристику", "photo": "Пожалуйста, отправьте новую фотографию:", "positive_traits": "Введите новые положительные черты:", "negative_traits": "Введите новые негативные черты:", } if edit_field in edit_prompts: await callback.message.answer(edit_prompts[edit_field]) await callback.answer() # Устанавливаем состояние для редактирования if edit_field == "phone": await state.set_state(EditStudentState.edit_phone) elif edit_field == "registration": await state.set_state(EditStudentState.edit_registration) elif edit_field == "actual": await state.set_state(EditStudentState.edit_actual) elif edit_field == "tag": await state.set_state(EditStudentState.edit_tag) elif edit_field == "diseases": await state.set_state(EditStudentState.edit_diseases) elif edit_field == "characteristic": await state.set_state(EditStudentState.edit_characteristic) elif edit_field == "photo": await state.set_state(EditStudentState.edit_photo) elif edit_field == "positive_traits": await state.set_state(EditStudentState.edit_positive_traits) elif edit_field == "negative_traits": await state.set_state(EditStudentState.edit_negative_traits) # Сохраняем student_id в контексте состояния await state.update_data(student_id=student_id) else: await callback.message.answer("Неверный запрос для редактирования.") await callback.answer() # Регистрация обработчиков для каждого состояния с правильным фильтром @dp.message(StateFilter(EditStudentState.edit_phone)) async def process_phone_edit(message: Message, state: FSMContext): new_phone = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем номер телефона в базе данных db.update_student_field(student_id, "phone_number", new_phone) await message.answer("Номер телефона успешно обновлен.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_registration)) async def process_registration_edit(message: Message, state: FSMContext): new_address = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем адрес прописки в базе данных db.update_student_field(student_id, "registration_address", new_address) await message.answer("Адрес прописки успешно обновлен.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_actual)) async def process_actual_edit(message: Message, state: FSMContext): new_address = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем фактический адрес проживания в базе данных db.update_student_field(student_id, "actual_address", new_address) await message.answer("Фактический адрес проживания успешно обновлен.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_tag)) async def process_tag_edit(message: Message, state: FSMContext): new_tag = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем Telegram тег в базе данных db.update_student_field(student_id, "telegram_tag", new_tag) await message.answer("Telegram тег успешно обновлен.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_diseases)) async def process_diseases_edit(message: Message, state: FSMContext): new_diseases = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем информацию о хронических заболеваниях в базе данных db.update_student_field(student_id, "chronic_diseases", new_diseases) await message.answer("Информация о хронических заболеваниях успешно обновлена.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_characteristic)) async def process_characteristic_edit(message: Message, state: FSMContext): new_characteristic = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем характеристику в базе данных db.update_student_field(student_id, "characteristic", new_characteristic) await message.answer("Характеристика успешно обновлена.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_photo), lambda message: message.content_type == ContentType.PHOTO) async def process_photo_edit(message: Message, state: FSMContext): # Получаем student_id из состояния data = await state.get_data() student_id = data['student_id'] # Получаем file_id новой фотографии photo_id = message.photo[-1].file_id # Сохраняем новый file_id в базе данных db.update_student_field(student_id, "photo_id", photo_id) await message.answer("Фотография успешно обновлена.") @dp.message(StateFilter(EditStudentState.edit_positive_traits)) async def process_positive_traits_edit(message: Message, state: FSMContext): new_traits = message.text.strip() data = await state.get_data() student_id = data["student_id"] db.update_student_field(student_id, "positive_traits", new_traits) await message.answer("Положительные черты успешно обновлены.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_negative_traits)) async def process_negative_traits_edit(message: Message, state: FSMContext): new_traits = message.text.strip() data = await state.get_data() student_id = data["student_id"] db.update_student_field(student_id, "negative_traits", new_traits) await message.answer("Негативные черты успешно обновлены.") await state.clear() # Обработчик команды /otsutstvuet @dp.message(Command(commands=['otsutstvuet'])) async def handle_otsutstvuet(message: Message): # Получаем текущий день в формате "дд.мм.гггг" current_day = datetime.now().strftime("%d.%m.%Y") # Получаем список отсутствующих студентов из базы данных absent_students = db.get_absent_students(current_day) if not absent_students: await message.answer("Сегодня никто не отсутствовал на занятиях.") else: # Создаем инлайн-клавиатуру со списком студентов, разделяя на блоки keyboard = InlineKeyboardMarkup(row_width=2) for student in absent_students: button = InlineKeyboardButton( text=f"{student['first_name']} {student['last_name']}", callback_data=f"confirm_{student['student_id']}" ) keyboard.add(button) await message.answer("Отсутствующие студенты на сегодня:", reply_markup=keyboard) # Обработчик нажатия на кнопку студента, чтобы подтвердить его присутствие @dp.callback_query(lambda call: call.data.startswith("confirm_")) async def handle_confirm_presence(callback: CallbackQuery): student_id = callback.data.split("_")[1] # Получаем данные о студенте student = db.get_student_by_id(int(student_id)) if student: # Формируем сообщение с информацией о подтверждении присутствия response = ( f"Студент: {student['first_name']} {student['last_name']}\n" "Этот студент пришел на следующую пару?" ) # Создаем клавиатуру для подтверждения confirm_keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="Да, пришел", callback_data=f"remove_{student_id}")], [InlineKeyboardButton(text="Нет, отсутствует", callback_data="cancel")] ]) await callback.message.answer(response, reply_markup=confirm_keyboard) await callback.answer() else: await callback.message.answer("Студент не найден.") await callback.answer() # Обработчик подтверждения удаления студента из списка отсутствующих @dp.callback_query(lambda call: call.data.startswith("remove_")) async def handle_remove_absence(callback: CallbackQuery): student_id = callback.data.split("_")[1] current_day = datetime.now().strftime("%d.%m.%Y") # Удаляем запись о студенте для текущего дня db.remove_absence(student_id, current_day) await callback.message.answer("Студент успешно удален из списка отсутствующих.") await callback.answer() # Обработчик отмены действия @dp.callback_query(lambda call: call.data == "cancel") async def handle_cancel(callback: CallbackQuery): await callback.message.answer("Действие отменено.") await callback.answer() # Обработчик команды /upload_student @dp.message(Command(commands=["upload_student"])) async def request_student_file(message: Message): await message.answer("Пожалуйста, загрузите CSV файл со списком студентов.") # Обработчик загрузки файла @dp.message(F.content_type == ContentType.DOCUMENT) async def handle_uploaded_file(message: Message): if not message.document.file_name.endswith('.csv'): await message.answer("Пожалуйста, загрузите файл в формате CSV.") return # Загружаем файл document_id = message.document.file_id file = await bot.get_file(document_id) file_path = file.file_path # Сохраняем CSV файл локально await bot.downloa Python import os import pandas as pd from dotenv import load_dotenv from aiogram import Bot, Dispatcher, F, Router from aiogram.filters import Command from aiogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton, CallbackQuery, ContentType, InputFile import asyncio from database.db import BotDB from utils.response import get_days_of_week, get_json from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from aiogram.fsm.storage.memory import MemoryStorage from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from aiogram.fsm.storage.memory import MemoryStorage from aiogram.filters.state import StateFilter from datetime import datetime from aiogram.types import ContentType, FSInputFile from utils.pdf_generator import generate_student_dossier import tempfile load_dotenv() BOT_TOKEN = os.getenv("BOT_TOKEN") if not BOT_TOKEN: raise ValueError("Токен бота не найден. Убедитесь, что BOT_TOKEN задан в .env файле.") # Создаем объект бота и диспетчера bot = Bot(token=BOT_TOKEN) dp = Dispatcher() # Подключение к базе данных db = BotDB('database/database.db') # Обработчик команды /start @dp.message(Command(commands=["start"])) async def send_welcome(message: Message): db.add_user(user_id=message.from_user.id, username=message.from_user.username, group_id="") await message.answer("Привет! Я твой телеграм-бот, используй команду /raspisanye для просмотра расписания.") # Обработчик команды /raspisanye @dp.message(Command(commands=["raspisanye"])) async def handle_raspisanye(message: Message): user = db.get_user(message.from_user.id) if not user: await message.answer("Ваши данные не найдены в базе. Пожалуйста, начните с команды /start.") return # Получаем список дней недели для расписания (из response.py) try: days_of_week = get_days_of_week(user['group_id']) except Exception as e: await message.answer(f"Ошибка при получении расписания: {str(e)}") return # Создаем кнопки для выбора дня недели keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=day, callback_data=f"schedule_{day}")] for day in days_of_week ]) await message.answer("Выберите день недели, чтобы увидеть расписание:", reply_markup=keyboard) # Обработчик callback'ов для выбора дня недели и отображения расписания @dp.callback_query(lambda call: call.data.startswith("schedule_")) async def handle_schedule_selection(callback: CallbackQuery): callback_data = callback.data.split("_") selected_day = callback_data[1] user = db.get_user(callback.from_user.id) if not user: await callback.message.answer("Ваши данные не найдены в базе. Пожалуйста, начните с команды /start.") await callback.answer() return # Получаем данные о расписании для конкретного дня недели try: schedule_data = get_json(user['group_id']) except Exception as e: await callback.message.answer(f"Ошибка при получении расписания: {str(e)}") await callback.answer() return if selected_day not in schedule_data: await callback.message.answer(f"Расписание на {selected_day} не найдено.") await callback.answer() return # Формируем сообщение с расписанием на выбранный день day_schedule = schedule_data[selected_day] response_text = f"Расписание на {selected_day}:\n\n" for pair_number, pair_info in day_schedule.items(): response_text += f"Пара {pair_number}: {pair_info['time']}\n" response_text += f"Предмет: {pair_info['name']}\n" # Проверяем наличие преподавателя и добавляем его, если доступен if 'teacher' in pair_info and pair_info['teacher']: response_text += f"{pair_info['teacher']}\n" # Проверяем наличие метода и добавляем его, если доступен if 'method' in pair_info and pair_info['method']: response_text += f"Метод: {pair_info['method']}\n" response_text += "\n" await callback.message.answer(response_text) await callback.answer() # Обработчик команды /propuski @dp.message(Command(commands=["propuski"])) async def start_absence_tracking(message: Message): user = db.get_user(message.from_user.id) if not user: await message.answer("Ваши данные не найдены в базе. Пожалуйста, начните с команды /start.") return # Получаем дни недели try: days_of_week = get_days_of_week(user['group_id']) except Exception as e: await message.answer(f"Ошибка при получении дней недели: {str(e)}") return # Создаем кнопки для выбора дня недели, используем точное значение, как оно в расписании keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=day, callback_data=f"absence_day_{day}")] for day in days_of_week ]) await message.answer("Выберите день недели, чтобы отметить пропуски:", reply_markup=keyboard) # Обработчик выбора дня для отметки пропусков @dp.callback_query(lambda call: call.data.startswith("absence_day_")) async def handle_absence_day(callback: CallbackQuery): callback_data = callback.data.split("_", 2) selected_day = callback_data[2] # Сохраняем точное название дня, чтобы оно совпадало с ключами в расписании user = db.get_user(callback.from_user.id) if not user: await callback.message.answer("Ваши данные не найдены в базе. Пожалуйста, начните с команды /start.") await callback.answer() return # Получаем данные о расписании для всей недели try: schedule_data = get_json(user['group_id']) except Exception as e: await callback.message.answer(f"Ошибка при получении расписания: {str(e)}") await callback.answer() return # Проверяем наличие расписания для выбранного дня if selected_day not in schedule_data: await callback.message.answer(f"Расписание на {selected_day} не найдено.") await callback.answer() return # Формируем кнопки с номерами пар day_schedule = schedule_data[selected_day] pairs_keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=f"Пара {pair_number}: {pair_info['name']}", callback_data=f"absence_pair_{selected_day}_{pair_number}")] for pair_number, pair_info in day_schedule.items() ]) await callback.message.answer("Выберите номер пары, для которой хотите отметить пропуски:", reply_markup=pairs_keyboard) await callback.answer() # Обработчик выбора пары для отметки пропусков @dp.callback_query(lambda call: call.data.startswith("absence_pair_")) async def handle_absence_pair(callback: CallbackQuery): callback_data = callback.data.split("_") selected_day = callback_data[2] pair_number = callback_data[3] # Получаем список всех студентов из базы данных students = db.get_all_students() if not students: await callback.message.answer("Студенты не найдены в базе данных.") await callback.answer() return # Формируем кнопки с именами студентов buttons = [ [InlineKeyboardButton(text=f"{student[1][:20]}", callback_data=f"abs_{selected_day[:3]}_{pair_number}_{student[0]}")] for student in students ] students_keyboard = InlineKeyboardMarkup(inline_keyboard=buttons) await callback.message.answer("Выберите студента, чтобы отметить пропуск:", reply_markup=students_keyboard) await callback.answer() # Обработчик выбора студента для отметки пропуска @dp.callback_query(lambda call: call.data.startswith("abs_")) async def handle_student_selection(callback: CallbackQuery): callback_data = callback.data.split("_") selected_day = callback_data[1] pair_number = callback_data[2] student_id = callback_data[3] # Спрашиваем о причине пропуска reason_markup = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="Опоздал (+1 час)", callback_data=f"reason_late_{student_id}_{selected_day}_{pair_number}")], [InlineKeyboardButton(text="Отсутствовал (+2 часа)", callback_data=f"reason_absent_{student_id}_{selected_day}_{pair_number}")] ]) await callback.message.answer("Укажите причину пропуска:", reply_markup=reason_markup) await callback.answer() # Обработчик выбора причины пропуска @dp.callback_query(lambda call: call.data.startswith("reason_")) async def handle_absence_reason(callback: CallbackQuery): callback_data = callback.data.split("_") reason_type = callback_data[1] # late или absent student_id = callback_data[2] selected_day = callback_data[3] pair_number = callback_data[4] # Спросим, была ли причина уважительной или нет respect_markup = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="Уважительная", callback_data=f"final_{reason_type}_respect_{student_id}_{selected_day}_{pair_number}")], [InlineKeyboardButton(text="Неуважительная", callback_data=f"final_{reason_type}_norespect_{student_id}_{selected_day}_{pair_number}")] ]) reason_text = "опоздал" if reason_type == "late" else "отсутствовал" await callback.message.answer(f"Вы выбрали, что студент {reason_text}. Укажите, была ли причина уважительной:", reply_markup=respect_markup) await callback.answer() # Обработчик финального выбора для записи в базу данных @dp.callback_query(lambda call: call.data.startswith("final_")) async def handle_final_selection(callback: CallbackQuery): callback_data = callback.data.split("_") reason_type = callback_data[1] # late или absent respect_type = callback_data[2] # respect или norespect student_id = callback_data[3] selected_day = callback_data[4] pair_number = callback_data[5] # Записываем данные в базу данных status = "Опоздал" if reason_type == "late" else "Отсутствовал" respect = "Уважительно" if respect_type == "respect" else "Неуважительно" db.add_attendance_record(student_id, selected_day, pair_number, status, respect) await callback.message.answer(f"Пропуск студента успешно отмечен: {status}, причина: {respect}.") await callback.answer() # Обработчик команды /proguli @dp.message(Command(commands=["proguli"])) async def handle_proguli(message: Message): students = db.get_all_students() if not students: await message.answer("Студенты не найдены в базе данных.") return # Создаем кнопки для каждого студента students_keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text=student[1], callback_data=f"proguli_student_{student[0]}")] for student in students ]) await message.answer("Выберите студента для просмотра досье:", reply_markup=students_keyboard) # Обработчик выбора студента для просмотра досье @dp.callback_query(lambda call: call.data.startswith("proguli_student_")) async def handle_student_profile(callback: CallbackQuery, state: FSMContext): callback_data = callback.data.split("_") student_id = callback_data[2] # Получаем данные о студенте, включая прогулы, характеристику и фото student = db.get_student_by_id(student_id) if not student: await callback.message.answer("Студент не найден в базе данных.") await callback.answer() return # Формируем текст для отображения досье студента profile_text = ( " <b>Досье студента:</b>\n" f" <b>ФИО:</b> {student['first_name']} {student['last_name']}\n" f" <b>Номер телефона:</b> {student['phone_number'] if student['phone_number'] else 'Не указано'}\n" f" <b>Адрес прописки:</b> {student['registration_address'] if student['registration_address'] else 'Не указано'}\n" f" <b>Фактический адрес проживания:</b> {student['actual_address'] if student['actual_address'] else 'Не указано'}\n" f" <b>Telegram тег:</b> {student['telegram_tag'] if student['telegram_tag'] else 'Не указан'}\n" f"⚕ <b>Хронические заболевания:</b> {student['chronic_diseases'] if student['chronic_diseases'] else 'Не указано'}\n" f" <b>Характеристика:</b> {student['characteristic']}\n\n" ) profile_text += f"✅ <b>Положительные черты:</b> {student.get('positive_traits', 'Не указаны')}\n" profile_text += f"❌ <b>Негативные черты:</b> {student.get('negative_traits', 'Не указаны')}\n" # Подсчет часов пропусков respectful_hours = sum(1 for absence in student["absences"] if absence["reason"].lower() == "уважительно") disrespectful_hours = sum(1 for absence in student["absences"] if absence["reason"].lower() != "уважительно") profile_text += ( f"⏳ <b>Часы пропусков (уважительно):</b> {respectful_hours} часов\n" f"⏱ <b>Часы пропусков (неуважительно):</b> {disrespectful_hours} часов\n\n" ) # Добавляем записи о прогулах, если они есть if student["absences"]: profile_text += " <b>Прогулы:</b>\n" for absence in student["absences"]: profile_text += f"- Дата: {absence['date']}, Пара: {absence['pair_number']}, Причина: {absence['reason']}\n" else: profile_text += "Прогулов не зафиксировано." # Формируем кнопки для редактирования данных edit_keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="Изменить номер телефона", callback_data=f"edit_phone_{student_id}")], [InlineKeyboardButton(text="Изменить адрес прописки", callback_data=f"edit_registration_{student_id}")], [InlineKeyboardButton(text="Изменить фактический адрес", callback_data=f"edit_actual_{student_id}")], [InlineKeyboardButton(text="Изменить телеграмм тег", callback_data=f"edit_tag_{student_id}")], [InlineKeyboardButton(text="Изменить хронические заболевания", callback_data=f"edit_diseases_{student_id}")], [InlineKeyboardButton(text="Изменить характеристику", callback_data=f"edit_characteristic_{student_id}")], [InlineKeyboardButton(text="Изменить фотографию", callback_data=f"edit_photo_{student_id}")], [InlineKeyboardButton(text="Изменить положительные черты", callback_data=f"edit_positive_traits_{student_id}")], [InlineKeyboardButton(text="Изменить негативные черты", callback_data=f"edit_negative_traits_{student_id}")], ]) # Отправляем фотографию, если она существует if student["photo_id"]: await callback.message.answer_photo(student["photo_id"], caption=profile_text, reply_markup=edit_keyboard, parse_mode="HTML") else: await callback.message.answer(profile_text, reply_markup=edit_keyboard, parse_mode="HTML") # Подтверждаем обработку колбэка await callback.answer() class EditStudentState(StatesGroup): edit_phone = State() edit_registration = State() edit_actual = State() edit_tag = State() edit_diseases = State() edit_characteristic = State() edit_photo = State() edit_positive_traits = State() edit_negative_traits = State() @dp.callback_query(lambda call: call.data.startswith("edit_")) async def handle_edit_request(callback: CallbackQuery, state: FSMContext): callback_data = callback.data.split("_") edit_field = callback_data[1] # Поле для редактирования (phone, registration, actual, tag, diseases) student_id = callback_data[2] # Устанавливаем prompt для каждой редактируемой информации edit_prompts = { "phone": "Введите новый номер телефона:", "registration": "Введите новый адрес прописки:", "actual": "Введите новый фактический адрес проживания:", "tag": "Введите новый Telegram тег:", "diseases": "Введите новую информацию о хронических заболеваниях:", "characteristic": "Введите новую характеристику", "photo": "Пожалуйста, отправьте новую фотографию:", "positive_traits": "Введите новые положительные черты:", "negative_traits": "Введите новые негативные черты:", } if edit_field in edit_prompts: await callback.message.answer(edit_prompts[edit_field]) await callback.answer() # Устанавливаем состояние для редактирования if edit_field == "phone": await state.set_state(EditStudentState.edit_phone) elif edit_field == "registration": await state.set_state(EditStudentState.edit_registration) elif edit_field == "actual": await state.set_state(EditStudentState.edit_actual) elif edit_field == "tag": await state.set_state(EditStudentState.edit_tag) elif edit_field == "diseases": await state.set_state(EditStudentState.edit_diseases) elif edit_field == "characteristic": await state.set_state(EditStudentState.edit_characteristic) elif edit_field == "photo": await state.set_state(EditStudentState.edit_photo) elif edit_field == "positive_traits": await state.set_state(EditStudentState.edit_positive_traits) elif edit_field == "negative_traits": await state.set_state(EditStudentState.edit_negative_traits) # Сохраняем student_id в контексте состояния await state.update_data(student_id=student_id) else: await callback.message.answer("Неверный запрос для редактирования.") await callback.answer() # Регистрация обработчиков для каждого состояния с правильным фильтром @dp.message(StateFilter(EditStudentState.edit_phone)) async def process_phone_edit(message: Message, state: FSMContext): new_phone = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем номер телефона в базе данных db.update_student_field(student_id, "phone_number", new_phone) await message.answer("Номер телефона успешно обновлен.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_registration)) async def process_registration_edit(message: Message, state: FSMContext): new_address = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем адрес прописки в базе данных db.update_student_field(student_id, "registration_address", new_address) await message.answer("Адрес прописки успешно обновлен.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_actual)) async def process_actual_edit(message: Message, state: FSMContext): new_address = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем фактический адрес проживания в базе данных db.update_student_field(student_id, "actual_address", new_address) await message.answer("Фактический адрес проживания успешно обновлен.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_tag)) async def process_tag_edit(message: Message, state: FSMContext): new_tag = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем Telegram тег в базе данных db.update_student_field(student_id, "telegram_tag", new_tag) await message.answer("Telegram тег успешно обновлен.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_diseases)) async def process_diseases_edit(message: Message, state: FSMContext): new_diseases = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем информацию о хронических заболеваниях в базе данных db.update_student_field(student_id, "chronic_diseases", new_diseases) await message.answer("Информация о хронических заболеваниях успешно обновлена.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_characteristic)) async def process_characteristic_edit(message: Message, state: FSMContext): new_characteristic = message.text.strip() data = await state.get_data() student_id = data['student_id'] # Обновляем характеристику в базе данных db.update_student_field(student_id, "characteristic", new_characteristic) await message.answer("Характеристика успешно обновлена.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_photo), lambda message: message.content_type == ContentType.PHOTO) async def process_photo_edit(message: Message, state: FSMContext): # Получаем student_id из состояния data = await state.get_data() student_id = data['student_id'] # Получаем file_id новой фотографии photo_id = message.photo[-1].file_id # Сохраняем новый file_id в базе данных db.update_student_field(student_id, "photo_id", photo_id) await message.answer("Фотография успешно обновлена.") @dp.message(StateFilter(EditStudentState.edit_positive_traits)) async def process_positive_traits_edit(message: Message, state: FSMContext): new_traits = message.text.strip() data = await state.get_data() student_id = data["student_id"] db.update_student_field(student_id, "positive_traits", new_traits) await message.answer("Положительные черты успешно обновлены.") await state.clear() @dp.message(StateFilter(EditStudentState.edit_negative_traits)) async def process_negative_traits_edit(message: Message, state: FSMContext): new_traits = message.text.strip() data = await state.get_data() student_id = data["student_id"] db.update_student_field(student_id, "negative_traits", new_traits) await message.answer("Негативные черты успешно обновлены.") await state.clear() # Обработчик команды /otsutstvuet @dp.message(Command(commands=['otsutstvuet'])) async def handle_otsutstvuet(message: Message): # Получаем текущий день в формате "дд.мм.гггг" current_day = datetime.now().strftime("%d.%m.%Y") # Получаем список отсутствующих студентов из базы данных absent_students = db.get_absent_students(current_day) if not absent_students: await message.answer("Сегодня никто не отсутствовал на занятиях.") else: # Создаем инлайн-клавиатуру со списком студентов, разделяя на блоки keyboard = InlineKeyboardMarkup(row_width=2) for student in absent_students: button = InlineKeyboardButton( text=f"{student['first_name']} {student['last_name']}", callback_data=f"confirm_{student['student_id']}" ) keyboard.add(button) await message.answer("Отсутствующие студенты на сегодня:", reply_markup=keyboard) # Обработчик нажатия на кнопку студента, чтобы подтвердить его присутствие @dp.callback_query(lambda call: call.data.startswith("confirm_")) async def handle_confirm_presence(callback: CallbackQuery): student_id = callback.data.split("_")[1] # Получаем данные о студенте student = db.get_student_by_id(int(student_id)) if student: # Формируем сообщение с информацией о подтверждении присутствия response = ( f"Студент: {student['first_name']} {student['last_name']}\n" "Этот студент пришел на следующую пару?" ) # Создаем клавиатуру для подтверждения confirm_keyboard = InlineKeyboardMarkup(inline_keyboard=[ [InlineKeyboardButton(text="Да, пришел", callback_data=f"remove_{student_id}")], [InlineKeyboardButton(text="Нет, отсутствует", callback_data="cancel")] ]) await callback.message.answer(response, reply_markup=confirm_keyboard) await callback.answer() else: await callback.message.answer("Студент не найден.") await callback.answer() # Обработчик подтверждения удаления студента из списка отсутствующих @dp.callback_query(lambda call: call.data.startswith("remove_")) async def handle_remove_absence(callback: CallbackQuery): student_id = callback.data.split("_")[1] current_day = datetime.now().strftime("%d.%m.%Y") # Удаляем запись о студенте для текущего дня db.remove_absence(student_id, current_day) await callback.message.answer("Студент успешно удален из списка отсутствующих.") await callback.answer() # Обработчик отмены действия @dp.callback_query(lambda call: call.data == "cancel") async def handle_cancel(callback: CallbackQuery): await callback.message.answer("Действие отменено.") await callback.answer() # Обработчик команды /upload_student @dp.message(Command(commands=["upload_student"])) async def request_student_file(message: Message): await message.answer("Пожалуйста, загрузите CSV файл со списком студентов.") # Обработчик загрузки файла @dp.message(F.content_type == ContentType.DOCUMENT) async def handle_uploaded_file(message: Message): if not message.document.file_name.endswith('.csv'): await message.answer("Пожалуйста, загрузите файл в формате CSV.") return # Загружаем файл document_id = message.document.file_id file = await bot.get_file(document_id) file_path = file.file_path # Сохраняем CSV файл локально await bot.downloa это код bot.py import sqlite3 from typing import Optional, Dict, List, Tuple class BotDB: def __init__(self, db_file: str): """Инициализация подключения к базе данных.""" self.conn = sqlite3.connect(db_file, check_same_thread=False) self.cursor = self.conn.cursor() self.create_tables() def create_tables(self) -> None: """Создание всех необходимых таблиц в базе данных.""" with open('database/database.sql', 'r', encoding='utf-8') as f: sql_script = f.read() self.cursor.executescript(sql_script) self.conn.commit() def add_user(self, user_id: int, username: str, group_id: str) -> None: """Добавление пользователя в базу данных.""" self.cursor.execute( "INSERT OR IGNORE INTO users (user_id, username, group_id) VALUES (?, ?, ?)", (user_id, username, group_id) ) self.conn.commit() def get_user(self, user_id: int) -> Optional[Dict]: """Получение информации о пользователе по его идентификатору.""" query = self.cursor.execute( "SELECT * FROM users WHERE user_id = ?", (user_id,) ) result = query.fetchone() if result: return { "user_id": result[0], "username": result[1], "group_id": result[2] } return None def add_student(self, first_name: str, last_name: str, phone_number: str = "", registration_address: str = "", actual_address: str = "", telegram_tag: str = "", chronic_diseases: str = "") -> None: """Добавление студента в базу данных.""" self.cursor.execute( "INSERT INTO students (first_name, last_name, phone_number, registration_address, " "actual_address, telegram_tag, chronic_diseases) VALUES (?, ?, ?, ?, ?, ?, ?)", (first_name, last_name, phone_number, registration_address, actual_address, telegram_tag, chronic_diseases) ) self.conn.commit() def get_all_students(self) -> List[Tuple[int, str]]: """Получение всех студентов из базы данных.""" query = self.cursor.execute( "SELECT student_id, first_name || ' ' || last_name FROM students" ) return query.fetchall() def add_attendance_record(self, student_id: int, selected_day: str, pair_number: str, status: str, respect: str) -> None: """Добавление записи о пропуске студента в базу данных.""" self.cursor.execute( "INSERT INTO attendance (student_id, date, pair_number, status, reason) VALUES (?, ?, ?, ?, ?)", (student_id, selected_day, pair_number, status, respect) ) self.conn.commit() def get_student_by_id(self, student_id: int) -> Optional[Dict]: """ Получение данных о студенте, включая записи о прогулов, характеристику, фотографию, положительные и негативные черты, по его идентификатору. Args: student_id (int): ID студента. Returns: Optional[Dict]: Словарь с данными о студенте или None, если студент не найден. """ # Получение основной информации о студенте query = self.cursor.execute( """ SELECT student_id, first_name, last_name, phone_number, registration_address, actual_address, telegram_tag, chronic_diseases, characteristic, photo_id, positive_traits, negative_traits FROM students WHERE student_id = ? """, (student_id,) ) student = query.fetchone() if not student: return None # Формирование словаря с основной информацией о студенте student_data = { "student_id": student[0], "first_name": student[1], "last_name": student[2], "phone_number": student[3], "registration_address": student[4], "actual_address": student[5], "telegram_tag": student[6], "chronic_diseases": student[7], "characteristic": student[8] if len(student) > 8 else "Не указано", "photo_id": student[9] if len(student) > 9 else None, "positive_traits": student[10] if len(student) > 10 else "Не указаны", "negative_traits": student[11] if len(student) > 11 else "Не указаны", } # Получение данных о прогулов студента attendance_query = self.cursor.execute( """ SELECT date, pair_number, reason FROM attendance WHERE student_id = ? ORDER BY date ASC """, (student_id,) ) absences = attendance_query.fetchall() # Формируем список прогулов student_data["absences"] = [ {"date": absence[0], "pair_number": absence[1], "reason": absence[2]} for absence in absences ] return student_data def get_student_attendance(self, student_id: int) -> List[Dict]: """Получение данных о пропусках студента по его идентификатору.""" query = self.cursor.execute( "SELECT * FROM attendance WHERE student_id = ?", (student_id,) ) result = query.fetchall() return [ { "id": record[0], "student_id": record[1], "date": record[2], "pair_number": record[3], "status": record[4], "reason": record[5], "hours": 1 if record[4] == "Опоздал" else 2 # Опоздание - 1 час, отсутствие - 2 часа } for record in result ] def update_student_field(self, student_id: int, field_name: str, field_value: str) -> None: """Обновление указанного поля студента.""" allowed_fields = {"phone_number", "registration_address", "actual_address", "telegram_tag", "chronic_diseases", "characteristic", "photo_id", "positive_traits", "negative_traits"} if field_name not in allowed_fields: raise ValueError("Некорректное поле для обновления.") # Выполняем обновление, если поле корректно query = f"UPDATE students SET {field_name} = ? WHERE student_id = ?" self.cursor.execute(query, (field_value, student_id)) self.conn.commit() def add_absence(self, student_id: int, date: str, pair_number: int, status: str, reason: str) -> None: """Добавление пропуска для студента.""" self.cursor.execute( """ INSERT INTO attendance (student_id, date, pair_number, status, reason) VALUES (?, ?, ?, ?, ?) """, (student_id, date, pair_number, status, reason) ) self.conn.commit() def remove_absence(self, student_id: int, date: str) -> None: """Удаление записи об отсутствии студента на конкретный день.""" self.cursor.execute( "DELETE FROM attendance WHERE student_id = ? AND date = ? AND status = 'Отсутствует'", (student_id, date) ) self.conn.commit() def update_characteristic(self, student_id: int, characteristic: str) -> None: """Обновление характеристики студента.""" self.cursor.execute( "UPDATE students SET characteristic = ? WHERE student_id = ?", (characteristic, student_id) ) self.conn.commit() def get_all_students_sorted(self) -> List[Dict[str, str]]: """ Возвращает список студентов, отсортированный по имени и фамилии. :return: Список студентов в формате [{"student_id": "ID", "first_name": "Имя", "last_name": "Фамилия"}] """ query = """ SELECT student_id, first_name, last_name FROM students ORDER BY first_name ASC, last_name ASC """ self.cursor.execute(query) students = self.cursor.fetchall() # Преобразуем данные в список словарей return [{"student_id": row[0], "first_name": row[1], "last_name": row[2] if len(row) > 2 else ""} for row in students] def get_student_data(self, student_id: int) -> Dict: """ Получение данных студента из базы данных по его ID. Args: student_id (int): ID студента. Returns: Dict: Словарь с данными о студенте. """ query = """ SELECT first_name, last_name, phone_number, registration_address, actual_address, telegram_tag, chronic_diseases, mother_name, mother_phone, father_name, father_phone, positive_traits, negative_traits FROM students WHERE student_id = ? """ self.cursor.execute(query, (student_id,)) student = self.cursor.fetchone() if student: return { "first_name": student[0], "last_name": student[1], "phone_number": student[2], "registration_address": student[3], "actual_address": student[4], "telegram_tag": student[5], "chronic_diseases": student[6], "mother_name": student[7], "mother_phone": student[8], "father_name": student[9], "father_phone": student[10], "positive_traits": student[11], "negative_traits": student[12], } return None def close(self) -> None: """Закрытие соединения с базой данных.""" self.conn.close() Python import sqlite3 from typing import Optional, Dict, List, Tuple class BotDB: def __init__(self, db_file: str): """Инициализация подключения к базе данных.""" self.conn = sqlite3.connect(db_file, check_same_thread=False) self.cursor = self.conn.cursor() self.create_tables() def create_tables(self) -> None: """Создание всех необходимых таблиц в базе данных.""" with open('database/database.sql', 'r', encoding='utf-8') as f: sql_script = f.read() self.cursor.executescript(sql_script) self.conn.commit() def add_user(self, user_id: int, username: str, group_id: str) -> None: """Добавление пользователя в базу данных.""" self.cursor.execute( "INSERT OR IGNORE INTO users (user_id, username, group_id) VALUES (?, ?, ?)", (user_id, username, group_id) ) self.conn.commit() def get_user(self, user_id: int) -> Optional[Dict]: """Получение информации о пользователе по его идентификатору.""" query = self.cursor.execute( "SELECT * FROM users WHERE user_id = ?", (user_id,) ) result = query.fetchone() if result: return { "user_id": result[0], "username": result[1], "group_id": result[2] } return None def add_student(self, first_name: str, last_name: str, phone_number: str = "", registration_address: str = "", actual_address: str = "", telegram_tag: str = "", chronic_diseases: str = "") -> None: """Добавление студента в базу данных.""" self.cursor.execute( "INSERT INTO students (first_name, last_name, phone_number, registration_address, " "actual_address, telegram_tag, chronic_diseases) VALUES (?, ?, ?, ?, ?, ?, ?)", (first_name, last_name, phone_number, registration_address, actual_address, telegram_tag, chronic_diseases) ) self.conn.commit() def get_all_students(self) -> List[Tuple[int, str]]: """Получение всех студентов из базы данных.""" query = self.cursor.execute( "SELECT student_id, first_name || ' ' || last_name FROM students" ) return query.fetchall() def add_attendance_record(self, student_id: int, selected_day: str, pair_number: str, status: str, respect: str) -> None: """Добавление записи о пропуске студента в базу данных.""" self.cursor.execute( "INSERT INTO attendance (student_id, date, pair_number, status, reason) VALUES (?, ?, ?, ?, ?)", (student_id, selected_day, pair_number, status, respect) ) self.conn.commit() def get_student_by_id(self, student_id: int) -> Optional[Dict]: """ Получение данных о студенте, включая записи о прогулов, характеристику, фотографию, положительные и негативные черты, по его идентификатору. Args: student_id (int): ID студента. Returns: Optional[Dict]: Словарь с данными о студенте или None, если студент не найден. """ # Получение основной информации о студенте query = self.cursor.execute( """ SELECT student_id, first_name, last_name, phone_number, registration_address, actual_address, telegram_tag, chronic_diseases, characteristic, photo_id, positive_traits, negative_traits FROM students WHERE student_id = ? """, (student_id,) ) student = query.fetchone() if not student: return None # Формирование словаря с основной информацией о студенте student_data = { "student_id": student[0], "first_name": student[1], "last_name": student[2], "phone_number": student[3], "registration_address": student[4], "actual_address": student[5], "telegram_tag": student[6], "chronic_diseases": student[7], "characteristic": student[8] if len(student) > 8 else "Не указано", "photo_id": student[9] if len(student) > 9 else None, "positive_traits": student[10] if len(student) > 10 else "Не указаны", "negative_traits": student[11] if len(student) > 11 else "Не указаны", } # Получение данных о прогулов студента attendance_query = self.cursor.execute( """ SELECT date, pair_number, reason FROM attendance WHERE student_id = ? ORDER BY date ASC """, (student_id,) ) absences = attendance_query.fetchall() # Формируем список прогулов student_data["absences"] = [ {"date": absence[0], "pair_number": absence[1], "reason": absence[2]} for absence in absences ] return student_data def get_student_attendance(self, student_id: int) -> List[Dict]: """Получение данных о пропусках студента по его идентификатору.""" query = self.cursor.execute( "SELECT * FROM attendance WHERE student_id = ?", (student_id,) ) result = query.fetchall() return [ { "id": record[0], "student_id": record[1], "date": record[2], "pair_number": record[3], "status": record[4], "reason": record[5], "hours": 1 if record[4] == "Опоздал" else 2 # Опоздание - 1 час, отсутствие - 2 часа } for record in result ] def update_student_field(self, student_id: int, field_name: str, field_value: str) -> None: """Обновление указанного поля студента.""" allowed_fields = {"phone_number", "registration_address", "actual_address", "telegram_tag", "chronic_diseases", "characteristic", "photo_id", "positive_traits", "negative_traits"} if field_name not in allowed_fields: raise ValueError("Некорректное поле для обновления.") # Выполняем обновление, если поле корректно query = f"UPDATE students SET {field_name} = ? WHERE student_id = ?" self.cursor.execute(query, (field_value, student_id)) self.conn.commit() def add_absence(self, student_id: int, date: str, pair_number: int, status: str, reason: str) -> None: """Добавление пропуска для студента.""" self.cursor.execute( """ INSERT INTO attendance (student_id, date, pair_number, status, reason) VALUES (?, ?, ?, ?, ?) """, (student_id, date, pair_number, status, reason) ) self.conn.commit() def remove_absence(self, student_id: int, date: str) -> None: """Удаление записи об отсутствии студента на конкретный день.""" self.cursor.execute( "DELETE FROM attendance WHERE student_id = ? AND date = ? AND status = 'Отсутствует'", (student_id, date) ) self.conn.commit() def update_characteristic(self, student_id: int, characteristic: str) -> None: """Обновление характеристики студента.""" self.cursor.execute( "UPDATE students SET characteristic = ? WHERE student_id = ?", (characteristic, student_id) ) self.conn.commit() def get_all_students_sorted(self) -> List[Dict[str, str]]: """ Возвращает список студентов, отсортированный по имени и фамилии. :return: Список студентов в формате [{"student_id": "ID", "first_name": "Имя", "last_name": "Фамилия"}] """ query = """ SELECT student_id, first_name, last_name FROM students ORDER BY first_name ASC, last_name ASC """ self.cursor.execute(query) students = self.cursor.fetchall() # Преобразуем данные в список словарей return [{"student_id": row[0], "first_name": row[1], "last_name": row[2] if len(row) > 2 else ""} for row in students] def get_student_data(self, student_id: int) -> Dict: """ Получение данных студента из базы данных по его ID. Args: student_id (int): ID студента. Returns: Dict: Словарь с данными о студенте. """ query = """ SELECT first_name, last_name, phone_number, registration_address, actual_address, telegram_tag, chronic_diseases, mother_name, mother_phone, father_name, father_phone, positive_traits, negative_traits FROM students WHERE student_id = ? """ self.cursor.execute(query, (student_id,)) student = self.cursor.fetchone() if student: return { "first_name": student[0], "last_name": student[1], "phone_number": student[2], "registration_address": student[3], "actual_address": student[4], "telegram_tag": student[5], "chronic_diseases": student[6], "mother_name": student[7], "mother_phone": student[8], "father_name": student[9], "father_phone": student[10], "positive_traits": student[11], "negative_traits": student[12], } return None def close(self) -> None: """Закрытие соединения с базой данных.""" self.conn.close() Это db.py При попытке Изменить положительные и отрицательные черты говорит "Неверный запрос для редактирования." хотя вроде всё правильно