import json import logging import queue import sqlite3 import threading import traceback import webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from journeyfm.config_store import get_secret, load_runtime_config, save_runtime_config from journeyfm.paths import data_path from journeyfm.update_service import run_update_job DEFAULT_HOST = "127.0.0.1" DEFAULT_PORT = 8765 _sync_state = {"running": False} _sync_lock = threading.Lock() class _QueueLogHandler(logging.Handler): def __init__(self, q): super().__init__() self.q = q self.setFormatter(logging.Formatter("%(levelname)s %(name)s: %(message)s")) def emit(self, record): try: self.q.put_nowait(self.format(record)) except Exception: pass # ── Data loaders ────────────────────────────────────────────── 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": {}, "song_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, scraped_songs FROM history") seen = set() for station_json, scraped_songs_json in c.fetchall(): try: for s in json.loads(station_json or "[]"): if s.get("success"): key = s.get("display_name") or s.get("station") or "Unknown" stats["station_counts"][key] = stats["station_counts"].get(key, 0) + int(s.get("scraped_count", 0)) except Exception: pass try: for song in json.loads(scraped_songs_json or "[]"): sname = song.get("source", "Unknown") title = song.get("title", "?") artist = song.get("artist", "?") k = (sname, artist, title) if k in seen: continue seen.add(k) stats["song_counts"].setdefault(sname, {}) dk = f"{artist} - {title}" stats["song_counts"][sname][dk] = stats["song_counts"][sname].get(dk, 0) + 1 except Exception: pass finally: conn.close() return stats def load_history_entries(db_path=None, limit=200): db_path = db_path or data_path("playlist_history.db") if not Path(db_path).exists(): return [] conn = sqlite3.connect(db_path) try: c = conn.cursor() c.execute( "SELECT id, date, status, scraped_count, matched_count, added_count," " missing_count, duplicate_count, skipped_count, station_breakdown" " FROM history ORDER BY date DESC LIMIT ?", (limit,) ) return [ {"id": r[0], "date": r[1], "status": r[2], "scraped_count": r[3], "matched_count": r[4], "added_count": r[5], "missing_count": r[6], "duplicate_count": r[7], "skipped_count": r[8], "station_breakdown": json.loads(r[9] or "[]")} for r in c.fetchall() ] except Exception: return [] finally: conn.close() def load_buy_list(buy_list_path=None): buy_list_path = buy_list_path or data_path("amazon_buy_list.txt") if not Path(buy_list_path).exists(): return [] try: with open(buy_list_path, "r", encoding="utf-8") as f: return [ln.strip() for ln in f if ln.strip() and not ln.startswith("http") and ln.strip() != "Songs not in your library - Amazon search links:"] except Exception: return [] def load_buy_list_rich(buy_list_path=None): buy_list_path = buy_list_path or data_path("amazon_buy_list.txt") if not Path(buy_list_path).exists(): return [] entries = [] try: with open(buy_list_path, "r", encoding="utf-8") as f: lines = [ln.strip() for ln in f if ln.strip()] i = 0 while i < len(lines): line = lines[i] if not line or line.startswith("Songs not in") or line.startswith("http"): i += 1 continue url = None if i + 1 < len(lines) and lines[i + 1].startswith("http"): url = lines[i + 1] i += 2 else: i += 1 entries.append({"song": line, "url": url}) except Exception: return [] return entries def dedup_buy_list(buy_list_path=None): buy_list_path = buy_list_path or data_path("amazon_buy_list.txt") entries = load_buy_list_rich(buy_list_path) seen = set() unique = [] for e in entries: key = e["song"].lower().strip() if key not in seen: seen.add(key) unique.append(e) removed = len(entries) - len(unique) with open(buy_list_path, "w", encoding="utf-8") as f: f.write("Songs not in your library - Amazon search links:\n\n") for e in unique: f.write(f"{e['song']}\n") if e["url"]: f.write(f"{e['url']}\n\n") else: f.write("\n") return {"original": len(entries), "removed": removed, "remaining": len(unique)} # ── Config ──────────────────────────────────────────────────── def get_web_config(): config = load_runtime_config() token = config.get("PLEX_TOKEN", "") config["_has_token"] = bool(token) config["PLEX_TOKEN"] = "" stations = config.get("SELECTED_STATIONS", []) if isinstance(stations, str): stations = [s.strip() for s in stations.split(",") if s.strip()] config["SELECTED_STATIONS"] = stations return config def save_web_config(data): config = {} config["SERVER_IP"] = str(data.get("SERVER_IP", "")).strip() config["PLAYLIST_NAME"] = str(data.get("PLAYLIST_NAME", "")).strip() stations = data.get("SELECTED_STATIONS", []) if isinstance(stations, str): stations = [s.strip() for s in stations.split(",") if s.strip()] config["SELECTED_STATIONS"] = stations config["AUTO_UPDATE"] = bool(data.get("AUTO_UPDATE", False)) try: config["UPDATE_INTERVAL"] = int(data.get("UPDATE_INTERVAL", 15)) except (ValueError, TypeError): config["UPDATE_INTERVAL"] = 15 config["UPDATE_UNIT"] = str(data.get("UPDATE_UNIT", "Minutes")) token = str(data.get("PLEX_TOKEN", "")).strip() config["PLEX_TOKEN"] = token if token else get_secret("PLEX_TOKEN") save_runtime_config(config) # ── Playlist ────────────────────────────────────────────────── def get_playlist_tracks(): from journeyfm.plex_service import connect_to_plex_server config = load_runtime_config() plex = connect_to_plex_server(config.get("PLEX_TOKEN", ""), config.get("SERVER_IP", "")) name = config.get("PLAYLIST_NAME", "") playlist = plex.playlist(name) tracks = [] for item in playlist.items(): tracks.append({ "title": item.title, "artist": getattr(item, "grandparentTitle", ""), "album": getattr(item, "parentTitle", ""), "rating_key": str(item.ratingKey), "duration_ms": item.duration or 0, }) return {"name": name, "tracks": tracks} def remove_playlist_tracks(rating_keys): from journeyfm.plex_service import connect_to_plex_server config = load_runtime_config() plex = connect_to_plex_server(config.get("PLEX_TOKEN", ""), config.get("SERVER_IP", "")) name = config.get("PLAYLIST_NAME", "") playlist = plex.playlist(name) keys = set(str(k) for k in rating_keys) to_remove = [item for item in playlist.items() if str(item.ratingKey) in keys] if to_remove: playlist.removeItems(to_remove) return len(to_remove) # ── HTML ────────────────────────────────────────────────────── DASHBOARD_HTML = r""" Journey FM
Plex Sync
Journey FM
Last sync
—
Overview
Scraped
—
Matched
—
Added
—
Missing
—
Duplicates
—
Total Runs
—
Station breakdown — all-time scrapes
Library match rate
—
in library
Top Tracks
# Station Track Plays
Loading…
Songs added per run — last 20
Run Log →
DateStatusScrapedMatched AddedMissingDupesSkipped
Loading…
Missing from Library
Loading…
Playlist
Artist Title Album
Click Load to fetch your playlist from Plex
Plex Connection
Local network address of your Plex server
—
Playlist
Stations
Auto-Update
Enable automatic updates
""" def render_dashboard_html(stats): return DASHBOARD_HTML def _build_handler(stats_supplier): class Handler(BaseHTTPRequestHandler): def do_GET(self): p = self.path.split("?")[0] if p in ("/", "/index.html"): self._html(DASHBOARD_HTML.encode("utf-8")) elif p == "/api/stats": self._json(stats_supplier()) elif p == "/api/config": self._json(get_web_config()) elif p == "/api/test-connection": try: from journeyfm.plex_service import connect_to_plex_server cfg = load_runtime_config() connect_to_plex_server(cfg.get("PLEX_TOKEN", ""), cfg.get("SERVER_IP", "")) self._json({"status": "ok", "message": "Connected to Plex successfully"}) except Exception as exc: self._json({"status": "error", "message": str(exc)}) elif p == "/api/sync-stream": self._sse_sync() elif p == "/api/refresh": try: result = run_update_job(load_runtime_config()) self._json({"status": "ok", "result": result}) except Exception as exc: self._json({"status": "error", "message": str(exc), "trace": traceback.format_exc()}, 500) elif p == "/api/preview": try: result = run_update_job(load_runtime_config(), dry_run=True, persist_history=False, write_buy_list=False) self._json(result) except Exception as exc: self._json({"status": "error", "message": str(exc)}, 500) elif p == "/api/buy-list": self._json(load_buy_list()) elif p == "/api/buy-list-rich": self._json(load_buy_list_rich()) elif p == "/api/buy-list-links": self._json([e["url"] for e in load_buy_list_rich() if e.get("url")]) elif p == "/api/history": self._json(load_history_entries()) elif p == "/api/playlist": try: self._json(get_playlist_tracks()) except Exception as exc: self._json({"error": str(exc), "tracks": []}) else: self.send_response(404) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"404 Not Found") def do_POST(self): p = self.path.split("?")[0] body = self._read_body() if p == "/api/config": try: save_web_config(json.loads(body)) self._json({"status": "ok"}) except Exception as exc: self._json({"status": "error", "message": str(exc)}, 400) elif p == "/api/playlist/remove": try: data = json.loads(body) removed = remove_playlist_tracks(data.get("rating_keys", [])) self._json({"status": "ok", "removed": removed}) except Exception as exc: self._json({"status": "error", "message": str(exc)}, 400) elif p == "/api/buy-list-dedup": try: self._json(dedup_buy_list()) except Exception as exc: self._json({"status": "error", "message": str(exc)}, 500) else: self.send_response(404) self.end_headers() def _sse_sync(self): with _sync_lock: if _sync_state["running"]: self._json({"status": "error", "message": "A sync is already in progress"}, 409) return _sync_state["running"] = True self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "keep-alive") self.send_header("X-Accel-Buffering", "no") self.end_headers() q = queue.Queue() handler = _QueueLogHandler(q) def run(): root = logging.getLogger() root.addHandler(handler) try: result = run_update_job(load_runtime_config()) q.put({"_done": True, "result": result}) except Exception as exc: q.put(f"ERROR: {exc}") q.put({"_done": True, "result": None, "error": str(exc)}) finally: root.removeHandler(handler) _sync_state["running"] = False threading.Thread(target=run, daemon=True).start() try: while True: try: item = q.get(timeout=120) except queue.Empty: self.wfile.write(b": ping\n\n") self.wfile.flush() continue if isinstance(item, dict) and item.get("_done"): payload = json.dumps({"result": item.get("result"), "error": item.get("error")}, default=str) self.wfile.write(f"event: done\ndata: {payload}\n\n".encode()) self.wfile.flush() break else: payload = json.dumps({"msg": str(item)}, default=str) self.wfile.write(f"data: {payload}\n\n".encode()) self.wfile.flush() except (BrokenPipeError, ConnectionResetError, OSError): pass finally: _sync_state["running"] = False def _read_body(self): length = int(self.headers.get("Content-Length", 0)) return self.rfile.read(length).decode("utf-8") if length else "{}" def _json(self, data, status=200): body = json.dumps(data, default=str).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _html(self, body): self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, format, *args): pass return Handler def start_dashboard_server(host=DEFAULT_HOST, port=DEFAULT_PORT, open_browser_if_possible=True): httpd = ThreadingHTTPServer((host, port), _build_handler(load_recent_stats)) def serve(): try: if open_browser_if_possible: webbrowser.open(f"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}/"