Yt Music

auth.json
.yt-music/
*.pyc
__pycache__/
.pytest_cache/
.DS_Store

[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts = -ra --strict-markers --tb=short
markers =
    integration: hits the real ytmusicapi or Playwright; skipped by default. Run with -m integration.
filterwarnings =
    error
    ignore::DeprecationWarning:http.server

#!/usr/bin/env python3
"""
YouTube Music CLI Helper
Auth: <skill-root>/.yt-music/auth.json by default
Usage: uv run --with ytmusicapi python scripts/helper.py <command> [args]
"""

import argparse
import hashlib
import json
import os
import tempfile
import sys
import time
from pathlib import Path
from typing import NoReturn


def _resolve_data_dir() -> Path:
    configured = os.environ.get("YT_MUSIC_DATA_DIR")
    if configured:
        return Path(configured).expanduser()
    return Path(__file__).resolve().parent.parent / ".yt-music"


SCRIPT_PATH = Path(__file__).resolve()
DATA_DIR = _resolve_data_dir()
AUTH_FILE = DATA_DIR / "auth.json"
AUTH_HEADER_KEYS = {"Authorization", "Cookie", "X-Goog-AuthUser", "x-origin"}


def auth_setup_instructions() -> dict:
    return {
        "required": True,
        "message": "Authentication is required before library, playlist, rating, upload, or account operations.",
        "agent_prompt": (
            "Authentication is required. Ask the user for either a Cookie string "
            "from a logged-in music.youtube.com request or a cookies JSON export file path."
        ),
        "language_policy": (
            "Mirror the user's language when replying. If the user's language is unclear, default to concise English."
        ),
        "reply_templates": {
            "short_en": (
                "You need to sign in to YouTube Music before I can continue with your "
                "library, playlists, account, uploads, or full playback.\n\n"
                "Please send me one of these:\n"
                "1. A Cookie string\n"
                "2. A cookies JSON file path"
            ),
            "cookie_string_en": (
                "Open a logged-in music.youtube.com page\n"
                "Open DevTools -> Network\n"
                "Filter /browse and reload the page\n"
                "Open any matching request\n"
                "Copy the Cookie request header value\n"
                "Send the full Cookie string back to me"
            ),
            "cookies_json_en": (
                "Use a cookie export extension such as Cookie-Editor on music.youtube.com\n"
                "Export cookies as JSON\n"
                "Save the exported file locally\n"
                "Send me the file path"
            ),
        },
        "cookie_string_steps": [
            "Open https://music.youtube.com in a logged-in browser session",
            "Open DevTools and go to the Network tab",
            "Filter requests with /browse and reload the page",
            "Open any matching request and copy the Cookie header value",
            "Send that Cookie string back to the agent",
        ],
        "cookies_json_steps": [
            "Open a cookie export extension such as Cookie-Editor on music.youtube.com",
            "Export cookies as JSON",
            "Save the JSON file locally",
            "Send the JSON file path back to the agent",
        ],
        "setup_commands": {
            "cookie_string": f"uv run --with ytmusicapi python {SCRIPT_PATH} auth setup --cookie '<cookie string>'",
            "cookies_json": f"uv run --with ytmusicapi python {SCRIPT_PATH} auth setup --cookies-file /path/to/cookies.json",
        },
    }


def _write_json(path: Path, data) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False, dir=path.parent) as tmp:
        json.dump(data, tmp, indent=2, ensure_ascii=False)
        tmp.write("\n")
    Path(tmp.name).replace(path)


def _sanitize_auth_headers(data: dict) -> dict:
    return {key: value for key, value in data.items() if key in AUTH_HEADER_KEYS and isinstance(value, str)}


def _migrate_legacy_auth_file() -> None:
    if not AUTH_FILE.exists():
        return
    try:
        raw = json.loads(AUTH_FILE.read_text())
    except Exception:
        return
    if not isinstance(raw, dict):
        return

    headers = _sanitize_auth_headers(raw)
    changed = set(raw) != set(headers)
    if changed:
        _write_json(AUTH_FILE, headers)


# ─── Auth ────────────────────────────────────────────────────────────────────

def build_auth_from_cookie(cookie_string: str) -> dict:
    """Compute SAPISIDHASH and build the auth headers dict from a raw cookie string."""
    sapisid = None
    for part in cookie_string.split(";"):
        part = part.strip()
        if part.startswith("SAPISID="):
            sapisid = part.split("=", 1)[1]
            break
    if not sapisid:
        # Try __Secure-3PAPISID as fallback
        for part in cookie_string.split(";"):
            part = part.strip()
            if part.startswith("__Secure-3PAPISID="):
                sapisid = part.split("=", 1)[1]
                break
    if not sapisid:
        bail("SAPISID not found in cookie string")

    ts = int(time.time())
    sha1 = hashlib.sha1(f"{ts} {sapisid} https://music.youtube.com".encode()).hexdigest()
    return {
        "Authorization": f"SAPISIDHASH {ts}_{sha1}",
        "Cookie": cookie_string.strip(),
        "X-Goog-AuthUser": "0",
        "x-origin": "https://music.youtube.com",
    }


def get_yt(require_auth=False):
    from ytmusicapi import YTMusic
    _migrate_legacy_auth_file()
    if AUTH_FILE.exists():
        return YTMusic(str(AUTH_FILE))
    if require_auth:
        details = {
            "error": "Auth required",
            "path": str(AUTH_FILE),
            **auth_setup_instructions(),
        }
        print(json.dumps(details, ensure_ascii=False, indent=2), file=sys.stderr)
        sys.exit(1)
    return YTMusic()  # unauthenticated — search/browse only


def bail(msg: str) -> NoReturn:
    print(json.dumps({"error": msg}, ensure_ascii=False), file=sys.stderr)
    sys.exit(1)


def out(data):
    print(json.dumps(data, ensure_ascii=False, indent=2))


# ─── Auth commands ────────────────────────────────────────────────────────────

def cmd_auth(args):
    if args.action == "check":
        _migrate_legacy_auth_file()
        if AUTH_FILE.exists():
            out({"status": "ok", "path": str(AUTH_FILE)})
        else:
            out({
                "status": "missing",
                "path": str(AUTH_FILE),
                **auth_setup_instructions(),
            })

    elif args.action == "setup":
        if getattr(args, "cookies_file", None):
            _import_cookies_json(args.cookies_file)
        else:
            cookie = args.cookie or sys.stdin.read().strip()
            if not cookie:
                bail("No cookie provided. Pass --cookie '<string>' or --cookies-file cookies.json")
            headers = build_auth_from_cookie(cookie)
            _write_json(AUTH_FILE, headers)
            out({"status": "saved", "path": str(AUTH_FILE)})

    elif args.action == "account":
        yt = get_yt(require_auth=True)
        out(yt.get_account_info())

    elif args.action == "remove":
        if AUTH_FILE.exists():
            AUTH_FILE.unlink()
            out({"status": "removed"})
        else:
            out({"status": "already_missing"})


def _import_cookies_json(path: str) -> None:
    """
    Import a full cookie JSON export (array of cookie objects from a browser extension
    such as Cookie-Editor or EditThisCookie) and derive API auth headers.

    Stores:
    - auth.json: Cookie / Authorization headers for ytmusicapi
    """
    p = Path(path)
    if not p.exists():
        bail(f"File not found: {path}")
    try:
        raw = json.loads(p.read_text())
        if not isinstance(raw, list):
            bail("Expected a JSON array of cookie objects (e.g. exported from Cookie-Editor)")
    except json.JSONDecodeError as e:
        bail(f"Could not parse JSON: {e}")

    cookie_parts: list = []
    for c in raw:
        if not isinstance(c, dict) or "name" not in c or "value" not in c:
            continue
        domain = str(c.get("domain", "")).strip()
        if not domain or not any(d in domain for d in ["youtube.com", "google.com"]):
            continue
        if "youtube.com" in domain:
            cookie_parts.append(f"{c['name']}={c['value']}")

    if not cookie_parts:
        bail("No YouTube/Google cookies found in the file")

    cookie_str = "; ".join(cookie_parts)
    try:
        auth_data: dict = build_auth_from_cookie(cookie_str)
    except SystemExit:
        auth_data = {"Cookie": cookie_str, "X-Goog-AuthUser": "0",
                     "x-origin": "https://music.youtube.com"}

    _write_json(AUTH_FILE, _sanitize_auth_headers(auth_data))
    out({"status": "saved", "path": str(AUTH_FILE), "cookies_imported": len(cookie_parts)})


