Suno Music Creator

__pycache__/
*.pyc
*.pyo
.env

MIT License

Copyright (c) 2026 joeseesun

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

#!/usr/bin/env python3
"""
Suno Token 自动刷新脚本 (CDP 简化版)

工作流程：
1. 检查 Chrome 是否开启远程调试（localhost:9222）
2. 查找是否有已登录的 suno.com 页面
3. 如果有，通过 CDP WebSocket 提取 JWT Token
4. 如果没有，提示用户登录 Suno
5. 自动更新 .env 并重启 suno-api 服务
"""
import json
import subprocess
import sys
import os
import asyncio
import websockets
from pathlib import Path

# 检查 Chrome 远程调试是否开启
def check_chrome_debugging():
    """检查 Chrome 是否开启了远程调试"""
    try:
        result = subprocess.run(
            ['curl', '-s', 'http://localhost:9222/json'],
            capture_output=True,
            text=True,
            timeout=3
        )
        if result.returncode == 0 and result.stdout:
            return True, json.loads(result.stdout)
        return False, None
    except Exception as e:
        return False, str(e)

# 查找 Suno 页面
def find_suno_page(pages):
    """在 Chrome 页面列表中查找 suno.com 页面"""
    for page in pages:
        if page.get('type') == 'page' and 'suno.com' in page.get('url', ''):
            return page
    return None

# 通过 CDP WebSocket 提取 Token
async def extract_token_via_cdp(ws_url):
    """通过 WebSocket CDP 提取 JWT Token"""
    try:
        async with websockets.connect(ws_url) as websocket:
            # 启用 Network domain
            await websocket.send(json.dumps({
                "id": 1,
                "method": "Network.enable"
            }))
            await websocket.recv()

            # 获取所有 cookies
            await websocket.send(json.dumps({
                "id": 2,
                "method": "Network.getAllCookies"
            }))
            response = await websocket.recv()
            data = json.loads(response)

            # 查找 __session cookie
            if 'result' in data and 'cookies' in data['result']:
                for cookie in data['result']['cookies']:
                    if cookie.get('name') == '__session' and 'suno.com' in cookie.get('domain', ''):
                        return cookie.get('value'), None

            # 如果没找到，尝试刷新页面并监听网络请求
            print("⏳ Cookie 中未找到 token，尝试触发网络请求...")
            await websocket.send(json.dumps({
                "id": 3,
                "method": "Page.reload"
            }))

            # 监听网络请求
            for _ in range(30):
                try:
                    msg = await asyncio.wait_for(websocket.recv(), timeout=2.0)
                    event = json.loads(msg)

                    if event.get('method') == 'Network.requestWillBeSent':
                        params = event.get('params', {})
                        request = params.get('request', {})
                        headers = request.get('headers', {})

                        if 'Authorization' in headers:
                            auth = headers['Authorization']
                            if auth.startswith('Bearer '):
                                token = auth.replace('Bearer ', '').strip()
                                return token, None
                except asyncio.TimeoutError:
                    break

            return None, "未能从网络请求中提取 token"

    except Exception as e:
        return None, f"CDP 连接失败: {e}"

# 主函数
async def main():
    print("🚀 Suno Token 自动刷新 (CDP 简化版)\n")

    # Step 1: 检查 Chrome 远程调试
    print("🔍 检查 Chrome 远程调试状态...")
    chrome_ok, pages = check_chrome_debugging()

    if not chrome_ok:
        print("❌ Chrome 远程调试未开启，尝试从 Dia 浏览器直接提取 cookies...")
        script_dir = Path(__file__).parent
        extractor_script = script_dir / "extract_browser_cookies.py"

        result = subprocess.run(
            [sys.executable, str(extractor_script), "--browser", "dia", "--json"],
            capture_output=True, text=True, timeout=15
        )

        if result.returncode == 0:
            import json as json_mod
            data = json_mod.loads(result.stdout)
            token = data.get("jwt_token", "")
            if token:
                print(f"✅ 从 Dia 浏览器提取到 Token!")
                print(f"   长度: {len(token)} 字符")
                print(f"   预览: {token[:30]}...{token[-20:]}\n")

                update_script = script_dir / "update_token.py"
                upd = subprocess.run(
                    [sys.executable, str(update_script), token],
                    capture_output=True, text=True
                )
                if upd.returncode == 0:
                    print(upd.stdout)
                    print("\n✅ Token 刷新完成!")
                    return 0
                else:
                    print(f"❌ 更新失败:\n{upd.stderr}")
                    return 1

        print("❌ Dia 浏览器提取也失败了")
        print("\n请使用以下命令启动 Chrome:")
        print("  /Applications/Google\\ Chrome.app/Contents/MacOS/Google\\ Chrome \\")
        print("    --remote-debugging-port=9222 \\")
        print("    --user-data-dir=/tmp/chrome-debug &")
        print("\n或者确保已登录 Suno 的 Dia 浏览器处于运行状态")
        return 1

    print(f"✅ Chrome 已连接，找到 {len(pages)} 个页面")

    # Step 2: 查找 Suno 页面
    print("🔍 查找 suno.com 页面...")
    suno_page = find_suno_page(pages)

    if not suno_page:
        print("❌ 没有找到 suno.com 页面")
        print("\n请执行以下操作:")
        print("1. 打开 Chrome 浏览器")
        print("2. 访问 https://suno.com/create")
        print("3. 确保已登录")
        print("4. 再次运行本脚本")
        return 1

    print(f"✅ 找到 Suno 页面: {suno_page['url']}")

    # Step 3: 提取 Token
    print("🔑 通过 CDP 提取 JWT Token...")
    ws_url = suno_page['webSocketDebuggerUrl']
    token, error = await extract_token_via_cdp(ws_url)

    if error or not token:
        print(f"❌ 提取失败: {error or '未知错误'}")
        print("\n可能的原因:")
        print("- 浏览器未登录 Suno")
        print("- Token 已过期，需要刷新页面")
        print("- 网络请求被拦截")
        return 1

    print(f"✅ 成功提取 Token!")
    print(f"   长度: {len(token)} 字符")
    print(f"   预览: {token[:30]}...{token[-20:]}\n")

    # Step 4: 调用 update_token.py 更新配置
    print("🔄 更新 .env 配置并重启服务...")
    script_dir = Path(__file__).parent
    update_script = script_dir / "update_token.py"

    result = subprocess.run(
        [sys.executable, str(update_script), token],
        capture_output=True,
        text=True
    )

    if result.returncode != 0:
        print(f"❌ 更新失败:\n{result.stderr}")
        return 1

    print(result.stdout)
    print("\n✅ Token 刷新完成!")
    return 0

if __name__ == "__main__":
    try:
        exit_code = asyncio.run(main())
        sys.exit(exit_code)
    except KeyboardInterrupt:
        print("\n\n⚠️  用户中断")
        sys.exit(1)
    except Exception as e:
        print(f"\n❌ 未预期的错误: {e}")
        sys.exit(1)

