Spaces:
Runtime error
Runtime error
File size: 3,064 Bytes
8d21059 | 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 | # database/subscribers.py
import logging
from database.connection import get_conn
logger = logging.getLogger(__name__)
def get_subscriber(user_id: int) -> dict | None:
conn = get_conn()
try:
c = conn.cursor()
c.execute("SELECT * FROM subscribers WHERE user_id=?", (user_id,))
row = c.fetchone()
return dict(row) if row else None
except Exception as e:
logger.error(f"get_subscriber({user_id}): {e}", exc_info=True)
return None
finally:
conn.close()
def upsert_subscriber(
user_id: int,
wants_free_games: bool,
wants_dlcs: bool,
wants_translations: bool,
) -> bool:
conn = get_conn()
try:
conn.execute("""
INSERT INTO subscribers
(user_id, wants_free_games, wants_dlcs, wants_translations, updated_at)
VALUES (?, ?, ?, ?, datetime('now','localtime'))
ON CONFLICT(user_id) DO UPDATE SET
wants_free_games = excluded.wants_free_games,
wants_dlcs = excluded.wants_dlcs,
wants_translations = excluded.wants_translations,
updated_at = datetime('now','localtime')
""", (user_id, int(wants_free_games), int(wants_dlcs), int(wants_translations)))
conn.commit()
return True
except Exception as e:
logger.error(f"upsert_subscriber({user_id}): {e}", exc_info=True)
return False
finally:
conn.close()
def remove_subscriber(user_id: int) -> bool:
conn = get_conn()
try:
c = conn.cursor()
c.execute("DELETE FROM subscribers WHERE user_id=?", (user_id,))
conn.commit()
return c.rowcount > 0
except Exception as e:
logger.error(f"remove_subscriber({user_id}): {e}", exc_info=True)
return False
finally:
conn.close()
def get_free_game_subscribers() -> list[dict]:
"""كل المشتركين الذين يريدون إشعارات الألعاب المجانية."""
conn = get_conn()
try:
c = conn.cursor()
c.execute("""
SELECT * FROM subscribers
WHERE wants_free_games=1 OR wants_dlcs=1
""")
return [dict(r) for r in c.fetchall()]
except Exception:
return []
finally:
conn.close()
def get_translation_subscribers() -> list[dict]:
"""كل المشتركين الذين يريدون إشعارات التعريبات الجديدة."""
conn = get_conn()
try:
c = conn.cursor()
c.execute("SELECT * FROM subscribers WHERE wants_translations=1")
return [dict(r) for r in c.fetchall()]
except Exception:
return []
finally:
conn.close()
def get_total_subscribers() -> int:
conn = get_conn()
try:
return conn.execute("""
SELECT COUNT(*) FROM subscribers
WHERE wants_free_games=1 OR wants_dlcs=1 OR wants_translations=1
""").fetchone()[0]
except Exception:
return 0
finally:
conn.close()
|