# ─── Search ───────────────────────────────────────────────────────────────────

def cmd_search(args):
    yt = get_yt()
    kwargs = dict(limit=args.limit)
    if args.type:
        kwargs["filter"] = args.type
    if args.library:
        kwargs["scope"] = "library"
    results = yt.search(args.query, **kwargs)
    out(results)


def cmd_suggest(args):
    yt = get_yt()
    out(yt.get_search_suggestions(args.query))


# ─── Library ──────────────────────────────────────────────────────────────────

def cmd_library(args):
    yt = get_yt(require_auth=True)
    sub = args.sub
    limit = args.limit

    dispatch = {
        "songs":         lambda: yt.get_library_songs(limit=limit, order=args.order),
        "liked":         lambda: yt.get_liked_songs(limit=limit),
        "playlists":     lambda: yt.get_library_playlists(limit=limit),
        "albums":        lambda: yt.get_library_albums(limit=limit, order=args.order),
        "artists":       lambda: yt.get_library_artists(limit=limit, order=args.order),
        "subscriptions": lambda: yt.get_library_subscriptions(limit=limit),
        "history":       lambda: yt.get_history(),
        "uploads":       lambda: yt.get_library_upload_songs(limit=limit),
    }

    if sub in dispatch:
        out(dispatch[sub]())
    else:
        # Overview: return counts/names of playlists
        playlists = yt.get_library_playlists(limit=limit)
        out({"playlists": playlists})


# ─── Playlist ─────────────────────────────────────────────────────────────────

def cmd_playlist(args):
    yt = get_yt(require_auth=True)
    action = args.action

    if action == "get":
        if not args.playlist_id:
            bail("playlist_id required for 'get'")
        out(yt.get_playlist(args.playlist_id, limit=args.limit))

    elif action == "create":
        if not args.title:
            bail("--title required for 'create'")
        video_ids = args.video_ids or None
        pl_id = yt.create_playlist(
            args.title,
            args.description or "",
            privacy_status=args.privacy or "PRIVATE",
            video_ids=video_ids,
        )
        out({"playlistId": pl_id, "title": args.title, "privacy": args.privacy or "PRIVATE"})

    elif action == "edit":
        if not args.playlist_id:
            bail("playlist_id required for 'edit'")
        kwargs = {}
        if args.title:
            kwargs["title"] = args.title
        if args.description is not None:
            kwargs["description"] = args.description
        if args.privacy:
            kwargs["privacyStatus"] = args.privacy
        result = yt.edit_playlist(args.playlist_id, **kwargs)
        out({"status": result})

    elif action == "delete":
        if not args.playlist_id:
            bail("playlist_id required for 'delete'")
        result = yt.delete_playlist(args.playlist_id)
        out({"status": result})

    elif action == "add":
        if not args.playlist_id or not args.video_ids:
            bail("playlist_id and video_ids required for 'add'")
        result = yt.add_playlist_items(
            args.playlist_id, args.video_ids,
            duplicates=args.duplicates,
        )
        out(result if isinstance(result, dict) else {"status": result})

    elif action == "add-playlist":
        # Add all songs from another playlist
        if not args.playlist_id or not args.source_playlist:
            bail("playlist_id and --source-playlist required for 'add-playlist'")
        result = yt.add_playlist_items(
            args.playlist_id,
            source_playlist=args.source_playlist,
        )
        out(result if isinstance(result, dict) else {"status": result})

    elif action == "remove":
        if not args.playlist_id or not args.video_ids:
            bail("playlist_id and video_ids required for 'remove'")
        pl = yt.get_playlist(args.playlist_id, limit=None)
        to_remove = [t for t in pl.get("tracks", []) if t.get("videoId") in args.video_ids]
        if not to_remove:
            out({"status": "not_found", "removed": 0})
            return
        result = yt.remove_playlist_items(args.playlist_id, to_remove)
        out({"status": result, "removed": len(to_remove)})

    elif action == "rate":
        if not args.playlist_id:
            bail("playlist_id required for 'rate'")
        result = yt.rate_playlist(args.playlist_id, args.rating or "LIKE")
        out({"status": result})


# ─── Artist ───────────────────────────────────────────────────────────────────

def cmd_artist(args):
    yt = get_yt()
    data = yt.get_artist(args.browse_id)
    out(data)


def cmd_artist_albums(args):
    yt = get_yt()
    # get_artist_albums needs channelId + params from get_artist response
    artist = yt.get_artist(args.browse_id)
    albums_data = artist.get("albums", {})
    browse_params = albums_data.get("params") if albums_data else None
    if browse_params:
        out(yt.get_artist_albums(args.browse_id, browse_params))
    else:
        out(albums_data.get("results", []))


# ─── Album ────────────────────────────────────────────────────────────────────

def cmd_album(args):
    yt = get_yt()
    # handle both browseId and playlistId
    browse_id = args.browse_id
    if browse_id.startswith("OLAK") or browse_id.startswith("PL"):
        browse_id = yt.get_album_browse_id(browse_id)
    out(yt.get_album(browse_id))


# ─── Song ─────────────────────────────────────────────────────────────────────

def cmd_song(args):
    yt = get_yt()
    out(yt.get_song(args.video_id))


def cmd_lyrics(args):
    yt = get_yt()
    watch = yt.get_watch_playlist(args.video_id)
    lyrics_id = watch.get("lyrics")
    if not lyrics_id:
        out({"error": "No lyrics available for this song"})
        return
    out(yt.get_lyrics(lyrics_id))


def cmd_related(args):
    yt = get_yt()
    watch = yt.get_watch_playlist(args.video_id)
    related_id = watch.get("related")
    if not related_id:
        out({"error": "No related songs found"})
        return
    out(yt.get_song_related(related_id))


def cmd_watch(args):
    yt = get_yt()
    kwargs = dict(limit=args.limit)
    if args.playlist_id:
        kwargs["playlistId"] = args.playlist_id
    out(yt.get_watch_playlist(args.video_id, **kwargs))


# ─── Rating & Library Status ──────────────────────────────────────────────────

def cmd_rate(args):
    yt = get_yt(require_auth=True)
    result = yt.rate_song(args.video_id, args.rating)
    out({"status": result, "videoId": args.video_id, "rating": args.rating})


# ─── Artist subscription ──────────────────────────────────────────────────────

def cmd_subscribe(args):
    yt = get_yt(require_auth=True)
    if args.action == "subscribe":
        result = yt.subscribe_artists(args.channel_ids)
    else:
        result = yt.unsubscribe_artists(args.channel_ids)
    out({"status": result})


# ─── Discover ─────────────────────────────────────────────────────────────────

def cmd_charts(args):
    yt = get_yt()
    out(yt.get_charts(country=args.country))


def cmd_moods(args):
    yt = get_yt()
    out(yt.get_mood_categories())


def cmd_mood_playlist(args):
    yt = get_yt()
    out(yt.get_mood_playlists(args.params))


def cmd_home(args):
    yt = get_yt(require_auth=True)
    out(yt.get_home(limit=args.limit))


# ─── History ──────────────────────────────────────────────────────────────────

def cmd_history(args):
    yt = get_yt(require_auth=True)
    if args.action == "list":
        out(yt.get_history())
    elif args.action == "remove":
        if not args.feedback_tokens:
            bail("feedback_tokens required for 'remove'")
        out({"status": yt.remove_history_items(args.feedback_tokens)})


# ─── Taste Profile ────────────────────────────────────────────────────────────

def cmd_taste(args):
    yt = get_yt(require_auth=True)
    if args.action == "get":
        out(yt.get_tasteprofile())
    elif args.action == "set":
        if not args.artists:
            bail("--artists required for 'set'")
        taste = yt.get_tasteprofile()
        result = yt.set_tasteprofile(args.artists, taste)
        out({"status": result})


# ─── User ─────────────────────────────────────────────────────────────────────

