Tag albums service: album lookup API for Home Assistant and assignment web page

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/tag_albums/__init__.py

@@ -0,0 +1 @@
+"""Tells Home Assistant which album to play for a scanned RFID tag."""

src/tag_albums/app.py

@@ -0,0 +1,123 @@
+"""HTTP API for Home Assistant and the web page for assigning albums."""
+
+import secrets
+from datetime import datetime
+from pathlib import Path
+
+import httpx
+from fastapi import Depends, FastAPI, Form, Header, HTTPException, Request
+from fastapi.responses import HTMLResponse, RedirectResponse
+from fastapi.templating import Jinja2Templates
+from pydantic import BaseModel
+
+from .config import Settings
+from .library import MusicLibrary
+from .store import Store, Tag
+
+templates = Jinja2Templates(directory=Path(__file__).parent / "templates")
+
+
+def _local_time(value: str | None) -> str:
+    if not value:
+        return ""
+    return datetime.fromisoformat(value).astimezone().strftime("%Y-%m-%d %H:%M")
+
+
+templates.env.filters["local_time"] = _local_time
+
+
+class ScanRequest(BaseModel):
+    tag_id: str
+
+
+def _tag_response(tag: Tag) -> dict:
+    if not tag.assigned:
+        return {"tag_id": tag.tag_id, "assigned": False}
+    return {"tag_id": tag.tag_id, "assigned": True, "artist": tag.artist, "album": tag.album}
+
+
+def create_app(settings: Settings, store: Store | None = None, library: MusicLibrary | None = None) -> FastAPI:
+    store = store or Store(settings.db_path)
+    library = library or MusicLibrary(settings.ha_url, settings.ha_token)
+    app = FastAPI(title="Tag albums", docs_url=None, redoc_url=None)
+
+    def require_token(authorization: str = Header(default="")) -> None:
+        scheme, _, token = authorization.partition(" ")
+        if scheme.lower() != "bearer" or not secrets.compare_digest(token, settings.api_token):
+            raise HTTPException(status_code=401, detail="Invalid or missing token")
+
+    # --- API for Home Assistant ---------------------------------------------
+
+    @app.post("/api/scans", dependencies=[Depends(require_token)])
+    def record_scan(scan: ScanRequest) -> dict:
+        """Record a Scan and return the Tag's Assignment, if any."""
+        if not scan.tag_id.strip():
+            raise HTTPException(status_code=422, detail="tag_id is empty")
+        return _tag_response(store.record_scan(scan.tag_id))
+
+    @app.get("/api/tags/{tag_id}", dependencies=[Depends(require_token)])
+    def get_tag(tag_id: str) -> dict:
+        """Read-only lookup: doesn't count as a Scan."""
+        tag = store.get(tag_id)
+        if tag is None:
+            raise HTTPException(status_code=404, detail="Unknown tag")
+        return _tag_response(tag)
+
+    # --- Web page -------------------------------------------------------------
+
+    @app.get("/", response_class=HTMLResponse)
+    def index(request: Request):
+        return templates.TemplateResponse(
+            request,
+            "index.html",
+            {"unassigned": store.unassigned(), "assigned": store.assigned()},
+        )
+
+    @app.get("/unassigned", response_class=HTMLResponse)
+    def unassigned(request: Request):
+        return templates.TemplateResponse(
+            request, "_unassigned.html", {"unassigned": store.unassigned()}
+        )
+
+    @app.get("/tags/{tag_id}", response_class=HTMLResponse)
+    def tag_page(request: Request, tag_id: str):
+        tag = store.get(tag_id)
+        if tag is None:
+            raise HTTPException(status_code=404, detail="Unknown tag")
+        return templates.TemplateResponse(request, "tag.html", {"tag": tag})
+
+    @app.get("/tags/{tag_id}/search", response_class=HTMLResponse)
+    def search(request: Request, tag_id: str, q: str = ""):
+        albums, error = [], None
+        if q.strip():
+            try:
+                albums = library.search_albums(q.strip())
+            except (httpx.HTTPError, RuntimeError) as e:
+                error = f"Search failed: {e}"
+        return templates.TemplateResponse(
+            request,
+            "_results.html",
+            {"tag_id": tag_id, "albums": albums, "error": error, "q": q},
+        )
+
+    @app.post("/tags/{tag_id}/assign")
+    def assign(tag_id: str, artist: str = Form(), album: str = Form(), image: str = Form(default="")):
+        store.assign(tag_id, artist, album, image)
+        return RedirectResponse("/", status_code=303)
+
+    @app.post("/tags/{tag_id}/unassign")
+    def unassign(tag_id: str):
+        store.unassign(tag_id)
+        return RedirectResponse("/", status_code=303)
+
+    @app.post("/tags/{tag_id}/delete")
+    def delete(tag_id: str):
+        store.delete(tag_id)
+        return RedirectResponse("/", status_code=303)
+
+    return app
+
+
+def app_from_env() -> FastAPI:
+    """Entry point for uvicorn: `uvicorn --factory tag_albums.app:app_from_env`."""
+    return create_app(Settings.from_env())

