565 lines
19 KiB
Python
565 lines
19 KiB
Python
"""
|
|
db/models.py
|
|
------------
|
|
All CRUD operations for LottoSight.
|
|
Functions:
|
|
Games — get_all_games(), get_game_by_name(), get_game_by_id(), set_game_active()
|
|
Draws — insert_draw(), draw_exists(), get_draws(), get_last_draw(), get_draw_count()
|
|
Predict — insert_prediction(), get_predictions()
|
|
Fetch — insert_fetch_log(), get_last_fetch_log(), get_fetch_logs()
|
|
"""
|
|
|
|
import logging
|
|
from db.database import get_connection
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# GAMES
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
_BUILTIN_GAMES = {"Powerball", "Mega Millions"}
|
|
|
|
|
|
def add_game(name: str, main_count: int, main_max: int,
|
|
bonus_count: int = 0, bonus_max: int = 0) -> int:
|
|
"""
|
|
Insert a new game row. Returns the new game id.
|
|
Raises ValueError if name is blank or already exists.
|
|
"""
|
|
name = name.strip()
|
|
if not name:
|
|
raise ValueError("Game name cannot be blank.")
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT id FROM games WHERE name = ?", (name,)
|
|
)
|
|
if cursor.fetchone():
|
|
raise ValueError(f"A game named '{name}' already exists.")
|
|
cursor.execute(
|
|
"""INSERT INTO games (name, main_count, main_max, bonus_count, bonus_max, active)
|
|
VALUES (?, ?, ?, ?, ?, 1)""",
|
|
(name, main_count, main_max, bonus_count, bonus_max),
|
|
)
|
|
conn.commit()
|
|
new_id = cursor.lastrowid
|
|
logger.info("[DB] Added custom game id=%d name='%s'", new_id, name)
|
|
return new_id
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def delete_game(game_id: int) -> bool:
|
|
"""
|
|
Delete a game by id. Returns True on success, False if the game has draws
|
|
(to protect data integrity) or is a built-in game.
|
|
"""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT name FROM games WHERE id = ?", (game_id,))
|
|
row = cursor.fetchone()
|
|
if row is None:
|
|
return False
|
|
if row["name"] in _BUILTIN_GAMES:
|
|
logger.warning("[DB] Refused to delete built-in game '%s'", row["name"])
|
|
return False
|
|
cursor.execute("SELECT COUNT(*) as cnt FROM draws WHERE game_id = ?", (game_id,))
|
|
if cursor.fetchone()["cnt"] > 0:
|
|
logger.warning("[DB] Refused to delete game id=%d — has draws", game_id)
|
|
return False
|
|
cursor.execute("DELETE FROM predictions WHERE game_id = ?", (game_id,))
|
|
cursor.execute("DELETE FROM games WHERE id = ?", (game_id,))
|
|
conn.commit()
|
|
logger.info("[DB] Deleted custom game id=%d", game_id)
|
|
return True
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_all_games(active_only=False):
|
|
"""
|
|
Return list of all games as sqlite3.Row objects.
|
|
Pass active_only=True to filter to enabled games only.
|
|
"""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
if active_only:
|
|
cursor.execute("SELECT * FROM games WHERE active = 1 ORDER BY name")
|
|
else:
|
|
cursor.execute("SELECT * FROM games ORDER BY name")
|
|
return cursor.fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_game_by_name(name):
|
|
"""Return single game row by name, or None if not found."""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM games WHERE name = ?", (name,))
|
|
return cursor.fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_game_by_id(game_id):
|
|
"""Return single game row by id, or None if not found."""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM games WHERE id = ?", (game_id,))
|
|
return cursor.fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def set_game_active(game_id, active: bool):
|
|
"""Enable or disable a game. active=True enables, False disables."""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"UPDATE games SET active = ? WHERE id = ?",
|
|
(1 if active else 0, game_id)
|
|
)
|
|
conn.commit()
|
|
logger.info("[DB] Game id=%d set active=%s", game_id, active)
|
|
except Exception as e:
|
|
conn.rollback()
|
|
logger.error("[ERROR] set_game_active failed: %s", e, exc_info=True)
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def add_custom_game(name, main_count, main_max, bonus_count=0, bonus_max=0):
|
|
"""Insert a custom game config. Returns new game id."""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO games (name, main_count, main_max, bonus_count, bonus_max, active)
|
|
VALUES (?, ?, ?, ?, ?, 1)
|
|
""", (name, main_count, main_max, bonus_count, bonus_max))
|
|
conn.commit()
|
|
new_id = cursor.lastrowid
|
|
logger.info("[DB] Custom game created: '%s' id=%d", name, new_id)
|
|
return new_id
|
|
except Exception as e:
|
|
conn.rollback()
|
|
logger.error("[ERROR] add_custom_game failed: %s", e, exc_info=True)
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# DRAWS
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def draw_exists(game_id, draw_date):
|
|
"""
|
|
Check if a draw already exists for this game + date.
|
|
Returns True if duplicate, False if new.
|
|
"""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT id FROM draws WHERE game_id = ? AND draw_date = ?",
|
|
(game_id, draw_date)
|
|
)
|
|
return cursor.fetchone() is not None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def insert_draw(game_id, draw_date, numbers, bonus=None, multiplier=None, source=None):
|
|
"""
|
|
Insert a single draw record.
|
|
- numbers: list of ints or comma-separated string
|
|
- bonus: int, str, or None
|
|
- Returns: 'inserted' or 'skipped' (duplicate)
|
|
"""
|
|
# Normalize numbers to comma-separated string
|
|
if isinstance(numbers, (list, tuple)):
|
|
numbers_str = ",".join(str(n) for n in numbers)
|
|
else:
|
|
numbers_str = str(numbers).strip()
|
|
|
|
bonus_str = str(bonus) if bonus is not None else None
|
|
multiplier_str = str(multiplier) if multiplier is not None else None
|
|
|
|
# Duplicate check
|
|
if draw_exists(game_id, draw_date):
|
|
logger.debug("[DB] Skipped duplicate: game_id=%d date=%s", game_id, draw_date)
|
|
return "skipped"
|
|
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO draws (game_id, draw_date, numbers, bonus, multiplier, source)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""", (game_id, draw_date, numbers_str, bonus_str, multiplier_str, source))
|
|
conn.commit()
|
|
logger.debug("[DB] Inserted draw: game_id=%d date=%s numbers=%s",
|
|
game_id, draw_date, numbers_str)
|
|
return "inserted"
|
|
except Exception as e:
|
|
conn.rollback()
|
|
logger.error("[ERROR] insert_draw failed: %s", e, exc_info=True)
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_draws(game_id, limit=None, date_from=None, date_to=None, order="DESC"):
|
|
"""
|
|
Fetch draw records for a game.
|
|
- date_from / date_to: ISO strings 'YYYY-MM-DD' (optional)
|
|
- order: 'DESC' (newest first) or 'ASC' (oldest first)
|
|
- limit: max rows to return (None = all)
|
|
Returns list of sqlite3.Row objects.
|
|
"""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
query = "SELECT * FROM draws WHERE game_id = ?"
|
|
params = [game_id]
|
|
|
|
if date_from:
|
|
query += " AND draw_date >= ?"
|
|
params.append(date_from)
|
|
if date_to:
|
|
query += " AND draw_date <= ?"
|
|
params.append(date_to)
|
|
|
|
order = "DESC" if order.upper() == "DESC" else "ASC"
|
|
query += f" ORDER BY draw_date {order}"
|
|
|
|
if limit:
|
|
query += " LIMIT ?"
|
|
params.append(limit)
|
|
|
|
cursor.execute(query, params)
|
|
return cursor.fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_last_draw(game_id):
|
|
"""Return the most recent draw record for a game, or None."""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT * FROM draws
|
|
WHERE game_id = ?
|
|
ORDER BY draw_date DESC
|
|
LIMIT 1
|
|
""", (game_id,))
|
|
return cursor.fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_draw_count(game_id):
|
|
"""Return total number of draw records for a game."""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT COUNT(*) as cnt FROM draws WHERE game_id = ?",
|
|
(game_id,)
|
|
)
|
|
row = cursor.fetchone()
|
|
return row["cnt"] if row else 0
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_draws_with_game(game_id=None, limit=None, date_from=None, date_to=None,
|
|
order="DESC", number=None):
|
|
"""
|
|
Like get_draws() but JOINs games so each row includes game_name.
|
|
game_id=None returns draws for all games.
|
|
number (int|str): if given, restrict to draws containing that ball number.
|
|
Used by the History screen.
|
|
"""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
query = """
|
|
SELECT d.*, g.name AS game_name
|
|
FROM draws d
|
|
JOIN games g ON g.id = d.game_id
|
|
WHERE 1=1
|
|
"""
|
|
params = []
|
|
if game_id is not None:
|
|
query += " AND d.game_id = ?"
|
|
params.append(game_id)
|
|
if date_from:
|
|
query += " AND d.draw_date >= ?"
|
|
params.append(date_from)
|
|
if date_to:
|
|
query += " AND d.draw_date <= ?"
|
|
params.append(date_to)
|
|
if number is not None:
|
|
# Wrap stored CSV in commas so boundary matches are exact
|
|
query += " AND ',' || d.numbers || ',' LIKE ?"
|
|
params.append(f"%,{int(number)},%")
|
|
|
|
order_sql = "DESC" if order.upper() == "DESC" else "ASC"
|
|
query += f" ORDER BY d.draw_date {order_sql}, g.name ASC"
|
|
|
|
if limit:
|
|
query += " LIMIT ?"
|
|
params.append(limit)
|
|
|
|
cursor.execute(query, params)
|
|
return cursor.fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_all_draws_numbers(game_id):
|
|
"""
|
|
Return all draws as list of dicts with parsed number lists.
|
|
Used by analyzer and predictor.
|
|
Format: [{"draw_date": str, "numbers": [int,...], "bonus": int|None}, ...]
|
|
"""
|
|
rows = get_draws(game_id, order="ASC")
|
|
result = []
|
|
for row in rows:
|
|
try:
|
|
numbers = [int(n.strip()) for n in row["numbers"].split(",")]
|
|
bonus = int(row["bonus"]) if row["bonus"] else None
|
|
result.append({
|
|
"draw_date": row["draw_date"],
|
|
"numbers": numbers,
|
|
"bonus": bonus,
|
|
"multiplier": row["multiplier"],
|
|
"source": row["source"],
|
|
})
|
|
except Exception as e:
|
|
logger.warning("[DB] Skipping malformed draw id=%d: %s", row["id"], e)
|
|
return result
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# PREDICTIONS
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def insert_prediction(game_id, strategy, numbers, bonus=None):
|
|
"""
|
|
Save a generated prediction to DB.
|
|
- numbers: list of ints or comma-separated string
|
|
Returns new prediction id.
|
|
"""
|
|
if isinstance(numbers, (list, tuple)):
|
|
numbers_str = ",".join(str(n) for n in numbers)
|
|
else:
|
|
numbers_str = str(numbers).strip()
|
|
|
|
bonus_str = str(bonus) if bonus is not None else None
|
|
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO predictions (game_id, strategy, numbers, bonus)
|
|
VALUES (?, ?, ?, ?)
|
|
""", (game_id, strategy, numbers_str, bonus_str))
|
|
conn.commit()
|
|
new_id = cursor.lastrowid
|
|
logger.info("[PREDICT] Saved prediction id=%d game_id=%d strategy='%s' numbers=%s bonus=%s",
|
|
new_id, game_id, strategy, numbers_str, bonus_str)
|
|
return new_id
|
|
except Exception as e:
|
|
conn.rollback()
|
|
logger.error("[ERROR] insert_prediction failed: %s", e, exc_info=True)
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_predictions(game_id=None, limit=50):
|
|
"""
|
|
Fetch saved predictions.
|
|
- game_id: filter by game (None = all games)
|
|
- limit: max rows
|
|
Returns list of sqlite3.Row objects, newest first.
|
|
"""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
if game_id:
|
|
cursor.execute("""
|
|
SELECT p.*, g.name as game_name
|
|
FROM predictions p
|
|
JOIN games g ON g.id = p.game_id
|
|
WHERE p.game_id = ?
|
|
ORDER BY p.created_at DESC, p.id DESC
|
|
LIMIT ?
|
|
""", (game_id, limit))
|
|
else:
|
|
cursor.execute("""
|
|
SELECT p.*, g.name as game_name
|
|
FROM predictions p
|
|
JOIN games g ON g.id = p.game_id
|
|
ORDER BY p.created_at DESC, p.id DESC
|
|
LIMIT ?
|
|
""", (limit,))
|
|
return cursor.fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def delete_prediction(pred_id: int):
|
|
"""Delete a single saved prediction by id."""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM predictions WHERE id = ?", (pred_id,))
|
|
conn.commit()
|
|
logger.info("[PREDICT] Deleted prediction id=%d", pred_id)
|
|
except Exception as e:
|
|
conn.rollback()
|
|
logger.error("[ERROR] delete_prediction failed: %s", e, exc_info=True)
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def delete_all_predictions(game_id=None):
|
|
"""Delete all saved predictions, optionally filtered by game."""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
if game_id is not None:
|
|
cursor.execute("DELETE FROM predictions WHERE game_id = ?", (game_id,))
|
|
else:
|
|
cursor.execute("DELETE FROM predictions")
|
|
conn.commit()
|
|
logger.info("[PREDICT] Cleared all predictions (game_id=%s)", game_id)
|
|
except Exception as e:
|
|
conn.rollback()
|
|
logger.error("[ERROR] delete_all_predictions failed: %s", e, exc_info=True)
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# FETCH LOG
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def insert_fetch_log(source, added, skipped, status="success", message=None):
|
|
"""
|
|
Log a fetch operation result.
|
|
Called after every auto or manual fetch attempt.
|
|
Returns new log id.
|
|
"""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO fetch_log (source, added, skipped, status, message)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", (source, added, skipped, status, message))
|
|
conn.commit()
|
|
new_id = cursor.lastrowid
|
|
logger.info("[FETCH] Log id=%d source='%s' added=%d skipped=%d status=%s",
|
|
new_id, source, added, skipped, status)
|
|
return new_id
|
|
except Exception as e:
|
|
conn.rollback()
|
|
logger.error("[ERROR] insert_fetch_log failed: %s", e, exc_info=True)
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_last_fetch_log(source=None):
|
|
"""
|
|
Return the most recent fetch log entry.
|
|
- source: filter by source name (None = any source)
|
|
Returns sqlite3.Row or None.
|
|
"""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
if source:
|
|
cursor.execute("""
|
|
SELECT * FROM fetch_log
|
|
WHERE source = ?
|
|
ORDER BY fetched_at DESC, id DESC
|
|
LIMIT 1
|
|
""", (source,))
|
|
else:
|
|
cursor.execute("""
|
|
SELECT * FROM fetch_log
|
|
ORDER BY fetched_at DESC, id DESC
|
|
LIMIT 1
|
|
""")
|
|
return cursor.fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_fetch_logs(limit=50):
|
|
"""
|
|
Return recent fetch log entries, newest first.
|
|
Used in Settings screen to show fetch history.
|
|
"""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT * FROM fetch_log
|
|
ORDER BY fetched_at DESC, id DESC
|
|
LIMIT ?
|
|
""", (limit,))
|
|
return cursor.fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_last_fetch_per_source():
|
|
"""
|
|
Return the most recent fetch log entry for each source.
|
|
Returns dict: {source_name: Row, ...}
|
|
Used by status bar and settings screen.
|
|
"""
|
|
conn = get_connection()
|
|
try:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT f.*
|
|
FROM fetch_log f
|
|
INNER JOIN (
|
|
SELECT source, MAX(fetched_at) as max_at
|
|
FROM fetch_log
|
|
GROUP BY source
|
|
) latest ON f.source = latest.source AND f.fetched_at = latest.max_at
|
|
ORDER BY f.source
|
|
""")
|
|
rows = cursor.fetchall()
|
|
return {row["source"]: row for row in rows}
|
|
finally:
|
|
conn.close()
|