def cmd_user(args):
    yt = get_yt()
    out(yt.get_user(args.channel_id))


# ─── Uploads ──────────────────────────────────────────────────────────────────

def cmd_upload(args):
    yt = get_yt(require_auth=True)
    if args.action == "list":
        out(yt.get_library_upload_songs(limit=args.limit))
    elif args.action == "upload":
        if not args.filepath:
            bail("--filepath required for 'upload'")
        out({"status": yt.upload_song(args.filepath)})
    elif args.action == "delete":
        if not args.entity_id:
            bail("--entity-id required for 'delete'")
        out({"status": yt.delete_upload_entity(args.entity_id)})


# ─── Parser ───────────────────────────────────────────────────────────────────

def build_parser():
    p = argparse.ArgumentParser(
        prog="yt-music",
        description="YouTube Music CLI — wraps ytmusicapi",
    )
    sub = p.add_subparsers(dest="command", metavar="COMMAND")

    # auth
    pa = sub.add_parser("auth", help="Authentication management")
    pa.add_argument("action", choices=["check", "setup", "account", "remove"])
    pa.add_argument("--cookie", help="Raw Cookie header string (for setup)")
    pa.add_argument("--cookies-file", dest="cookies_file",
                    help="JSON cookie export from a browser extension (for setup, e.g. Cookie-Editor)")
    pa.set_defaults(func=cmd_auth)

    # search
    ps = sub.add_parser("search", help="Search YouTube Music")
    ps.add_argument("query")
    ps.add_argument("--type", "-t", choices=[
        "songs", "artists", "albums", "playlists",
        "videos", "podcasts", "episodes", "profiles",
    ])
    ps.add_argument("--limit", "-l", type=int, default=10)
    ps.add_argument("--library", action="store_true", help="Search within your library")
    ps.set_defaults(func=cmd_search)

    # suggest
    psg = sub.add_parser("suggest", help="Search autocomplete suggestions")
    psg.add_argument("query")
    psg.set_defaults(func=cmd_suggest)

    # library
    plib = sub.add_parser("library", help="Browse your library")
    plib.add_argument("sub", nargs="?", default="playlists",
                      choices=["songs", "liked", "playlists", "albums",
                               "artists", "subscriptions", "history", "uploads"])
    plib.add_argument("--limit", "-l", type=int, default=25)
    plib.add_argument("--order", choices=["a_to_z", "z_to_a", "recently_added"])
    plib.set_defaults(func=cmd_library)

    # playlist
    ppl = sub.add_parser("playlist", help="Playlist management")
    ppl.add_argument("action", choices=["get", "create", "edit", "delete",
                                         "add", "add-playlist", "remove", "rate"])
    ppl.add_argument("playlist_id", nargs="?")
    ppl.add_argument("video_ids", nargs="*")
    ppl.add_argument("--title")
    ppl.add_argument("--description")
    ppl.add_argument("--privacy", choices=["PUBLIC", "PRIVATE", "UNLISTED"])
    ppl.add_argument("--source-playlist", dest="source_playlist")
    ppl.add_argument("--rating", choices=["LIKE", "DISLIKE", "INDIFFERENT"])
    ppl.add_argument("--duplicates", action="store_true")
    ppl.add_argument("--limit", "-l", type=int, default=100)
    ppl.set_defaults(func=cmd_playlist)

    # artist
    par = sub.add_parser("artist", help="Artist profile")
    par.add_argument("browse_id")
    par.set_defaults(func=cmd_artist)

    paa = sub.add_parser("artist-albums", help="All albums by an artist")
    paa.add_argument("browse_id")
    paa.set_defaults(func=cmd_artist_albums)

    # album
    pal = sub.add_parser("album", help="Album details and tracklist")
    pal.add_argument("browse_id")
    pal.set_defaults(func=cmd_album)

    # song
    pso = sub.add_parser("song", help="Song metadata")
    pso.add_argument("video_id")
    pso.set_defaults(func=cmd_song)

    # lyrics
    ply = sub.add_parser("lyrics", help="Song lyrics")
    ply.add_argument("video_id")
    ply.set_defaults(func=cmd_lyrics)

    # related
    pre = sub.add_parser("related", help="Related songs")
    pre.add_argument("video_id")
    pre.set_defaults(func=cmd_related)

    # watch
    pwt = sub.add_parser("watch", help="'Up next' radio/recommendations")
    pwt.add_argument("video_id")
    pwt.add_argument("--playlist-id", dest="playlist_id")
    pwt.add_argument("--limit", "-l", type=int, default=10)
    pwt.set_defaults(func=cmd_watch)

    # rate
    prt = sub.add_parser("rate", help="Like / dislike / unlike a song")
    prt.add_argument("video_id")
    prt.add_argument("rating", choices=["LIKE", "DISLIKE", "INDIFFERENT"])
    prt.set_defaults(func=cmd_rate)

    # subscribe
    psb = sub.add_parser("subscribe", help="Follow / unfollow artists")
    psb.add_argument("action", choices=["subscribe", "unsubscribe"])
    psb.add_argument("channel_ids", nargs="+")
    psb.set_defaults(func=cmd_subscribe)

    # charts
    pch = sub.add_parser("charts", help="Trending charts")
    pch.add_argument("--country", "-c", default="ZZ",
                     help="ISO country code (ZZ = global)")
    pch.set_defaults(func=cmd_charts)

    # moods
    pmd = sub.add_parser("moods", help="Mood & genre categories")
    pmd.set_defaults(func=cmd_moods)

    pmdp = sub.add_parser("mood-playlist", help="Playlists for a mood/genre")
    pmdp.add_argument("params", help="params value from 'moods' output")
    pmdp.set_defaults(func=cmd_mood_playlist)

    # home
    phm = sub.add_parser("home", help="Personalised home feed")
    phm.add_argument("--limit", "-l", type=int, default=3)
    phm.set_defaults(func=cmd_home)

    # history
    phi = sub.add_parser("history", help="Listening history")
    phi.add_argument("action", nargs="?", default="list", choices=["list", "remove"])
    phi.add_argument("feedback_tokens", nargs="*")
    phi.set_defaults(func=cmd_history)

    # taste
    ptaste = sub.add_parser("taste", help="Taste profile preferences")
    ptaste.add_argument("action", choices=["get", "set"])
    ptaste.add_argument("--artists", nargs="+")
    ptaste.set_defaults(func=cmd_taste)

    # user
    pusr = sub.add_parser("user", help="Public user profile")
    pusr.add_argument("channel_id")
    pusr.set_defaults(func=cmd_user)

    # upload
    pup = sub.add_parser("upload", help="Personal music uploads")
    pup.add_argument("action", choices=["list", "upload", "delete"])
    pup.add_argument("--filepath")
    pup.add_argument("--entity-id", dest="entity_id")
    pup.add_argument("--limit", "-l", type=int, default=25)
    pup.set_defaults(func=cmd_upload)

    return p


def main():
    parser = build_parser()
    args = parser.parse_args()
    if not args.command:
        parser.print_help()
        sys.exit(0)
    args.func(args)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Persistent Playwright browser daemon for YouTube Music playback control.
