Spaces:
Running
Running
| import sqlite3 | |
| import os | |
| from datetime import datetime | |
| class Database: | |
| def __init__(self, db_name="bot_database.db"): | |
| self.db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), db_name) | |
| self.init_db() | |
| def get_connection(self): | |
| return sqlite3.connect(self.db_path) | |
| def init_db(self): | |
| """Ma'lumotlar bazasi jadvallarini yaratish.""" | |
| with self.get_connection() as conn: | |
| cursor = conn.cursor() | |
| # Foydalanuvchilar jadvali | |
| cursor.execute(''' | |
| CREATE TABLE IF NOT EXISTS users ( | |
| user_id INTEGER PRIMARY KEY, | |
| username TEXT, | |
| first_name TEXT, | |
| joined_at DATETIME, | |
| default_filter TEXT DEFAULT 'retro' | |
| ) | |
| ''') | |
| # Amallar tarixi jadvali | |
| cursor.execute(''' | |
| CREATE TABLE IF NOT EXISTS history ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, | |
| user_id INTEGER, | |
| media_type TEXT, | |
| filter_type TEXT, | |
| processed_at DATETIME, | |
| file_id TEXT, | |
| FOREIGN KEY (user_id) REFERENCES users (user_id) | |
| ) | |
| ''') | |
| conn.commit() | |
| def add_user(self, user_id, username, first_name): | |
| """Yangi foydalanuvchini ro'yxatga olish yoki yangilash.""" | |
| with self.get_connection() as conn: | |
| cursor = conn.cursor() | |
| cursor.execute(''' | |
| INSERT OR REPLACE INTO users (user_id, username, first_name, joined_at) | |
| VALUES (?, ?, ?, COALESCE((SELECT joined_at FROM users WHERE user_id = ?), ?)) | |
| ''', (user_id, username, first_name, user_id, datetime.now().isoformat())) | |
| conn.commit() | |
| def log_history(self, user_id, media_type, filter_type, file_id): | |
| """Ishlov berish tarixini saqlash.""" | |
| with self.get_connection() as conn: | |
| cursor = conn.cursor() | |
| cursor.execute(''' | |
| INSERT INTO history (user_id, media_type, filter_type, processed_at, file_id) | |
| VALUES (?, ?, ?, ?, ?) | |
| ''', (user_id, media_type, filter_type, datetime.now().isoformat(), file_id)) | |
| conn.commit() | |
| def get_user_history(self, user_id, limit=5): | |
| """Foydalanuvchining oxirgi amallarini olish.""" | |
| with self.get_connection() as conn: | |
| cursor = conn.cursor() | |
| cursor.execute(''' | |
| SELECT media_type, filter_type, processed_at | |
| FROM history | |
| WHERE user_id = ? | |
| ORDER BY processed_at DESC | |
| LIMIT ? | |
| ''', (user_id, limit)) | |
| return cursor.fetchall() | |
| def set_default_filter(self, user_id, filter_type): | |
| """Foydalanuvchi uchun standart filtrni o'rnatish.""" | |
| with self.get_connection() as conn: | |
| cursor = conn.cursor() | |
| cursor.execute('UPDATE users SET default_filter = ? WHERE user_id = ?', (filter_type, user_id)) | |
| conn.commit() | |
| def get_user_settings(self, user_id): | |
| """Foydalanuvchi sozlamalarini olish.""" | |
| with self.get_connection() as conn: | |
| cursor = conn.cursor() | |
| cursor.execute('SELECT default_filter FROM users WHERE user_id = ?', (user_id,)) | |
| result = cursor.fetchone() | |
| return result[0] if result else 'retro' | |
| def get_stats(self): | |
| """Bot statistikasini qaytaradi (jami userlar va amallar).""" | |
| with self.get_connection() as conn: | |
| cursor = conn.cursor() | |
| cursor.execute('SELECT COUNT(*) FROM users') | |
| total_users = cursor.fetchone()[0] | |
| cursor.execute('SELECT COUNT(*) FROM history') | |
| total_processed = cursor.fetchone()[0] | |
| # Oxirgi 24 soatdagi faollik | |
| cursor.execute(''' | |
| SELECT COUNT(*) FROM history | |
| WHERE processed_at > datetime("now", "-1 day") | |
| ''') | |
| daily_active = cursor.fetchone()[0] | |
| return { | |
| "total_users": total_users, | |
| "total_processed": total_processed, | |
| "daily_active": daily_active | |
| } | |
| def get_all_users(self): | |
| """Xabar tarqatish uchun barcha user ID-larini qaytaradi.""" | |
| with self.get_connection() as conn: | |
| cursor = conn.cursor() | |
| cursor.execute('SELECT user_id FROM users') | |
| return [row[0] for row in cursor.fetchall()] | |
| # Global DB instance | |
| db = Database() | |