src/tag_albums/cli.py

@@ -0,0 +1,42 @@
+"""Command line: run the service, or import Assignments from tag_albums.yaml."""
+
+import argparse
+import os
+
+import yaml
+
+from .store import Store
+
+
+def import_yaml(path: str, db_path: str) -> int:
+    with open(path) as f:
+        entries = yaml.safe_load(f) or {}
+    store = Store(db_path)
+    try:
+        for tag_id, entry in entries.items():
+            store.assign(str(tag_id), entry["artist"], entry["album"])
+    finally:
+        store.close()
+    return len(entries)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(prog="tag-albums")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    serve = sub.add_parser("serve", help="Run the web service")
+    serve.add_argument("--host", default="0.0.0.0")
+    serve.add_argument("--port", type=int, default=8087)
+
+    imp = sub.add_parser("import-yaml", help="Import Assignments from a tag_albums.yaml file")
+    imp.add_argument("path")
+
+    args = parser.parse_args()
+    if args.command == "serve":
+        import uvicorn
+
+        uvicorn.run("tag_albums.app:app_from_env", factory=True, host=args.host, port=args.port)
+    elif args.command == "import-yaml":
+        db_path = os.environ.get("DB_PATH", "tag_albums.sqlite3")
+        count = import_yaml(args.path, db_path)
+        print(f"Imported {count} assignments into {db_path}")

src/tag_albums/config.py

@@ -0,0 +1,22 @@
+import os
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Settings:
+    db_path: str
+    api_token: str
+    ha_url: str
+    ha_token: str
+
+    @classmethod
+    def from_env(cls) -> "Settings":
+        missing = [k for k in ("API_TOKEN", "HA_URL", "HA_TOKEN") if not os.environ.get(k)]
+        if missing:
+            raise RuntimeError(f"Missing environment variables: {', '.join(missing)}")
+        return cls(
+            db_path=os.environ.get("DB_PATH", "tag_albums.sqlite3"),
+            api_token=os.environ["API_TOKEN"],
+            ha_url=os.environ["HA_URL"].rstrip("/"),
+            ha_token=os.environ["HA_TOKEN"],
+        )

src/tag_albums/library.py