"""

from __future__ import annotations

import argparse
import json
import os
import secrets
import shutil
import subprocess
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any

YTM_URL = "https://music.youtube.com"
YTM_DAEMON_HOST = "127.0.0.1"

KEYS = {
    "toggle": "k",
    "next": "Shift+N",
    "prev": "Shift+P",
}

SELECTORS = {
    "app": "ytmusic-app",
    "player_bar": "ytmusic-player-bar",
    "shuffle": ".shuffle",
    "repeat": ".repeat",
}


def _resolve_data_dir() -> Path:
    configured = os.environ.get("YT_MUSIC_DATA_DIR")
    if configured:
        return Path(configured).expanduser()
    return Path(__file__).resolve().parent.parent / ".yt-music"


DATA_DIR = _resolve_data_dir()
PROFILE_DIR = DATA_DIR / "playwright-profile"
STATE_FILE = DATA_DIR / "player-daemon.json"


def _write_json(path: Path, data: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n")
    tmp.replace(path)


def _remove_state_file(pid: int) -> None:
    try:
        raw = json.loads(STATE_FILE.read_text())
    except Exception:
        raw = None
    if isinstance(raw, dict) and raw.get("pid") == pid:
        try:
            STATE_FILE.unlink()
        except FileNotFoundError:
            pass


def _candidate_paths() -> list[str]:
    system = os.uname().sysname if hasattr(os, "uname") else ""
    if system == "Darwin":
        return [
            "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
            str(Path.home() / "Applications/Google Chrome.app/Contents/MacOS/Google Chrome"),
        ]
    if os.name == "nt":
        local_app_data = os.environ.get("LOCALAPPDATA", "")
        program_files = os.environ.get("PROGRAMFILES", "")
        program_files_x86 = os.environ.get("PROGRAMFILES(X86)", "")
        return [
            str(Path(local_app_data) / "Google/Chrome/Application/chrome.exe"),
            str(Path(program_files) / "Google/Chrome/Application/chrome.exe"),
            str(Path(program_files_x86) / "Google/Chrome/Application/chrome.exe"),
        ]
    return [
        "google-chrome",
        "google-chrome-stable",
        "chromium",
        "chromium-browser",
    ]


def _find_browser() -> str | None:
    for candidate in _candidate_paths():
        if Path(candidate).exists():
            return candidate
        resolved = shutil.which(candidate)
        if resolved:
            return resolved
    return None


def _browser_version(browser_path: str | None) -> str | None:
    if not browser_path:
        return None
    result = subprocess.run([browser_path, "--version"], capture_output=True, text=True, check=False)
    if result.returncode != 0:
        return None
    return result.stdout.strip() or None


class PlayerError(RuntimeError):
    pass


class YTMusicRuntime:
    def __init__(self, user_data_dir: Path):
        from playwright.sync_api import sync_playwright

        self._lock = threading.RLock()
        self.user_data_dir = user_data_dir
        self.browser_path = _find_browser()
        self.browser_version = _browser_version(self.browser_path)
        self._pw_cm = sync_playwright()
        self._pw = self._pw_cm.start()
        launch_kwargs: dict[str, Any] = {
            "user_data_dir": str(self.user_data_dir),
            "headless": False,
            "args": ["--start-maximized"],
            "no_viewport": True,
        }
        if self.browser_path:
            launch_kwargs["executable_path"] = self.browser_path
        self.context = self._pw.chromium.launch_persistent_context(**launch_kwargs)
        self.context.set_default_timeout(15000)
        self.page = self._ensure_page()
        self._ensure_ytm_loaded()

    def close(self) -> None:
        with self._lock:
            self.context.close()
            self._pw_cm.stop()

    def health(self) -> dict[str, Any]:
        with self._lock:
            page = self._ensure_page()
            player_loaded = self._has_player(page)
            return {
                "daemon": "running",
                "mode": "playwright-persistent",
                "browser_path": self.browser_path,
                "browser_version": self.browser_version,
                "profile_dir": str(self.user_data_dir),
                "page_url": page.url,
                "player_loaded": player_loaded,
                "playing": self._is_playing(page) if player_loaded else False,
            }

    def handle(self, payload: dict[str, Any]) -> dict[str, Any]:
        action = str(payload.get("action") or "").strip()
        if not action:
            raise PlayerError("Missing action")

        with self._lock:
            page = self._ensure_page()

            if action == "open":
                video_id = str(payload.get("video_id") or "").strip()
                if not video_id:
                    raise PlayerError("Missing video_id")
                return self._cmd_open(page, video_id)
            if action in {"toggle", "play", "pause", "next", "prev"}:
                return self._cmd_control(page, action)
            if action == "volume":
                return self._cmd_volume(page, int(payload.get("level", 0)))
            if action == "seek":
                return self._cmd_seek(page, float(payload.get("seconds", 0)))
            if action == "shuffle":
                return self._cmd_shuffle(page)
            if action == "repeat":
                return self._cmd_repeat(page)
            if action == "status":
                return self._cmd_status(page)
            raise PlayerError(f"Unsupported action: {action}")

    def _ensure_page(self):
        if getattr(self, "page", None) and not self.page.is_closed():
            return self.page
        for page in self.context.pages:
            if not page.is_closed() and "music.youtube.com" in page.url:
                self.page = page
                return page
        for page in self.context.pages:
            if not page.is_closed():
                self.page = page
                return page
        self.page = self.context.new_page()
        return self.page

    def _ensure_ytm_loaded(self) -> None:
        page = self._ensure_page()
        if "music.youtube.com" not in page.url:
            page.goto(YTM_URL, wait_until="domcontentloaded", timeout=20000)
        try:
            page.wait_for_selector(SELECTORS["app"], timeout=15000)
        except Exception:
            pass
        try:
            page.bring_to_front()
        except Exception:
            pass

    def _wait_for_player(self, page, timeout: int = 15000) -> None:
        try:
            page.wait_for_selector(SELECTORS["player_bar"], timeout=timeout)
        except Exception:
            raise PlayerError(
                "Player bar not found. Open music.youtube.com in the launched browser, sign in if needed, "
                "then retry."
            )

    def _has_player(self, page) -> bool:
        try:
            return bool(page.locator(SELECTORS["player_bar"]).count())
        except Exception:
            return False

    def _send_key(self, page, key: str) -> None:
        page.focus("body")
        page.keyboard.press(key)
        time.sleep(0.3)

    def _is_playing(self, page) -> bool:
        try:
            return bool(page.evaluate("""
                (() => {
                    const v = document.querySelector('video');
                    return v ? !v.paused : false;
                })()
            """))
        except Exception:
            return False

    def _empty_status(self, page) -> dict[str, Any]:
        return {
            "title": None,
            "artist": None,
            "playing": False,
            "position": "0:00",
            "duration": "0:00",
            "position_seconds": 0,
            "duration_seconds": 0,
            "volume": None,
            "page_url": page.url,
            "player_loaded": False,
        }

    def _get_status(self, page) -> dict[str, Any]:
        try:
            data = page.evaluate("""
                (() => {
                    const video = document.querySelector('video');
                    const titleEl = document.querySelector('.title.ytmusic-player-bar');
                    const artistEl = document.querySelector('.byline-wrapper.ytmusic-player-bar a');

                    const title = titleEl ? titleEl.innerText.trim() : null;
                    const artist = artistEl ? artistEl.innerText.trim() : null;
                    const cur = video ? Math.floor(video.currentTime) : 0;
                    const dur = video ? Math.floor(video.duration) || 0 : 0;
                    const fmt = s => {
                        const m = Math.floor(s / 60);
                        return m + ':' + (s % 60 + '').padStart(2, '0');
                    };

                    return {
                        title,
                        artist,
                        playing: video ? !video.paused : false,
                        position: fmt(cur),
                        duration: fmt(dur),
                        position_seconds: cur,
                        duration_seconds: dur,
                        volume: video ? Math.round(video.volume * 100) : null,
                    };
                })()
            """)
        except Exception as exc:
            raise PlayerError(str(exc)) from exc
        data["page_url"] = page.url
        data["player_loaded"] = self._has_player(page)
        return data

    def _attempt_start_playback(self, page) -> str:
        try:
            started = page.evaluate("""
                async () => {
                    const video = document.querySelector('video');
                    if (!video) return false;
                    try {
                        await video.play();
                        return !video.paused;
                    } catch {
                        return false;
                    }
                }
            """)
            if started:
                return "video.play()"
        except Exception:
            pass

        for method, action in [
            ("play_button", lambda: page.locator("tp-yt-paper-icon-button.play-pause-button").first.click(timeout=3000)),
            ("toggle_key", lambda: self._send_key(page, KEYS["toggle"])),
        ]:
            try:
                action()
                time.sleep(0.8)
                if self._is_playing(page):
                    return method
            except Exception:
                continue
        return "failed"

    def _ensure_playing(self, page, timeout_ms: int = 12000) -> tuple[bool, list[str]]:
        deadline = time.time() + (timeout_ms / 1000)
        actions: list[str] = []
        while time.time() < deadline:
            if self._is_playing(page):
                return True, actions
            action = self._attempt_start_playback(page)
            if action != "failed":
                actions.append(action)
            if self._is_playing(page):
                return True, actions
            time.sleep(1.0)
        return self._is_playing(page), actions

    def _cmd_open(self, page, video_id: str) -> dict[str, Any]:
        url = f"{YTM_URL}/watch?v={video_id}"
        page.goto(url, wait_until="domcontentloaded", timeout=20000)
        self._wait_for_player(page, timeout=20000)
        verified, recovery_actions = self._ensure_playing(page)
        return {
            "action": "open",
            "url": url,
            "mode": "playwright-daemon",
            "playback_verified": verified,
            "recovery_actions": recovery_actions,
            **self._get_status(page),
        }

    def _cmd_control(self, page, action: str) -> dict[str, Any]:
        self._ensure_ytm_loaded()
        self._wait_for_player(page)

        if action == "toggle":
            self._send_key(page, KEYS["toggle"])
            time.sleep(0.3)
        elif action == "play":
            if not self._is_playing(page):
                self._send_key(page, KEYS["toggle"])
                time.sleep(0.3)
                self._ensure_playing(page, timeout_ms=5000)
        elif action == "pause":
            if self._is_playing(page):
                self._send_key(page, KEYS["toggle"])
                time.sleep(0.3)
        elif action == "next":
            self._send_key(page, KEYS["next"])
            time.sleep(1.5)
        elif action == "prev":
            self._send_key(page, KEYS["prev"])
            time.sleep(1.5)

        return {"action": action, **self._get_status(page)}

    def _cmd_volume(self, page, level: int) -> dict[str, Any]:
        self._ensure_ytm_loaded()
        self._wait_for_player(page)
        level = max(0, min(100, level))
        result = page.evaluate(
            """
            (targetLevel) => {
                const v = document.querySelector('video');
                if (!v) return 'video_not_found';
                v.volume = targetLevel / 100;
                v.muted = false;
                return Math.round(v.volume * 100);
            }
            """,
            level,
        )
        return {"action": "volume", "level": level, "result": result, **self._get_status(page)}

    def _cmd_seek(self, page, seconds: float) -> dict[str, Any]:
        self._ensure_ytm_loaded()
        self._wait_for_player(page)
        seconds = max(0, seconds)
        result = page.evaluate(
            """
            (targetSeconds) => {
                const v = document.querySelector('video');
                if (!v) return 'video_not_found';
                v.currentTime = targetSeconds;
                return Math.floor(v.currentTime);
            }
            """,
            seconds,
        )
        time.sleep(0.3)
        return {"action": "seek", "target_seconds": seconds, "result": result, **self._get_status(page)}

    def _cmd_shuffle(self, page) -> dict[str, Any]:
        self._ensure_ytm_loaded()
        self._wait_for_player(page)
        try:
            btn = page.locator(SELECTORS["shuffle"])
            btn.click()
            time.sleep(0.3)
            active = btn.get_attribute("aria-checked") == "true"
        except Exception as exc:
            raise PlayerError(f"Could not find shuffle button: {exc}") from exc
        return {"action": "shuffle", "shuffle_on": active, **self._get_status(page)}

    def _cmd_repeat(self, page) -> dict[str, Any]:
        self._ensure_ytm_loaded()
        self._wait_for_player(page)
        try:
            btn = page.locator(SELECTORS["repeat"])
            btn.click()
            time.sleep(0.3)
            mode = btn.get_attribute("aria-label") or "unknown"
        except Exception as exc:
            raise PlayerError(f"Could not find repeat button: {exc}") from exc
        return {"action": "repeat", "mode": mode, **self._get_status(page)}

    def _cmd_status(self, page) -> dict[str, Any]:
        self._ensure_ytm_loaded()
        if not self._has_player(page):
            return {"action": "status", **self._empty_status(page)}
        return {"action": "status", **self._get_status(page)}


class RequestHandler(BaseHTTPRequestHandler):
    server: "PlayerHTTPServer"

    def log_message(self, format: str, *args) -> None:
        return

    def _auth_ok(self) -> bool:
        return self.headers.get("X-YTMUSIC-Token") == self.server.token

    def _read_json(self) -> dict[str, Any]:
        length = int(self.headers.get("Content-Length", "0") or "0")
        raw = self.rfile.read(length) if length else b"{}"
        try:
            data = json.loads(raw.decode("utf-8"))
        except json.JSONDecodeError as exc:
            raise PlayerError(f"Invalid JSON payload: {exc}") from exc
        if not isinstance(data, dict):
            raise PlayerError("JSON payload must be an object")
        return data

    def _send(self, status: int, data: dict[str, Any]) -> None:
        body = json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        if not self._auth_ok():
            self._send(403, {"error": "Forbidden"})
            return
        if self.path != "/health":
            self._send(404, {"error": "Not found"})
            return
        self._send(200, self.server.runtime.health())

    def do_POST(self) -> None:
        if not self._auth_ok():
            self._send(403, {"error": "Forbidden"})
            return
        try:
            payload = self._read_json()
            if self.path == "/command":
                response = self.server.runtime.handle(payload)
                self._send(200, response)
                return
            if self.path == "/shutdown":
                self._send(200, {"action": "daemon-stop", "daemon": "stopping"})
                threading.Thread(target=self.server.shutdown, daemon=True).start()
                return
            self._send(404, {"error": "Not found"})
        except PlayerError as exc:
            self._send(400, {"error": str(exc)})
        except Exception as exc:
            self._send(500, {"error": f"Unhandled daemon error: {exc}"})


class PlayerHTTPServer(ThreadingHTTPServer):
    daemon_threads = True

    def __init__(self, server_address, runtime: YTMusicRuntime, token: str):
        super().__init__(server_address, RequestHandler)
        self.runtime = runtime
        self.token = token


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="yt-music-player-daemon",
        description="Run the persistent Playwright daemon for YouTube Music playback control",
    )
    parser.add_argument("--host", default=YTM_DAEMON_HOST)
    parser.add_argument("--port", type=int, default=0)
    parser.add_argument("--user-data-dir", type=Path, default=PROFILE_DIR)
    return parser


def main() -> None:
    args = build_parser().parse_args()
    args.user_data_dir = args.user_data_dir.expanduser().resolve()
    args.user_data_dir.mkdir(parents=True, exist_ok=True)
    DATA_DIR.mkdir(parents=True, exist_ok=True)

    token = secrets.token_hex(16)
    runtime = YTMusicRuntime(args.user_data_dir)
    server = PlayerHTTPServer((args.host, args.port), runtime=runtime, token=token)

    state = {
        "pid": os.getpid(),
        "host": args.host,
        "port": server.server_address[1],
        "token": token,
        "profile_dir": str(args.user_data_dir),
        "started_at": int(time.time()),
        "mode": "playwright-persistent",
    }
    _write_json(STATE_FILE, state)

    try:
        server.serve_forever(poll_interval=0.5)
    finally:
        try:
            server.server_close()
        finally:
            runtime.close()
            _remove_state_file(os.getpid())


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
YouTube Music playback client backed by a persistent Playwright daemon.

Regular playback commands auto-start the daemon when needed.
"""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any, NoReturn

