audioserver/bandcamp-api/app.py

#!/usr/bin/env python3
"""Bandcamp credential store + login-test micro-service."""

import json
import os
import re
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import unquote

import requests as _requests

CONFIG_FILE         = os.environ.get("CONFIG_FILE",         "/config/bandcamp.json")
SPOTIFY_CONFIG_FILE = os.environ.get("SPOTIFY_CONFIG_FILE", "/config/spotify.json")


# ── Config helpers ────────────────────────────────────────────────────────────

def _read():
    if os.path.exists(CONFIG_FILE):
        with open(CONFIG_FILE) as f:
            return json.load(f)
    return {}


def _write(data: dict):
    os.makedirs(os.path.dirname(CONFIG_FILE) or ".", exist_ok=True)
    with open(CONFIG_FILE, "w") as f:
        json.dump(data, f, indent=2)


# ── Spotify credential helpers ────────────────────────────────────────────────

def _spotify_read() -> dict:
    if os.path.exists(SPOTIFY_CONFIG_FILE):
        with open(SPOTIFY_CONFIG_FILE) as f:
            return json.load(f)
    return {}


def _spotify_write(data: dict):
    os.makedirs(os.path.dirname(SPOTIFY_CONFIG_FILE) or ".", exist_ok=True)
    with open(SPOTIFY_CONFIG_FILE, "w") as f:
        json.dump(data, f, indent=2)


# ── Bandcamp cookie auth ─────────────────────────────────────────────────────

def _apply_cookies(session, cookie_str: str):
    for part in cookie_str.split(';'):
        part = part.strip()
        if '=' in part:
            name, _, value = part.partition('=')
            session.cookies.set(name.strip(), value.strip(), domain='.bandcamp.com')


def _bc_verify_cookies(cookie_str: str) -> dict:
    session = _requests.Session()
    session.headers["User-Agent"] = (
        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/124.0 Safari/537.36"
    )
    _apply_cookies(session, cookie_str)

    # Try js_account_details cookie first (avoids an HTTP round-trip)
    fan_id = None
    username = ''
    js_raw = session.cookies.get('js_account_details', '')
    if js_raw:
        try:
            data = json.loads(unquote(js_raw))
            fan_id = data.get('fan_id') or data.get('id')
            username = data.get('username', '')
        except Exception:
            pass

    # Fall back to fetching the homepage
    if not fan_id:
        try:
            r = session.get("https://bandcamp.com/", timeout=15)
            m = re.search(r'"fan_id"\s*:\s*(\d+)', r.text)
            if m:
                fan_id = int(m.group(1))
            if not username:
                m2 = re.search(r'"username"\s*:\s*"([^"]+)"', r.text)
                if m2:
                    username = m2.group(1)
        except Exception as exc:
            return {"ok": False, "error": f"Could not reach bandcamp.com: {exc}"}

    if not fan_id:
        return {"ok": False, "error": "Cookies invalid or expired — no fan account found"}

    # Verify with a lightweight API call
    try:
        r = session.post(
            "https://bandcamp.com/api/fan/2/collection_summary",
            json={"fan_id": fan_id},
            timeout=15,
        )
        if not r.ok:
            return {"ok": False, "error": "Cookies rejected by Bandcamp (may have expired)"}
    except Exception as exc:
        return {"ok": False, "error": str(exc)}

    return {"ok": True, "username": username}


# ── Bandcamp login ────────────────────────────────────────────────────────────