#!/usr/bin/env python3
"""
清理 Suno SRT 文件中的结构标记，只保留实际歌词
用于 MTV 视频制作
"""

import re
import sys
from pathlib import Path


def is_suno_tag(text: str) -> bool:
    """判断是否为 Suno 结构标记"""
    text = text.strip()

    # 空行
    if not text:
        return True

    # 方括号标记: [Verse 1], [Male Vocals: Powerful], [Hook: Anthemic], etc.
    if text.startswith('[') and text.endswith(']'):
        return True

    # 括号里的背景声: (Hey! 三百杯), (换美酒!), etc.
    # 保留这些，因为这是歌词的一部分

    return False


def clean_srt(input_file: Path, output_file: Path = None):
    """清理 SRT 文件，去除 Suno 标记"""

    if not input_file.exists():
        print(f"❌ 文件不存在: {input_file}")
        return False

    if output_file is None:
        # 默认输出文件名: 原文件名.clean.srt
        output_file = input_file.with_suffix('.clean.srt')

    print(f"📖 Reading: {input_file}")

    with open(input_file, 'r', encoding='utf-8') as f:
        content = f.read()

    # 按空行分割为字幕块
    blocks = content.strip().split('\n\n')

    cleaned_blocks = []
    new_index = 1

    for block in blocks:
        lines = block.strip().split('\n')
        if len(lines) < 3:
            continue

        # 解析字幕块
        index_line = lines[0]
        time_line = lines[1]
        text_lines = lines[2:]

        # 过滤掉 Suno 标记
        cleaned_text_lines = [line for line in text_lines if not is_suno_tag(line)]

        # 如果没有剩余文本，跳过这个块
        if not cleaned_text_lines:
            continue

        # 重建字幕块（重新编号）
        cleaned_block = f"{new_index}\n{time_line}\n" + "\n".join(cleaned_text_lines)
        cleaned_blocks.append(cleaned_block)
        new_index += 1

    # 写入清理后的文件
    output_content = "\n\n".join(cleaned_blocks) + "\n"

    with open(output_file, 'w', encoding='utf-8') as f:
        f.write(output_content)

    print(f"✅ Cleaned SRT saved: {output_file}")
    print(f"   Original blocks: {len(blocks)}")
    print(f"   Cleaned blocks: {len(cleaned_blocks)}")
    print(f"   Removed: {len(blocks) - len(cleaned_blocks)} blocks")

    return True


def main():
    if len(sys.argv) < 2:
        print("Usage: python clean_srt_for_mtv.py <input.srt> [output.srt]")
        print("\nExample:")
        print("  python clean_srt_for_mtv.py 将进酒.srt")
        print("  python clean_srt_for_mtv.py 将进酒.srt 将进酒_clean.srt")
        sys.exit(1)

    input_file = Path(sys.argv[1])
    output_file = Path(sys.argv[2]) if len(sys.argv) > 2 else None

    clean_srt(input_file, output_file)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
自动从浏览器提取 Suno.com 的 cookies 和 JWT token

支持的浏览器：
- Chrome / Chromium
- Brave
- Edge
- Arc
- Firefox
- Safari

Usage:
    python extract_browser_cookies.py [--browser chrome|firefox|safari|arc]