YTM_DAEMON_HOST = "127.0.0.1"
YTM_DAEMON_START_TIMEOUT = 25.0


def _resolve_data_dir() -> Path:
    configured = os.environ.get("YT_MUSIC_DATA_DIR")
    if configured:
        return Path(configured).expanduser()
    return Path(__file__).resolve().parent.parent / ".yt-music"


SCRIPT_DIR = Path(__file__).resolve().parent
DATA_DIR = _resolve_data_dir()
STATE_FILE = DATA_DIR / "player-daemon.json"
LOG_FILE = DATA_DIR / "player-daemon.log"
DAEMON_SCRIPT = SCRIPT_DIR / "player_daemon.py"


def bail(msg: str) -> NoReturn:
    print(json.dumps({"error": msg}, ensure_ascii=False))
    sys.exit(1)


def out(data: dict[str, Any]) -> None:
    print(json.dumps(data, ensure_ascii=False, indent=2))


def _load_state() -> dict[str, Any] | None:
    if not STATE_FILE.exists():
        return None
    try:
        raw = json.loads(STATE_FILE.read_text())
    except Exception:
        return None
    return raw if isinstance(raw, dict) else None


def _clear_state() -> None:
    try:
        STATE_FILE.unlink()
    except FileNotFoundError:
        pass


