Everything in Hi Beatz,
as clean JSON.

#!/usr/bin/env python3
"""
hb_token.py - reads your Hi Beatz gamer ID and session token from a phone
connected over USB, and prints the command that turns them into an API key.

    python3 hb_token.py

The script only reads. Nothing is written to the phone, nothing is
installed, and the game is not modified. No root required.

What it needs
    Python 3.8 or newer
    adb from Google's Android platform tools, on your PATH
    A phone with USB debugging enabled and this computer authorised

Useful options
    --gamer-id <n>   your ID from the game profile; makes the search exact
    --serial <id>    pick a device when more than one is attached
    --adb <path>     adb is not on your PATH
    --file <path>    parse a save file you already have, skip the phone
    --describe       describe the save format without writing anything
    --json           machine readable output
    --verbose        show every extraction attempt

Nothing is written to disk. The save file is held in memory, read, and
dropped when the script exits.

adb is not downloaded either. It has to be installed already; the script
looks on your PATH and in the usual Android Studio locations.
"""

from __future__ import annotations

import argparse
import json
import re
import shutil
import struct
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path

PACKAGE = "com.funtapx.magic.piano.rhythm.tiles.music.game"
SAVE_NAME = "GlobalData.bytes"
API_BASE = "https://api.beatz-nodefox.de"

# Game IDs and tokens observed so far are 8 to 10 digits. Anything outside
# that range is almost certainly a coincidence in the binary.
PLAUSIBLE = range(10_000_000, 4_000_000_000)


# ──────────────────────────────────────────────────────────────── output

class Out:
    """Small console helper. Colours are dropped when piped to a file."""

    def __init__(self, colour: bool, verbose: bool, quiet: bool):
        self.colour = colour
        self.verbose = verbose
        self.quiet = quiet

    def _c(self, code: str, text: str) -> str:
        return f"\033[{code}m{text}\033[0m" if self.colour else text

    def head(self, text: str) -> None:
        if not self.quiet:
            print(self._c("1", text))

    def info(self, label: str, value: str) -> None:
        if not self.quiet:
            print(f"{label:<12}{value}")

    def step(self, text: str) -> None:
        if self.verbose and not self.quiet:
            print(self._c("2", f"  · {text}"))

    def warn(self, text: str) -> None:
        print(self._c("33", f"warning: {text}"), file=sys.stderr)

    def fail(self, text: str, *hints: str) -> None:
        print(self._c("31", f"error: {text}"), file=sys.stderr)
        for h in hints:
            print(f"  {h}", file=sys.stderr)
        sys.exit(1)


# ──────────────────────────────────────────────────────────────── adb

@dataclass
class Adb:
    binary: str
    serial: str | None = None
    out: Out = field(default=None, repr=False)

    def run(self, *args: str, timeout: int = 30) -> subprocess.CompletedProcess:
        cmd = [self.binary]
        if self.serial:
            cmd += ["-s", self.serial]
        cmd += list(args)
        self.out.step("adb " + " ".join(args))
        return subprocess.run(cmd, capture_output=True, timeout=timeout)

    def devices(self) -> list[tuple[str, str]]:
        """[(serial, state)] for everything adb currently sees."""
        proc = subprocess.run([self.binary, "devices"], capture_output=True, timeout=20)
        found = []
        for line in proc.stdout.decode("utf-8", "replace").splitlines()[1:]:
            parts = line.split()
            if len(parts) >= 2:
                found.append((parts[0], parts[1]))
        return found


# Where to look for adb. Two situations cover almost everyone: it is
# installed and on PATH, or the platform tools were unpacked somewhere and
# this script was dropped next to adb. Both work without any arguments.
def adb_search_paths() -> list[Path]:
    home = Path.home()
    exe = "adb.exe" if sys.platform == "win32" else "adb"

    roots = [
        Path(__file__).resolve().parent,          # script dropped into platform-tools
        Path.cwd(),                               # or run from that folder
        home / "Android/Sdk/platform-tools",
        home / "Library/Android/sdk/platform-tools",       # macOS, Android Studio
        home / "AppData/Local/Android/Sdk/platform-tools", # Windows, Android Studio
        Path("/usr/lib/android-sdk/platform-tools"),       # Debian, Ubuntu
        Path("/opt/android-sdk/platform-tools"),
        Path("/usr/local/share/android-sdk/platform-tools"),
        Path("C:/Android/platform-tools"),
        Path("C:/platform-tools"),
        Path(__file__).resolve().parent / "platform-tools",
        Path.cwd() / "platform-tools",
    ]
    return [r / exe for r in roots]