@@ -0,0 +1,62 @@
+"""Album search in the Music Assistant library, through Home Assistant."""
+
+import ssl
+from dataclasses import dataclass
+
+import httpx
+
+
+@dataclass(frozen=True)
+class Album:
+    artist: str
+    album: str
+    image: str | None
+
+
+class MusicLibrary:
+    def __init__(self, ha_url: str, ha_token: str, client: httpx.Client | None = None):
+        self._client = client or httpx.Client(
+            base_url=ha_url,
+            headers={"Authorization": f"Bearer {ha_token}"},
+            timeout=10,
+            # System trust store rather than certifi, so a home CA installed
+            # on the host (or pointed to by SSL_CERT_FILE) is trusted.
+            verify=ssl.create_default_context(),
+        )
+        self._entry_id: str | None = None
+
+    def _music_assistant_entry(self) -> str:
+        if self._entry_id is None:
+            r = self._client.get(
+                "/api/config/config_entries/entry", params={"domain": "music_assistant"}
+            )
+            r.raise_for_status()
+            entries = r.json()
+            if not entries:
+                raise RuntimeError("Music Assistant is not set up in Home Assistant")
+            self._entry_id = entries[0]["entry_id"]
+        return self._entry_id
+
+    def search_albums(self, query: str, limit: int = 12) -> list[Album]:
+        r = self._client.post(
+            "/api/services/music_assistant/search",
+            params={"return_response": ""},
+            json={
+                "config_entry_id": self._music_assistant_entry(),
+                "name": query,
+                "media_type": ["album"],
+                "limit": limit,
+            },
+        )
+        r.raise_for_status()
+        albums = r.json().get("service_response", {}).get("albums", [])
+        return [
+            Album(
+                # Music Assistant's play_media takes a single artist name
+                # to disambiguate the album, so keep the primary one.
+                artist=next((a["name"] for a in item.get("artists", [])), ""),
+                album=item["name"],
+                image=item.get("image"),
+            )
+            for item in albums
+        ]

src/tag_albums/store.py

@@ -0,0 +1,116 @@
+"""SQLite storage for Tags and their Assignments.
+
+A Tag row always exists once the Tag has been scanned (or imported); its
+Assignment is the artist/album columns, all NULL for an Unassigned tag.
+"""
+
+import sqlite3
+from dataclasses import dataclass
+from datetime import UTC, datetime
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS tags (
+    tag_id      TEXT PRIMARY KEY,
+    first_seen  TEXT NOT NULL,
+    last_seen   TEXT,
+    scan_count  INTEGER NOT NULL DEFAULT 0,
+    artist      TEXT,
+    album       TEXT,
+    image       TEXT,
+    assigned_at TEXT
+)
+"""
+
+
+@dataclass(frozen=True)
+class Tag:
+    tag_id: str
+    first_seen: str
+    last_seen: str | None
+    scan_count: int
+    artist: str | None
+    album: str | None
+    image: str | None
+    assigned_at: str | None
+
+    @property
+    def assigned(self) -> bool:
+        return self.album is not None
+
+
+def normalize_tag_id(tag_id: str) -> str:
+    return tag_id.strip().upper()
+
+
+def _now() -> str:
+    return datetime.now(UTC).isoformat(timespec="seconds")
+
+
+class Store:
+    def __init__(self, path: str):
+        self._conn = sqlite3.connect(path, check_same_thread=False, isolation_level=None)
+        self._conn.row_factory = sqlite3.Row
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._conn.execute(SCHEMA)
+
+    def close(self) -> None:
+        self._conn.close()
+
+    def get(self, tag_id: str) -> Tag | None:
+        row = self._conn.execute(
+            "SELECT * FROM tags WHERE tag_id = ?", (normalize_tag_id(tag_id),)
+        ).fetchone()
+        return Tag(**row) if row else None
+
+    def record_scan(self, tag_id: str) -> Tag:
+        """Record a Scan, creating the Tag as an Unassigned tag on first sight."""
+        tag_id = normalize_tag_id(tag_id)
+        now = _now()
+        self._conn.execute(
+            """
+            INSERT INTO tags (tag_id, first_seen, last_seen, scan_count)
+            VALUES (?, ?, ?, 1)
+            ON CONFLICT (tag_id) DO UPDATE
+            SET last_seen = excluded.last_seen, scan_count = scan_count + 1
+            """,
+            (tag_id, now, now),
+        )
+        return self.get(tag_id)
+
+    def assign(self, tag_id: str, artist: str, album: str, image: str | None = None) -> Tag:
+        """Create or replace the Tag's Assignment, creating the Tag if needed."""
+        tag_id = normalize_tag_id(tag_id)
+        now = _now()
+        self._conn.execute(
+            """
+            INSERT INTO tags (tag_id, first_seen, artist, album, image, assigned_at)
+            VALUES (?, ?, ?, ?, ?, ?)
+            ON CONFLICT (tag_id) DO UPDATE
+            SET artist = excluded.artist, album = excluded.album,
+                image = excluded.image, assigned_at = excluded.assigned_at
+            """,
+            (tag_id, now, artist, album, image or None, now),
+        )
+        return self.get(tag_id)
+
+    def unassign(self, tag_id: str) -> None:
+        self._conn.execute(
+            "UPDATE tags SET artist = NULL, album = NULL, image = NULL, assigned_at = NULL"
+            " WHERE tag_id = ?",
+            (normalize_tag_id(tag_id),),
+        )
+
+    def delete(self, tag_id: str) -> None:
+        self._conn.execute("DELETE FROM tags WHERE tag_id = ?", (normalize_tag_id(tag_id),))
+
+    def unassigned(self) -> list[Tag]:
+        rows = self._conn.execute(
+            "SELECT * FROM tags WHERE album IS NULL ORDER BY last_seen DESC, first_seen DESC"
+        )
+        return [Tag(**r) for r in rows]
+
+    def assigned(self) -> list[Tag]:
+        rows = self._conn.execute(
+            "SELECT * FROM tags WHERE album IS NOT NULL ORDER BY artist COLLATE NOCASE, album COLLATE NOCASE"
+        )
+        return [Tag(**r) for r in rows]