def _request(
    state: dict[str, Any],
    path: str,
    payload: dict[str, Any] | None = None,
    timeout: float = 5.0,
) -> dict[str, Any]:
    url = f"http://{YTM_DAEMON_HOST}:{state['port']}{path}"
    body = None if payload is None else json.dumps(payload).encode("utf-8")
    headers = {
        "Accept": "application/json",
        "X-YTMUSIC-Token": str(state.get("token", "")),
    }
    if body is not None:
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=body, headers=headers)
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        data = json.loads(resp.read().decode("utf-8"))
    if not isinstance(data, dict):
        raise RuntimeError("Daemon returned a non-object JSON payload")
    return data


def _probe(state: dict[str, Any]) -> dict[str, Any] | None:
    try:
        return _request(state, "/health", timeout=2.0)
    except (OSError, urllib.error.URLError, urllib.error.HTTPError, RuntimeError, json.JSONDecodeError):
        return None


def _wait_for_daemon(deadline: float) -> dict[str, Any]:
    while time.time() < deadline:
        state = _load_state()
        if state:
            health = _probe(state)
            if health:
                return {"state": state, "health": health}
        time.sleep(0.4)
    raise RuntimeError(
        "Timed out waiting for the playback daemon to start. Check .yt-music/player-daemon.log for details."
    )


def _start_daemon() -> dict[str, Any]:
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    if not DAEMON_SCRIPT.exists():
        raise RuntimeError(f"Daemon script not found: {DAEMON_SCRIPT}")

    with LOG_FILE.open("ab") as log_file:
        subprocess.Popen(
            [sys.executable, str(DAEMON_SCRIPT)],
            cwd=str(SCRIPT_DIR),
            stdin=subprocess.DEVNULL,
            stdout=log_file,
            stderr=subprocess.STDOUT,
            start_new_session=True,
            env=os.environ.copy(),
        )

    return _wait_for_daemon(time.time() + YTM_DAEMON_START_TIMEOUT)


def _ensure_daemon() -> dict[str, Any]:
    state = _load_state()
    if state:
        health = _probe(state)
        if health:
            return {"state": state, "health": health}
        _clear_state()
    return _start_daemon()


def _cmd_remote(args: argparse.Namespace) -> None:
    daemon = _ensure_daemon()
    payload: dict[str, Any] = {"action": args.action}
    if hasattr(args, "video_id"):
        payload["video_id"] = args.video_id
    if hasattr(args, "level"):
        payload["level"] = args.level
    if hasattr(args, "seconds"):
        payload["seconds"] = args.seconds

    try:
        response = _request(daemon["state"], "/command", payload=payload, timeout=20.0)
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        bail(detail or f"Daemon request failed: HTTP {exc.code}")
    except (OSError, urllib.error.URLError, RuntimeError, json.JSONDecodeError) as exc:
        bail(f"Could not reach the playback daemon: {exc}")
    out(response)


def _cmd_daemon_start(args: argparse.Namespace) -> None:
    daemon = _ensure_daemon()
    out({"action": "daemon-start", **daemon["health"]})


def _cmd_daemon_status(args: argparse.Namespace) -> None:
    state = _load_state()
    if not state:
        out({"action": "daemon-status", "daemon": "stopped"})
        return
    health = _probe(state)
    if not health:
        _clear_state()
        out({"action": "daemon-status", "daemon": "stopped"})
        return
    out({"action": "daemon-status", **health})


def _cmd_daemon_stop(args: argparse.Namespace) -> None:
    state = _load_state()
    if not state:
        out({"action": "daemon-stop", "daemon": "stopped"})
        return

    try:
        response = _request(state, "/shutdown", payload={"action": "daemon-stop"}, timeout=5.0)
    except (OSError, urllib.error.URLError, urllib.error.HTTPError, RuntimeError, json.JSONDecodeError):
        _clear_state()
        out({"action": "daemon-stop", "daemon": "stopped"})
        return

    deadline = time.time() + 5.0
    while time.time() < deadline:
        if not STATE_FILE.exists():
            break
        time.sleep(0.2)
    _clear_state()
    out(response)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="yt-music-player",
        description="Control YouTube Music playback via a persistent Playwright browser daemon",
    )
    parser.add_argument("--chrome-port", help=argparse.SUPPRESS)
    sub = parser.add_subparsers(dest="action", metavar="ACTION")

    open_cmd = sub.add_parser("open", help="Open and play a song by videoId")
    open_cmd.add_argument("video_id")
    open_cmd.set_defaults(func=_cmd_remote)

    for name, help_text in [
        ("toggle", "Toggle play/pause"),
        ("play", "Resume playback"),
        ("pause", "Pause playback"),
        ("next", "Skip to next track"),
        ("prev", "Go to previous track"),
        ("status", "Show current playback status"),
        ("shuffle", "Toggle shuffle mode"),
        ("repeat", "Cycle repeat mode"),
    ]:
        cmd = sub.add_parser(name, help=help_text)
        cmd.set_defaults(func=_cmd_remote)

    volume_cmd = sub.add_parser("volume", help="Set volume (0-100)")
    volume_cmd.add_argument("level", type=int)
    volume_cmd.set_defaults(func=_cmd_remote)

    seek_cmd = sub.add_parser("seek", help="Seek to position in seconds")
    seek_cmd.add_argument("seconds", type=float)
    seek_cmd.set_defaults(func=_cmd_remote)

    daemon_start = sub.add_parser("daemon-start", help="Start the persistent playback daemon")
    daemon_start.set_defaults(func=_cmd_daemon_start)

    daemon_status = sub.add_parser("daemon-status", help="Show daemon status without auto-starting it")
    daemon_status.set_defaults(func=_cmd_daemon_status)

    daemon_stop = sub.add_parser("daemon-stop", help="Stop the persistent playback daemon")
    daemon_stop.set_defaults(func=_cmd_daemon_stop)

    return parser


def main() -> None:
    parser = build_parser()
    args = parser.parse_args()
    if not args.action:
        parser.print_help()
        sys.exit(0)
    args.func(args)


if __name__ == "__main__":
    main()

"""
Shared pytest fixtures for the yt-music skill test suite.

Goals:
- Never touch the user's real ~/.yt-music/ data dir — every test gets a temp dir
  via the YT_MUSIC_DATA_DIR env var.
- Never import the real `ytmusicapi` or `playwright` packages at test time —
  fake modules are installed in sys.modules before the scripts are imported.
- Each test gets a freshly-reloaded `helper` / `player` / `player_daemon` module
  so module-level constants (DATA_DIR, AUTH_FILE, STATE_FILE) reflect the per-test env.
"""

from __future__ import annotations

import importlib
import sys
import types
from pathlib import Path
from unittest.mock import MagicMock

import pytest

SKILL_ROOT = Path(__file__).resolve().parent.parent
SCRIPTS_DIR = SKILL_ROOT / "scripts"

# Ensure scripts/ is importable as a top-level package path.
if str(SCRIPTS_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPTS_DIR))


def _purge(*names: str) -> None:
    for name in names:
        sys.modules.pop(name, None)


@pytest.fixture
def data_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    """Per-test data dir; the scripts honor YT_MUSIC_DATA_DIR at import time."""
    target = tmp_path / "yt-music-data"
    target.mkdir(parents=True, exist_ok=True)
    monkeypatch.setenv("YT_MUSIC_DATA_DIR", str(target))
    return target


@pytest.fixture
def fake_yt_instance() -> MagicMock:
    """A MagicMock standing in for a YTMusic client. Tests assert on its calls."""
    return MagicMock(name="YTMusicInstance")


@pytest.fixture
def fake_ytmusicapi(monkeypatch: pytest.MonkeyPatch, fake_yt_instance: MagicMock):
    """
    Install a fake `ytmusicapi` module in sys.modules.

    `helper.get_yt()` does a lazy `from ytmusicapi import YTMusic` and instantiates
    it with either an auth file path or no args. We capture both call shapes and
    always return `fake_yt_instance` so test assertions can inspect method calls.
    """
    fake_module = types.ModuleType("ytmusicapi")
    yt_factory = MagicMock(name="YTMusicFactory", return_value=fake_yt_instance)
    fake_module.YTMusic = yt_factory  # type: ignore[attr-defined]
    monkeypatch.setitem(sys.modules, "ytmusicapi", fake_module)
    yield yt_factory
    sys.modules.pop("ytmusicapi", None)


