import sqlite3 import os from typing import Optional, Dict, Any class DatabaseManager: def __init__(self, db_path: str = "bot_database.db"): self.db_path = db_path self._init_db() def _get_connection(self): return sqlite3.connect(self.db_path) def _init_db(self): with self._get_connection() as conn: cursor = conn.cursor() # Table for user states and collected form data cursor.execute(""" CREATE TABLE IF NOT EXISTS user_states ( user_id INTEGER PRIMARY KEY, state TEXT NOT NULL DEFAULT 'FREE_CHAT', name TEXT, phone TEXT, question TEXT, consent_1 TEXT, consent_2 TEXT, consent_3 TEXT, last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Migration checks for existing databases for col in ["consent_1", "consent_2", "consent_3"]: try: cursor.execute(f"ALTER TABLE user_states ADD COLUMN {col} TEXT") except sqlite3.OperationalError: # Column already exists pass # Table for logs / dialogue history if needed cursor.execute(""" CREATE TABLE IF NOT EXISTS dialogue_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, sender TEXT, -- 'user' or 'bot' message TEXT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Table for dynamic services cursor.execute(""" CREATE TABLE IF NOT EXISTS services ( id TEXT PRIMARY KEY, name TEXT NOT NULL, price TEXT NOT NULL, description TEXT NOT NULL, is_active INTEGER DEFAULT 1 ) """) # Seed default services if empty cursor.execute("SELECT COUNT(*) FROM services") if cursor.fetchone()[0] == 0: default_services = { "1": { "name": "Тренировка по Dota 2 🎮", "price": "1 500 ₽/час", "description": "Индивидуальное занятие с опытным тренером, разбор реплеев, позиционка и микроконтроль." }, "2": { "name": "Тренировка по CS2 🔫", "price": "1 500 ₽/час", "description": "Анализ ошибок, тренировка стрельбы (aim), раскидки гранат и командное взаимодействие." }, "3": { "name": "Пакет услуг 'Стандарт' 📦", "price": "7 500 ₽", "description": "5 индивидуальных тренировок по выбранной дисциплине + базовый план развития." }, "4": { "name": "Пакет услуг 'Pro' 🌟", "price": "14 000 ₽", "description": "10 индивидуальных тренировок + детальный разбор матчей + постоянная поддержка тренера." } } for key, svc in default_services.items(): cursor.execute( "INSERT INTO services (id, name, price, description) VALUES (?, ?, ?, ?)", (key, svc["name"], svc["price"], svc["description"]) ) # Table for web registrations cursor.execute(""" CREATE TABLE IF NOT EXISTS web_registrations ( id TEXT PRIMARY KEY, name TEXT, phone TEXT, email TEXT, vk_link TEXT, agree_oferta INTEGER, agree_policy INTEGER, agree_newsletter INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() def get_user_data(self, user_id: int) -> Dict[str, Any]: """Retrieve user state and collected data.""" with self._get_connection() as conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute( "SELECT state, name, phone, question, consent_1, consent_2, consent_3 FROM user_states WHERE user_id = ?", (user_id,) ) row = cursor.fetchone() if row: return dict(row) else: # Create default entry if not exists cursor.execute( "INSERT INTO user_states (user_id, state) VALUES (?, 'FREE_CHAT')", (user_id,) ) conn.commit() return { "state": "FREE_CHAT", "name": None, "phone": None, "question": None, "consent_1": None, "consent_2": None, "consent_3": None } def update_user_state(self, user_id: int, state: str): """Update only the user state.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "INSERT INTO user_states (user_id, state) VALUES (?, ?) " "ON CONFLICT(user_id) DO UPDATE SET state = excluded.state, last_updated = CURRENT_TIMESTAMP", (user_id, state) ) conn.commit() def update_user_field(self, user_id: int, field: str, value: Optional[str]): """Update a specific field in the form.""" allowed_fields = ["name", "phone", "question", "state", "consent_1", "consent_2", "consent_3"] if field not in allowed_fields: raise ValueError(f"Invalid field name: {field}") with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( f"INSERT INTO user_states (user_id, {field}) VALUES (?, ?) " f"ON CONFLICT(user_id) DO UPDATE SET {field} = excluded.{field}, last_updated = CURRENT_TIMESTAMP", (user_id, value) ) conn.commit() def reset_form(self, user_id: int): """Reset form fields and return to free chat state.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "UPDATE user_states SET state = 'FREE_CHAT', " "name = NULL, phone = NULL, question = NULL, " "consent_1 = NULL, consent_2 = NULL, consent_3 = NULL " "WHERE user_id = ?", (user_id,) ) conn.commit() def log_message(self, user_id: int, sender: str, message: str): """Log messages for dialogue context.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "INSERT INTO dialogue_logs (user_id, sender, message) VALUES (?, ?, ?)", (user_id, sender, message) ) conn.commit() def get_dialogue_history(self, user_id: int, limit: int = 15) -> list: """Get recent dialogue history to provide context to the operator.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT sender, message, timestamp FROM dialogue_logs WHERE user_id = ? ORDER BY id DESC LIMIT ?", (user_id, limit) ) rows = cursor.fetchall() return rows[::-1] # Return in chronological order def get_services(self) -> Dict[str, Dict[str, str]]: """Загрузить все активные услуги из базы данных.""" with self._get_connection() as conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute("SELECT id, name, price, description FROM services WHERE is_active = 1") rows = cursor.fetchall() services = {} for row in rows: services[row["id"]] = { "name": row["name"], "price": row["price"], "description": row["description"] } return services def update_service(self, service_id: str, field: str, value: str): """Обновить конкретное поле существующей услуги.""" allowed_fields = ["name", "price", "description"] if field not in allowed_fields: raise ValueError(f"Invalid field for service: {field}") with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( f"UPDATE services SET {field} = ? WHERE id = ?", (value, service_id) ) conn.commit() def add_service(self, name: str, price: str, description: str) -> str: """Добавить новую услугу и вернуть её новый ID.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT id FROM services") ids = [] for r in cursor.fetchall(): try: ids.append(int(r[0])) except ValueError: pass next_id = str(max(ids) + 1) if ids else "1" cursor.execute( "INSERT INTO services (id, name, price, description) VALUES (?, ?, ?, ?)", (next_id, name, price, description) ) conn.commit() return next_id def delete_service(self, service_id: str): """Мягкое удаление (деактивация) услуги.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "UPDATE services SET is_active = 0 WHERE id = ?", (service_id,) ) conn.commit() def add_web_registration(self, reg_id: str, data: dict): """Сохранить заявку с веб-формы в базу данных SQLite.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO web_registrations ( id, name, phone, email, vk_link, agree_oferta, agree_policy, agree_newsletter ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( reg_id, data.get("name", "").strip(), data.get("phone", "").strip(), data.get("email", "").strip(), data.get("vk", "").strip(), 1 if data.get("agreeOferta") else 0, 1 if data.get("agreePolicy") else 0, 1 if data.get("agreeNewsletter") else 0 )) conn.commit()