src/tag_albums/templates/_results.html

@@ -0,0 +1,20 @@
+{% if error %}
+<p class="muted">{{ error }}</p>
+{% elif albums %}
+<ul class="rows">
+  {% for a in albums %}
+  <li>
+    {% if a.image %}<img class="cover" src="{{ a.image }}" alt="" loading="lazy">{% else %}<span class="cover"></span>{% endif %}
+    <div class="grow"><strong>{{ a.album }}</strong> · {{ a.artist }}</div>
+    <form class="inline" method="post" action="/tags/{{ tag_id }}/assign">
+      <input type="hidden" name="artist" value="{{ a.artist }}">
+      <input type="hidden" name="album" value="{{ a.album }}">
+      <input type="hidden" name="image" value="{{ a.image or '' }}">
+      <button class="primary">Assign</button>
+    </form>
+  </li>
+  {% endfor %}
+</ul>
+{% elif q %}
+<p class="muted">No albums found for “{{ q }}”.</p>
+{% endif %}

src/tag_albums/templates/_unassigned.html

@@ -0,0 +1,19 @@
+<ul class="rows">
+  {% for tag in unassigned %}
+  <li>
+    <div class="grow">
+      <div><code>{{ tag.tag_id }}</code></div>
+      <div class="muted">
+        {% if tag.last_seen %}last scanned {{ tag.last_seen|local_time }} · {{ tag.scan_count }} scan{{ "s" if tag.scan_count != 1 }}{% else %}never scanned{% endif %}
+      </div>
+    </div>
+    <a href="/tags/{{ tag.tag_id }}">Assign</a>
+    <form class="inline" method="post" action="/tags/{{ tag.tag_id }}/delete"
+          onsubmit="return confirm('Delete {{ tag.tag_id }}? It comes back if it is scanned again.')">
+      <button class="danger">Delete</button>
+    </form>
+  </li>
+  {% else %}
+  <li class="empty muted">No unassigned tags.</li>
+  {% endfor %}
+</ul>

src/tag_albums/templates/base.html