@pytest.fixture
def helper_mod(data_dir: Path, fake_ytmusicapi: MagicMock):
    """
    Fresh import of `helper.py` with the temp data dir env var set and a fake
    ytmusicapi already in sys.modules. Returns the module.
    """
    _purge("helper")
    module = importlib.import_module("helper")
    assert Path(module.DATA_DIR) == data_dir, (
        f"helper.DATA_DIR ({module.DATA_DIR}) did not pick up YT_MUSIC_DATA_DIR ({data_dir})"
    )
    yield module
    _purge("helper")


@pytest.fixture
def player_mod(data_dir: Path):
    """Fresh import of `player.py` with the temp data dir env var set."""
    _purge("player")
    module = importlib.import_module("player")
    assert Path(module.DATA_DIR) == data_dir
    yield module
    _purge("player")


@pytest.fixture
def fake_playwright(monkeypatch: pytest.MonkeyPatch) -> types.ModuleType:
    """
    Install fake `playwright` + `playwright.sync_api` modules so that importing
    `player_daemon` never reaches the real Playwright. The fake `sync_playwright`
    is a no-op context manager; tests that instantiate YTMusicRuntime should
    additionally patch its internals — most unit tests only touch pure helpers
    that never reach Playwright at all.
    """
    pkg = types.ModuleType("playwright")
    sync_api = types.ModuleType("playwright.sync_api")
    sync_api.sync_playwright = MagicMock(name="sync_playwright")  # type: ignore[attr-defined]
    pkg.sync_api = sync_api  # type: ignore[attr-defined]
    monkeypatch.setitem(sys.modules, "playwright", pkg)
    monkeypatch.setitem(sys.modules, "playwright.sync_api", sync_api)
    yield sync_api
    sys.modules.pop("playwright", None)
    sys.modules.pop("playwright.sync_api", None)


@pytest.fixture
def player_daemon_mod(data_dir: Path, fake_playwright: types.ModuleType):
    """Fresh import of `player_daemon.py` for pure-helper tests."""
    _purge("player_daemon")
    module = importlib.import_module("player_daemon")
    assert Path(module.DATA_DIR) == data_dir
    yield module
    _purge("player_daemon")


def run_cli(module, *argv: str, monkeypatch: pytest.MonkeyPatch):
    """Invoke a script's `main()` with patched sys.argv, returning the SystemExit code."""
    monkeypatch.setattr(sys, "argv", [getattr(module, "__name__", "script"), *argv])
    try:
        module.main()
        return 0
    except SystemExit as exc:  # bail() and parser exits
        code = exc.code
        if code is None:
            return 0
        if isinstance(code, int):
            return code
        return 1

"""Tests for the auth subsystem inside helper.py."""

from __future__ import annotations

import hashlib
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock

import pytest

from .conftest import run_cli


# ─── build_auth_from_cookie ──────────────────────────────────────────────────

def test_build_auth_from_cookie_with_sapisid(helper_mod):
    """Happy path: full Cookie header with SAPISID yields all four required keys."""
    cookie = "Foo=1; SAPISID=mysapisidvalue; Bar=2"
    auth = helper_mod.build_auth_from_cookie(cookie)
    assert set(auth.keys()) == {"Authorization", "Cookie", "X-Goog-AuthUser", "x-origin"}
    assert auth["Cookie"] == cookie
    assert auth["x-origin"] == "https://music.youtube.com"
    assert auth["X-Goog-AuthUser"] == "0"
    # SAPISIDHASH format: "SAPISIDHASH <ts>_<sha1hex>"
    assert auth["Authorization"].startswith("SAPISIDHASH ")
    ts_hash = auth["Authorization"].split(" ", 1)[1]
    ts_str, sha = ts_hash.split("_", 1)
    expected = hashlib.sha1(
        f"{ts_str} mysapisidvalue https://music.youtube.com".encode()
    ).hexdigest()
    assert sha == expected


def test_build_auth_from_cookie_falls_back_to_3papisid(helper_mod):
    """When SAPISID is missing, __Secure-3PAPISID must be accepted as fallback."""
    cookie = "__Secure-3PAPISID=alt_value; SID=abc"
    auth = helper_mod.build_auth_from_cookie(cookie)
    assert auth["Cookie"] == cookie
    assert auth["Authorization"].startswith("SAPISIDHASH ")


def test_build_auth_from_cookie_missing_sapisid_exits(helper_mod, capsys):
    """No SAPISID and no __Secure-3PAPISID → bail with structured error JSON."""
    with pytest.raises(SystemExit) as ei:
        helper_mod.build_auth_from_cookie("Foo=1; Bar=2")
    assert ei.value.code == 1
    err = capsys.readouterr().err
    body = json.loads(err)
    assert "SAPISID" in body["error"]


# ─── auth_setup_instructions ─────────────────────────────────────────────────

def test_auth_setup_instructions_shape(helper_mod):
    """The bundled guidance must include all the artifacts the agent flow depends on."""
    info = helper_mod.auth_setup_instructions()
    assert info["required"] is True
    assert "agent_prompt" in info
    assert "language_policy" in info
    # English fallbacks
    for key in ("short_en", "cookie_string_en", "cookies_json_en"):
        assert key in info["reply_templates"]
    # Step lists
    assert isinstance(info["cookie_string_steps"], list) and info["cookie_string_steps"]
    assert isinstance(info["cookies_json_steps"], list) and info["cookies_json_steps"]
    # Concrete setup commands
    for key in ("cookie_string", "cookies_json"):
        assert key in info["setup_commands"]


# ─── _sanitize_auth_headers ──────────────────────────────────────────────────

def test_sanitize_auth_headers_keeps_only_allowed(helper_mod):
    raw = {
        "Authorization": "SAPISIDHASH 1_a",
        "Cookie": "x=1",
        "X-Goog-AuthUser": "0",
        "x-origin": "https://music.youtube.com",
        "garbage": "drop me",
        "another": 123,  # non-str also dropped
    }
    sanitized = helper_mod._sanitize_auth_headers(raw)
    assert sanitized == {
        "Authorization": "SAPISIDHASH 1_a",
        "Cookie": "x=1",
        "X-Goog-AuthUser": "0",
        "x-origin": "https://music.youtube.com",
    }


# ─── _migrate_legacy_auth_file ───────────────────────────────────────────────

def test_migrate_legacy_auth_file_strips_extra_keys(helper_mod, data_dir):
    legacy = {
        "Authorization": "SAPISIDHASH 1_a",
        "Cookie": "x=1",
        "X-Goog-AuthUser": "0",
        "x-origin": "https://music.youtube.com",
        "old_field": "remove me",
    }
    helper_mod.AUTH_FILE.write_text(json.dumps(legacy))
    helper_mod._migrate_legacy_auth_file()
    after = json.loads(helper_mod.AUTH_FILE.read_text())
    assert "old_field" not in after
    assert after["Authorization"] == "SAPISIDHASH 1_a"


def test_migrate_legacy_auth_file_noop_when_missing(helper_mod):
    """No file → nothing to do, no exception."""
    helper_mod._migrate_legacy_auth_file()
    assert not helper_mod.AUTH_FILE.exists()


def test_migrate_legacy_auth_file_ignores_invalid_json(helper_mod):
    helper_mod.AUTH_FILE.parent.mkdir(parents=True, exist_ok=True)
    helper_mod.AUTH_FILE.write_text("not-valid-json{")
    helper_mod._migrate_legacy_auth_file()
    # File unchanged because we can't parse it.
    assert helper_mod.AUTH_FILE.read_text() == "not-valid-json{"


# ─── cmd_auth check ──────────────────────────────────────────────────────────

