"""library-gui plugin (AUD-161): local web dashboard for the library. Serves a read-only single-page dashboard on 127.0.0.1 showing the library with per-item download status — which kinds (audio / cover / chapter / pdf) are already recorded in which formats, and which are still missing — plus an expandable detail panel per title, sortable columns and marketplace/ownership filters. Data comes exclusively from the ephemeral plugin broker via ``invoke`` (re-entrancy, scope ``invoke``); the plugin never sees auth material. Install: ``audible plugin add [--symlink] `` (or the raw GitHub URL) and make ``audible_plugin_sdk`` importable (``pip install /sdk/python`` or PYTHONPATH). Then:: audible -m all library-gui [--no-open] [--port N] The session is pinned to the invoking ``-a``/``-m``/``-s`` selection (one account per GUI session) — start with ``-m all`` to see every marketplace of the account. Tip for a sync-free dashboard: a settings bundle with ``[settings.gui.db] auto_sync = "none"`` started as ``audible -m all -s gui library-gui``. Stop with Ctrl-C. """ import argparse import hmac import json import secrets import sys import threading import webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, urlparse MANIFEST = { "name": "library-gui", "version": "0.1.0", "description": "Local web dashboard: library + download status", "scopes": ["invoke"], "help": "usage: audible [-m all] library-gui [--no-open] [--port N]", } # Answer the discovery probe before importing the SDK: the manifest # needs nothing from the broker, so `plugin list` shows this plugin as # intact even when audible_plugin_sdk is not importable (yet). if __name__ == "__main__" and "--audible-describe" in sys.argv: print(json.dumps(MANIFEST)) raise SystemExit(0) try: from audible_plugin_sdk import Broker, BrokerError, run except ImportError: print( "error: audible_plugin_sdk is not importable — install it with\n" " pip install /sdk/python\n" "or add /sdk/python to PYTHONPATH.", file=sys.stderr, ) raise SystemExit(2) from None #: Artifact kinds tracked in the downloads table (db/schema.rs). KINDS = ("audio", "cover", "chapter", "pdf") def slim_item(doc): """The doc fields the table and the per-item detail panel need — keeps the browser payload small (full docs easily reach several MB per library).""" def names(field): return [n.get("name", "") for n in (doc.get(field) or []) if n.get("name")] images = doc.get("product_images") or {} cover = images.get("500") or next(iter(images.values()), None) series = (doc.get("series") or [{}])[0] rating = (doc.get("rating") or {}).get("overall_distribution") or {} ladders = [ " › ".join(c.get("name", "") for c in (ladder.get("ladder") or [])) for ladder in doc.get("category_ladders") or [] ] return { "title": doc.get("title"), "subtitle": doc.get("subtitle"), "authors": names("authors"), "narrators": names("narrators"), "publisher": doc.get("publisher_name"), "language": doc.get("language"), "release_date": doc.get("release_date") or doc.get("issue_date"), "purchase_date": doc.get("purchase_date"), "runtime_min": doc.get("runtime_length_min"), "series": series.get("title"), "series_sequence": series.get("sequence"), "rating": rating.get("display_average_rating"), "num_ratings": rating.get("num_ratings"), "summary": doc.get("publisher_summary"), "categories": [ladder for ladder in ladders if ladder], "is_archived": bool(doc.get("is_archived")), # Same ownership marker as `library list --borrowed` (AUD-153): # origin_type == "Purchase" is owned; absent/other = borrowed. "owned": doc.get("origin_type") == "Purchase", "cover": cover, } class Api: """Broker-backed data endpoints. Each call runs one built-in via ``/v1/invoke`` with ``-o json`` and reshapes the result.""" def __init__(self, broker): self.broker = broker def _invoke_json(self, argv, empty): """Invoke a built-in and parse its JSON stdout. `-o json` emits the constant envelope `{"error", "warnings", "result"}` (AUD-279): the payload sits under `"result"`; a failure raises (the envelope's error mirrors the stderr detail).""" reply = self.broker.invoke(argv) stdout = reply.get("stdout", "") if reply.get("code") != 0: detail = (reply.get("stderr") or stdout or "").strip() raise BrokerError(500, f"`audible {' '.join(argv)}` failed: {detail[:300]}") if not stdout.strip(): # Defensive: with the envelope a successful run always prints # one document; treat a bare stdout as the empty result. return empty result = json.loads(stdout).get("result") return empty if result is None else result def library(self): # `library list` is the (mp, asin) source of truth; the export's # raw docs contribute display fields (cover, authors, …), joined # by asin. A same-asin title in two marketplaces shares one doc — # fine for display. `--kind book,podcast`: the dashboard shows # books and podcast shows; standalone episodes are deliberately # left out to keep the view uncluttered (AUD-173). listing = self._invoke_json( ["library", "list", "--limit", "0", "--kind", "book,podcast"], empty=[] ) export = self._invoke_json( ["library", "export", "--kind", "book,podcast"], empty={} ) docs = {} for doc in export.get("items", []): docs.setdefault(doc.get("asin"), doc) items = [] for row in listing: slim = slim_item(docs.get(row.get("asin"), {})) slim["asin"] = row.get("asin") slim["marketplace"] = row.get("mp") slim["title"] = slim["title"] or row.get("title") items.append(slim) return {"count": len(items), "items": items} def downloads(self): return self._invoke_json(["db", "downloads", "list"], empty=[]) class Handler(BaseHTTPRequestHandler): server_version = "audible-library-gui/0.1" # Injected by main(): the shared Api and the session token. api = None token = None def log_message(self, fmt, *args): # keep the terminal quiet pass def _authorized(self): query = parse_qs(urlparse(self.path).query) sent = (query.get("token") or [self.headers.get("X-Token", "")])[0] return hmac.compare_digest(sent, self.token) def _reply(self, status, content_type, body): self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _reply_json(self, status, value): self._reply(status, "application/json", json.dumps(value).encode()) def do_GET(self): if not self._authorized(): self._reply_json(403, {"error": "missing or wrong token"}) return route = urlparse(self.path).path try: if route == "/": self._reply(200, "text/html; charset=utf-8", PAGE.encode()) elif route == "/api/library": self._reply_json(200, self.api.library()) elif route == "/api/downloads": self._reply_json(200, self.api.downloads()) else: self._reply_json(404, {"error": f"no route {route}"}) except BrokerError as error: self._reply_json(502, {"error": str(error)}) except Exception as error: # surface, never crash the server thread self._reply_json(500, {"error": f"{type(error).__name__}: {error}"}) def main(argv): parser = argparse.ArgumentParser( prog="audible library-gui", epilog="Account/marketplace/settings are fixed when the plugin starts: " "put the selectors BEFORE the plugin name, e.g. `audible -m all library-gui`.", ) parser.add_argument( "--no-open", action="store_true", help="do not open the browser automatically (URL is printed either way)", ) parser.add_argument( "--port", type=int, default=0, help="fixed port instead of a random free one", ) # Catch selector flags placed after the plugin name — the broker has # already pinned the invoking -a/-m/-s selection (AUD-123), so guide # the user instead of failing with "unrecognized arguments". for flags in (("-a", "--account"), ("-m", "--marketplace"), ("-s", "--settings")): parser.add_argument(*flags, dest="selector", help=argparse.SUPPRESS) args = parser.parse_args(argv) if args.selector is not None: parser.error( "selectors are fixed when the plugin starts — put them before " "the plugin name: audible -m all library-gui" ) Handler.api = Api(Broker()) Handler.token = secrets.token_urlsafe(16) server = ThreadingHTTPServer(("127.0.0.1", args.port), Handler) url = f"http://127.0.0.1:{server.server_address[1]}/?token={Handler.token}" print(f"audible library-gui listening on {url}", flush=True) print("Ctrl-C stops the dashboard.", flush=True) if not args.no_open: threading.Timer(0.3, webbrowser.open, [url]).start() try: server.serve_forever() except KeyboardInterrupt: print("\nbye") finally: server.server_close() return 0 # --- embedded single-page app (no dependencies, no CDN; the JS reads # --- the session token from the page URL) ----------------------------- PAGE = r""" audible — library

audible library

loading library…
""" if __name__ == "__main__": raise SystemExit(run(MANIFEST, main))