import json import sqlite3 import threading import time import webbrowser from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path from journeyfm.paths import data_path from journeyfm.config_store import load_runtime_config from journeyfm.update_service import run_update_job DEFAULT_HOST = "127.0.0.1" DEFAULT_PORT = 8765 def load_recent_stats(db_path=None): db_path = db_path or data_path("playlist_history.db") stats = { "total_updates": 0, "total_scraped": 0, "total_matched": 0, "total_added": 0, "total_missing": 0, "total_duplicates": 0, "total_skipped": 0, "last_attempted": None, "last_success": None, "station_counts": {}, } if not Path(db_path).exists(): return stats conn = sqlite3.connect(db_path) try: c = conn.cursor() c.execute( "SELECT COUNT(*), SUM(scraped_count), SUM(matched_count), SUM(added_count), SUM(missing_count), SUM(duplicate_count), SUM(skipped_count) FROM history" ) row = c.fetchone() or () stats["total_updates"] = int(row[0] or 0) stats["total_scraped"] = int(row[1] or 0) stats["total_matched"] = int(row[2] or 0) stats["total_added"] = int(row[3] or 0) stats["total_missing"] = int(row[4] or 0) stats["total_duplicates"] = int(row[5] or 0) stats["total_skipped"] = int(row[6] or 0) c.execute("SELECT MAX(date), MAX(CASE WHEN status='success' THEN date END) FROM history") last_row = c.fetchone() or () stats["last_attempted"] = last_row[0] stats["last_success"] = last_row[1] c.execute("SELECT station_breakdown FROM history") station_rows = c.fetchall() for (station_json,) in station_rows: try: for station in json.loads(station_json or "[]"): if station.get("success"): key = station.get("display_name") or station.get("station") or "Unknown" stats["station_counts"][key] = stats["station_counts"].get(key, 0) + int(station.get("scraped_count", 0)) except Exception: continue finally: conn.close() return stats def render_dashboard_html(stats): title = "Journey FM Song Count Command Center" station_rows = "".join( f"{name}{count}" for name, count in sorted(stats["station_counts"].items()) ) overall_count = sum(stats["station_counts"].values()) template = """

Journey FM Song Count Command Center

Live track count breakdown by station with overall totals and history-driven health metrics.

Overall Find

Total Scraped

Total Matched

Missing / Buy List

Duplicates

Updates Run

Station	Scraped this run

Download JSON stats

""" return ( template .replace("", title) .replace("", f"{overall_count:,}") .replace("", f"{stats['total_scraped']:,}") .replace("", f"{stats['total_matched']:,}") .replace("", f"{stats['total_missing']:,}") .replace("", f"{stats['total_duplicates']:,}") .replace("", f"{stats['total_updates']:,}") .replace("", station_rows or 'No station data yet') .replace("", stats['last_attempted'] or 'N/A') .replace("", stats['last_success'] or 'N/A') ) def _build_handler(stats_supplier): class DashboardHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path in ("/", "/stats", "/index.html"): stats = stats_supplier() content = render_dashboard_html(stats).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(content))) self.end_headers() self.wfile.write(content) elif self.path == "/api/stats": stats = stats_supplier() content = json.dumps(stats, default=str).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(content))) self.end_headers() self.wfile.write(content) elif self.path == "/api/refresh": try: result = run_update_job(load_runtime_config()) response = {"status": "ok", "result": result} content = json.dumps(response, default=str).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(content))) self.end_headers() self.wfile.write(content) except Exception as exc: response = {"status": "error", "message": str(exc)} content = json.dumps(response, default=str).encode("utf-8") self.send_response(500) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(content))) self.end_headers() self.wfile.write(content) else: self.send_response(404) self.send_header("Content-Type", "text/plain; charset=utf-8") self.end_headers() self.wfile.write(b"404 Not Found") def log_message(self, format, *args): # Keep server quiet; logging handled elsewhere pass return DashboardHandler def start_dashboard_server(host=DEFAULT_HOST, port=DEFAULT_PORT, open_browser_if_possible=True): server_address = (host, port) httpd = HTTPServer(server_address, _build_handler(load_recent_stats)) def serve(): try: if open_browser_if_possible: webbrowser.open(f"http://{host}:{port}") print(f"Dashboard available at http://{host}:{port}/") httpd.serve_forever() except KeyboardInterrupt: pass finally: httpd.server_close() thread = threading.Thread(target=serve, daemon=True) thread.start() return httpd, thread def stop_dashboard_server(httpd): try: httpd.shutdown() except Exception: pass def get_dashboard_url(host=DEFAULT_HOST, port=DEFAULT_PORT): return f"http://{host}:{port}/"