def install_hint() -> list[str]:
    """The one command that installs adb on the system we are running on."""
    if sys.platform == "darwin":
        return ["Install it with Homebrew:",
                "  brew install --cask android-platform-tools"]
    if sys.platform == "win32":
        return ["Install it with winget:",
                "  winget install Google.PlatformTools",
                "or unzip the platform tools and run this script from that folder."]
    return ["Install it from your package manager:",
            "  sudo apt install android-sdk-platform-tools     (Debian, Ubuntu)",
            "  sudo dnf install android-tools                  (Fedora)",
            "  sudo pacman -S android-tools                    (Arch)"]


def locate_adb(explicit: str | None, out: Out) -> str:
    if explicit:
        if Path(explicit).is_file():
            return explicit
        out.fail(f"no adb at {explicit}")

    found = shutil.which("adb")
    if found:
        out.step(f"adb found on PATH: {found}")
        return found

    for candidate in adb_search_paths():
        if candidate.is_file():
            out.step(f"adb found at {candidate}")
            return str(candidate)

    out.fail(
        "adb not found",
        "This script does not download anything; adb has to be on the machine",
        "already. It ships in Google's Android platform tools.",
        "",
        *install_hint(),
        "",
        "Or download the zip, unpack it, and point the script at it:",
        "  https://developer.android.com/tools/releases/platform-tools",
        "  python3 hb_token.py --adb /path/to/platform-tools/adb",
    )


def pick_device(adb: Adb, out: Out) -> str:
    try:
        devices = adb.devices()
    except FileNotFoundError:
        out.fail(f"could not run {adb.binary}")
    except subprocess.TimeoutExpired:
        out.fail("adb did not respond", "Unplug the cable, plug it back in and retry.")

    ready = [s for s, state in devices if state == "device"]
    waiting = [s for s, state in devices if state == "unauthorized"]

    if waiting and not ready:
        out.fail(
            "the phone has not authorised this computer",
            "Look at the phone screen and confirm the USB debugging dialog.",
            "Tick 'Always allow from this computer' so it stops asking.",
        )

    if not ready:
        out.fail(
            "no phone found",
            "Check that USB debugging is on under Developer options.",
            "Try another cable; some charging cables carry no data lines.",
            "Then run: adb devices",
        )

    if len(ready) > 1 and not adb.serial:
        out.fail(
            f"{len(ready)} devices attached",
            "Pick one with --serial, for example:",
            f"  python3 hb_token.py --serial {ready[0]}",
        )

    return adb.serial or ready[0]


# ────────────────────────────────────────────────────────── extraction

def read_save(adb: Adb, out: Out) -> bytes:
    """
    Tries the known ways of reading the save file, in order of likelihood.

    Which one works depends on the Android version and how the game was
    built, so instead of assuming, every route is attempted and the first
    one that returns something usable wins.
    """
    attempts = [
        ("app private storage via run-as",
         ("exec-out", "run-as", PACKAGE, "cat", f"files/{SAVE_NAME}")),
        ("app folder on shared storage",
         ("exec-out", "cat", f"/sdcard/Android/data/{PACKAGE}/files/{SAVE_NAME}")),
        ("app files folder on shared storage",
         ("exec-out", "cat", f"/storage/emulated/0/Android/data/{PACKAGE}/files/{SAVE_NAME}")),
        ("run-as with shared storage path",
         ("exec-out", "run-as", PACKAGE, "cat",
          f"/sdcard/Android/data/{PACKAGE}/files/{SAVE_NAME}")),
    ]

    errors = []
    for label, args in attempts:
        out.step(f"trying {label}")
        try:
            proc = adb.run(*args)
        except subprocess.TimeoutExpired:
            errors.append(f"{label}: timed out")
            continue

        data = proc.stdout
        # run-as prints its complaint on stdout on some builds, so a short
        # answer that looks like text is treated as a failure, not as data.
        if len(data) > 64 and not data.lstrip()[:1].isalpha():
            out.step(f"got {len(data)} bytes from {label}")
            return data
        if len(data) > 512:
            out.step(f"got {len(data)} bytes from {label}")
            return data

        reason = (proc.stderr or data).decode("utf-8", "replace").strip()
        errors.append(f"{label}: {reason[:120] or 'empty response'}")

    out.fail(
        "could not read the save file",
        *[f"- {e}" for e in errors],
        "",
        "Newer Android versions block access to app folders. If none of the",
        "routes worked, copy the file off the phone yourself and pass it in:",
        f"  python3 hb_token.py --file {SAVE_NAME}",
    )


