Spaces:
Running
Running
File size: 4,724 Bytes
c1d5bd6 54844e5 c1d5bd6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | 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()
|