cube/apps/music/scripts/import_guitar.py

#!/usr/bin/env python3
"""
从 /data/guitar/ 把旧 guitar app 的曲库导入到 music API。

跑法（在 oci 上，源数据原地）:
    python3 import_guitar.py --src /data/guitar --api https://music.famzheng.me

每个 mp3 = 一个 piece：
- title / artist 从文件名 "{title} - {artist}.mp3" 拆
- lyrics 来自同名 .lrc（UTF-8）
- chord png 来自 chords/{sanitize(artist)}-{sanitize(title)}.png（如有）
- play_count 来自 playcounts.json[mp3 文件名]
- category 设为 "流行"（旧 guitar 里都是流行/吉他歌）

幂等：以 (title, artist) 作主键去重，已有 piece 跳过附件重传。
"""

import argparse
import json
import re
import sys
import urllib.request
import urllib.parse
import urllib.error
import os
from pathlib import Path

CATEGORY = "流行"


def sanitize(text: str) -> str:
    """匹配旧 chord_server.py 的 sanitize：替换非法文件名字符为 -"""
    return re.sub(r'[<>:"/\\|?*]', '-', text).strip()


def parse_filename(stem: str) -> tuple[str, str]:
    """'{title} - {artist}' → (title, artist)"""
    parts = stem.split(' - ')
    title = parts[0].strip()
    artist = ' - '.join(parts[1:]).strip() if len(parts) > 1 else ''
    return title, artist


def http_request(method: str, url: str, *, json_body=None, file_field=None,
                 timeout: int = 600) -> dict:
    """简单 HTTP 客户端，return JSON dict。失败抛异常。"""
    headers = {}
    data = None
    if json_body is not None:
        data = json.dumps(json_body).encode('utf-8')
        headers['Content-Type'] = 'application/json'
    elif file_field is not None:
        boundary = '----music-import-' + os.urandom(8).hex()
        body = []
        for field_name, filename, content, mime in file_field:
            body.append(f'--{boundary}\r\n'.encode())
            disp = (f'Content-Disposition: form-data; name="{field_name}"; '
                    f'filename="{filename}"\r\n').encode()
            body.append(disp)
            body.append(f'Content-Type: {mime}\r\n\r\n'.encode())
            body.append(content)
            body.append(b'\r\n')
        body.append(f'--{boundary}--\r\n'.encode())
        data = b''.join(body)
        headers['Content-Type'] = f'multipart/form-data; boundary={boundary}'
    req = urllib.request.Request(url, data=data, method=method, headers=headers)
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read().decode('utf-8'))


def list_pieces(api: str) -> list:
    return http_request('GET', f'{api}/api/pieces')


def create_piece(api: str, title: str, artist: str, lyrics: str | None) -> int:
    body = {
        'title': title,
        'artist': artist or None,
        'category': CATEGORY,
        'lyrics': lyrics or None,
    }
    r = http_request('POST', f'{api}/api/pieces', json_body=body)
    return int(r['id'])


def patch_play_count(api: str, piece_id: int, count: int):
    http_request('PATCH', f'{api}/api/pieces/{piece_id}',
                 json_body={'play_count': count})


def upload_attachment(api: str, piece_id: int, path: Path,
                      mime: str, role: str | None = None) -> dict:
    url = f'{api}/api/pieces/{piece_id}/attachments'
    if role:
        url += '?role=' + urllib.parse.quote(role)
    bytes_content = path.read_bytes()
    return http_request('POST', url, file_field=[
        ('files', path.name, bytes_content, mime),
    ])


def detect_audio_mime(path: Path) -> str:
    ext = path.suffix.lower()
    return {
        '.mp3': 'audio/mpeg',
        '.m4a': 'audio/mp4',
        '.flac': 'audio/flac',
        '.ogg': 'audio/ogg',
        '.wav': 'audio/wav',
    }.get(ext, 'application/octet-stream')


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--src', default='/data/guitar', help='guitar 数据根目录')
    ap.add_argument('--api', required=True, help='music API base，e.g. https://music.famzheng.me')
    ap.add_argument('--dry-run', action='store_true', help='只打印不实际调用')
    ap.add_argument('--limit', type=int, default=0, help='测试用，最多导入 N 首（0=全部）')
    args = ap.parse_args()

    src = Path(args.src)
    if not src.exists():
        print(f'[!] {src} 不存在', file=sys.stderr)
        sys.exit(1)

    chords_dir = src / 'chords'
    playcounts_file = src / 'playcounts.json'
    playcounts = {}
    if playcounts_file.exists():
        try:
            playcounts = json.loads(playcounts_file.read_text())
        except Exception as e:
            print(f'[!] 读 playcounts.json 失败: {e}', file=sys.stderr)

    # 已有 pieces 去重
    existing = {}
    if not args.dry_run:
        try:
            for p in list_pieces(args.api):
                key = (p['title'], p.get('artist') or '')
                existing[key] = p['id']
            print(f'[i] 远端已有 {len(existing)} 首曲目，重复的会跳过创建')
        except urllib.error.URLError as e:
            print(f'[!] 连不上 {args.api}: {e}', file=sys.stderr)
            sys.exit(1)

    mp3s = sorted(src.glob('*.mp3'))
    if args.limit:
        mp3s = mp3s[:args.limit]

    total = len(mp3s)
    ok = skipped = failed = 0

    for i, mp3 in enumerate(mp3s, 1):
        stem = mp3.stem
        title, artist = parse_filename(stem)
        if not title:
            print(f'[!] [{i}/{total}] 无法解析文件名: {mp3.name}', file=sys.stderr)
            failed += 1
            continue

        key = (title, artist)
        prefix = f'[{i}/{total}] {title} - {artist}'

        if key in existing:
            print(f'  ↻ {prefix} (已存在 id={existing[key]}, 跳过)')
            skipped += 1
            continue

        # lrc
        lrc_path = mp3.with_suffix('.lrc')
        lyrics = None
        if lrc_path.exists():
            try:
                lyrics = lrc_path.read_text(encoding='utf-8')
            except UnicodeDecodeError:
                lyrics = lrc_path.read_text(encoding='gbk', errors='replace')

        # chord png
        chord_path = chords_dir / f'{sanitize(artist)}-{sanitize(title)}.png'
        if not chord_path.exists():
            chord_path = None

        play_count = int(playcounts.get(mp3.name, 0))

        if args.dry_run:
            print(f'  + {prefix}'
                  f'{" [词]" if lyrics else ""}'
                  f'{" [谱]" if chord_path else ""}'
                  f'{f" [{play_count}次]" if play_count else ""}')
            ok += 1
            continue

        try:
            piece_id = create_piece(args.api, title, artist, lyrics)
            print(f'  + {prefix} → id={piece_id}', end='', flush=True)

            upload_attachment(args.api, piece_id, mp3,
                              detect_audio_mime(mp3))
            print(' [audio]', end='', flush=True)

            if chord_path:
                upload_attachment(args.api, piece_id, chord_path,
                                  'image/png', role='chord')
                print(' [chord]', end='', flush=True)

            if play_count:
                patch_play_count(args.api, piece_id, play_count)
                print(f' [{play_count}次]', end='', flush=True)

            print()
            ok += 1
        except Exception as e:
            print(f'\n[!] {prefix} 失败: {e}', file=sys.stderr)
            failed += 1

    print()
    print(f'完成: ok={ok}  skipped={skipped}  failed={failed}  total={total}')


if __name__ == '__main__':
    main()