#!/usr/bin/env python3 """ Memory Lane FM — Discord Rich Presence (gateway edition) ----------------------------------------------------------- Sets your Discord Rich Presence directly over Discord's gateway, using your Discord account token — the same underlying mechanism apps like Kizzy use. Unlike the local-RPC companion script, this does NOT need a local Discord desktop client running at all: it authenticates as your account over the network, so it works identically whether Discord is open on desktop, Android, iOS, or nowhere — one script, every platform, including a headless server if you'd rather not keep a personal device running it. *** READ BEFORE USING *** This uses your real Discord ACCOUNT token, not a scoped/revocable API token — the credential that can fully act as you, not just set a status. Automating a user account this way ("self-botting") is against Discord's Terms of Service. Enforcement is understood to target automation abuse (mass messaging, scraping, spam) rather than presence-only use, but it isn't zero-risk, and it's your account and your call, not something this script can make risk-free. Treat your account token exactly like your password — anyone who gets it controls your whole account. Getting your Discord account token — three ways, in order of trust: 1. RECOMMENDED — real browser login (default, no setup). If you don't set DISCORD_TOKEN, this script opens an actual Chromium window pointed at discord.com's real login page. You log in there directly — password, 2FA, any CAPTCHA challenge are all handled by Discord's own page, never by this script. Once you're logged in, the script captures the resulting token from your browser's own authenticated requests and closes the window. Needs: pip install playwright playwright install chromium The captured token is cached locally afterward (see MLFM_DISCORD_TOKEN_CACHE below) so you don't need to re-login every run — delete that file to force a fresh login. 2. Reuse a token you already have from somewhere else (e.g. Kizzy) — set DISCORD_TOKEN / MLFM_DISCORD_TOKEN and this script skips the browser entirely. 3. Memory Lane FM's own site can also do this login exchange for you server-side (see the "Log in with Discord" option in your profile panel) — convenient, but your password briefly passes through the station's server code to make that one request, unlike option 1 where it only ever touches Discord's real page. Not what this script uses; mentioned here so you know it's the less-local alternative if you'd rather not run anything at all. Setup: 1. pip install websockets requests (add: pip install playwright && playwright install chromium — only if you want option 1 above; skip if you already have a token) 2. Get your Memory Lane FM API token the usual way — "Copy My API Token" wherever it shows on the site for your account. 3. Get a Discord Application ID + upload a large_image asset, same as the desktop RPC script (DISCORD_CLIENT_ID / LARGE_IMAGE_KEY below). 4. Fill in MLFM_TOKEN and DISCORD_CLIENT_ID below (or as env vars MLFM_TOKEN / MLFM_DISCORD_CLIENT_ID). Leave DISCORD_TOKEN blank to use the browser-login flow, or fill it in if you already have one. 5. Run: python3 mlfm_discord_presence_gateway.py On Android/iOS, run it inside a Python environment app (Termux, Pyto, a-Shell, etc.) — same script, same command. (Browser login needs a device that can actually display a Chromium window, so on a headless/mobile-terminal setup, use option 2 or 3 instead.) """ import asyncio import json import os import sys import time import logging import requests try: import websockets except ImportError: sys.exit("Missing dependency: pip install websockets") # ── Config — fill these in, or set as environment variables ──────────────── SERVER_URL = os.environ.get("MLFM_SERVER_URL", "https://memorylane.un-nerv.org/") MLFM_TOKEN = os.environ.get("MLFM_TOKEN", "") DISCORD_TOKEN = os.environ.get("MLFM_DISCORD_TOKEN", "") # your Discord ACCOUNT token — leave blank to log in via browser instead DISCORD_CLIENT_ID = os.environ.get("MLFM_DISCORD_CLIENT_ID", "") LARGE_IMAGE_KEY = "memorylane_logo_1024" # must match an uploaded Rich Presence art asset name # Where a browser-obtained token gets cached so you don't have to log in # every run. Just as sensitive as the token itself — written with 0600 # permissions, but treat it like a password file, and delete it any time # you want to force a fresh login (or log out of that Discord session). TOKEN_CACHE_PATH = os.environ.get("MLFM_DISCORD_TOKEN_CACHE", "mlfm_discord_gateway_token.txt") GATEWAY_URL = "wss://gateway.discord.gg/?v=10&encoding=json" POLL_INTERVAL_SECS = 15 MIN_RECONNECT_DELAY = 5 MAX_RECONNECT_DELAY = 60 BROWSER_LOGIN_TIMEOUT_SECS = 600 # 10 minutes to finish logging in (2FA/captcha take time) logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) log = logging.getLogger("mlfm-presence-gw") def _load_cached_token() -> str: if os.path.exists(TOKEN_CACHE_PATH): try: with open(TOKEN_CACHE_PATH, "r", encoding="utf-8") as f: return f.read().strip() except OSError: pass return "" def _save_cached_token(token: str): try: fd = os.open(TOKEN_CACHE_PATH, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(token) log.info(f"Token cached locally at {TOKEN_CACHE_PATH} (0600) — delete that file to force a fresh login next run.") except OSError as e: log.warning(f"Couldn't cache token to disk ({e}) — you'll need to log in again next run.") async def _browser_login() -> str: """Opens a real Chromium window at Discord's actual login page and waits for you to log in directly with Discord — password, 2FA, and any CAPTCHA challenge are handled entirely by Discord's own page, never by this script. Captures the resulting account token by watching for it on your browser's own authenticated API requests once login succeeds, then closes the window. Best-effort: this watches for an 'authorization' header on requests to discord.com/api, which is how Discord's real web client sends its token on every call once you're logged in. If Discord changes how their web client authenticates, this may need updating — that's a Discord-side change, not something under this script's control.""" try: from playwright.async_api import async_playwright except ImportError: sys.exit("Missing dependency for browser login: pip install playwright (then: playwright install chromium)") log.info("Opening a browser window — log in to Discord normally. " "Password, 2FA, and any CAPTCHA are all handled by Discord's own page.") captured = {"token": None} def _on_request(request): if captured["token"]: return auth = request.headers.get("authorization") if auth and "discord.com/api" in request.url: captured["token"] = auth async with async_playwright() as p: browser = await p.chromium.launch(headless=False) page = await browser.new_page() page.on("request", _on_request) await page.goto("https://discord.com/login") waited = 0 while not captured["token"] and waited < BROWSER_LOGIN_TIMEOUT_SECS: await asyncio.sleep(1) waited += 1 await browser.close() if not captured["token"]: sys.exit("Didn't capture a token — the login window closed, or timed out, before finishing login.") log.info("Logged in — captured your Discord token.") return captured["token"] def fetch_presence_status() -> dict | None: """Same generic-token presence check as the RPC companion script — one MLFM_TOKEN works no matter whose account it belongs to.""" try: resp = requests.get( f"{SERVER_URL.rstrip('/')}/api/user/presence", headers={"X-Token": MLFM_TOKEN}, timeout=10, ) if resp.status_code == 401: log.error("Server rejected MLFM_TOKEN (401). Copy a fresh one from the site.") return None resp.raise_for_status() return resp.json() except requests.RequestException as e: log.warning(f"Couldn't reach server: {e}") return None def _safe_rp_string(value: str, fallback: str) -> str: """Discord Rich Presence requires details/state to be 2-128 chars.""" value = (value or "").strip() if len(value) < 2: return fallback if len(value) > 128: return value[:125] + "..." return value def _state_for(status: dict, default: str) -> str: if status.get("is_admin"): return "Hosting Memory Lane FM" if status.get("is_asuka") and status.get("asuka_live"): return "Live on air" if status.get("is_static") and status.get("static_live"): return "Live on air" if status.get("is_nightmare") and status.get("nightmare_live"): return "Live on air" return default # Presences set over the gateway (what this script does, as a user # account announcing its own activity) need an asset's raw numeric ID # in assets.large_image, NOT its human-readable key name — that # name-to-ID resolution only happens automatically through the official # local-RPC/IPC path (what the desktop companion script uses via # pypresence). This fetches Discord's public asset list for the # configured Application once and caches it, rather than sending the # plain name Discord's gateway path won't understand. _ASSET_CACHE = {"id": None, "fetched": False} def _resolve_asset_id() -> str | None: if _ASSET_CACHE["fetched"]: return _ASSET_CACHE["id"] _ASSET_CACHE["fetched"] = True try: resp = requests.get( f"https://discord.com/api/v9/oauth2/applications/{DISCORD_CLIENT_ID}/assets", timeout=10, ) resp.raise_for_status() for asset in resp.json(): if isinstance(asset, dict) and asset.get("name") == LARGE_IMAGE_KEY: _ASSET_CACHE["id"] = asset.get("id") break if not _ASSET_CACHE["id"]: log.warning(f"Couldn't find an asset named '{LARGE_IMAGE_KEY}' under this Application ID — " "presence will show without the art asset. Double-check DISCORD_CLIENT_ID and " "that the asset is uploaded under that exact Application.") except requests.RequestException as e: log.warning(f"Couldn't fetch asset list from Discord ({e}) — presence will show without the art asset.") return _ASSET_CACHE["id"] def _activity_payload(status: dict | None, start_ms: int) -> dict | None: """Build the 'activities' entry for the gateway PRESENCE_UPDATE op, or None if nothing should be shown right now. start_ms is passed in rather than computed here so it can stay fixed across a whole continuous listening session instead of resetting Discord's shown elapsed time every poll.""" if not status or not status.get("listening"): return None track = _safe_rp_string(status.get("now_playing"), "Memory Lane FM") show = _state_for(status, _safe_rp_string(status.get("show"), "Memory Lane FM")) activity = { "name": "Memory Lane FM", "type": 2, # Listening. EXPERIMENT: paired with a party field below — # a plain Listening activity (no party) was confirmed to # drop the large_image art asset over the desktop RPC # path, and a party (even fake) is a known trick to get # Discord's client to render it as a normal activity # instead of assuming Spotify-specific handling. If this # still drops the art, revert type back to 0 (Playing). "application_id": DISCORD_CLIENT_ID, "details": track, "state": show, "timestamps": {"start": start_ms}, "party": {"id": "mlfm-listening", "size": [1, 1]}, } asset_id = _resolve_asset_id() if asset_id: activity["assets"] = {"large_image": asset_id, "large_text": "Memory Lane FM"} return activity async def _heartbeat_loop(ws, interval_secs: float): while True: await asyncio.sleep(interval_secs) await ws.send(json.dumps({"op": 1, "d": None})) async def _presence_loop(ws): """Polls Memory Lane FM and pushes a PRESENCE_UPDATE (op 3) only when what we'd show actually changes — not on every poll tick. Sending an identical update every cycle forever, for a long-running connection, both resets Discord's shown elapsed-time clock every cycle and risks tripping Discord's gateway-level rate limit on presence updates, which can leave a connection unable to push further updates — including a legitimate "resumed listening" one — until it reconnects. Only sending on real change avoids both.""" last_key = None # (track, show) of what we last actually sent, or None if cleared session_start_ms = None while True: status = fetch_presence_status() listening = bool(status) and bool(status.get("listening")) if listening: track = _safe_rp_string(status.get("now_playing"), "Memory Lane FM") show = _state_for(status, _safe_rp_string(status.get("show"), "Memory Lane FM")) key = (track, show) if key != last_key: if last_key is None: session_start_ms = int(time.time() * 1000) # only reset when (re)starting a session log.info(f"Presence updated: {track} — {show}") activity = _activity_payload(status, session_start_ms) await ws.send(json.dumps({ "op": 3, "d": {"status": "online", "afk": False, "since": 0, "activities": [activity] if activity else []}, })) last_key = key else: if last_key is not None: await ws.send(json.dumps({ "op": 3, "d": {"status": "online", "afk": False, "since": 0, "activities": []}, })) log.info("Cleared presence (not listening).") last_key = None session_start_ms = None await asyncio.sleep(POLL_INTERVAL_SECS) async def _run_session(): """One gateway connection lifecycle: connect, IDENTIFY, run heartbeat + presence loops until the connection drops. No session RESUME on reconnect — always a fresh IDENTIFY, which costs a brief presence flicker on reconnect but keeps this far simpler and more robust than implementing RESUME/sequence tracking.""" async with websockets.connect(GATEWAY_URL, max_size=None) as ws: hello = json.loads(await ws.recv()) heartbeat_interval = hello["d"]["heartbeat_interval"] / 1000 await ws.send(json.dumps({ "op": 2, "d": { "token": DISCORD_TOKEN, "properties": {"os": "linux", "browser": "chrome", "device": ""}, "presence": {"status": "online", "afk": False, "since": 0, "activities": []}, }, })) log.info("Connected to Discord gateway.") hb_task = asyncio.create_task(_heartbeat_loop(ws, heartbeat_interval)) pres_task = asyncio.create_task(_presence_loop(ws)) try: async for _msg in ws: pass # we only need to keep reading so the connection stays alive finally: hb_task.cancel() pres_task.cancel() async def run(): global DISCORD_TOKEN if not MLFM_TOKEN: sys.exit("Set MLFM_TOKEN (env var) or fill in MLFM_TOKEN in the script.") if not DISCORD_CLIENT_ID: sys.exit("Set MLFM_DISCORD_CLIENT_ID (env var) or fill in DISCORD_CLIENT_ID in the script.") if not DISCORD_TOKEN: DISCORD_TOKEN = _load_cached_token() if not DISCORD_TOKEN: DISCORD_TOKEN = await _browser_login() _save_cached_token(DISCORD_TOKEN) backoff = MIN_RECONNECT_DELAY while True: try: await _run_session() backoff = MIN_RECONNECT_DELAY # clean disconnect — reset backoff except (websockets.ConnectionClosed, OSError) as e: log.warning(f"Gateway connection lost ({e}). Reconnecting in {backoff}s.") except Exception as e: log.warning(f"Unexpected error ({e}). Reconnecting in {backoff}s.") await asyncio.sleep(backoff) backoff = min(backoff * 2, MAX_RECONNECT_DELAY) if __name__ == "__main__": try: asyncio.run(run()) except KeyboardInterrupt: print() log.info("Stopped.")