# ─────────────────────────────────────────────────────────────── parsing

@dataclass
class Credentials:
    gamer_id: int
    token: int
    how: str


def parse_text(data: bytes, out: Out) -> Credentials | None:
    """
    Unencrypted saves keep their fields as readable text. Look for the
    field names the game uses, in any of the spellings seen so far.
    """
    text = data.decode("utf-8", "ignore")

    id_names = ("gamerId", "gamer_id", "playerId", "player_id", "uid")
    tk_names = ("token", "sessionId", "session_id", "loginToken")

    def grab(names: tuple[str, ...]) -> int | None:
        for name in names:
            m = re.search(rf'"{name}"\s*[:=]\s*"?(\d{{6,12}})"?', text)
            if m:
                value = int(m.group(1))
                if value in PLAUSIBLE:
                    out.step(f"text match: {name} = {value}")
                    return value
        return None

    gamer_id = grab(id_names)
    token = grab(tk_names)

    if gamer_id and token:
        return Credentials(gamer_id, token, "field names in the save file")
    return None


def find_anchored(data: bytes, gamer_id: int, out: Out) -> Credentials | None:
    """
    The reliable route: you read your own ID off your game profile and
    pass it in. We locate it in the file and take the 32 bit value next
    to it as the token, which is how the game stores the pair.
    """
    needle = struct.pack("<I", gamer_id)
    at = data.find(needle)

    while at != -1:
        for label, off in (("after", at + 4), ("before", at - 4)):
            if 0 <= off <= len(data) - 4:
                (value,) = struct.unpack_from("<I", data, off)
                if value in PLAUSIBLE and value != gamer_id:
                    out.step(f"anchor at 0x{at:x}, token {label} it: {value}")
                    return Credentials(
                        gamer_id, value,
                        f"gamer ID you supplied, token stored {label} it at 0x{off:x}",
                    )
        at = data.find(needle, at + 1)

    # Text form of the same idea, for readable saves.
    m = re.search(rf"{gamer_id}\D{{1,20}}?(\d{{6,12}})", data.decode("utf-8", "ignore"))
    if m and int(m.group(1)) in PLAUSIBLE:
        return Credentials(gamer_id, int(m.group(1)),
                           "gamer ID you supplied, token found next to it as text")
    return None


def parse_binary(data: bytes, out: Out) -> Credentials | None:
    """
    Last resort without an anchor: look for two plausible 32 bit values
    side by side, on aligned offsets only. Save files are not random, so
    a single hit is trustworthy; several hits are not, and we say so
    instead of picking one.
    """
    hits: list[tuple[int, int, int]] = []

    for off in range(0, len(data) - 8, 4):
        a, b = struct.unpack_from("<II", data, off)
        if a in PLAUSIBLE and b in PLAUSIBLE and a != b:
            hits.append((off, a, b))

    for off, a, b in hits[:12]:
        out.step(f"binary candidate at 0x{off:x}: {a}, {b}")

    if len(hits) == 1:
        off, a, b = hits[0]
        return Credentials(a, b, f"adjacent 32 bit values at offset 0x{off:x}")

    if len(hits) > 1:
        out.warn(f"{len(hits)} possible pairs; too ambiguous to choose one.")
    return None