"""

import argparse
import json
import os
import platform
import shutil
import sqlite3
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Optional

# 浏览器配置
BROWSER_PATHS = {
    "dia": {
        "darwin": "~/Library/Application Support/Dia/User Data/Default/Cookies",
    },
    "chrome": {
        "darwin": "~/Library/Application Support/Google/Chrome/Default/Cookies",
        "linux": "~/.config/google-chrome/Default/Cookies",
    },
    "brave": {
        "darwin": "~/Library/Application Support/BraveSoftware/Brave-Browser/Default/Cookies",
        "linux": "~/.config/BraveSoftware/Brave-Browser/Default/Cookies",
    },
    "edge": {
        "darwin": "~/Library/Application Support/Microsoft Edge/Default/Cookies",
        "linux": "~/.config/microsoft-edge/Default/Cookies",
    },
    "arc": {
        "darwin": "~/Library/Application Support/Arc/User Data/Default/Cookies",
    },
    "firefox": {
        "darwin": "~/Library/Application Support/Firefox/Profiles/*/cookies.sqlite",
        "linux": "~/.mozilla/firefox/*/cookies.sqlite",
    },
    "safari": {
        "darwin": "~/Library/Cookies/Cookies.binarycookies",
    },
}


def get_os():
    """获取操作系统类型"""
    system = platform.system().lower()
    if system == "darwin":
        return "darwin"
    elif system == "linux":
        return "linux"
    else:
        return None


def find_cookie_db(browser: str) -> Optional[Path]:
    """查找浏览器的 cookie 数据库"""
    os_type = get_os()
    if not os_type or browser not in BROWSER_PATHS:
        return None

    if os_type not in BROWSER_PATHS[browser]:
        return None

    path_pattern = BROWSER_PATHS[browser][os_type]
    expanded_path = Path(path_pattern).expanduser()

    # 如果路径包含通配符（Firefox）
    if "*" in str(expanded_path):
        import glob
        matches = glob.glob(str(expanded_path))
        if matches:
            return Path(matches[0])
        return None

    # 普通路径
    if expanded_path.exists():
        return expanded_path

    return None


def decrypt_chrome_cookie_mac(encrypted_value: bytes) -> str:
    """解密 macOS Chrome 的加密 cookie"""
    try:
        # macOS Chrome 使用 'v10' 或 'v11' 前缀
        if encrypted_value[:3] in (b'v10', b'v11'):
            try:
                from Crypto.Cipher import AES
                from Crypto.Protocol.KDF import PBKDF2

                # 获取 Chrome Safe Storage 密码
                result = subprocess.run(
                    ['security', 'find-generic-password',
                     '-w', '-s', 'Chrome Safe Storage'],
                    capture_output=True,
                    text=True,
                    timeout=5
                )

                if result.returncode != 0:
                    # 尝试不指定 -a 参数
                    result = subprocess.run(
                        ['security', 'find-generic-password',
                         '-w', '-s', 'Chrome Safe Storage', '-a', 'Chrome'],
                        capture_output=True,
                        text=True,
                        timeout=5
                    )

                if result.returncode == 0:
                    password = result.stdout.strip().encode('utf-8')

                    # 使用 PBKDF2 派生密钥
                    salt = b'saltysalt'
                    iterations = 1003
                    key = PBKDF2(password, salt, dkLen=16, count=iterations)

                    # 解密（v10 使用 AES-128-CBC）
                    cipher = AES.new(key, AES.MODE_CBC, IV=b' ' * 16)
                    decrypted = cipher.decrypt(encrypted_value[3:])

                    # 移除 PKCS7 padding
                    padding_length = decrypted[-1]
                    if isinstance(padding_length, int) and 1 <= padding_length <= 16:
                        decrypted = decrypted[:-padding_length]

                    decoded = decrypted.decode('utf-8', errors='ignore')

                    # JWT token 通常以 'eyJ' 开头，清理前面的无效字符
                    if 'eyJ' in decoded:
                        decoded = decoded[decoded.index('eyJ'):]

                    return decoded
            except Exception as e:
                # 解密失败，返回空字符串
                return ""

        # 如果不是加密的（没有v10/v11前缀），尝试直接解码
        try:
            return encrypted_value.decode('utf-8', errors='ignore')
        except:
            return ""
    except Exception:
        return ""


def decrypt_dia_cookie_mac(encrypted_value: bytes) -> str:
    """解密 macOS Dia 浏览器的加密 cookie

    Dia 格式与标准 Chromium 不同：
      v10 + 16字节nonce(丢弃) + AES-CBC密文
    解密后明文还有 16字节header，真实值从第17字节开始
    """
    try:
        if encrypted_value[:3] not in (b'v10', b'v11'):
            return encrypted_value.decode('utf-8', errors='ignore')

        from Crypto.Cipher import AES
        import hashlib

        result = subprocess.run(
            ['security', 'find-generic-password', '-w', '-s', 'Dia Safe Storage'],
            capture_output=True, text=True, timeout=5
        )
        if result.returncode != 0:
            return ""

        password = result.stdout.strip().encode('utf-8')
        key = hashlib.pbkdf2_hmac('sha1', password, b'saltysalt', 1003, dklen=16)

        # 跳过 v10(3字节) + nonce(16字节)，剩余才是密文
        ciphertext = encrypted_value[3 + 16:]
        if len(ciphertext) == 0:
            return ""

        cipher = AES.new(key, AES.MODE_CBC, IV=b' ' * 16)
        decrypted = cipher.decrypt(ciphertext)
        pad = decrypted[-1]
        result_bytes = decrypted[:-pad] if 1 <= pad <= 16 else decrypted

        # 跳过前16字节的明文header，剩余才是真实cookie值
        value = result_bytes[16:].decode('utf-8', errors='ignore')

        # JWT token 以 'eyJ' 开头，找到起始位置
        if 'eyJ' in value:
            value = value[value.index('eyJ'):]

        return value
    except Exception:
        return ""


def extract_from_chromium(cookie_db: Path) -> dict:
    """从 Chromium 系列浏览器提取 cookies"""
    # 复制数据库文件（避免锁定）
    with tempfile.NamedTemporaryFile(delete=False, suffix='.db') as tmp:
        tmp_path = Path(tmp.name)

    try:
        shutil.copy2(cookie_db, tmp_path)

        conn = sqlite3.connect(tmp_path)
        cursor = conn.cursor()

        # 查询 suno.com 的 cookies
        cursor.execute("""
            SELECT name, encrypted_value, value
            FROM cookies
            WHERE host_key LIKE '%suno.com%'
        """)

        cookies = {}
        os_type = get_os()

        for name, encrypted_value, plain_value in cursor.fetchall():
            # 尝试解密
            if encrypted_value and os_type == "darwin":
                value = decrypt_chrome_cookie_mac(encrypted_value)
            else:
                value = plain_value or ""

            if value:
                cookies[name] = value

        conn.close()

        # 提取 JWT token
        jwt_token = cookies.get("__session", "")

        return {
            "jwt_token": jwt_token,
            "cookies": cookies,
            "cookie_string": "; ".join([f"{k}={v}" for k, v in cookies.items()])
        }

    finally:
        tmp_path.unlink(missing_ok=True)


def extract_from_dia(cookie_db: Path) -> dict:
    """从 Dia 浏览器提取 cookies"""
    with tempfile.NamedTemporaryFile(delete=False, suffix='.db') as tmp:
        tmp_path = Path(tmp.name)

    try:
        shutil.copy2(cookie_db, tmp_path)

        conn = sqlite3.connect(tmp_path)
        cursor = conn.cursor()

        cursor.execute("""
            SELECT name, encrypted_value, value
            FROM cookies
            WHERE host_key LIKE '%suno.com%'
        """)

        cookies = {}
        os_type = get_os()

        for name, encrypted_value, plain_value in cursor.fetchall():
            if encrypted_value and os_type == "darwin":
                value = decrypt_dia_cookie_mac(encrypted_value)
            else:
                value = plain_value or ""

            if value:
                cookies[name] = value

        conn.close()

        jwt_token = cookies.get("__session", "")

        return {
            "jwt_token": jwt_token,
            "cookies": cookies,
            "cookie_string": "; ".join([f"{k}={v}" for k, v in cookies.items()])
        }

    finally:
        tmp_path.unlink(missing_ok=True)


def extract_from_firefox(cookie_db: Path) -> dict:
    """从 Firefox 提取 cookies"""
    with tempfile.NamedTemporaryFile(delete=False, suffix='.sqlite') as tmp:
        tmp_path = Path(tmp.name)

    try:
        shutil.copy2(cookie_db, tmp_path)

        conn = sqlite3.connect(tmp_path)
        cursor = conn.cursor()

        # Firefox 的 cookie 表结构不同
        cursor.execute("""
            SELECT name, value
            FROM moz_cookies
            WHERE host LIKE '%suno.com%'
        """)

        cookies = {}
        for name, value in cursor.fetchall():
            cookies[name] = value

        conn.close()

        jwt_token = cookies.get("__session", "")

        return {
            "jwt_token": jwt_token,
            "cookies": cookies,
            "cookie_string": "; ".join([f"{k}={v}" for k, v in cookies.items()])
        }

    finally:
        tmp_path.unlink(missing_ok=True)


def auto_detect_browser() -> Optional[str]:
    """自动检测可用的浏览器"""
    browsers = ["dia", "chrome", "arc", "brave", "edge", "firefox"]

    for browser in browsers:
        cookie_db = find_cookie_db(browser)
        if cookie_db:
            return browser

    return None


def extract_cookies(browser: Optional[str] = None) -> dict:
    """提取浏览器 cookies"""

    # 自动检测浏览器
    if not browser:
        browser = auto_detect_browser()
        if not browser:
            return {"error": "No supported browser found"}

    # 查找 cookie 数据库
    cookie_db = find_cookie_db(browser)
    if not cookie_db:
        return {"error": f"Cookie database not found for {browser}"}

    print(f"Found {browser} cookies at: {cookie_db}", file=sys.stderr)

    # 根据浏览器类型提取
    if browser in ["chrome", "brave", "edge", "arc"]:
        return extract_from_chromium(cookie_db)
    elif browser == "dia":
        return extract_from_dia(cookie_db)
    elif browser == "firefox":
        return extract_from_firefox(cookie_db)
    else:
        return {"error": f"Unsupported browser: {browser}"}


def main():
    parser = argparse.ArgumentParser(
        description="Extract Suno.com cookies from browser"
    )
    parser.add_argument(
        "--browser", "-b",
        choices=["dia", "chrome", "brave", "edge", "arc", "firefox", "safari"],
        help="Browser to extract from (auto-detect if not specified)"
    )
    parser.add_argument(
        "--json", "-j",
        action="store_true",
        help="Output as JSON"
    )

    args = parser.parse_args()

    # 提取 cookies
    result = extract_cookies(args.browser)

    if "error" in result:
        print(f"Error: {result['error']}", file=sys.stderr)
        sys.exit(1)

    # 输出结果
    if args.json:
        print(json.dumps(result, indent=2))
    else:
        print(f"JWT Token: {result['jwt_token'][:50]}..." if result['jwt_token'] else "No JWT token found")
        print(f"\nCookies found: {len(result['cookies'])}")
        for name, value in result['cookies'].items():
            preview = value[:50] + "..." if len(value) > 50 else value
            print(f"  {name}: {preview}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Suno Aligned Lyrics Fetcher - 获取精准同步歌词

通过本地 suno-api 代理获取带时间戳的同步歌词，生成 LRC/SRT/MD 格式文件。
无需手动管理 JWT Token，由本地服务器自动处理鉴权。

Usage:
    python fetch_aligned_lyrics.py <song_id>
    python fetch_aligned_lyrics.py <song_id1> <song_id2>
    python fetch_aligned_lyrics.py <song_id> --format lrc
    python fetch_aligned_lyrics.py <song_id> --format srt
    python fetch_aligned_lyrics.py <song_id> --format md
    python fetch_aligned_lyrics.py <song_id> --format all   # 默认：lrc + srt + md
    python fetch_aligned_lyrics.py <song_id> --output ~/Music/lyrics
"""

