import sqlite3 import os import json 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", "call_time", "vac_data"]: 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": "Counter-Strike 2 🔫", "price": "от 870 ₽ / занятие", "description": "Тренировки по CS2: aim, раскидки гранат, командная тактика. Разовое занятие 870 ₽, абонементы от 2 790 ₽." }, "2": { "name": "Dota 2 🎮", "price": "от 870 ₽ / занятие", "description": "Тренировки по Dota 2: разбор реплеев, позиционка, фарм и тимплей. Разовое занятие 870 ₽, абонементы от 2 790 ₽." }, "3": { "name": "World of Tanks 🛡️", "price": "от 870 ₽ / занятие", "description": "Тренировки по World of Tanks: позиционирование, сведение, тактика боёв. Разовое занятие 870 ₽, абонементы от 2 790 ₽." } } 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 completed leads (заявки) — единый источник для статистики cursor.execute(""" CREATE TABLE IF NOT EXISTS leads ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, source TEXT, -- 'bot' | 'web' name TEXT, phone TEXT, service TEXT, call_time TEXT, promo TEXT, marketing INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Table for HR/recruitment applications (отклики на вакансии и набор в состав) cursor.execute(""" CREATE TABLE IF NOT EXISTS applications ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, track TEXT, -- 'player' | 'worker' discipline TEXT, position TEXT, rank TEXT, role TEXT, experience TEXT, contact TEXT, status TEXT DEFAULT 'новый', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Миграция: статус для существующих таблиц applications try: cursor.execute("ALTER TABLE applications ADD COLUMN status TEXT DEFAULT 'новый'") except sqlite3.OperationalError: pass # Table for runtime-added admins (добавляются супер-админом из бота) cursor.execute(""" CREATE TABLE IF NOT EXISTS admins ( user_id INTEGER PRIMARY KEY, added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Table for editable quick FAQ answers (быстрые ответы по ключевым словам) cursor.execute(""" CREATE TABLE IF NOT EXISTS custom_faq ( id INTEGER PRIMARY KEY AUTOINCREMENT, keyword TEXT NOT NULL, answer TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # 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, call_time 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, "call_time": 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", "call_time"] 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 progress and return to free chat, but KEEP name/phone for returning users.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "UPDATE user_states SET state = 'FREE_CHAT', " "question = NULL, call_time = NULL, vac_data = NULL, " "consent_1 = NULL, consent_2 = NULL, consent_3 = NULL " "WHERE user_id = ?", (user_id,) ) conn.commit() def reset_form_full(self, user_id: int): """Полностью очистить данные пользователя (включая имя/телефон) — для 'Заполнить заново'.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "UPDATE user_states SET state = 'FREE_CHAT', " "name = NULL, phone = NULL, question = NULL, call_time = 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 count_web_registrations(self) -> int: """Количество заявок с сайта (для генерации следующего ID независимо от Excel).""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT COUNT(*) FROM web_registrations") return cursor.fetchone()[0] def add_lead(self, user_id: int, source: str, name: str, phone: str, service: Optional[str] = None, call_time: Optional[str] = None, promo: Optional[str] = None, marketing: bool = False): """Записать завершённую заявку в единый журнал leads (для статистики и истории).""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "INSERT INTO leads (user_id, source, name, phone, service, call_time, promo, marketing) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (user_id, source, name, phone, service, call_time, promo, 1 if marketing else 0) ) conn.commit() def get_marketing_subscribers(self) -> list: """Список user_id, давших согласие на маркетинговую рассылку (consent_2 = 'Согласен').""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT user_id FROM user_states WHERE consent_2 = 'Согласен'") return [row[0] for row in cursor.fetchall()] def get_lead_stats(self) -> Dict[str, Any]: """Сводная статистика по заявкам из журнала leads.""" with self._get_connection() as conn: cursor = conn.cursor() def scalar(query, params=()): cursor.execute(query, params) return cursor.fetchone()[0] stats = { "total": scalar("SELECT COUNT(*) FROM leads"), "today": scalar("SELECT COUNT(*) FROM leads WHERE date(created_at) = date('now', 'localtime')"), "last7": scalar("SELECT COUNT(*) FROM leads WHERE created_at >= datetime('now', '-7 days')"), "from_bot": scalar("SELECT COUNT(*) FROM leads WHERE source = 'bot'"), "from_web": scalar("SELECT COUNT(*) FROM leads WHERE source = 'web'"), "marketing_yes": scalar("SELECT COUNT(*) FROM leads WHERE marketing = 1"), } cursor.execute( "SELECT service, COUNT(*) FROM leads WHERE service IS NOT NULL AND service != '' " "GROUP BY service ORDER BY COUNT(*) DESC" ) stats["by_service"] = cursor.fetchall() return stats def get_dialogue_stats(self) -> Dict[str, Any]: """Статистика диалогов: уникальные пользователи и объём сообщений.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT COUNT(DISTINCT user_id) FROM dialogue_logs") unique_users = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM dialogue_logs WHERE sender = 'user'") user_messages = cursor.fetchone()[0] return {"unique_users": unique_users, "user_messages": user_messages} def get_vac_data(self, user_id: int) -> Dict[str, Any]: """Черновик отклика на вакансию (хранится как JSON в user_states.vac_data).""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT vac_data FROM user_states WHERE user_id = ?", (user_id,)) row = cursor.fetchone() if row and row[0]: try: return json.loads(row[0]) except (ValueError, TypeError): return {} return {} def set_vac_field(self, user_id: int, key: str, value: Optional[str]): """Обновить одно поле в черновике отклика.""" data = self.get_vac_data(user_id) data[key] = value with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "INSERT INTO user_states (user_id, vac_data) VALUES (?, ?) " "ON CONFLICT(user_id) DO UPDATE SET vac_data = excluded.vac_data, last_updated = CURRENT_TIMESTAMP", (user_id, json.dumps(data, ensure_ascii=False)) ) conn.commit() def clear_vac_data(self, user_id: int): """Очистить черновик отклика.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "INSERT INTO user_states (user_id, vac_data) VALUES (?, NULL) " "ON CONFLICT(user_id) DO UPDATE SET vac_data = NULL, last_updated = CURRENT_TIMESTAMP", (user_id,) ) conn.commit() def add_application(self, user_id: int, track: str, data: dict): """Сохранить завершённый отклик (игрок/сотрудник) в журнал applications.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute( "INSERT INTO applications (user_id, track, discipline, position, rank, role, experience, contact) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", ( user_id, track, data.get("discipline"), data.get("position"), data.get("rank"), data.get("role"), data.get("experience"), data.get("contact") ) ) conn.commit() # --- Администраторы (добавленные в рантайме супер-админом) --- def add_admin(self, user_id: int) -> bool: with self._get_connection() as conn: cursor = conn.cursor() try: cursor.execute("INSERT INTO admins (user_id) VALUES (?)", (user_id,)) conn.commit() return True except sqlite3.IntegrityError: return False # уже админ def remove_admin(self, user_id: int) -> bool: with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM admins WHERE user_id = ?", (user_id,)) conn.commit() return cursor.rowcount > 0 def get_admins(self) -> list: with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT user_id FROM admins ORDER BY user_id") return [r[0] for r in cursor.fetchall()] # --- Быстрые ответы (custom FAQ) --- def add_custom_faq(self, keyword: str, answer: str) -> int: with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("INSERT INTO custom_faq (keyword, answer) VALUES (?, ?)", (keyword.strip().lower(), answer.strip())) conn.commit() return cursor.lastrowid def get_custom_faq(self) -> list: """Список (id, keyword, answer) всех быстрых ответов.""" with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT id, keyword, answer FROM custom_faq ORDER BY id") return cursor.fetchall() def delete_custom_faq(self, faq_id: int) -> bool: with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM custom_faq WHERE id = ?", (faq_id,)) conn.commit() return cursor.rowcount > 0 def match_custom_faq(self, text_lower: str) -> Optional[str]: """Возвращает ответ, если в тексте встречается одно из ключевых слов.""" for _id, keyword, answer in self.get_custom_faq(): if keyword and keyword in text_lower: return answer return None # --- Отклики: статусы и статистика --- def set_application_status(self, app_id: int, status: str) -> bool: with self._get_connection() as conn: cursor = conn.cursor() cursor.execute("UPDATE applications SET status = ? WHERE id = ?", (status, app_id)) conn.commit() return cursor.rowcount > 0 def get_recent_applications(self, limit: int = 10) -> list: with self._get_connection() as conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute( "SELECT id, track, discipline, position, contact, status, created_at " "FROM applications ORDER BY id DESC LIMIT ?", (limit,) ) return [dict(r) for r in cursor.fetchall()] def get_application_stats(self) -> Dict[str, Any]: """Статистика откликов: всего, игроки/сотрудники, по дисциплинам и должностям.""" with self._get_connection() as conn: cursor = conn.cursor() stats = { "total": cursor.execute("SELECT COUNT(*) FROM applications").fetchone()[0], "players": cursor.execute("SELECT COUNT(*) FROM applications WHERE track='player'").fetchone()[0], "workers": cursor.execute("SELECT COUNT(*) FROM applications WHERE track='worker'").fetchone()[0], } cursor.execute("SELECT discipline, COUNT(*) FROM applications WHERE track='player' " "AND discipline IS NOT NULL GROUP BY discipline ORDER BY COUNT(*) DESC") stats["by_discipline"] = cursor.fetchall() cursor.execute("SELECT position, COUNT(*) FROM applications WHERE track='worker' " "AND position IS NOT NULL GROUP BY position ORDER BY COUNT(*) DESC") stats["by_position"] = cursor.fetchall() return stats 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()