cube/apps/music/chord/chord_server.py

"""
chord-fetcher sidecar 的 HTTP service。

跟 music 主容器同 pod，监听 :8001。被 music backend 通过 localhost 调用。
worker 单线程串行（chromium 一次跑一个，省资源），文件落 /data/chord-fetch/{piece_id}.png。
"""

import json
import logging
import queue
import sys
import threading
import os
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse

# 调试热更：/data 是 PVC mount，重启容器不丢；放 yopu.py 在 /data/chord-overrides/
# 启动时把它放最高优先级，方便不重 build image 直接 hot-fix selector。
_OVERRIDE_DIR = Path('/data/chord-overrides')
_OVERRIDE_DIR.mkdir(parents=True, exist_ok=True)
if (_OVERRIDE_DIR / 'yopu.py').exists():
    sys.path.insert(0, str(_OVERRIDE_DIR))
    print(f"[chord-server] using yopu.py override from {_OVERRIDE_DIR}")

import yopu  # noqa: E402

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s %(levelname)s %(name)s: %(message)s')
logger = logging.getLogger('chord-server')

OUT_DIR = Path(os.getenv('CHORD_OUT_DIR', '/data/chord-fetch'))
OUT_DIR.mkdir(parents=True, exist_ok=True)

app = FastAPI()

VALID_MODES = ('letters', 'functional')

# in-memory job state. (piece_id, mode) -> {status, error, title, artist}
state: dict[tuple[int, str], dict] = {}
state_lock = threading.Lock()
job_q: queue.Queue = queue.Queue()


def out_path(piece_id: int, mode: str) -> Path:
    return OUT_DIR / f"{piece_id}-{mode}.png"


def worker():
    while True:
        piece_id, mode, title, artist = job_q.get()
        key = (piece_id, mode)
        with state_lock:
            state[key] = {'status': 'processing', 'error': '', 'title': title, 'artist': artist}
        logger.info("[piece=%d mode=%s] start: title=%r artist=%r", piece_id, mode, title, artist)
        try:
            ok, msg = yopu.fetch_chord_chart(title, artist, str(out_path(piece_id, mode)), mode=mode)
            with state_lock:
                if ok:
                    state[key] = {'status': 'completed', 'error': '', 'title': title, 'artist': artist}
                    logger.info("[piece=%d mode=%s] completed: %s", piece_id, mode, msg)
                else:
                    state[key] = {'status': 'failed', 'error': msg, 'title': title, 'artist': artist}
                    logger.warning("[piece=%d mode=%s] failed: %s", piece_id, mode, msg)
        except Exception as e:
            logger.exception("[piece=%d mode=%s] worker crash", piece_id, mode)
            with state_lock:
                state[key] = {'status': 'failed', 'error': str(e), 'title': title, 'artist': artist}
        finally:
            job_q.task_done()


threading.Thread(target=worker, daemon=True).start()


@app.get('/healthz')
def healthz():
    return {'ok': True}


@app.post('/fetch')
def fetch(piece_id: int, title: str, artist: str = '', mode: str = 'functional'):
    """加入 fetch 队列。
    mode='letters' = 弹唱谱字母版；mode='functional' = 数字级数版。
    幂等：已 completed 且文件还在，直接返回 completed。"""
    if piece_id <= 0 or not title.strip():
        raise HTTPException(400, 'piece_id / title required')
    if mode not in VALID_MODES:
        raise HTTPException(400, f'mode must be one of {VALID_MODES}')

    key = (piece_id, mode)
    with state_lock:
        cur = state.get(key, {})
        if cur.get('status') == 'completed' and out_path(piece_id, mode).exists():
            return {'status': 'completed'}
        if cur.get('status') in ('pending', 'processing'):
            return {'status': cur['status']}
        state[key] = {'status': 'pending', 'error': '', 'title': title, 'artist': artist}

    job_q.put((piece_id, mode, title.strip(), artist.strip()))
    return {'status': 'pending'}


@app.get('/status/{piece_id}/{mode}')
def status(piece_id: int, mode: str):
    if mode not in VALID_MODES:
        raise HTTPException(400, f'mode must be one of {VALID_MODES}')
    key = (piece_id, mode)
    with state_lock:
        cur = state.get(key, {})
    file_exists = out_path(piece_id, mode).exists()
    if cur.get('status') == 'completed' and not file_exists:
        return {'status': 'failed', 'error': 'png 文件丢了', 'mode': mode}
    if not cur and file_exists:
        return {'status': 'completed', 'mode': mode, 'file_exists': True}
    return {
        'status': cur.get('status', 'none'),
        'error': cur.get('error', ''),
        'mode': mode,
        'file_exists': file_exists,
    }


@app.get('/image/{piece_id}/{mode}')
def image(piece_id: int, mode: str):
    if mode not in VALID_MODES:
        raise HTTPException(400, f'mode must be one of {VALID_MODES}')
    p = out_path(piece_id, mode)
    if not p.exists():
        raise HTTPException(404, 'not found')
    return FileResponse(p, media_type='image/png')


@app.delete('/state/{piece_id}/{mode}')
def reset(piece_id: int, mode: str):
    """music backend import 完后清状态 + 删 png（防 PVC 越积越多）。"""
    if mode not in VALID_MODES:
        raise HTTPException(400, f'mode must be one of {VALID_MODES}')
    with state_lock:
        state.pop((piece_id, mode), None)
    p = out_path(piece_id, mode)
    if p.exists():
        try:
            p.unlink()
        except Exception as e:
            logger.warning("[piece=%d mode=%s] cleanup unlink: %s", piece_id, mode, e)
    return {'ok': True}