def test_cmd_auth_check_missing(helper_mod, capsys, monkeypatch):
    code = run_cli(helper_mod, "auth", "check", monkeypatch=monkeypatch)
    assert code == 0
    body = json.loads(capsys.readouterr().out)
    assert body["status"] == "missing"
    # Hands the agent the full guidance payload.
    assert body["required"] is True
    assert "agent_prompt" in body


def test_cmd_auth_check_ok(helper_mod, capsys, monkeypatch):
    helper_mod.AUTH_FILE.parent.mkdir(parents=True, exist_ok=True)
    helper_mod.AUTH_FILE.write_text(json.dumps({
        "Authorization": "x", "Cookie": "y", "X-Goog-AuthUser": "0", "x-origin": "z",
    }))
    code = run_cli(helper_mod, "auth", "check", monkeypatch=monkeypatch)
    assert code == 0
    body = json.loads(capsys.readouterr().out)
    assert body == {"status": "ok", "path": str(helper_mod.AUTH_FILE)}


# ─── cmd_auth setup --cookie ─────────────────────────────────────────────────

def test_cmd_auth_setup_cookie_string_persists(helper_mod, capsys, monkeypatch):
    code = run_cli(
        helper_mod, "auth", "setup", "--cookie", "SAPISID=mysapisidvalue; Foo=1",
        monkeypatch=monkeypatch,
    )
    assert code == 0
    body = json.loads(capsys.readouterr().out)
    assert body == {"status": "saved", "path": str(helper_mod.AUTH_FILE)}
    persisted = json.loads(helper_mod.AUTH_FILE.read_text())
    assert persisted["Cookie"] == "SAPISID=mysapisidvalue; Foo=1"
    assert persisted["Authorization"].startswith("SAPISIDHASH ")


def test_cmd_auth_setup_empty_cookie_fails(helper_mod, capsys, monkeypatch):
    monkeypatch.setattr(sys, "stdin", _make_stdin(""))
    code = run_cli(helper_mod, "auth", "setup", monkeypatch=monkeypatch)
    assert code == 1
    err = capsys.readouterr().err
    assert "No cookie provided" in err


# ─── cmd_auth setup --cookies-file ───────────────────────────────────────────

def test_cmd_auth_setup_cookies_file_happy_path(helper_mod, tmp_path: Path, capsys, monkeypatch):
    cookies_json = tmp_path / "cookies.json"
    cookies_json.write_text(json.dumps([
        {"name": "SAPISID", "value": "mysapisidvalue", "domain": ".youtube.com"},
        {"name": "SID",      "value": "abc",           "domain": ".youtube.com"},
        {"name": "junk",     "value": "no",            "domain": "bing.com"},  # filtered
    ]))
    code = run_cli(
        helper_mod, "auth", "setup", "--cookies-file", str(cookies_json),
        monkeypatch=monkeypatch,
    )
    assert code == 0
    body = json.loads(capsys.readouterr().out)
    assert body["status"] == "saved"
    assert body["cookies_imported"] == 2
    persisted = json.loads(helper_mod.AUTH_FILE.read_text())
    assert "SAPISID=mysapisidvalue" in persisted["Cookie"]
    assert "SID=abc" in persisted["Cookie"]
    assert "junk=no" not in persisted["Cookie"]


def test_cmd_auth_setup_cookies_file_missing(helper_mod, capsys, monkeypatch):
    code = run_cli(
        helper_mod, "auth", "setup", "--cookies-file", "/no/such/file.json",
        monkeypatch=monkeypatch,
    )
    assert code == 1
    assert "File not found" in capsys.readouterr().err


def test_cmd_auth_setup_cookies_file_not_a_list(helper_mod, tmp_path, capsys, monkeypatch):
    bad = tmp_path / "bad.json"
    bad.write_text(json.dumps({"not": "an array"}))
    code = run_cli(
        helper_mod, "auth", "setup", "--cookies-file", str(bad),
        monkeypatch=monkeypatch,
    )
    assert code == 1
    assert "JSON array" in capsys.readouterr().err


def test_cmd_auth_setup_cookies_file_invalid_json(helper_mod, tmp_path, capsys, monkeypatch):
    bad = tmp_path / "bad.json"
    bad.write_text("{not json")
    code = run_cli(
        helper_mod, "auth", "setup", "--cookies-file", str(bad),
        monkeypatch=monkeypatch,
    )
    assert code == 1
    assert "Could not parse JSON" in capsys.readouterr().err


def test_cmd_auth_setup_cookies_file_no_matching_cookies(helper_mod, tmp_path, capsys, monkeypatch):
    empty = tmp_path / "empty.json"
    empty.write_text(json.dumps([
        {"name": "x", "value": "1", "domain": "bing.com"},
    ]))
    code = run_cli(
        helper_mod, "auth", "setup", "--cookies-file", str(empty),
        monkeypatch=monkeypatch,
    )
    assert code == 1
    assert "No YouTube/Google cookies" in capsys.readouterr().err


def test_cmd_auth_setup_cookies_file_falls_back_when_no_sapisid(helper_mod, tmp_path, capsys, monkeypatch):
    """If neither SAPISID nor __Secure-3PAPISID is in the export, persist the headers
    we _do_ have (Cookie / X-Goog-AuthUser / x-origin) but skip Authorization."""
    cookies_json = tmp_path / "cookies.json"
    cookies_json.write_text(json.dumps([
        {"name": "SID", "value": "abc", "domain": ".youtube.com"},
    ]))
    code = run_cli(
        helper_mod, "auth", "setup", "--cookies-file", str(cookies_json),
        monkeypatch=monkeypatch,
    )
    assert code == 0
    body = json.loads(capsys.readouterr().out)
    assert body["status"] == "saved"
    persisted = json.loads(helper_mod.AUTH_FILE.read_text())
    assert "Authorization" not in persisted  # no SAPISID → no SAPISIDHASH
    assert persisted["Cookie"] == "SID=abc"
    assert persisted["x-origin"] == "https://music.youtube.com"


# ─── cmd_auth remove ─────────────────────────────────────────────────────────

def test_cmd_auth_remove_existing(helper_mod, capsys, monkeypatch):
    helper_mod.AUTH_FILE.parent.mkdir(parents=True, exist_ok=True)
    helper_mod.AUTH_FILE.write_text("{}")
    code = run_cli(helper_mod, "auth", "remove", monkeypatch=monkeypatch)
    assert code == 0
    body = json.loads(capsys.readouterr().out)
    assert body == {"status": "removed"}
    assert not helper_mod.AUTH_FILE.exists()


def test_cmd_auth_remove_when_absent(helper_mod, capsys, monkeypatch):
    code = run_cli(helper_mod, "auth", "remove", monkeypatch=monkeypatch)
    assert code == 0
    body = json.loads(capsys.readouterr().out)
    assert body == {"status": "already_missing"}


# ─── cmd_auth account ────────────────────────────────────────────────────────

def test_cmd_auth_account_requires_auth_file(helper_mod, capsys, monkeypatch):
    """Without auth.json, the require_auth path exits 1 with structured guidance."""
    code = run_cli(helper_mod, "auth", "account", monkeypatch=monkeypatch)
    assert code == 1
    err = json.loads(capsys.readouterr().err)
    assert err["error"] == "Auth required"
    assert "agent_prompt" in err


def test_cmd_auth_account_returns_account_info(
    helper_mod, fake_yt_instance, capsys, monkeypatch,
):
    helper_mod.AUTH_FILE.parent.mkdir(parents=True, exist_ok=True)
    helper_mod.AUTH_FILE.write_text(json.dumps({
        "Authorization": "x", "Cookie": "y", "X-Goog-AuthUser": "0", "x-origin": "z",
    }))
    fake_yt_instance.get_account_info.return_value = {"accountName": "Demo User"}
    code = run_cli(helper_mod, "auth", "account", monkeypatch=monkeypatch)
    assert code == 0
    body = json.loads(capsys.readouterr().out)
    assert body == {"accountName": "Demo User"}
    fake_yt_instance.get_account_info.assert_called_once_with()


# ─── helpers ─────────────────────────────────────────────────────────────────

class _StdinShim:
    def __init__(self, text: str) -> None:
        self._text = text

    def read(self) -> str:
        return self._text


def _make_stdin(text: str) -> _StdinShim:
    return _StdinShim(text)