import argparse
import json
import os
import sys
import urllib.request
import urllib.error
from pathlib import Path
from typing import Dict, List, Optional, Tuple

# 配置
SUNO_API_BASE = os.environ.get("SUNO_API_URL", "http://localhost:3000")
DEFAULT_OUTPUT_DIR = Path(os.environ.get("SUNO_MUSIC_DIR", str(Path.home() / "Music" / "Suno")))


def fetch_aligned_lyrics(song_id: str) -> Optional[Dict]:
    """通过本地 suno-api 代理获取带时间戳的同步歌词"""
    url = f"{SUNO_API_BASE}/api/get_aligned_lyrics?song_id={song_id}"
    try:
        with urllib.request.urlopen(url, timeout=30) as response:
            return json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        if e.code == 404:
            print(f"⚠️  No aligned lyrics found for {song_id}", file=sys.stderr)
        else:
            error_body = e.read().decode("utf-8")
            print(f"❌ HTTP {e.code}: {error_body}", file=sys.stderr)
        return None
    except Exception as e:
        print(f"❌ Error fetching aligned lyrics: {e}", file=sys.stderr)
        return None


def fetch_song_metadata(song_id: str) -> Optional[Dict]:
    """通过本地 /api/get 获取歌曲元数据（标题、纯文本歌词等）"""
    url = f"{SUNO_API_BASE}/api/get?ids={song_id}"
    try:
        with urllib.request.urlopen(url, timeout=30) as response:
            data = json.loads(response.read().decode("utf-8"))
            # /api/get 返回列表
            if isinstance(data, list) and data:
                return data[0]
            return data
    except Exception as e:
        print(f"⚠️  Could not fetch song metadata: {e}", file=sys.stderr)
        return None


def parse_aligned_lyrics(data) -> Tuple[List[Dict], Optional[float]]:
    """解析 aligned_lyrics 数据，返回 (lines, duration)。

    本地 suno-api 返回词级别的 list：
    [{"word": "花非花，\n", "start_s": 1.2, "end_s": 2.3}, ...]
    每个 word 可能包含换行符和 Suno 结构标签（[Intro], [Verse 1] 等）。
    注意：结构标签可能被拆分到多个相邻 word item，需先合并再过滤。
    """
    import re

    # 兼容 list（本地 API）和 dict（旧格式）
    if isinstance(data, list):
        word_items = data
    elif isinstance(data, dict):
        word_items = data.get("aligned_lyrics", [])
        if not word_items and "data" in data:
            word_items = data["data"].get("aligned_lyrics", [])
    else:
        return [], None

    if not word_items:
        return [], None

    # 提取时长（dict 格式才有）
    duration = None
    if isinstance(data, dict):
        for key in ("duration_s", "duration"):
            val = data.get(key)
            if isinstance(val, (int, float)) and val > 0:
                duration = float(val)
                break

    # Step 1: 合并所有词，建立字符级时间戳映射
    # 结构标签可能被拆分到多个 word item（如 "[Verse" 在一个，" 1]" 在下一个），
    # 合并后再做正则才能正确匹配完整的 [...] 标签。
    joined = ""
    char_ts: List[Tuple[float, float]] = []  # 每个字符对应的 (start_s, end_s)
    for item in word_items:
        raw = item.get("word", "") or item.get("text", "")
        s = float(item.get("start_s") or 0)
        e = float(item.get("end_s") or 0)
        joined += raw
        char_ts.extend([(s, e)] * len(raw))

    # Step 2: 标记所有结构标签字符位置（合并后正则能匹配跨 item 的标签）
    tag_positions: set = set()
    for m in re.finditer(r'\[[^\]]*\]', joined):
        tag_positions.update(range(m.start(), m.end()))

    # Step 3: 逐字符扫描，按 \n 分行，跳过标签字符
    lines = []
    current_chars: List[str] = []
    current_start: Optional[float] = None
    current_end: Optional[float] = None

    for i, ch in enumerate(joined):
        if i in tag_positions:
            continue  # 跳过结构标签字符

        if ch == "\n":
            text = "".join(current_chars).strip()
            if text and current_start is not None and current_end is not None:
                lines.append({"text": text, "start_s": current_start, "end_s": current_end})
            current_chars = []
            current_start = None
            current_end = None
        else:
            ts = char_ts[i]
            if not current_chars:
                current_start = ts[0]
            current_end = ts[1]
            current_chars.append(ch)

    # 处理最后一行（没有尾随换行符时）
    text = "".join(current_chars).strip()
    if text and current_start is not None and current_end is not None:
        lines.append({"text": text, "start_s": current_start, "end_s": current_end})

    return lines, duration


