diff --git a/journeyfm/web_service.py b/journeyfm/web_service.py index 1528421..1a4bc29 100644 --- a/journeyfm/web_service.py +++ b/journeyfm/web_service.py @@ -1,46 +1,55 @@ import json +import logging +import queue import sqlite3 import threading -import time import traceback import webbrowser -from http.server import BaseHTTPRequestHandler, HTTPServer +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path -from journeyfm.config_store import load_runtime_config +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 -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 +_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": {}, + "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" + "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) @@ -50,486 +59,1157 @@ def load_recent_stats(db_path=None): 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") - station_rows = c.fetchall() - for station_json, scraped_songs_json in station_rows: + seen = set() + for station_json, scraped_songs_json in c.fetchall(): 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)) + 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: - songs = json.loads(scraped_songs_json or "[]") - if "seen_song_keys" not in stats: - stats["seen_song_keys"] = set() - for song in songs: - station_name = song.get("source", "Unknown") + for song in json.loads(scraped_songs_json or "[]"): + sname = song.get("source", "Unknown") title = song.get("title", "?") artist = song.get("artist", "?") - key = (station_name, artist, title) - if key in stats["seen_song_keys"]: + k = (sname, artist, title) + if k in seen: continue - stats["seen_song_keys"].add(key) - display_key = f"{artist} - {title}" - stats["song_counts"].setdefault(station_name, {}) - stats["song_counts"][station_name][display_key] = stats["song_counts"][station_name].get(display_key, 0) + 1 + 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=100): +def load_history_entries(db_path=None, limit=200): db_path = db_path or data_path("playlist_history.db") - rows = [] if not Path(db_path).exists(): - return rows + 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,) + "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,) ) - result = c.fetchall() - for r in result: - rows.append({ - "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 '[]'), - }) + 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() - return rows 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: + with open(buy_list_path, "r", encoding="utf-8") as f: lines = [ln.strip() for ln in f if ln.strip()] - for ln in lines: - if ln.startswith('http'): - continue - entries.append(ln) + 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 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()) +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)} - # Build a flattened station-song ranking table for display - top_song_rows = [] - for station_name, songs in sorted(stats.get("song_counts", {}).items()): - sorted_songs = sorted(songs.items(), key=lambda kv: kv[1], reverse=True) - for song_name, count in sorted_songs[:10]: - top_song_rows.append(f"{station_name}{song_name}{count}") - top_song_rows_html = ''.join(top_song_rows) or 'No song counts yet' - template = """ - +# ── 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""" - - - <!--TITLE--> - + + +Journey FM + + + + -
-

Journey FM Song Count Command Center

-

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

-
- - - - - - - +
+
+ Plex Sync +
Journey FM
+
+ +
Last sync
—
+
+ +
+
+
Overview
+
+ + +
+
+ +
+ + +
+
+
Scraped
—
+
Matched
—
+
Added
—
+
Missing
—
+
Duplicates
—
+
Total Runs
—
+
+
+
+
Station breakdown — all-time scrapes
+
+ + +
- -
-
-

Overall Find

-

Total Scraped

-

Total Matched

-

Missing / Buy List

-

Duplicates

-

Updates Run

+
+
Library match rate
+
+ +
+
—
+
in library
+
- -
-

Station totals

-
- - - -
StationScraped this run
-
-
- -
-

Top tracks per station (historical play counts)

-
- - - -
StationSongCount
-
-
- -
-

App Controls (desktop features)

- - - - -
-
- -
-

Buy List

-
    -
    - -
    -

    Run History

    -
    - - - -
    DateStatusScrapedMatchedAddedMissingDuplicatesSkipped
    -
    -
    - -
    -

    Raw API snapshot

    - 
    Loading…
    -
    - -
    - Last attempted run:
    - Last successful 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("", top_song_rows_html) - .replace("", stats['last_attempted'] or 'N/A') - .replace("", stats['last_success'] or 'N/A') - ) + +def render_dashboard_html(stats): + return DASHBOARD_HTML def _build_handler(stats_supplier): - class DashboardHandler(BaseHTTPRequestHandler): + class Handler(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": + 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()) - 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) + self._json({"status": "ok", "result": result}) except Exception as exc: - trace = traceback.format_exc() - response = { - "status": "error", - "message": str(exc), - "trace": trace, - } - 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) - elif self.path == "/api/preview": + 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) - content = json.dumps(result, 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) + self._json(result) 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) - elif self.path == "/api/buy-list": + 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: - entries = load_buy_list() - content = json.dumps(entries, 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) + self._json(get_playlist_tracks()) 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) - elif self.path == "/api/history": - try: - history = load_history_entries() - content = json.dumps(history, 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) + self._json({"error": str(exc), "tracks": []}) else: self.send_response(404) - self.send_header("Content-Type", "text/plain; charset=utf-8") + 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): - # Keep server quiet; logging handled elsewhere pass - return DashboardHandler + return Handler 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)) + httpd = ThreadingHTTPServer((host, port), _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