#!/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"

# The game keeps one save folder per account, named after the gamer ID,
# under shared storage. adb may read there without root, which is what
# makes this work on a normal phone.
SAVE_ROOT = f"/sdcard/Android/data/{PACKAGE}/files/Save"

# Tail of GlobalData.bytes:
#   <int32 playerId> <len> token <len> version
# Token and version are length prefixed ASCII, the ID is a plain int32.
SAVE_TAIL = re.compile(
    rb"(.{4})[\x01-\x40]([0-9]{4,12})[\x01-\x20](\d+(?:\.\d+){1,3})\s*$", re.S
)

# 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 list_accounts(adb: Adb, out: Out) -> tuple[list[str], str]:
    """
    Save folder names. Each is a gamer ID; "0" is the state before login.

    Returns the folders and why the list is empty when it is, because the
    two reasons need opposite advice: a blocked folder is a phone problem,
    a missing folder means the game was never logged into here.
    """
    try:
        proc = adb.run("shell", f"ls {SAVE_ROOT}")
    except subprocess.TimeoutExpired:
        return [], "timeout"

    text = (proc.stdout + proc.stderr).decode("utf-8", "replace")

    if "Permission denied" in text:
        return [], "denied"
    if "No such file" in text:
        return [], "missing"

    names = [n.strip() for n in text.split() if n.strip()]
    accounts = sorted((n for n in names if n.isdigit() and n != "0"), reverse=True)
    if "0" in names:
        accounts.append("0")          # try the pre-login state last
    return accounts, ("empty" if not accounts else "ok")


def read_save(adb: Adb, out: Out) -> tuple[bytes, str | None]:
    """
    Reads the save file over adb.

    Returns the file contents and, when it could be determined, the gamer ID
    taken from the folder name. exec-out is used rather than shell so the
    bytes arrive unaltered on Windows.
    """
    accounts, why = list_accounts(adb, out)

    if why == "denied":
        out.fail(
            "this phone does not let adb read the app folder",
            f"Blocked: {SAVE_ROOT}",
            "",
            "Some manufacturers lock /Android/data down beyond what stock",
            "Android does. The file is still there, it just cannot be reached",
            "this way. Copy it across on the phone itself and pass it in:",
            "",
            "  1. Open the phone's file manager",
            f"  2. Go to Android/data/{PACKAGE}/files/Save",
            "  3. Open the folder named after your gamer ID",
            "  4. Copy GlobalData.bytes to Downloads",
            "  5. adb pull /sdcard/Download/GlobalData.bytes",
            "     python3 hb_token.py --file GlobalData.bytes",
        )

    if why in ("missing", "empty"):
        out.fail(
            "no save folder for any account",
            f"Looked in {SAVE_ROOT}",
            "",
            "Log into the game and let it reach the main menu, then run this",
            "again. The folder is created on first login.",
            "",
            "To check by hand:",
            f"  adb shell ls {SAVE_ROOT}",
        )

    if not accounts:
        out.fail("could not list the save folder", f"Looked in {SAVE_ROOT}")

    out.step(f"save folders: {', '.join(accounts)}")

    problems = []
    for account in accounts:
        path = f"{SAVE_ROOT}/{account}/{SAVE_NAME}"
        out.step(f"reading {path}")
        try:
            proc = adb.run("exec-out", f"cat {path}", timeout=60)
        except subprocess.TimeoutExpired:
            problems.append(f"{account}: timed out")
            continue

        data = proc.stdout
        if len(data) > 32:
            out.step(f"got {len(data)} bytes from account {account}")
            return data, (account if account != "0" else None)

        reason = (proc.stderr or data).decode("utf-8", "replace").strip()
        problems.append(f"{account}: {reason[:100] or 'empty file'}")

    out.fail(
        "save folders exist but none of them could be read",
        *[f"- {p}" for p in problems],
    )


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

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


def parse_save_tail(data: bytes, out: Out) -> Credentials | None:
    """
    The normal case: the game writes ID, token and version at the very end
    of the save file, so match there first.
    """
    m = SAVE_TAIL.search(data)
    if not m:
        return None

    (gamer_id,) = struct.unpack("<i", m.group(1))
    token = int(m.group(2))
    version = m.group(3).decode("ascii", "ignore")
    out.step(f"save tail matched, game version {version}")

    if gamer_id > 0 and token in PLAUSIBLE:
        return Credentials(gamer_id, token, f"end of the save file (game {version})")
    return None


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_save_tail, 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 ────────────────────────────────────────────
    folder_id: int | None = None

    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, from_folder = read_save(adb, out)
        # The folder is named after the account, so it doubles as a check
        # and as a fallback if the file itself will not give up the ID.
        if from_folder and args.gamer_id is None:
            args.gamer_id = None          # let the parser try on its own first
            folder_id = int(from_folder)
        else:
            folder_id = None

    if args.describe:
        describe(data, out)

    # ── find the values ──────────────────────────────────────────────
    creds = parse(data, args.gamer_id or folder_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)