def fmt_lrc_time(seconds: float) -> str:
    """[mm:ss.xx]"""
    m = int(seconds // 60)
    s = int(seconds % 60)
    h = int((seconds % 1) * 100)
    return f"[{m:02d}:{s:02d}.{h:02d}]"


def fmt_srt_time(seconds: float) -> str:
    """hh:mm:ss,mmm"""
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = int(seconds % 60)
    ms = int((seconds % 1) * 1000)
    return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"


def fmt_md_time(seconds: float) -> str:
    """[mm:ss] 简洁格式，方便 AI 读取"""
    m = int(seconds // 60)
    s = int(seconds % 60)
    return f"[{m:02d}:{s:02d}]"


def generate_lrc(lines: List[Dict], title: str = "") -> str:
    parts = []
    if title:
        parts.append(f"[ti:{title}]")
    parts.append("[by:Suno AI]")
    parts.append("")
    for line in lines:
        parts.append(f"{fmt_lrc_time(line['start_s'])}{line['text']}")
    return "\n".join(parts)


def generate_srt(lines: List[Dict]) -> str:
    parts = []
    for i, line in enumerate(lines, 1):
        parts.append(str(i))
        parts.append(f"{fmt_srt_time(line['start_s'])} --> {fmt_srt_time(line['end_s'])}")
        parts.append(line["text"])
        parts.append("")
    return "\n".join(parts)


def generate_md(lines: List[Dict], title: str = "", song_id: str = "") -> str:
    """生成 Markdown 格式带时间戳歌词，方便 AI 读取"""
    parts = []
    if title:
        parts.append(f"# {title}")
        parts.append("")
    if song_id:
        parts.append(f"**Song ID**: `{song_id}`")
        parts.append("")
    parts.append("## 带时间戳歌词")
    parts.append("")
    parts.append("| 时间 | 结束 | 歌词 |")
    parts.append("|------|------|------|")
    for line in lines:
        start = fmt_md_time(line["start_s"])
        end = fmt_md_time(line["end_s"])
        text = line["text"].replace("|", "\\|")
        parts.append(f"| {start} | {end} | {text} |")
    parts.append("")
    parts.append("## 纯文本歌词（无时间戳）")
    parts.append("")
    for line in lines:
        parts.append(line["text"])
    return "\n".join(parts)


def save_lyrics(song_id: str, lines: List[Dict], output_dir: Path,
                format_type: str = "all", title: str = "", name: str = "") -> List[str]:
    """保存歌词文件，返回保存的文件路径列表"""
    output_dir.mkdir(parents=True, exist_ok=True)

    # 优先使用显式传入的 name（对应 MP3 文件名词干），其次从 title 生成
    if name:
        base_name = name
    else:
        safe_title = "".join(c for c in title if c.isalnum() or c in " -_").strip()
        base_name = safe_title if safe_title else song_id[:8]

    saved = []

    if format_type in ("lrc", "all", "both"):
        path = output_dir / f"{base_name}.lrc"
        path.write_text(generate_lrc(lines, title=title), encoding="utf-8")
        saved.append(str(path))
        print(f"✅ LRC saved: {path}")

    if format_type in ("srt", "all", "both"):
        path = output_dir / f"{base_name}.srt"
        path.write_text(generate_srt(lines), encoding="utf-8")
        saved.append(str(path))
        print(f"✅ SRT saved: {path}")

    if format_type in ("md", "all"):
        path = output_dir / f"{base_name}.lyrics.md"
        path.write_text(generate_md(lines, title=title, song_id=song_id), encoding="utf-8")
        saved.append(str(path))
        print(f"✅ MD  saved: {path}")

    return saved


def process_song(song_id: str, output_dir: Path, format_type: str = "all", name: str = "") -> bool:
    print(f"\n🎵 Processing song: {song_id}")

    # 1. 获取精准时间戳歌词
    print("📡 Fetching aligned lyrics via local API...")
    lyrics_data = fetch_aligned_lyrics(song_id)

    if not lyrics_data:
        # 降级：通过本地 /api/get 获取纯文本歌词
        print("⚠️  No aligned lyrics, falling back to plain lyrics...")
        meta = fetch_song_metadata(song_id)
        if meta:
            title = meta.get("title", song_id[:8])
            prompt = meta.get("metadata", {}).get("prompt", "") or meta.get("lyric", "")
            if prompt:
                base = name if name else "".join(c for c in title if c.isalnum() or c in " -_").strip()
                txt_path = output_dir / f"{base}.txt"
                txt_path.write_text(prompt, encoding="utf-8")
                print(f"📝 Plain lyrics saved: {txt_path}")
                return True
        print("❌ No lyrics found", file=sys.stderr)
        return False

    # 2. 解析
    lines, duration = parse_aligned_lyrics(lyrics_data)
    if not lines:
        print("⚠️  No valid lyrics lines found", file=sys.stderr)
        return False

    print(f"✅ Found {len(lines)} lines with timestamps")
    if duration:
        print(f"⏱️  Duration: {duration:.2f}s")

    # 3. 获取歌曲标题（用于 MD header，不影响文件名）
    meta = fetch_song_metadata(song_id)
    title = meta.get("title", song_id[:8]) if meta else song_id[:8]

    # 4. 保存（name 优先用于文件名）
    saved = save_lyrics(song_id, lines, output_dir, format_type, title, name=name)
    return len(saved) > 0


def main():
    parser = argparse.ArgumentParser(description="Fetch aligned lyrics from Suno (via local API proxy)")
    parser.add_argument("song_ids", nargs="+", help="Song ID(s)")
    parser.add_argument(
        "--format", "-f",
        choices=["lrc", "srt", "md", "all", "both"],
        default="all",
        help="Output format: lrc / srt / md / all (default: all = lrc+srt+md)"
    )
    parser.add_argument(
        "--output", "-o",
        type=Path,
        default=DEFAULT_OUTPUT_DIR,
        help=f"Output directory (default: {DEFAULT_OUTPUT_DIR})"
    )
    parser.add_argument(
        "--name", "-n",
        default="",
        help="Base filename (without extension), overrides title-derived name. Used to match MP3 filename."
    )

    args = parser.parse_args()

    success = 0
    for song_id in args.song_ids:
        if process_song(song_id, args.output, args.format, name=args.name):
            success += 1

    print(f"\n{'='*50}")
    print(f"✅ Processed {success}/{len(args.song_ids)} song(s)")
    print(f"📁 Output: {args.output}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Suno Music Generator - 调用 Suno API 生成音乐

Usage:
    python generate_music.py '{"title":"歌名","prompt":"歌词","tags":"风格","negative_tags":"排除风格"}'

    # 或者只传歌词，使用默认设置
    python generate_music.py --prompt "歌词内容" --title "歌名" --tags "pop, rock"
"""

import argparse
import json
import os
import subprocess
import sys
import time
import urllib.request
import urllib.error
from pathlib import Path

# 配置
SUNO_API_URL = os.environ.get("SUNO_API_URL", "http://localhost:3000")
DEFAULT_MODEL = "chirp-crow"  # V5
MUSIC_OUTPUT_DIR = Path(os.environ.get("SUNO_MUSIC_DIR", str(Path.home() / "Music" / "Suno")))


def check_token_validity() -> dict:
    """检查当前 token 是否有效（通过调用 /api/get_limit 确认服务器认证状态）"""
    url = f"{SUNO_API_URL}/api/get_limit"
    try:
        with urllib.request.urlopen(url, timeout=10) as response:
            data = json.loads(response.read().decode("utf-8"))
            if "credits_left" in data:
                return {"valid": True, "remaining_minutes": 9999, "credits_left": data["credits_left"]}
            return {"valid": False, "error": "Unexpected response from /api/get_limit"}
    except urllib.error.HTTPError as e:
        if e.code in (401, 403):
            return {"valid": False, "reason": "expired"}
        return {"valid": False, "error": f"HTTP {e.code}"}
    except Exception as e:
        return {"valid": False, "error": str(e)}


def auto_refresh_token_cdp() -> bool:
    """使用 CDP 自动刷新 token"""
    script_dir = Path(__file__).parent
    cdp_script = script_dir / "refresh_suno_token_cdp.py"

    if not cdp_script.exists():
        print(f"❌ CDP refresh script not found: {cdp_script}")
        return False

    print("🔄 Token expired, auto-refreshing via CDP...")
    print("💡 This will connect to your Chrome browser (remote debugging)")
    try:
        result = subprocess.run(
            [sys.executable, str(cdp_script)],
            capture_output=True,
            text=True,
            timeout=120  # CDP 刷新可能需要更长时间
        )

        if result.returncode == 0 and "Token refresh completed successfully" in result.stdout:
            print("✅ Token refreshed successfully!")
            return True
        else:
            print(f"❌ Token refresh failed")
            if result.stdout:
                print(result.stdout)
            if result.stderr:
                print(result.stderr)
            return False
    except subprocess.TimeoutExpired:
        print("❌ Token refresh timeout (120s)")
        return False
    except Exception as e:
        print(f"❌ Auto-refresh error: {e}")
        return False


def generate_music(data: dict, retry_on_401: bool = True) -> dict:
    """调用 Suno API 生成音乐"""
    url = f"{SUNO_API_URL}/api/custom_generate"

    # 确保必要字段
    payload = {
        "title": data.get("title", "Untitled"),
        "prompt": data.get("prompt", ""),
        "tags": data.get("tags", "pop"),
        "negative_tags": data.get("negative_tags", ""),
        "make_instrumental": data.get("make_instrumental", False),
        "model": data.get("model", DEFAULT_MODEL),
        "wait_audio": data.get("wait_audio", False),
    }

    # 发送请求
    req = urllib.request.Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST"
    )

    try:
        with urllib.request.urlopen(req, timeout=60) as response:
            return json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        # 如果是 401 且允许重试，尝试自动刷新 token
        if e.code == 401 and retry_on_401:
            if auto_refresh_token_cdp():
                # 重试一次
                print("🔄 Retrying with new token...")
                time.sleep(2)
                return generate_music(data, retry_on_401=False)

        error_body = e.read().decode("utf-8")
        return {"error": f"HTTP {e.code}: {error_body}"}
    except urllib.error.URLError as e:
        return {"error": f"Connection failed: {e.reason}"}

def get_song_status(song_ids: list) -> dict:
    """查询歌曲生成状态"""
    ids_param = ",".join(song_ids)
    url = f"{SUNO_API_URL}/api/get?ids={ids_param}"

    try:
        with urllib.request.urlopen(url, timeout=30) as response:
            return json.loads(response.read().decode("utf-8"))
    except Exception as e:
        return {"error": str(e)}


def download_song(audio_url: str, title: str, song_id: str, lyrics_data: dict = None) -> str:
    """下载歌曲到 AI音乐/{歌名}/ 子目录，同时保存歌词"""
    # 生成安全的标题（用于文件夹名和文件名）
    safe_title = "".join(c for c in title if c.isalnum() or c in " -_").strip()
    if not safe_title:
        safe_title = song_id[:8]

    # 每首歌独立子目录
    song_dir = MUSIC_OUTPUT_DIR / safe_title
    song_dir.mkdir(parents=True, exist_ok=True)

    filename = f"{safe_title}.mp3"
    filepath = song_dir / filename

    # 避免重名（同一首歌的多个版本放在同一目录）
    counter = 1
    while filepath.exists():
        filename = f"{safe_title}-{counter}.mp3"
        filepath = song_dir / filename
        counter += 1

    print(f"📥 Downloading to {filepath}...")
    try:
        req = urllib.request.Request(audio_url, headers={"User-Agent": "Mozilla/5.0"})
        with urllib.request.urlopen(req, timeout=120) as response:
            with open(filepath, "wb") as f:
                f.write(response.read())

        # 保存歌词文件（与 MP3 同名，扩展名为 .lyrics.json）
        if lyrics_data:
            lyrics_path = filepath.with_suffix(".lyrics.json")
            lyrics_content = {
                "title": title,
                "prompt": lyrics_data.get("prompt", ""),
                "tags": lyrics_data.get("tags", ""),
                "negative_tags": lyrics_data.get("negative_tags", ""),
                "model": lyrics_data.get("model", DEFAULT_MODEL),
                "song_id": song_id,
                "audio_file": filepath.name,
                "generated_at": time.strftime("%Y-%m-%dT%H:%M:%S")
            }
            with open(lyrics_path, "w", encoding="utf-8") as f:
                json.dump(lyrics_content, f, ensure_ascii=False, indent=2)
            print(f"📝 Lyrics saved to {lyrics_path}")

        # 自动获取带时间戳的歌词 (SRT/LRC/MD)
        extra_files = []
        try:
            script_dir = Path(__file__).parent
            fetch_script = script_dir / "fetch_aligned_lyrics.py"

            if fetch_script.exists():
                result = subprocess.run(
                    [sys.executable, str(fetch_script), song_id,
                     "--output", str(song_dir), "--name", filepath.stem],
                    capture_output=True,
                    text=True,
                    timeout=30
                )

                if result.returncode == 0:
                    # 收集所有生成的歌词文件
                    for ext in (".lrc", ".srt", ".lyrics.md"):
                        candidate = song_dir / (filepath.stem + ext)
                        if candidate.exists():
                            extra_files.append(candidate)
                else:
                    if result.stderr and "No aligned lyrics found" not in result.stderr:
                        print(f"⚠️  Aligned lyrics fetch failed (non-critical)")
        except Exception as e:
            print(f"⚠️  Could not fetch aligned lyrics (non-critical): {e}")

        return {"mp3": str(filepath), "extras": extra_files}
    except Exception as e:
        return f"Download failed: {e}"


def wait_and_download(song_ids: list, max_wait: int = 300, lyrics_data: dict = None) -> list:
    """等待歌曲生成完成并下载"""
    print(f"⏳ Waiting for {len(song_ids)} song(s) to complete...")
    start_time = time.time()
    downloaded = []

    while time.time() - start_time < max_wait:
        result = get_song_status(song_ids)
        if "error" in result:
            print(f"❌ Error checking status: {result['error']}")
            time.sleep(10)
            continue

        all_done = True
        for song in result:
            status = song.get("status", "")
            title = song.get("title", "Unknown")
            song_id = song.get("id", "")

            if status in ("complete", "streaming"):
                audio_url = song.get("audio_url", "")
                if audio_url and song_id not in [d.get("id") for d in downloaded]:
                    result = download_song(audio_url, title, song_id, lyrics_data)
                    mp3_path = result["mp3"] if isinstance(result, dict) else result
                    extras = result.get("extras", []) if isinstance(result, dict) else []
                    downloaded.append({"id": song_id, "title": title, "path": mp3_path, "extras": extras})
                    print(f"✅ Downloaded: {title} → {mp3_path}")
            elif status == "error":
                print(f"❌ Generation failed: {title} - {song.get('error_message', 'Unknown error')}")
            else:
                all_done = False
                print(f"⏳ {title}: {status}...")

        if all_done:
            break
        time.sleep(15)

    return downloaded

def main():
    parser = argparse.ArgumentParser(description="Generate music using Suno API")
    parser.add_argument("json_data", nargs="?", help="JSON data with title, prompt, tags, etc.")
    parser.add_argument("--prompt", "-p", help="Lyrics/prompt")
    parser.add_argument("--title", "-t", help="Song title")
    parser.add_argument("--tags", help="Style tags (comma separated)")
    parser.add_argument("--negative-tags", help="Excluded style tags")
    parser.add_argument("--model", "-m", default=DEFAULT_MODEL, help="Model version")
    parser.add_argument("--wait", "-w", action="store_true", help="Wait for audio generation")
    parser.add_argument("--instrumental", "-i", action="store_true", help="Instrumental only")
    parser.add_argument("--status", "-s", help="Check status of song IDs (comma separated)")
    parser.add_argument("--download", "-d", action="store_true", help="Wait and download after generation")

    args = parser.parse_args()

    # 查询状态模式
    if args.status:
        song_ids = [s.strip() for s in args.status.split(",")]
        if args.download:
            # 等待并下载
            downloaded = wait_and_download(song_ids)
            if downloaded:
                print()
                print(f"📁 Songs saved to: {MUSIC_OUTPUT_DIR}")
                for d in downloaded:
                    print(f"\n   🎵 {d['title']}")
                    print(f"      MP3 : {d['path']}")
                    for f in d.get("extras", []):
                        ext = Path(f).suffix.lstrip(".")
                        print(f"      {ext.upper():<5}: {f}")
        else:
            result = get_song_status(song_ids)
            print(json.dumps(result, indent=2, ensure_ascii=False))
        return

    # 构建数据
    if args.json_data:
        try:
            data = json.loads(args.json_data)
        except json.JSONDecodeError as e:
            print(f"Error: Invalid JSON - {e}", file=sys.stderr)
            sys.exit(1)
    elif args.prompt:
        data = {
            "title": args.title or "Untitled",
            "prompt": args.prompt,
            "tags": args.tags or "pop",
            "negative_tags": args.negative_tags or "",
            "make_instrumental": args.instrumental,
            "model": args.model,
            "wait_audio": args.wait,
        }
    else:
        parser.print_help()
        sys.exit(1)

    # 🔍 生成前检查 API 是否可用（通过 /api/get_limit 验证）
    print("🔍 Checking API availability...")
    token_status = check_token_validity()

    if not token_status.get("valid"):
        reason = token_status.get('reason') or token_status.get('error', 'Unknown')
        print(f"⚠️  API check failed: {reason}")
        print("⚠️  Will try generation anyway — server manages auth internally")
    else:
        credits = token_status.get("credits_left", "?")
        print(f"✅ API ready (credits left: {credits})")

    # 生成音乐
    print(f"🎵 Generating music with {data.get('model', DEFAULT_MODEL)}...")
    print(f"   Title: {data.get('title', 'Untitled')}")
    print(f"   Tags: {data.get('tags', 'pop')}")
    print()

    result = generate_music(data)

    if "error" in result:
        print(f"❌ Error: {result['error']}", file=sys.stderr)
        sys.exit(1)

    # 输出结果
    print("✅ Music generation started!")
    print()

    if isinstance(result, list):
        for song in result:
            print(f"   Song ID: {song.get('id')}")
            print(f"   Title: {song.get('title')}")
            print(f"   Status: {song.get('status')}")
            print(f"   Model: {song.get('model_name')}")
            print()

        song_ids = [s.get("id") for s in result if s.get("id")]

        # 如果指定了 --download，等待并下载
        if args.download and song_ids:
            print()
            downloaded = wait_and_download(song_ids, lyrics_data=data)
            if downloaded:
                print()
                print(f"📁 Songs saved to: {MUSIC_OUTPUT_DIR}")
                for d in downloaded:
                    print(f"   - {d['title']}: {d['path']}")
        elif song_ids:
            # 提供查询命令
            print("📋 Check status with:")
            print(f"   python {__file__} --status {','.join(song_ids)}")
            print()
            print("   Or download when ready:")
            print(f"   python {__file__} --status {','.join(song_ids)} --download")
            print()
            print(f"📁 Music will be saved to: {MUSIC_OUTPUT_DIR}")
    else:
        print(json.dumps(result, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Suno Token Auto Refresh - 连接到已运行的 Chrome
Connect to existing Chrome via CDP (Chrome DevTools Protocol)

使用方法：
1. 先启动 Chrome（开启远程调试）：
   "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" \
     --remote-debugging-port=9222 \
     --user-data-dir="/tmp/chrome-debug-profile"

   注意：必须使用临时数据目录（/tmp/chrome-debug-profile），
        使用默认数据目录会导致 Chrome 拒绝开启远程调试端口。

2. 在 Chrome 中手动登录 Suno（只需要做一次）

3. 运行此脚本：
   python refresh_suno_token_cdp.py
"""

import os
import sys
import time
import subprocess
from pathlib import Path
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout

# 配置路径
SCRIPT_DIR = Path(__file__).parent
ENV_FILE = Path.home() / "Dropbox/code/suno-api/.env"

# Chrome CDP 地址
CDP_URL = "http://localhost:9222"

# Suno 网站
SUNO_APP_URL = "https://suno.com/create"


class SunoTokenRefresher:
    def __init__(self):
        self.captured_token = None

    def extract_token_from_request(self, request):
        """从网络请求中提取 JWT Token"""
        try:
            if "studio-api.prod.suno.com" in request.url or "suno.com/api" in request.url:
                headers = request.headers
                auth_header = headers.get("authorization", "")

                if auth_header.startswith("Bearer "):
                    token = auth_header.replace("Bearer ", "").strip()
                    if len(token) > 100:  # JWT Token 很长
                        print(f"🎯 Captured JWT Token from: {request.url[:80]}...")
                        self.captured_token = token
                        return token
        except Exception as e:
            pass
        return None

    def update_env_file(self, token):
        """更新 .env 文件中的 SUNO_COOKIE"""
        if not ENV_FILE.exists():
            print(f"❌ .env file not found: {ENV_FILE}")
            return False

        try:
            # 读取现有 .env 内容
            with open(ENV_FILE, 'r') as f:
                lines = f.readlines()

            # 更新 SUNO_COOKIE
            new_lines = []
            found = False
            for line in lines:
                if line.strip().startswith('SUNO_COOKIE='):
                    new_lines.append(f'SUNO_COOKIE="__session={token}"\n')
                    found = True
                else:
                    new_lines.append(line)

            # 如果没找到，追加
            if not found:
                new_lines.append(f'SUNO_COOKIE="__session={token}"\n')

            # 写回文件
            with open(ENV_FILE, 'w') as f:
                f.writelines(new_lines)

            print(f"✅ Updated {ENV_FILE}")
            return True
        except Exception as e:
            print(f"❌ Failed to update .env: {e}")
            return False

    def restart_suno_api(self):
        """重启 suno-api 服务"""
        print("🔄 Restarting suno-api service...")
        try:
            # 杀掉现有进程
            subprocess.run(
                "pkill -f 'node.*suno-api' || true",
                shell=True,
                check=False
            )
            time.sleep(2)

            # 启动新进程（后台运行）
            suno_api_dir = Path.home() / "Dropbox/code/suno-api"
            subprocess.Popen(
                "npm run dev > /dev/null 2>&1 &",
                shell=True,
                cwd=suno_api_dir
            )

            print("✅ suno-api service restarted")
            time.sleep(5)
            return True
        except Exception as e:
            print(f"❌ Failed to restart service: {e}")
            return False

    def verify_token(self, token):
        """验证 Token 是否有效"""
        print("🔍 Verifying token...")
        try:
            result = subprocess.run(
                f'python {SCRIPT_DIR}/update_token.py --check',
                shell=True,
                capture_output=True,
                text=True
            )

            if "Valid: Yes" in result.stdout:
                print("✅ Token is valid!")
                for line in result.stdout.split('\n'):
                    if 'Remaining:' in line or 'Expires:' in line:
                        print(f"   {line.strip()}")
                return True
            else:
                print("⚠️  Token validation uncertain")
                print(result.stdout)
                return False
        except Exception as e:
            print(f"❌ Verification failed: {e}")
            return False

    def check_chrome_running(self):
        """检查 Chrome 是否在运行并开启了远程调试"""
        print("🔍 Checking Chrome remote debugging...")
        try:
            import requests
            response = requests.get(f"{CDP_URL}/json/version", timeout=2)
            if response.status_code == 200:
                data = response.json()
                print(f"✅ Chrome connected: {data.get('Browser', 'Unknown')}")
                return True
        except Exception as e:
            print(f"❌ Cannot connect to Chrome at {CDP_URL}")
            print(f"   Error: {e}")
            return False

    def run(self):
        """主运行流程"""
        print("🚀 Suno Token Auto Refresh (CDP Mode)\n")

        # 检查 Chrome 是否运行
        if not self.check_chrome_running():
            print("\n💡 请先启动 Chrome（开启远程调试）：\n")
            print('"/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" --remote-debugging-port=9222\n')
            print("然后重新运行此脚本。")
            return

        try:
            with sync_playwright() as p:
                print("🌐 Connecting to Chrome via CDP...")
                browser = p.chromium.connect_over_cdp(CDP_URL)

                # 获取所有已打开的上下文
                contexts = browser.contexts
                if not contexts:
                    print("❌ No browser context found")
                    print("💡 请先在 Chrome 中打开一个窗口")
                    return

                context = contexts[0]
                pages = context.pages

                # 如果没有页面，创建新页面
                if not pages:
                    page = context.new_page()
                else:
                    # 使用第一个页面
                    page = pages[0]

                # 设置请求监听
                page.on("request", self.extract_token_from_request)

                print(f"📍 Current URL: {page.url}")

                # 如果不在 Suno，则导航到 Suno
                if "suno.com" not in page.url:
                    print(f"📍 Navigating to {SUNO_APP_URL}...")
                    try:
                        page.goto(SUNO_APP_URL, wait_until="domcontentloaded", timeout=60000)
                    except Exception as e:
                        print(f"⚠️  Navigation warning: {e}")

                print("\n" + "="*60)
                print("👉 请检查 Chrome 窗口：")
                print("   1. 确保已登录 Suno")
                print("   2. 如果未登录，请现在登录")
                print("   3. 脚本将在 60 秒后自动继续")
                print("="*60)

                print("\n⏳ 等待 60 秒...")
                wait_time = 60
                for i in range(wait_time):
                    time.sleep(1)
                    if i > 0 and i % 15 == 0:
                        remaining = wait_time - i
                        print(f"   还剩 {remaining} 秒...")

                print("\n✅ 继续执行...\n")

                # 访问创建页面触发 API 请求
                if "/create" not in page.url:
                    print(f"📍 Navigating to {SUNO_APP_URL}...")
                    try:
                        page.goto(SUNO_APP_URL, wait_until="domcontentloaded", timeout=60000)
                    except Exception as e:
                        print(f"⚠️  Navigation warning: {e}")

                # 等待 API 请求
                print("⏳ Waiting for API requests (15 seconds)...")
                time.sleep(15)

                # 如果还没捕获到，尝试交互触发
                if not self.captured_token:
                    print("🔄 Trying to trigger more API requests...")
                    try:
                        # 滚动页面
                        page.mouse.wheel(0, 500)
                        time.sleep(3)

                        # 刷新页面
                        print("🔄 Refreshing page...")
                        page.reload(wait_until="domcontentloaded", timeout=60000)
                        time.sleep(10)
                    except Exception as e:
                        print(f"⚠️  Interaction error: {e}")

                token = self.captured_token

                if token:
                    print(f"\n✅ Token captured! (Length: {len(token)})")
                    print(f"   Token preview: {token[:50]}...{token[-50:]}\n")

                    # 更新 .env 文件
                    if self.update_env_file(token):
                        # 重启服务
                        self.restart_suno_api()

                        # 验证 Token
                        self.verify_token(token)

                        print("\n" + "="*60)
                        print("✅ Token refresh completed successfully!")
                        print("="*60)
                    else:
                        print("\n❌ Failed to update configuration")
                else:
                    print("\n❌ Failed to capture token")
                    print("💡 Possible reasons:")
                    print("   1. Not logged in to Suno")
                    print("   2. Network issue")
                    print("   3. Try manually triggering requests (click around the page)")

                browser.close()

        except Exception as e:
            print(f"\n❌ Error: {e}")
            import traceback
            traceback.print_exc()


def main():
    refresher = SunoTokenRefresher()
    refresher.run()


if __name__ == "__main__":
    main()