@@ -0,0 +1,37 @@
+<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1">
+  <title>{% block title %}Tag albums{% endblock %}</title>
+  <script src="https://cdn.jsdelivr.net/npm/htmx.org@2.0.4/dist/htmx.min.js"></script>
+  <style>
+    :root { --bg: #fafafa; --fg: #1d1d1f; --muted: #6e6e73; --line: #e2e2e6; --card: #fff; --accent: #2f6fde; --danger: #c0392b; }
+    @media (prefers-color-scheme: dark) {
+      :root { --bg: #151517; --fg: #f2f2f4; --muted: #9a9aa1; --line: #2c2c31; --card: #1e1e22; --accent: #6c9cff; --danger: #ff6b5e; }
+    }
+    * { box-sizing: border-box; }
+    body { margin: 0 auto; max-width: 860px; padding: 24px 16px; background: var(--bg); color: var(--fg);
+           font: 15px/1.45 system-ui, -apple-system, "Segoe UI", sans-serif; }
+    h1 { font-size: 1.4rem; margin: 0 0 20px; }
+    h2 { font-size: 1.05rem; margin: 28px 0 10px; }
+    a { color: var(--accent); }
+    .muted { color: var(--muted); font-size: .9em; }
+    code { font-size: .9em; }
+    ul.rows { list-style: none; margin: 0; padding: 0; border: 1px solid var(--line); border-radius: 10px; background: var(--card); }
+    ul.rows li { display: flex; gap: 12px; align-items: center; padding: 10px 12px; border-top: 1px solid var(--line); }
+    ul.rows li:first-child { border-top: 0; }
+    .grow { flex: 1; min-width: 0; }
+    img.cover { width: 44px; height: 44px; border-radius: 6px; object-fit: cover; background: var(--line); flex: none; }
+    form.inline { display: inline; }
+    button { font: inherit; padding: 5px 12px; border-radius: 7px; border: 1px solid var(--line); background: var(--card); color: var(--fg); cursor: pointer; }
+    button.primary { background: var(--accent); border-color: var(--accent); color: #fff; }
+    button.danger { color: var(--danger); }
+    input[type=search] { font: inherit; width: 100%; padding: 8px 10px; border-radius: 8px; border: 1px solid var(--line); background: var(--card); color: var(--fg); }
+    .empty { padding: 14px 12px; }
+  </style>
+</head>
+<body>
+  {% block body %}{% endblock %}
+</body>
+</html>

src/tag_albums/templates/index.html

@@ -0,0 +1,30 @@
+{% extends "base.html" %}
+{% block body %}
+<h1>Tag albums</h1>
+
+<h2>Unassigned tags</h2>
+<p class="muted">Scan a new card on the reader and it shows up here.</p>
+<div hx-get="/unassigned" hx-trigger="every 3s" hx-swap="innerHTML">
+  {% include "_unassigned.html" %}
+</div>
+
+<h2>Assignments <span class="muted">({{ assigned|length }})</span></h2>
+<ul class="rows">
+  {% for tag in assigned %}
+  <li>
+    {% if tag.image %}<img class="cover" src="{{ tag.image }}" alt="" loading="lazy">{% else %}<span class="cover"></span>{% endif %}
+    <div class="grow">
+      <div><strong>{{ tag.album }}</strong> · {{ tag.artist }}</div>
+      <div class="muted"><code>{{ tag.tag_id }}</code>{% if tag.last_seen %} · last scanned {{ tag.last_seen|local_time }}{% endif %}</div>
+    </div>
+    <a href="/tags/{{ tag.tag_id }}">Change</a>
+    <form class="inline" method="post" action="/tags/{{ tag.tag_id }}/unassign"
+          onsubmit="return confirm('Remove the album from {{ tag.tag_id }}?')">
+      <button class="danger">Unassign</button>
+    </form>
+  </li>
+  {% else %}
+  <li class="empty muted">No assignments yet.</li>
+  {% endfor %}
+</ul>
+{% endblock %}

src/tag_albums/templates/tag.html

@@ -0,0 +1,16 @@
+{% extends "base.html" %}
+{% block title %}{{ tag.tag_id }} · Tag albums{% endblock %}
+{% block body %}
+<p><a href="/">← All tags</a></p>
+<h1><code>{{ tag.tag_id }}</code></h1>
+<p class="muted">
+  {% if tag.assigned %}Currently plays <strong>{{ tag.album }}</strong> · {{ tag.artist }}.{% else %}No album assigned yet.{% endif %}
+</p>
+
+<h2>Search the music library</h2>
+<input type="search" name="q" placeholder="Album or artist…" autofocus autocomplete="off"
+       hx-get="/tags/{{ tag.tag_id }}/search" hx-trigger="input changed delay:400ms, search"
+       hx-target="#results" hx-indicator="#searching">
+<p id="searching" class="muted htmx-indicator">Searching…</p>
+<div id="results"></div>
+{% endblock %}