def parse(data: bytes, gamer_id: int | None, out: Out) -> Credentials:
    if gamer_id is not None:
        found = find_anchored(data, gamer_id, out)
        if found:
            return found
        out.fail(
            f"gamer ID {gamer_id} does not appear in the save file",
            "Check the number on your game profile, or drop --gamer-id to",
            "let the script search on its own.",
        )

    for strategy in (parse_text, parse_binary):
        found = strategy(data, out)
        if found:
            return found

    out.fail(
        "the save file was read but the values could not be identified",
        f"file size: {len(data)} bytes",
        "",
        "Fastest fix: your gamer ID is shown on your profile in the game.",
        "Pass it in and the token is found next to it:",
        "  python3 hb_token.py --gamer-id 100000001",
        "",
        "If that also fails, the save format has changed. Run",
        "  python3 hb_token.py --describe",
        "and report that description; it contains no token.",
    )


# ───────────────────────────────────────────────────────────── diagnosis

def describe(data: bytes, out: Out) -> None:
    """
    Prints the shape of the save file so an unrecognised format can be
    reported without sending the file itself, which holds a live token.
    """
    printable = sum(1 for b in data[:4096] if 32 <= b < 127 or b in (9, 10, 13))
    ratio = printable / min(len(data), 4096) if data else 0

    print()
    print("save file description")
    print(f"  size          {len(data)} bytes")
    print(f"  first bytes   {data[:16].hex(' ')}")
    print(f"  looks like    {'text' if ratio > 0.85 else 'binary'} "
          f"({ratio:.0%} printable)")

    for marker, what in ((b"ES3", "Easy Save 3"), (b"{", "JSON"), (b"PK", "zip")):
        if data[:8].lstrip().startswith(marker):
            print(f"  format hint   {what}")
            break

    keys = sorted(set(re.findall(rb'"([A-Za-z_][A-Za-z0-9_]{2,24})"\s*[:=]', data[:65536])))
    if keys:
        names = ", ".join(k.decode("ascii", "ignore") for k in keys[:25])
        print(f"  field names   {names}")
    print()
    print("  Paste this description when reporting an unrecognised format.")
    print("  It contains no token and no personal data.")
    print()


# ────────────────────────────────────────────────────────────────── main

def main() -> None:
    ap = argparse.ArgumentParser(
        prog="hb_token.py",
        description="Read your Hi Beatz gamer ID and session token over USB.",
    )
    ap.add_argument("--serial", help="device serial, when several are attached")
    ap.add_argument("--adb", help="path to adb, if it is not on your PATH")
    ap.add_argument("--file", help="parse an existing save file instead of the phone")
    ap.add_argument("--describe", action="store_true",
                    help="describe the save file without writing anything anywhere")
    ap.add_argument("--gamer-id", type=int, metavar="N",
                    help="your ID from the game profile; makes the search exact")
    ap.add_argument("--json", action="store_true", help="print JSON instead of text")
    ap.add_argument("--verbose", action="store_true", help="show every attempt")
    args = ap.parse_args()

    out = Out(
        colour=sys.stdout.isatty() and not args.json,
        verbose=args.verbose,
        quiet=args.json,
    )

    out.head("Hi Beatz - token reader")

    # ── get the save file ────────────────────────────────────────────
    if args.file:
        try:
            with open(args.file, "rb") as fh:
                data = fh.read()
        except OSError as exc:
            out.fail(f"could not read {args.file}: {exc}")
        out.info("source", args.file)
    else:
        adb_path = locate_adb(args.adb, out)
        adb = Adb(adb_path, args.serial, out)
        serial = pick_device(adb, out)
        adb.serial = serial
        out.info("device", serial)
        data = read_save(adb, out)

    if args.describe:
        describe(data, out)

    # ── find the values ──────────────────────────────────────────────
    creds = parse(data, args.gamer_id, out)

    if args.json:
        print(json.dumps({
            "gamer_id": creds.gamer_id,
            "token": creds.token,
            "found_by": creds.how,
        }, indent=2))
        return

    out.info("gamer_id", str(creds.gamer_id))
    out.info("token", str(creds.token))
    if args.verbose:
        out.info("found by", creds.how)

    body = json.dumps({"gamer_id": creds.gamer_id, "token": creds.token})
    print()
    print("# paste this to create your key:")
    print(f"curl -X POST {API_BASE}/v1/auth \\")
    print("  -H 'Content-Type: application/json' \\")
    print(f"  -d '{body}'")
    print()
    print("# the token expires the next time you log in on your phone.")


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(130)