def _bc_login(email: str, password: str) -> dict:
    session = _requests.Session()
    session.headers["User-Agent"] = (
        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/124.0 Safari/537.36"
    )

    # Step 1 — obtain CSRF token from meta tag
    try:
        r = session.get("https://bandcamp.com/login", timeout=15)
    except Exception as exc:
        return {"ok": False, "error": f"Could not reach bandcamp.com: {exc}"}

    m = re.search(r'<meta\s+name=["\']csrf-token["\']\s+content=["\']([^"\']+)["\']', r.text, re.IGNORECASE)
    if not m:
        m = re.search(r'<meta\s+content=["\']([^"\']+)["\']\s+name=["\']csrf-token["\']', r.text, re.IGNORECASE)
    if not m:
        return {"ok": False, "error": "Could not obtain CSRF token from Bandcamp"}
    csrf = m.group(1)

    # Step 2 — POST login using field names from Bandcamp's login JS
    try:
        r = session.post(
            "https://bandcamp.com/login_cb",
            data={
                "user.name": email,
                "login.password": password,
                "login.twofactor": "",
                "login.twofactor_remember": "",
                "login.from": "",
                "to_band_path": "",
            },
            headers={
                "X-CSRF-TOKEN": csrf,
                "Referer": "https://bandcamp.com/login",
                "Accept": "application/json, text/javascript, */*",
                "Origin": "https://bandcamp.com",
            },
            timeout=15,
        )
        if not r.text:
            return {"ok": False, "error": "Empty response from Bandcamp"}
        payload = r.json()
    except Exception as exc:
        return {"ok": False, "error": str(exc)}

    if payload.get("ok") == 1 or payload.get("account"):
        username = (payload.get("account") or {}).get("username", "")
        return {"ok": True, "username": username}

    # Extract error from various response shapes
    err = payload.get("error") or payload.get("error_message")
    if not err and payload.get("errors"):
        errors = payload["errors"]
        if isinstance(errors, list):
            parts = [f"{e.get('field','')}: {e.get('reason','')}" for e in errors if isinstance(e, dict)]
            err = "; ".join(parts) if parts else str(errors)
        else:
            err = str(errors)
    err = err or "Login failed"
    return {"ok": False, "error": err}


# ── HTTP handler ──────────────────────────────────────────────────────────────

class _Handler(BaseHTTPRequestHandler):
    def log_message(self, *_):
        pass

    def _json(self, status: int, data: dict):
        body = json.dumps(data).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Allow", "GET, POST, OPTIONS")
        self.end_headers()

    def do_GET(self):
        if self.path in ("/credentials", "/credentials/"):
            cfg = _read()
            self._json(200, {
                "email": cfg.get("email", ""),
                "password_set": bool(cfg.get("password")),
                "cookies_set": bool(cfg.get("cookies")),
            })
        elif self.path in ("/spotify/credentials", "/spotify/credentials/"):
            cfg = _spotify_read()
            self._json(200, {
                "client_id": cfg.get("client_id", ""),
                "client_secret_set": bool(cfg.get("client_secret")),
            })
        elif self.path in ("/health", "/health/"):
            self._json(200, {"ok": True})
        else:
            self._json(404, {"error": "not found"})

    def do_POST(self):
        length = min(int(self.headers.get("Content-Length", 0)), 65536)
        try:
            body = json.loads(self.rfile.read(length) or b"{}")
        except json.JSONDecodeError:
            self._json(400, {"error": "invalid JSON"})
            return

        if self.path in ("/credentials", "/credentials/"):
            cfg = _read()
            if "email" in body:
                cfg["email"] = body["email"].strip()
            if "password" in body and body["password"]:
                cfg["password"] = body["password"]
            if "cookies" in body:
                if body["cookies"]:
                    cfg["cookies"] = body["cookies"].strip()
                else:
                    cfg.pop("cookies", None)
            _write(cfg)
            self._json(200, {"ok": True})

        elif self.path in ("/test-login", "/test-login/"):
            cfg = _read()
            cookies = cfg.get("cookies", "")
            if cookies:
                result = _bc_verify_cookies(cookies)
                self._json(200, result)
                return
            email = cfg.get("email", "")
            password = cfg.get("password", "")
            if not email or not password:
                self._json(400, {"ok": False, "error": "No credentials saved — save cookies or email/password first"})
                return
            result = _bc_login(email, password)
            self._json(200, result)

        elif self.path in ("/spotify/credentials", "/spotify/credentials/"):
            cfg = _spotify_read()
            if "client_id" in body:
                cfg["client_id"] = body["client_id"].strip()
            if "client_secret" in body and body["client_secret"]:
                cfg["client_secret"] = body["client_secret"].strip()
            _spotify_write(cfg)
            self._json(200, {"ok": True})

        else:
            self._json(404, {"error": "not found"})


if __name__ == "__main__":
    port = int(os.environ.get("PORT", 8091))
    server = HTTPServer(("0.0.0.0", port), _Handler)
    print(f"bandcamp-api listening on :{port}", flush=True)
    server.serve_forever()