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 (needed for VK bot sync) 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 (VK bot) for col in ["consent_1", "consent_2", "consent_3"]: try: cursor.execute(f"ALTER TABLE user_states ADD COLUMN {col} TEXT") except sqlite3.OperationalError: 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 ) """) # Migration check: add price_numeric to services table try: cursor.execute("ALTER TABLE services ADD COLUMN price_numeric REAL DEFAULT 0.0") except sqlite3.OperationalError: pass # Seed default bot services if empty (legacy) 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, price_numeric, description) VALUES (?, ?, ?, ?, ?)", (key, svc["name"], svc["price"], float(svc["price"].replace(" ", "").replace("₽/час", "").replace("₽", "")), svc["description"]) ) # Seed default web services if they don't exist web_services = { "web_1": {"name": "Разовая тренировка", "price": "870 ₽", "price_numeric": 870.0, "description": "counter-strike, dota, tanks"}, "web_2": {"name": "Абонемент на 4 занятия (Групповой)", "price": "2 790 ₽", "price_numeric": 2790.0, "description": "counter-strike, dota, tanks"}, "web_3": {"name": "Абонемент на 8 занятий (Групповой)", "price": "4 990 ₽", "price_numeric": 4990.0, "description": "counter-strike, dota, tanks"}, "web_4": {"name": "Абонемент на 4 занятия (Индивидуальный)", "price": "3 290 ₽", "price_numeric": 3290.0, "description": "counter-strike, dota, tanks"}, "web_5": {"name": "Абонемент на 8 занятий (Индивидуальный)", "price": "5 990 ₽", "price_numeric": 5990.0, "description": "counter-strike, dota, tanks"} } for sid, svc in web_services.items(): cursor.execute("SELECT COUNT(*) FROM services WHERE id = ?", (sid,)) if cursor.fetchone()[0] == 0: cursor.execute( "INSERT INTO services (id, name, price, price_numeric, description) VALUES (?, ?, ?, ?, ?)", (sid, svc["name"], svc["price"], svc["price_numeric"], svc["description"]) ) # Table for web registrations with payments support 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, service_name TEXT, amount REAL, payment_status TEXT DEFAULT 'pending', yookassa_payment_id TEXT, paid_at TIMESTAMP, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Run migration on web_registrations to add new columns if they don't exist for col, col_type in [ ("service_name", "TEXT"), ("amount", "REAL"), ("payment_status", "TEXT DEFAULT 'pending'"), ("yookassa_payment_id", "TEXT"), ("paid_at", "TIMESTAMP") ]: try: cursor.execute(f"ALTER TABLE web_registrations ADD COLUMN {col} {col_type}") except sqlite3.OperationalError: pass conn.commit() def get_services(self) -> Dict[str, Dict[str, Any]]: with self._get_connection() as conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute("SELECT id, name, price, price_numeric, description FROM services WHERE is_active = 1") rows = cursor.fetchall() services = {} for row in rows: services[row["id"]] = { "name": row["name"], "price": row["price"], "price_numeric": row["price_numeric"] if "price_numeric" in row.keys() else 0.0, "description": row["description"] } return services 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, service_name, amount, payment_status, yookassa_payment_id ) 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, data.get("serviceName", "Разовая тренировка").strip(), float(data.get("servicePrice", 870)), data.get("paymentStatus", "pending"), data.get("yookassaPaymentId", None) )) conn.commit() def update_web_payment_status(self, reg_id: str, status: str, yookassa_payment_id: str = None, paid_at: str = None): """Обновить статус оплаты и сопутствующие данные для заявки.""" with self._get_connection() as conn: cursor = conn.cursor() if yookassa_payment_id and paid_at: cursor.execute(""" UPDATE web_registrations SET payment_status = ?, yookassa_payment_id = ?, paid_at = ? WHERE id = ? """, (status, yookassa_payment_id, paid_at, reg_id)) elif yookassa_payment_id: cursor.execute(""" UPDATE web_registrations SET payment_status = ?, yookassa_payment_id = ? WHERE id = ? """, (status, yookassa_payment_id, reg_id)) else: cursor.execute(""" UPDATE web_registrations SET payment_status = ? WHERE id = ? """, (status, reg_id)) conn.commit() def get_web_registration(self, reg_id: str) -> Optional[Dict[str, Any]]: """Получить детали одной веб-регистрации.""" with self._get_connection() as conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute("SELECT * FROM web_registrations WHERE id = ?", (reg_id,)) row = cursor.fetchone() return dict(row) if row else None def get_all_web_registrations(self) -> list: """Получить все веб-регистрации для админ-панели.""" with self._get_connection() as conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute("SELECT * FROM web_registrations ORDER BY created_at DESC") rows = cursor.fetchall() return [dict(row) for row in rows] def get_all_services_list(self) -> list: """Получить полный список всех услуг (включая неактивные) для админ-панели.""" with self._get_connection() as conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute("SELECT id, name, price, price_numeric, description, is_active FROM services") rows = cursor.fetchall() return [dict(row) for row in rows] def add_service(self, service_id: str, name: str, price: str, price_numeric: float, description: str) -> bool: """Добавить новую услугу в базу данных.""" with self._get_connection() as conn: cursor = conn.cursor() try: cursor.execute(""" INSERT INTO services (id, name, price, price_numeric, description, is_active) VALUES (?, ?, ?, ?, ?, 1) """, (service_id.strip(), name.strip(), price.strip(), float(price_numeric), description.strip())) conn.commit() return True except sqlite3.IntegrityError: return False def update_service(self, service_id: str, name: str, price: str, price_numeric: float, description: str, is_active: int) -> bool: """Обновить поля существующей услуги по её ID.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute(""" UPDATE services SET name = ?, price = ?, price_numeric = ?, description = ?, is_active = ? WHERE id = ? """, (name.strip(), price.strip(), float(price_numeric), description.strip(), int(is_active), service_id.strip())) conn.commit() return cursor.rowcount > 0 def delete_service(self, service_id: str) -> bool: """Удалить услугу из базы данных.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM services WHERE id = ?", (service_id.strip(),)) conn.commit() return cursor.rowcount > 0