""" Статистика использования ботов (SQLite). """ import os import sqlite3 from datetime import datetime, date from pathlib import Path DB_PATH = os.environ.get('STATS_DB', '/app/data/tournaments.db') def _get_conn(): p = Path(DB_PATH) try: p.parent.mkdir(parents=True, exist_ok=True) except PermissionError: # Fallback: use tempdir or cwd (local testing) fallback = Path(__file__).parent.parent / 'data' / 'tournaments.db' fallback.parent.mkdir(parents=True, exist_ok=True) return _get_conn_at(fallback) return _get_conn_at(p) def _get_conn_at(db_path: Path): conn = sqlite3.connect(str(db_path)) conn.execute('PRAGMA journal_mode=WAL') conn.execute(''' CREATE TABLE IF NOT EXISTS users ( user_id INTEGER PRIMARY KEY, username TEXT, first_name TEXT, last_name TEXT, first_seen TEXT DEFAULT (datetime('now')), last_seen TEXT DEFAULT (datetime('now')) ) ''') conn.execute(''' CREATE TABLE IF NOT EXISTS requests ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, url TEXT, tournament_name TEXT, timestamp TEXT DEFAULT (datetime('now')), success INTEGER DEFAULT 1, error TEXT ) ''') conn.commit() return conn def record_user(user_id: int, username: str = '', first_name: str = '', last_name: str = ''): conn = _get_conn() now = datetime.utcnow().isoformat() existing = conn.execute('SELECT user_id FROM users WHERE user_id = ?', (user_id,)).fetchone() if existing: conn.execute('UPDATE users SET username=?, first_name=?, ' 'last_name=?, last_seen=? WHERE user_id=?', (username, first_name, last_name, now, user_id)) else: conn.execute('INSERT INTO users (user_id, username, first_name, ' 'last_name, first_seen, last_seen) VALUES (?,?,?,?,?,?)', (user_id, username, first_name, last_name, now, now)) conn.commit() conn.close() def record_request(user_id: int, url: str = '', tournament_name: str = '', success: bool = True, error: str = ''): conn = _get_conn() conn.execute('INSERT INTO requests (user_id, url, tournament_name, ' 'timestamp, success, error) VALUES (?,?,?,datetime(\'now\'),?,?)', (user_id, url, tournament_name, 1 if success else 0, error)) conn.commit() conn.close() def get_stats() -> dict: conn = _get_conn() unique_users = conn.execute( 'SELECT COUNT(*) FROM users').fetchone()[0] total_requests = conn.execute( 'SELECT COUNT(*) FROM requests').fetchone()[0] today = date.today().isoformat() today_requests = conn.execute( "SELECT COUNT(*) FROM requests WHERE date(timestamp) = ?", (today,)).fetchone()[0] today_unique_users = conn.execute( "SELECT COUNT(DISTINCT user_id) FROM requests WHERE date(timestamp) = ?", (today,)).fetchone()[0] conn.close() return { 'unique_users': unique_users, 'total_requests': total_requests, 'today_requests': today_requests, 'today_unique_users': today_unique_users, }