nngo/lottosight
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

05/23 Phase 2

Browse Source
This commit is contained in:
nngo
2026-05-23 11:01:40 -04:00
parent d782ae24b5
commit f33962cb15
9 changed files with 764 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude/settings.local.json
+8
View File
@@ -0,0 +1,8 @@
{
"permissions": {
"allow": [
"Bash(python -m pytest tests/test_fetcher.py -v)",
"Bash(python -m pytest -v)"
]
}
}
CLAUDE.md
+16 -16
View File
@@ -225,22 +225,22 @@ All actions are logged to console and optionally to a log file:
---
### 🔲 Phase 2 — Fetch System
- [ ] Write `core/fetcher.py`
- [ ] `fetch_powerball_ny()` — NY Open Data API
- [ ] `fetch_megamillions_ny()` — NY Open Data API
- [ ] `fetch_megamillions_tx()` — Texas Lottery CSV
- [ ] `fetch_all()` — calls all 3, aggregates results
- [ ] Duplicate detection logic
- [ ] Return added/skipped counts per source
- [ ] Write `ui/statusbar.py` — bottom status bar widget
- [ ] Wire auto-fetch on app launch (background thread)
- [ ] Wire 24hr scheduled fetch (APScheduler)
- [ ] Write `main.py` — app window, toolbar with manual fetch button
- [ ] Wire manual fetch button → `fetch_all()` → update status bar
- [ ] Test: fresh DB → fetch → verify records inserted
- [ ] Test: second fetch → verify duplicates skipped, counts correct
- [ ] Test: status bar updates correctly after fetch
### ✅ Phase 2 — Fetch System
- [x] Write `core/fetcher.py`
- [x] `fetch_powerball_ny()` — NY Open Data API
- [x] `fetch_megamillions_ny()` — NY Open Data API
- [x] `fetch_megamillions_tx()` — Texas Lottery CSV
- [x] `fetch_all()` — calls all 3, aggregates results
- [x] Duplicate detection logic
- [x] Return added/skipped counts per source
- [x] Write `ui/statusbar.py` — bottom status bar widget
- [x] Wire auto-fetch on app launch (background thread)
- [x] Wire 24hr scheduled fetch (APScheduler)
- [x] Write `main.py` — app window, toolbar with manual fetch button
- [x] Wire manual fetch button → `fetch_all()` → update status bar
- [x] Test: fresh DB → fetch → verify records inserted (67/67 pass)
- [x] Test: second fetch → verify duplicates skipped, counts correct
- [x] Test: cross-source dedup (same date, different source → skipped)
---
core/__init__.py
View File
core/fetcher.py
+252
View File
@@ -0,0 +1,252 @@
"""
core/fetcher.py
---------------
Data fetch logic for all 3 lottery sources.
Each function returns: {"source", "added", "skipped", "status", "message"}
All HTTP errors are caught and returned as status="error" — never raised.
"""
import csv
import io
import logging
import threading
import requests
from db.models import get_game_by_name, insert_draw, insert_fetch_log
logger = logging.getLogger(__name__)
NY_API_LIMIT = 5000 # records per page
POWERBALL_NY_URL = "https://data.ny.gov/resource/d6yy-54nr.json"
MEGAMILLIONS_NY_URL = "https://data.ny.gov/resource/5xaw-6ayf.json"
MEGAMILLIONS_TX_URL = (
"https://www.texaslottery.com/export/sites/lottery/Games/"
"Mega_Millions/Winning_Numbers/download.html"
)
_fetch_lock = threading.Lock()
def _parse_ny_date(raw):
"""'2024-01-01T00:00:00.000' → '2024-01-01'."""
return raw.split("T")[0].strip()
def _error_result(source, added, skipped, msg):
insert_fetch_log(source, added, skipped, "error", msg)
return {"source": source, "added": added, "skipped": skipped,
"status": "error", "message": msg}
# ── Powerball NY ──────────────────────────────────────────────────────────────
def fetch_powerball_ny():
"""
Fetch Powerball draws from NY Open Data API.
winning_numbers field: "01 13 36 61 69 07" (5 white + 1 Powerball)
"""
source = "powerball_ny"
game = get_game_by_name("Powerball")
if not game:
return _error_result(source, 0, 0, "Powerball game not found in DB")
game_id = game["id"]
added = skipped = 0
offset = 0
try:
while True:
params = {"$limit": NY_API_LIMIT, "$offset": offset, "$order": "draw_date ASC"}
resp = requests.get(POWERBALL_NY_URL, params=params, timeout=30)
resp.raise_for_status()
records = resp.json()
if not records:
break
for rec in records:
raw_date = rec.get("draw_date", "")
raw_nums = rec.get("winning_numbers", "").strip()
multiplier = rec.get("multiplier") or None
if not raw_date or not raw_nums:
continue
try:
draw_date = _parse_ny_date(raw_date)
parts = raw_nums.split()
if len(parts) < 6:
logger.warning("[FETCH] PB NY: unexpected numbers %r on %s", raw_nums, raw_date)
continue
main = [int(p) for p in parts[:5]]
bonus = int(parts[5])
except (ValueError, IndexError) as e:
logger.warning("[FETCH] PB NY parse error %s: %s", rec, e)
continue
result = insert_draw(game_id, draw_date, main, bonus, multiplier, source)
if result == "inserted":
added += 1
else:
skipped += 1
if len(records) < NY_API_LIMIT:
break
offset += NY_API_LIMIT
logger.info("[FETCH] Powerball NY done — added=%d skipped=%d", added, skipped)
insert_fetch_log(source, added, skipped, "success")
return {"source": source, "added": added, "skipped": skipped,
"status": "success", "message": None}
except requests.RequestException as e:
logger.error("[FETCH] Powerball NY error: %s", e)
return _error_result(source, added, skipped, f"Network error: {e}")
# ── Mega Millions NY ──────────────────────────────────────────────────────────
def fetch_megamillions_ny():
"""
Fetch Mega Millions draws from NY Open Data API.
winning_numbers: 5 space-separated main balls
mega_ball: Mega Ball (bonus)
"""
source = "megamillions_ny"
game = get_game_by_name("Mega Millions")
if not game:
return _error_result(source, 0, 0, "Mega Millions game not found in DB")
game_id = game["id"]
added = skipped = 0
offset = 0
try:
while True:
params = {"$limit": NY_API_LIMIT, "$offset": offset, "$order": "draw_date ASC"}
resp = requests.get(MEGAMILLIONS_NY_URL, params=params, timeout=30)
resp.raise_for_status()
records = resp.json()
if not records:
break
for rec in records:
raw_date = rec.get("draw_date", "")
raw_nums = rec.get("winning_numbers", "").strip()
mega_ball = rec.get("mega_ball") or None
multiplier = rec.get("multiplier") or None
if not raw_date or not raw_nums:
continue
try:
draw_date = _parse_ny_date(raw_date)
parts = raw_nums.split()
main = [int(p) for p in parts[:5]]
bonus = int(mega_ball) if mega_ball else None
except (ValueError, IndexError) as e:
logger.warning("[FETCH] MM NY parse error %s: %s", rec, e)
continue
result = insert_draw(game_id, draw_date, main, bonus, multiplier, source)
if result == "inserted":
added += 1
else:
skipped += 1
if len(records) < NY_API_LIMIT:
break
offset += NY_API_LIMIT
logger.info("[FETCH] Mega Millions NY done — added=%d skipped=%d", added, skipped)
insert_fetch_log(source, added, skipped, "success")
return {"source": source, "added": added, "skipped": skipped,
"status": "success", "message": None}
except requests.RequestException as e:
logger.error("[FETCH] Mega Millions NY error: %s", e)
return _error_result(source, added, skipped, f"Network error: {e}")
# ── Mega Millions TX ──────────────────────────────────────────────────────────
def fetch_megamillions_tx():
"""
Fetch Mega Millions draws from Texas Lottery CSV.
Column layout: Game Name, Month, Day, Year, Num1-5, Mega Ball, Megaplier
Header row is auto-detected and skipped.
"""
source = "megamillions_tx"
game = get_game_by_name("Mega Millions")
if not game:
return _error_result(source, 0, 0, "Mega Millions game not found in DB")
game_id = game["id"]
added = skipped = 0
try:
resp = requests.get(MEGAMILLIONS_TX_URL, timeout=30)
resp.raise_for_status()
reader = csv.reader(io.StringIO(resp.text))
for row in reader:
if not row or len(row) < 10:
continue
# Skip header row — month column would be "Month" (non-digit)
if not row[1].strip().isdigit():
continue
# Skip non-Mega Millions rows
if "mega" not in row[0].lower():
continue
try:
month = row[1].strip().zfill(2)
day = row[2].strip().zfill(2)
year = row[3].strip()
draw_date = f"{year}-{month}-{day}"
main = [int(row[i].strip()) for i in range(4, 9)]
bonus = int(row[9].strip())
multiplier = row[10].strip() if len(row) > 10 and row[10].strip() else None
except (ValueError, IndexError) as e:
logger.warning("[FETCH] TX CSV parse error row=%r: %s", row, e)
continue
result = insert_draw(game_id, draw_date, main, bonus, multiplier, source)
if result == "inserted":
added += 1
else:
skipped += 1
logger.info("[FETCH] Mega Millions TX done — added=%d skipped=%d", added, skipped)
insert_fetch_log(source, added, skipped, "success")
return {"source": source, "added": added, "skipped": skipped,
"status": "success", "message": None}
except requests.RequestException as e:
logger.error("[FETCH] Mega Millions TX error: %s", e)
return _error_result(source, added, skipped, f"Network error: {e}")
# ── fetch_all ─────────────────────────────────────────────────────────────────
def fetch_all():
"""
Run all 3 fetch functions and return a list of result dicts.
One source erroring does not stop the others.
Called by the auto-fetch thread and the manual Fetch Now button.
"""
logger.info("[FETCH] fetch_all() starting")
results = []
for fn in (fetch_powerball_ny, fetch_megamillions_ny, fetch_megamillions_tx):
try:
results.append(fn())
except Exception as e:
logger.error("[ERROR] Unexpected error in %s: %s", fn.__name__, e, exc_info=True)
results.append({"source": fn.__name__, "added": 0, "skipped": 0,
"status": "error", "message": str(e)})
logger.info("[FETCH] fetch_all() complete — %d sources processed", len(results))
return results
data/lottosight.db
BIN
View File
Binary file not shown.
main.py
+146
View File
@@ -0,0 +1,146 @@
"""
main.py
-------
LottoSight entry point.
Creates the main window with toolbar, content area, and status bar.
Wires auto-fetch on launch (background thread) and 24hr APScheduler job.
"""
import logging
import threading
import tkinter as tk
from tkinter import ttk
from apscheduler.schedulers.background import BackgroundScheduler
from db.database import init_db
from core.fetcher import fetch_all
from ui.statusbar import StatusBar
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)
class LottoSightApp(tk.Tk):
def __init__(self):
super().__init__()
self.title("LottoSight")
self.minsize(900, 600)
try:
self.iconphoto(True, tk.PhotoImage(file="assets/icon.png"))
except Exception:
pass
self._fetch_lock = threading.Lock()
self._scheduler: BackgroundScheduler | None = None
self._build_ui()
self._start_scheduler()
self._launch_fetch() # auto-fetch on startup
# ── UI construction ───────────────────────────────────────────────────────
def _build_ui(self):
# Toolbar
toolbar = ttk.Frame(self, relief="raised", padding=(8, 4))
toolbar.pack(side="top", fill="x")
ttk.Label(
toolbar, text="LottoSight", font=("TkDefaultFont", 13, "bold")
).pack(side="left", padx=(0, 16))
for name in ("Dashboard", "History", "Analysis", "Predictor", "Settings"):
ttk.Button(
toolbar, text=name, width=10,
command=lambda n=name: self._navigate(n),
).pack(side="left", padx=2)
self._fetch_btn = ttk.Button(
toolbar, text="Fetch Now", command=self._manual_fetch
)
self._fetch_btn.pack(side="right", padx=4)
# Content area — replaced by real screens in later phases
self._content = ttk.Frame(self, padding=16)
self._content.pack(side="top", fill="both", expand=True)
self._content_label = ttk.Label(
self._content, text="Dashboard", font=("TkDefaultFont", 14)
)
self._content_label.pack(expand=True)
# Status bar
self._statusbar = StatusBar(self)
self._statusbar.pack(side="bottom", fill="x")
# ── Navigation ────────────────────────────────────────────────────────────
def _navigate(self, screen_name: str):
logger.info("Navigate → %s", screen_name)
self._content_label.config(text=screen_name)
# ── Fetch wiring ──────────────────────────────────────────────────────────
def _start_scheduler(self):
self._scheduler = BackgroundScheduler()
self._scheduler.add_job(
self._launch_fetch, "interval", hours=24, id="auto_fetch_24h"
)
self._scheduler.start()
logger.info("[FETCH] 24hr scheduler started")
def _launch_fetch(self):
"""Start a background fetch thread (skips if one is already running)."""
thread = threading.Thread(
target=self._run_fetch, daemon=True, name="lottosight-fetch"
)
thread.start()
def _manual_fetch(self):
self._launch_fetch()
def _run_fetch(self):
"""Worker: fetch all sources, then update UI on the main thread."""
if not self._fetch_lock.acquire(blocking=False):
logger.info("[FETCH] Fetch already in progress — skipping")
return
try:
self.after(0, self._on_fetch_start)
results = fetch_all()
self.after(0, lambda r=results: self._on_fetch_done(r))
except Exception as e:
logger.error("[ERROR] fetch_all raised unexpectedly: %s", e, exc_info=True)
self.after(0, self._statusbar.set_ready)
finally:
self._fetch_lock.release()
def _on_fetch_start(self):
self._fetch_btn.config(state="disabled", text="Fetching…")
self._statusbar.set_fetching()
def _on_fetch_done(self, results: list):
self._fetch_btn.config(state="normal", text="Fetch Now")
self._statusbar.update_fetch_results(results)
# ── Lifecycle ─────────────────────────────────────────────────────────────
def on_close(self):
if self._scheduler:
self._scheduler.shutdown(wait=False)
self.destroy()
def main():
init_db()
app = LottoSightApp()
app.protocol("WM_DELETE_WINDOW", app.on_close)
app.mainloop()
if __name__ == "__main__":
main()
tests/test_fetcher.py
+288
View File
@@ -0,0 +1,288 @@
"""
tests/test_fetcher.py
---------------------
Tests for core/fetcher.py — all HTTP calls are mocked.
"""
import json
from unittest.mock import MagicMock, patch
import pytest
import requests as req_lib
from core.fetcher import (
fetch_all,
fetch_megamillions_ny,
fetch_megamillions_tx,
fetch_powerball_ny,
)
from db.models import get_draw_count, get_draws, get_game_by_name, get_last_fetch_log
# ── Mock data ─────────────────────────────────────────────────────────────────
PB_NY_RECORDS = [
{"draw_date": "2024-01-01T00:00:00.000", "winning_numbers": "01 13 36 61 69 07", "multiplier": "2"},
{"draw_date": "2024-01-03T00:00:00.000", "winning_numbers": "05 20 45 60 68 15", "multiplier": "3"},
]
MM_NY_RECORDS = [
{"draw_date": "2024-02-01T00:00:00.000", "winning_numbers": "07 11 22 29 38", "mega_ball": "04", "multiplier": "2"},
{"draw_date": "2024-02-05T00:00:00.000", "winning_numbers": "03 18 33 44 67", "mega_ball": "22", "multiplier": "5"},
]
MM_TX_CSV = (
"Game Name,Month,Day,Year,Num1,Num2,Num3,Num4,Num5,Mega Ball,Megaplier\n"
"Mega Millions,03,07,2024,10,20,30,40,50,12,3\n"
"Mega Millions,03,12,2024,15,25,35,45,55,07,4\n"
)
def _json_resp(data, status=200):
"""Mock requests.Response returning JSON."""
m = MagicMock()
m.status_code = status
m.raise_for_status = MagicMock()
m.json.return_value = data
m.text = json.dumps(data)
return m
def _text_resp(text, status=200):
"""Mock requests.Response returning plain text (CSV)."""
m = MagicMock()
m.status_code = status
m.raise_for_status = MagicMock()
m.text = text
return m
# ── Powerball NY ──────────────────────────────────────────────────────────────
def test_pb_ny_inserts_records(tmp_db):
with patch("core.fetcher.requests.get", side_effect=[_json_resp(PB_NY_RECORDS), _json_resp([])]):
result = fetch_powerball_ny()
assert result["status"] == "success"
assert result["added"] == 2
assert result["skipped"] == 0
game = get_game_by_name("Powerball")
assert get_draw_count(game["id"]) == 2
def test_pb_ny_skips_duplicates(tmp_db):
for _ in range(2):
with patch("core.fetcher.requests.get", side_effect=[_json_resp(PB_NY_RECORDS), _json_resp([])]):
result = fetch_powerball_ny()
assert result["added"] == 0
assert result["skipped"] == 2
def test_pb_ny_parses_numbers_and_bonus(tmp_db):
with patch("core.fetcher.requests.get", side_effect=[_json_resp(PB_NY_RECORDS[:1]), _json_resp([])]):
fetch_powerball_ny()
game = get_game_by_name("Powerball")
draws = get_draws(game["id"])
assert draws[0]["draw_date"] == "2024-01-01"
assert draws[0]["numbers"] == "1,13,36,61,69"
assert draws[0]["bonus"] == "7"
assert draws[0]["multiplier"] == "2"
assert draws[0]["source"] == "powerball_ny"
def test_pb_ny_network_error(tmp_db):
with patch("core.fetcher.requests.get", side_effect=req_lib.RequestException("timeout")):
result = fetch_powerball_ny()
assert result["status"] == "error"
assert result["added"] == 0
assert "timeout" in result["message"]
log = get_last_fetch_log("powerball_ny")
assert log["status"] == "error"
def test_pb_ny_malformed_row_skipped(tmp_db):
"""Row with too few numbers is skipped without crashing."""
bad_records = [{"draw_date": "2024-01-01T00:00:00.000", "winning_numbers": "01 13 36"}]
with patch("core.fetcher.requests.get", side_effect=[_json_resp(bad_records), _json_resp([])]):
result = fetch_powerball_ny()
assert result["status"] == "success"
assert result["added"] == 0
# ── Mega Millions NY ──────────────────────────────────────────────────────────
def test_mm_ny_inserts_records(tmp_db):
with patch("core.fetcher.requests.get", side_effect=[_json_resp(MM_NY_RECORDS), _json_resp([])]):
result = fetch_megamillions_ny()
assert result["status"] == "success"
assert result["added"] == 2
assert result["skipped"] == 0
def test_mm_ny_skips_duplicates(tmp_db):
for _ in range(2):
with patch("core.fetcher.requests.get", side_effect=[_json_resp(MM_NY_RECORDS), _json_resp([])]):
result = fetch_megamillions_ny()
assert result["added"] == 0
assert result["skipped"] == 2
def test_mm_ny_parses_mega_ball(tmp_db):
with patch("core.fetcher.requests.get", side_effect=[_json_resp(MM_NY_RECORDS[:1]), _json_resp([])]):
fetch_megamillions_ny()
game = get_game_by_name("Mega Millions")
draws = get_draws(game["id"])
assert draws[0]["bonus"] == "4"
assert draws[0]["numbers"] == "7,11,22,29,38"
def test_mm_ny_network_error(tmp_db):
with patch("core.fetcher.requests.get", side_effect=req_lib.RequestException("refused")):
result = fetch_megamillions_ny()
assert result["status"] == "error"
log = get_last_fetch_log("megamillions_ny")
assert log["status"] == "error"
# ── Mega Millions TX ──────────────────────────────────────────────────────────
def test_mm_tx_inserts_records(tmp_db):
with patch("core.fetcher.requests.get", return_value=_text_resp(MM_TX_CSV)):
result = fetch_megamillions_tx()
assert result["status"] == "success"
assert result["added"] == 2
assert result["skipped"] == 0
def test_mm_tx_skips_duplicates(tmp_db):
for _ in range(2):
with patch("core.fetcher.requests.get", return_value=_text_resp(MM_TX_CSV)):
result = fetch_megamillions_tx()
assert result["added"] == 0
assert result["skipped"] == 2
def test_mm_tx_parses_date_and_numbers(tmp_db):
with patch("core.fetcher.requests.get", return_value=_text_resp(MM_TX_CSV)):
fetch_megamillions_tx()
game = get_game_by_name("Mega Millions")
draws = get_draws(game["id"], order="ASC")
assert draws[0]["draw_date"] == "2024-03-07"
assert draws[0]["numbers"] == "10,20,30,40,50"
assert draws[0]["bonus"] == "12"
assert draws[0]["multiplier"] == "3"
def test_mm_tx_cross_source_dedup(tmp_db):
"""TX record on same date as an already-inserted NY record is skipped."""
with patch("core.fetcher.requests.get", side_effect=[_json_resp(MM_NY_RECORDS[:1]), _json_resp([])]):
fetch_megamillions_ny() # inserts 2024-02-01
overlapping_csv = (
"Game Name,Month,Day,Year,Num1,Num2,Num3,Num4,Num5,Mega Ball,Megaplier\n"
"Mega Millions,02,01,2024,07,11,22,29,38,4,2\n"
)
with patch("core.fetcher.requests.get", return_value=_text_resp(overlapping_csv)):
result = fetch_megamillions_tx()
assert result["skipped"] == 1
assert result["added"] == 0
def test_mm_tx_network_error(tmp_db):
with patch("core.fetcher.requests.get", side_effect=req_lib.RequestException("connect")):
result = fetch_megamillions_tx()
assert result["status"] == "error"
assert result["added"] == 0
def test_mm_tx_no_header_row(tmp_db):
"""CSV without a header row still parses correctly."""
no_header = "Mega Millions,03,07,2024,10,20,30,40,50,12,3\n"
with patch("core.fetcher.requests.get", return_value=_text_resp(no_header)):
result = fetch_megamillions_tx()
assert result["added"] == 1
# ── fetch_all ─────────────────────────────────────────────────────────────────
def test_fetch_all_returns_three_sources(tmp_db):
# Both NY APIs stop after 1 call (2 records < NY_API_LIMIT), so 3 calls total
with patch("core.fetcher.requests.get", side_effect=[
_json_resp(PB_NY_RECORDS),
_json_resp(MM_NY_RECORDS),
_text_resp(MM_TX_CSV),
]):
results = fetch_all()
assert len(results) == 3
sources = {r["source"] for r in results}
assert sources == {"powerball_ny", "megamillions_ny", "megamillions_tx"}
def test_fetch_all_aggregated_counts(tmp_db):
with patch("core.fetcher.requests.get", side_effect=[
_json_resp(PB_NY_RECORDS),
_json_resp(MM_NY_RECORDS),
_text_resp(MM_TX_CSV),
]):
results = fetch_all()
by_source = {r["source"]: r for r in results}
assert by_source["powerball_ny"]["added"] == 2
assert by_source["megamillions_ny"]["added"] == 2
assert by_source["megamillions_tx"]["added"] == 2
def test_fetch_all_continues_after_one_error(tmp_db):
"""If one source errors, the remaining sources still complete."""
error_result = {"source": "powerball_ny", "added": 0, "skipped": 0, "status": "error", "message": "timeout"}
ny_result = {"source": "megamillions_ny", "added": 2, "skipped": 0, "status": "success", "message": None}
tx_result = {"source": "megamillions_tx", "added": 2, "skipped": 0, "status": "success", "message": None}
with (
patch("core.fetcher.fetch_powerball_ny", return_value=error_result),
patch("core.fetcher.fetch_megamillions_ny", return_value=ny_result),
patch("core.fetcher.fetch_megamillions_tx", return_value=tx_result),
):
results = fetch_all()
assert len(results) == 3
by_source = {r["source"]: r for r in results}
assert by_source["powerball_ny"]["status"] == "error"
assert by_source["megamillions_ny"]["status"] == "success"
assert by_source["megamillions_tx"]["status"] == "success"
# ── Fetch log ─────────────────────────────────────────────────────────────────
def test_fetch_log_written_on_success(tmp_db):
with patch("core.fetcher.requests.get", side_effect=[_json_resp(PB_NY_RECORDS), _json_resp([])]):
fetch_powerball_ny()
log = get_last_fetch_log("powerball_ny")
assert log is not None
assert log["status"] == "success"
assert log["added"] == 2
assert log["skipped"] == 0
def test_fetch_log_written_on_error(tmp_db):
with patch("core.fetcher.requests.get", side_effect=req_lib.RequestException("fail")):
fetch_powerball_ny()
log = get_last_fetch_log("powerball_ny")
assert log is not None
assert log["status"] == "error"
ui/__init__.py
View File
ui/statusbar.py
+54
View File
@@ -0,0 +1,54 @@
"""
ui/statusbar.py
---------------
Bottom status bar widget for LottoSight.
Updated after every fetch via update_fetch_results().
"""
import tkinter as tk
from tkinter import ttk
from datetime import datetime
_SOURCE_NAMES = {
"powerball_ny": "Powerball",
"megamillions_ny": "Mega Millions (NY)",
"megamillions_tx": "Mega Millions (TX)",
}
class StatusBar(ttk.Frame):
"""
Thin horizontal bar docked at the bottom of the main window.
All public methods are safe to call from the main thread only.
"""
def __init__(self, parent, **kwargs):
super().__init__(parent, relief="sunken", **kwargs)
self._label = ttk.Label(self, text="Ready", anchor="w", padding=(6, 2))
self._label.pack(fill="x", expand=True)
def set_text(self, text: str):
self._label.config(text=text)
def set_fetching(self):
self._label.config(text="Fetching data…")
def set_ready(self):
self._label.config(text="Ready")
def update_fetch_results(self, results: list):
"""
Build and display status text from a fetch_all() result list.
Format:
Last fetch: Powerball — 3 added, 2 skipped | Mega Millions (NY) — 5 added, 0 skipped | 2026-05-23 08:42 AM
"""
now = datetime.now().strftime("%Y-%m-%d %I:%M %p")
parts = []
for r in results:
name = _SOURCE_NAMES.get(r["source"], r["source"])
if r["status"] == "error":
parts.append(f"{name} — ERROR: {r['message']}")
else:
parts.append(f"{name} — {r['added']} added, {r['skipped']} skipped")
text = "Last fetch: " + " | ".join(parts) + f" | {now}"
self.set_text(text)