dorm_alarm/sql.py

import sqlite3
import asyncio
from typing import List, Tuple, Optional
from datetime import datetime

class DatabaseManager:    
    def __init__(self, db_path: str = "notifications.db"):
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """Создание таблиц при первом запуске"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            # Таблица пользователей с настройками уведомлений
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS users (
                    user_id INTEGER PRIMARY KEY,
                    username TEXT,
                    interval_minutes INTEGER DEFAULT 60,
                    is_active BOOLEAN DEFAULT 1,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            ''')

            cursor.execute('''
                CREATE TABLE IF NOT EXISTS notifications_log (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER,
                    sent_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    status TEXT DEFAULT 'sent',
                    FOREIGN KEY (user_id) REFERENCES users (user_id)
                )
            ''')
            conn.commit()
    
    def add_user(self, user_id: int, username: str) -> bool:
        """Добавление нового пользователя в систему"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute(
                    "INSERT OR REPLACE INTO users (user_id, username) VALUES (?, ?)",
                    (user_id, username)
                )
                conn.commit()
                return True
        except sqlite3.Error:
            return False
    
    def update_interval(self, user_id: int, interval_minutes: int) -> bool:
        """Обновление интервала уведомлений для пользователя"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute(
                    "UPDATE users SET interval_minutes = ? WHERE user_id = ?",
                    (interval_minutes, user_id)
                )
                conn.commit()
                return cursor.rowcount > 0
        except sqlite3.Error:
            return False
    
    def toggle_notifications(self, user_id: int) -> Optional[bool]:
        """Переключение статуса активности уведомлений"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute(
                    "UPDATE users SET is_active = NOT is_active WHERE user_id = ?",
                    (user_id,)
                )
                cursor.execute(
                    "SELECT is_active FROM users WHERE user_id = ?",
                    (user_id,)
                )
                result = cursor.fetchone()
                conn.commit()
                return bool(result[0]) if result else None
        except sqlite3.Error:
            return None
    
    def get_active_users(self) -> List[Tuple[int, int]]:
        """Получение списка активных пользователей с их интервалами"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute(
                    "SELECT user_id, interval_minutes FROM users WHERE is_active = 1"
                )
                return cursor.fetchall()
        except sqlite3.Error:
            return []
    
    def get_user_info(self, user_id: int) -> Optional[Tuple]:
        """Получение информации о пользователе"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute(
                    "SELECT user_id, username, interval_minutes, is_active FROM users WHERE user_id = ?",
                    (user_id,)
                )
                return cursor.fetchone()
        except sqlite3.Error:
            return None
    
    def log_notification(self, user_id: int, status: str = "sent"):
        """Логирование отправленного уведомления"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute(
                    "INSERT INTO notifications_log (user_id, status) VALUES (?, ?)",
                    (user_id, status)
                )
                conn.commit()
        except sqlite3.Error:
            pass  # Ошибки логирования не должны прерывать основной процесс
    
    def get_user_count(self) -> int:
        """Получение общего количества пользователей"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute("SELECT COUNT(*) FROM users")
                result = cursor.fetchone()
                return result[0] if result else 0
        except sqlite3.Error:
            return 0