booth-musicgpt/helpers.py

"""
helpers.py
==========
Utility and configuration helpers for the Theme Song Booth Flask app.

These functions are stateless (or use Flask's current_app / session context)
and are imported by app.py. Keeping them here reduces the size of the route file.

Key function groups:
  - Genre/decade parsing and style formatting
  - Admin auth, settings persistence, encryption helpers
  - Email sending (SMTP) and ntfy notifications
  - Hermes API key management and prompt callback signing
  - MusicGPT API client: generation, polling, file downloads, cost tracking
  - Gokapi integration: stems upload, zip bundling, share link generation
  - MP3 metadata tagging (mutagen)
"""

import os
import re
import shutil
import smtplib
import ssl
import time
import base64
import hmac
import hashlib
import secrets
import json
from email.message import EmailMessage
from pathlib import Path

from flask import session, current_app, flash
from werkzeug.utils import secure_filename

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC

from mutagen.mp3 import MP3
from mutagen.easyid3 import EasyID3

import requests

from models import update_request
from config import Config


# ---------------------------------------------------------------------------
# Genre / decade helpers
# ---------------------------------------------------------------------------

_GENRES_PATH = Path('/mnt/Storage/Music Genres.txt')
_DECADES_PATH = Path('/mnt/Storage/Decades.txt')
_FALLBACK_GENRES_PATH = Path(__file__).parent / 'lists' / 'music_genres.txt'
_FALLBACK_DECADES_PATH = Path(__file__).parent / 'lists' / 'decades.txt'


def _load_lines(path: Path) -> list[str]:
    """Load a text file and return non-empty stripped lines."""
    if not path.exists():
        return []
    lines = path.read_text(encoding='utf-8').splitlines()
    return [line.strip() for line in lines if line.strip()]


def _load_list(primary: Path, fallback: Path) -> list[str]:
    """Load from the primary path, falling back to the bundled copy."""
    lines = _load_lines(primary)
    if lines:
        return lines
    return _load_lines(fallback)


MUSIC_GENRES = _load_list(_GENRES_PATH, _FALLBACK_GENRES_PATH)
DECADES = _load_list(_DECADES_PATH, _FALLBACK_DECADES_PATH)


def parse_style_genre(style_genre: str | None) -> dict:
    """
    Split a stored combined style string into decade, basic, and additional.
    The stored format is 'Decade, Basic, Additional' (additional may be empty).
    """
    parts = [p.strip() for p in (style_genre or '').split(',') if p.strip()]
    return {
        'decade': parts[0] if len(parts) > 0 else '',
        'basic_style': parts[1] if len(parts) > 1 else '',
        'additional_style': ', '.join(parts[2:]) if len(parts) > 2 else '',
    }


def build_style_genre(decade: str, basic_style: str, additional_style: str) -> str:
    """Build the combined style_genre string stored in the database."""
    parts = [p.strip() for p in [decade, basic_style, additional_style] if p.strip()]
    return ', '.join(parts)


# ---------------------------------------------------------------------------
# Auth / validation helpers
# ---------------------------------------------------------------------------

def is_admin():
    """Return True if the current browser session is logged in as admin."""
    return session.get('admin') is True


def require_admin():
    """Redirect to the admin login page if the user is not logged in."""
    from flask import redirect, url_for
    if not is_admin():
        return redirect(url_for('admin_login'))


def admin_password_ok(pw):
    """Check the submitted admin password against the configured one."""
    return pw and pw == current_app.config['ADMIN_PASSWORD']


def is_valid_email(email):
    """Return True if the given string looks like a valid email address."""
    if not email:
        return False
    pattern = r"^[^\s@]+@[^\s@]+\.[^\s@]+$"
    return re.match(pattern, email) is not None


# ---------------------------------------------------------------------------
# File upload helpers
# ---------------------------------------------------------------------------

def allowed_file(filename):
    """Return True if the uploaded filename has an allowed extension (mp3)."""
    return '.' in filename and filename.rsplit('.', 1)[1].lower() in current_app.config['ALLOWED_EXTENSIONS']


def upload_path(request_id):
    """Return the per-request upload directory path, creating it if necessary."""
    p = Path(current_app.config['UPLOAD_FOLDER']) / str(request_id)
    p.mkdir(parents=True, exist_ok=True)
    return p


def save_upload(request_id, file_obj, version, song_title=None):
    """
    Save an uploaded MP3 file for a request, preserving the original filename
    with a version prefix. Applies the configured metadata tags.
    """
    if not file_obj or file_obj.filename == '':
        return None
    if not allowed_file(file_obj.filename):
        flash('Only MP3 files are allowed.', 'error')
        return None
    original = secure_filename(file_obj.filename)
    filename = f"{version.upper()} - {original}"
    p = upload_path(request_id)
    dest = p / filename
    file_obj.save(dest)
    apply_mp3_tags(str(dest), song_title)
    return str(dest)


def apply_mp3_tags(path, title=None):
    """
    Write common ID3 tags on an uploaded MP3 using the runtime metadata defaults.
    Failures are logged as a warning and do not block the upload.
    """
    cfg = load_booth_settings()
    try:
        audio = MP3(path)
        if audio.tags is None:
            audio.add_tags()
        if not isinstance(audio.tags, EasyID3):
            audio.tags = EasyID3()
        tags = audio.tags
        if title:
            tags['title'] = title
        if cfg.get('artist'):
            tags['artist'] = cfg['artist']
        if cfg.get('album'):
            tags['album'] = cfg['album']
        if cfg.get('year'):
            tags['date'] = str(cfg['year'])
        audio.save()
        if cfg.get('comment'):
            from mutagen.id3 import COMM, TXXX
            audio2 = MP3(path)
            if audio2.tags is None:
                audio2.add_tags()
            audio2.tags["COMM"] = COMM(encoding=3, lang='eng', desc='Comment', text=cfg['comment'])
            audio2.tags["TXXX:Comment"] = TXXX(encoding=3, desc='Comment', text=cfg['comment'])
            audio2.save()
    except Exception as e:
        try:
            flash(f'Warning: could not tag MP3: {e}', 'error')
        except RuntimeError:
            import logging
            logging.getLogger('app').warning('Could not tag MP3 %s: %s', path, e)


# ---------------------------------------------------------------------------
# Encryption / settings helpers
# ---------------------------------------------------------------------------

def _get_fernet():
    """Derive a Fernet key from the Flask SECRET_KEY so stored values are encrypted."""
    secret = current_app.config['SECRET_KEY'].encode()
    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=32,
        salt=b'theme-song-booth-v1',
        iterations=480000,
    )
    key = base64.urlsafe_b64encode(kdf.derive(secret))
    return Fernet(key)


def encrypt_value(value):
    """Encrypt a string using the Flask SECRET_KEY. Returns base64 ciphertext."""
    if not value:
        return ''
    return _get_fernet().encrypt(value.encode()).decode()


def decrypt_value(ciphertext):
    """Decrypt a string previously encrypted by encrypt_value."""
    if not ciphertext:
        return ''
    try:
        return _get_fernet().decrypt(ciphertext.encode()).decode()
    except Exception:
        return ''


def decrypt_value_legacy(ciphertext):
    """Decrypt or return plaintext. Tolerates unencrypted legacy values."""
    if not ciphertext:
        return ''
    plaintext = decrypt_value(ciphertext)
    if plaintext:
        return plaintext
    if not ciphertext.endswith('='):
        return ciphertext
    return ''


def settings_file_path():
    """Return the path to the persistent runtime settings JSON file."""
    return Path(current_app.config['DATABASE']).parent / current_app.config['SETTINGS_FILE']


def load_booth_settings():
    """Load persistent runtime settings from JSON file inside the upload parent."""
    cfg_path = settings_file_path()
    if cfg_path.exists():
        try:
            cfg = json.loads(cfg_path.read_text())
            for key in ('artist', 'album', 'year', 'comment', 'smtp_host', 'smtp_port', 'smtp_user', 'smtp_from'):
                if cfg.get(key) is None:
                    cfg[key] = ''
            return cfg
        except (json.JSONDecodeError, OSError):
            pass
    return {}


def save_booth_settings(settings):
    """Persist runtime settings to JSON file."""
    cfg_path = settings_file_path()
    try:
        cfg_path.write_text(json.dumps(settings, indent=2))
    except OSError as e:
        flash(f'Warning: could not save settings: {e}', 'error')


# ---------------------------------------------------------------------------
# Config getters
# ---------------------------------------------------------------------------

def get_email_config():
    """Return the effective SMTP configuration."""
    cfg = load_booth_settings()
    return {
        'SMTP_HOST': cfg.get('smtp_host', current_app.config['SMTP_HOST']),
        'SMTP_PORT': int(cfg.get('smtp_port') or current_app.config['SMTP_PORT']),
        'SMTP_USER': cfg.get('smtp_user', current_app.config['SMTP_USER']),
        'SMTP_PASS': decrypt_value(cfg.get('smtp_pass', '')) or current_app.config['SMTP_PASS'],
        'SMTP_FROM': cfg.get('smtp_from', current_app.config['SMTP_FROM']),
    }


def get_refresh_seconds():
    """Return the dashboard auto-refresh interval in seconds (10, 20, or 30)."""
    cfg = load_booth_settings()
    try:
        val = int(cfg.get('refresh_seconds', 10))
    except (ValueError, TypeError):
        val = 10
    return val if val in (10, 20, 30) else 10


def get_kiosk_cycle_seconds():
    """Return the kiosk slide cycle interval in seconds."""
    cfg = load_booth_settings()
    try:
        val = int(cfg.get('kiosk_cycle_seconds', 10))
    except (ValueError, TypeError):
        val = 10
    if val == 0:
        return 0
    return max(5, val)


def get_kiosk_mode():
    """Return 'qr', 'prices', 'queue', or 'cycle' based on kiosk_cycle_seconds setting."""
    cfg = load_booth_settings()
    try:
        val = int(cfg.get('kiosk_cycle_seconds', 10))
    except (ValueError, TypeError):
        val = 10
    if val == -1:
        return 'qr'
    if val == 0:
        return 'prices'
    if val == 1:
        return 'queue'
    return 'cycle'


def get_max_revisions():
    """Return the effective max revisions as an integer."""
    cfg = load_booth_settings()
    try:
        val = int(cfg.get('max_revisions', current_app.config.get('MAX_REVISIONS', 2)))
    except (ValueError, TypeError):
        val = current_app.config.get('MAX_REVISIONS', 2)
    return max(0, val)


def get_callback_expiry_hours():
    """Return the Hermes signed callback token lifetime in hours (default 168 = 7 days)."""
    cfg = load_booth_settings()
    try:
        val = int(cfg.get('callback_expiry_hours', 168))
    except (ValueError, TypeError):
        val = 168
    return max(1, val)


def get_musicgpt_api_key():
    """Return the MusicGPT API key from env only."""
    return current_app.config.get('MUSICGPT_API_KEY', '')


def get_musicgpt_default_model():
    """Return the configured default MusicGPT model, falling back to a supported model."""
    default = current_app.config.get('MUSICGPT_DEFAULT_MODEL', 'v6-pro')
    models = get_musicgpt_models()
    return default if default in models else models[-1]


def get_musicgpt_models():
    """Return the list of supported MusicGPT models (filtered by API key capability)."""
    return list(current_app.config.get('MUSICGPT_MODELS', ['v6', 'v6-pro']))


def build_musicgpt_webhook_url():
    """Build the public webhook URL for MusicGPT async callbacks from PUBLIC_BASE_URL."""
    base = current_app.config.get('PUBLIC_BASE_URL', '').rstrip('/')
    return f"{base}/api/musicgpt/webhook"


def get_hermes_api_key():
    """Return the effective Hermes API key."""
    env_key = current_app.config.get('HERMES_API_KEY', '')
    if env_key:
        return env_key
    cfg = load_booth_settings()
    return decrypt_value_legacy(cfg.get('hermes_api_key', ''))


def set_hermes_api_key(key):
    """Persist a new Hermes API key (encrypted) to runtime settings."""
    cfg = load_booth_settings()
    cfg['hermes_api_key'] = encrypt_value(key)
    save_booth_settings(cfg)


def generate_hermes_api_key():
    """Generate a new random API key for Hermes callback authentication."""
    return secrets.token_urlsafe(32)


def mask_api_key(key):
    """Return a masked version of the API key showing only the last 6 characters."""
    if not key:
        return 'Not set'
    if len(key) <= 6:
        return '*' * len(key)
    return '*' * (len(key) - 6) + key[-6:]


def get_ntfy_config():
    """Return the effective ntfy server URL, topic, and access token from runtime settings."""
    cfg = load_booth_settings()
    return {
        'server': cfg.get('ntfy_server', ''),
        'topic': cfg.get('ntfy_topic', ''),
        'token': decrypt_value(cfg.get('ntfy_token', '')) or '',
    }


def get_hermes_webhook_config():
    """Return the Hermes webhook URL and HMAC secret from runtime settings."""
    cfg = load_booth_settings()
    return {
        'url': cfg.get('hermes_webhook_url', '').strip(),
        'secret': decrypt_value(cfg.get('hermes_webhook_secret', '')) or '',
    }


def set_hermes_webhook_config(url, secret):
    """Persist Hermes webhook URL and secret (encrypted) to runtime settings."""
    cfg = load_booth_settings()
    cfg['hermes_webhook_url'] = url.strip().rstrip('/')
    if secret:
        cfg['hermes_webhook_secret'] = encrypt_value(secret)
    save_booth_settings(cfg)


def send_hermes_webhook(rid, req, callback_url=None):
    """
    POST a song request payload to the configured Hermes webhook.

    Returns (success: bool, message: str). On success, Hermes receives the
    customer data and can generate a Suno/MusicGPT prompt. If callback_url is
    provided, Hermes can POST the generated prompt directly back to
    /api/prompt/<rid>.
    """
    cfg = get_hermes_webhook_config()
    url = cfg.get('url', '')
    secret = cfg.get('secret', '')
    if not url or not secret:
        return False, 'Hermes webhook is not configured in /admin/settings'

    style_genre = req.get('style_genre') or ''
    style_parts = [p.strip() for p in style_genre.split(',') if p.strip()]
    style_sentence = ''
    if style_parts:
        parts = []
        if style_parts[0]:
            parts.append(f"{style_parts[0]}-era")
        if len(style_parts) > 1 and style_parts[1]:
            parts.append(style_parts[1])
        if len(style_parts) > 2:
            parts.append(f"with {', '.join(style_parts[2:])} influences")
        style_sentence = ' '.join(parts)

    def _bool(value):
        if isinstance(value, bool):
            return value
        return bool(int(value or 0))

    payload = {
        'event_type': 'song_request',
        'request_id': rid,
        'email': req.get('email', ''),
        'name': req.get('name', ''),
        'pronouns': req.get('pronouns', ''),
        'hobbies': req.get('hobbies', ''),
        'notable_facts': req.get('notable_facts', ''),
        'style_genre': style_sentence or style_genre,
        'vocal_gender': req.get('vocal_gender', ''),
        'extra_requests': req.get('extra_requests', ''),
        'stems_interest': _bool(req.get('stems_interest')),
        'is_revision': bool(req.get('revision_count', 0)) and bool(req.get('revision_note')),
        'revision_count': int(req.get('revision_count') or 0),
        'revision_note': req.get('revision_note', ''),
        'previous_title': req.get('suno_title', ''),
        'previous_style': req.get('suno_style', ''),
        'previous_lyrics': req.get('suno_lyrics', ''),
        'callback_url': callback_url or '',
    }

    body = json.dumps(payload, separators=(',', ':')).encode('utf-8')
    timestamp = str(int(time.time()))
    sig_data = f"{timestamp}.{body.decode('utf-8')}"
    signature = hmac.new(secret.encode('utf-8'), sig_data.encode('utf-8'), hashlib.sha256).hexdigest()

    headers = {
        'Content-Type': 'application/json',
        'X-Webhook-Signature-V2': signature,
        'X-Webhook-Timestamp': timestamp,
    }
    try:
        resp = requests.post(url, data=body, headers=headers, timeout=15)
        if resp.status_code in (200, 202):
            return True, f"Sent to Hermes (HTTP {resp.status_code})"
        return False, f"Hermes webhook returned HTTP {resp.status_code}: {resp.text[:200]}"
    except Exception as e:
        return False, f"Failed to reach Hermes webhook: {e}"


def send_ntfy(message, title='Theme Song Booth', priority='default', tags='bell'):
    """Send a push notification to the configured ntfy topic, if configured."""
    ntfy = get_ntfy_config()
    server = ntfy.get('server', '').rstrip('/')
    topic = ntfy.get('topic', '').strip()
    if not server or not topic:
        return False

    url = f"{server}/{topic}"
    headers = {
        'Title': title,
        'Priority': priority,
        'Tags': tags,
    }
    token = ntfy.get('token', '')
    if token:
        headers['Authorization'] = f'Bearer {token}'
    try:
        resp = requests.post(url, data=message.encode('utf-8'), headers=headers, timeout=10)
        return resp.status_code in (200, 202)
    except Exception:
        return False


# ---------------------------------------------------------------------------
# Signed callback helpers
# ---------------------------------------------------------------------------

def sign_prompt_callback(rid, expires_at=None):
    """
    Create a signed callback token for a specific request ID.
    The signature is HMAC-SHA256 over "rid:expires_at" using APP_SECRET_KEY.
    """
    secret = current_app.config['SECRET_KEY'].encode()
    if expires_at is None:
        expires_at = int(time.time()) + get_callback_expiry_hours() * 3600
    payload = f"{rid}:{expires_at}"
    sig = hmac.new(secret, payload.encode(), hashlib.sha256).hexdigest()[:16]
    return f"{rid}:{expires_at}:{sig}"


def verify_prompt_callback(token):
    """Verify a signed callback token. Returns (rid, ok) tuple."""
    if not token:
        return None, False
    parts = token.split(':')
    if len(parts) != 3:
        return None, False
    try:
        rid = int(parts[0])
        expires_at = int(parts[1])
    except ValueError:
        return None, False
    if int(time.time()) > expires_at:
        return None, False
    expected = sign_prompt_callback(rid, expires_at)
    if not hmac.compare_digest(expected, token):
        return None, False
    return rid, True


def build_prompt_callback_url(rid):
    """Build the full callback URL an operator pastes into Hermes for a request."""
    from flask import url_for
    token = sign_prompt_callback(rid)
    return f"{current_app.config['PUBLIC_BASE_URL']}/api/prompt/{rid}?token={token}"


# ---------------------------------------------------------------------------
# Email helpers
# ---------------------------------------------------------------------------

def send_email(to, subject, body, attachments=None, inline_images=None):
    """Send an email using the configured or runtime SMTP settings."""
    cfg = get_email_config()
    if not cfg['SMTP_PASS']:
        raise RuntimeError('SMTP password is not configured')

    msg = EmailMessage()
    msg['From'] = cfg['SMTP_FROM']
    msg['To'] = to
    msg['Subject'] = subject
    msg.set_content(body)

    html_body = body.replace('\n', '<br>\n')
    if inline_images:
        for _, cid in inline_images:
            html_body += f'<br><img src="cid:{cid}" alt="Dionysis Media" style="max-width:200px;margin-top:1rem;"/>'
    html_body += f'<br><br><hr style="border:none;border-top:1px solid #ddd;"/><p style="font-size:0.9rem;color:#555;">Dionysis Media: stories, sound, and a little divine chaos — <a href="https://dionysismedia.ca/">https://dionysismedia.ca/</a></p>'
    msg.add_alternative(html_body, subtype='html')

    if inline_images:
        for path, cid in inline_images:
            with open(path, 'rb') as f:
                data = f.read()
            ext = Path(path).suffix.lower().lstrip('.')
            subtype = ext if ext in ('png', 'jpg', 'jpeg', 'gif', 'webp') else 'png'
            msg.get_payload()[1].add_related(data, maintype='image', subtype=subtype, cid=f'<{cid}>')

    if attachments:
        for path, name in attachments:
            with open(path, 'rb') as f:
                data = f.read()
            msg.add_attachment(data, maintype='audio', subtype='mpeg', filename=name)

    with smtplib.SMTP_SSL(cfg['SMTP_HOST'], cfg['SMTP_PORT'], context=ssl.create_default_context()) as server:
        server.login(cfg['SMTP_USER'], cfg['SMTP_PASS'])
        server.send_message(msg)


def build_signature_images():
    """Return inline image tuple list for static/DM-Logo_email.png."""
    logo_path = Path(current_app.root_path) / 'static' / 'DM-Logo_email.png'
    if not logo_path.exists():
        return []
    return [(str(logo_path), 'dm-logo')]


# ---------------------------------------------------------------------------
# Booth state
# ---------------------------------------------------------------------------

def get_booth_open():
    """Return True if the booth is currently marked as open in runtime settings."""
    cfg = load_booth_settings()
    return cfg.get('booth_open', True)


def get_musicgpt_autopoll_enabled():
    """Return True if automatic MusicGPT polling is enabled (default True)."""
    cfg = load_booth_settings()
    return cfg.get('musicgpt_autopoll', True)


def set_musicgpt_autopoll_enabled(enabled):
    """Persist the automatic MusicGPT polling toggle."""
    cfg = load_booth_settings()
    cfg['musicgpt_autopoll'] = bool(enabled)
    save_booth_settings(cfg)


# ---------------------------------------------------------------------------
# MusicGPT API client
# ---------------------------------------------------------------------------

MUSICGPT_API_BASE = "https://api.musicgpt.com/api/public"


def _musicgpt_headers():
    """Return authorization headers for MusicGPT API calls."""
    return {
        "Authorization": get_musicgpt_api_key(),
        "Content-Type": "application/json"
    }


def musicgpt_generate_request(rid, title, music_style, lyrics, gender=None, model=None):
    """
    Queue a Music AI v2 generation request.
    Returns (task_id, conversion_id_1, conversion_id_2, credit_estimate, error_message).
    """
    url = f"{MUSICGPT_API_BASE}/v2/MusicAI"
    if model is None:
        model = get_musicgpt_default_model()
    payload = {
        "title": title,
        "music_style": music_style,
        "lyrics": lyrics,
        "make_instrumental": False,
        "model": model,
        "webhook_url": build_musicgpt_webhook_url(),
    }
    if gender and gender.lower() in ("male", "female", "neutral"):
        payload["gender"] = gender.lower()
    try:
        resp = requests.post(url, json=payload, headers=_musicgpt_headers(), timeout=30)
        data = resp.json()
        if resp.status_code == 200 and data.get("success"):
            return (
                data.get("task_id"),
                data.get("conversion_id_1"),
                data.get("conversion_id_2"),
                data.get("credit_estimate"),
                None,
            )
        return None, None, None, None, data.get("error") or f"HTTP {resp.status_code}"
    except Exception as e:
        return None, None, None, None, str(e)


def musicgpt_poll_status(task_id):
    """
    Poll the MusicGPT API for a generation task status.

    MusicGPT's /v1/byId endpoint returns the whole task record, including the
    'conversion' object with the current status and any available audio URLs.
    Returns a dict with keys: status, message, conversion, or error.
    """
    url = f"{MUSICGPT_API_BASE}/v1/byId"
    params = {"conversionType": "MUSIC_AI", "task_id": task_id}
    try:
        resp = requests.get(url, headers={"Authorization": get_musicgpt_api_key()}, params=params, timeout=20)
        data = resp.json()
        if resp.status_code == 200:
            return data
        return {"error": data.get("error") or f"HTTP {resp.status_code}"}
    except Exception as e:
        return {"error": str(e)}


def musicgpt_queue_stems(rid, audio_file_path, stems=None):
    """
    Queue a stem extraction job by uploading the MP3 file directly.

    The MusicGPT Extraction API's audio_url field only accepts YouTube URLs,
    not direct MP3 URLs. We use the audio_file upload option instead.

    Returns (task_id, conversion_id, credit_estimate, error_message).
    """
    url = f"{MUSICGPT_API_BASE}/v2/Extraction"
    if stems is None:
        stems = ["vocals", "instrumental"]
    payload = {
        "stems": json.dumps(stems),
        "webhook_url": build_musicgpt_webhook_url(),
    }
    try:
        with open(audio_file_path, "rb") as f:
            files = {"audio_file": f}
            resp = requests.post(url, data=payload, files=files, headers={"Authorization": get_musicgpt_api_key()}, timeout=60)
        try:
            data = resp.json()
        except Exception:
            return None, None, None, f"HTTP {resp.status_code}: {resp.text[:300]}"
        if resp.status_code == 200 and data.get("success"):
            return data.get("task_id"), data.get("conversion_id"), data.get("credit_estimate"), None
        # Capture as much detail as possible from the error response.
        error_msg = data.get("error") or data.get("message") or f"HTTP {resp.status_code}"
        if data.get("details"):
            error_msg += f" — {data['details']}"
        return None, None, None, error_msg
    except Exception as e:
        return None, None, None, str(e)


def _download_file(url, dest):
    """Download a file from url to dest. Returns True on success."""
    try:
        with requests.get(url, stream=True, timeout=120) as r:
            r.raise_for_status()
            with open(dest, "wb") as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)
        return True
    except Exception:
        return False


def download_musicgpt_outputs(req, data, version=None):
    """
    Download the MP3 and WAV outputs from a completed MusicGPT webhook/poll payload.

    MusicGPT returns audio URLs under different keys depending on whether the
    payload is for one conversion or the combined task result. This function
    normalizes those keys, downloads each available file to the request upload
    directory, applies MP3 metadata tags, and updates the request row.

    Updates the request row with local paths and returns a dict of saved paths.
    `data` is the conversion dict from the API. For per-conversion webhooks,
    pass `version='A' or 'B'`. For combined payloads, version is auto-detected
    from conversion_path_1/2 keys.
    """
    request_id = req["id"]
    upload_dir = upload_path(request_id)
    song_title = req.get("suno_title") or req.get("title")
    saved = {}

    # Map version label to the database column names used to store local paths.
    def _fields(v):
        return ("song_a_path", "song_a_wav_path") if v == "A" else ("song_b_path", "song_b_wav_path")

    # Determine which version(s) are present in the payload.
    versions = []
    if version in ("A", "B"):
        versions = [version]
    elif data.get("conversion_path_1") or data.get("conversion_path_wav_1"):
        if data.get("conversion_path_2") or data.get("conversion_path_wav_2"):
            versions = ["A", "B"]
        else:
            versions = ["A"]
    elif data.get("conversion_path_2") or data.get("conversion_path_wav_2"):
        versions = ["B"]
    elif data.get("conversion_path") or data.get("conversion_path_wav"):
        # Per-conversion webhook without explicit version: use the requested version or A.
        versions = [version if version in ("A", "B") else "A"]

    for v in versions:
        field_mp3, field_wav = _fields(v)
        url_key = "conversion_path_1" if v == "A" else "conversion_path_2"
        wav_key = "conversion_path_wav_1" if v == "A" else "conversion_path_wav_2"
        # Also support per-conversion webhook keys without _1/_2 suffix.
        # When MusicGPT sends one webhook per conversion it usually uses the plain
        # 'conversion_path' / 'conversion_path_wav' keys; we still need to know
        # whether to treat it as version A or B. The caller-provided version helps.
        if not (data.get(url_key) or data.get(wav_key)):
            if v == "A" or version == "A":
                mp3_url = data.get("conversion_path")
                wav_url = data.get("conversion_path_wav")
            else:
                mp3_url = data.get("conversion_path")
                wav_url = data.get("conversion_path_wav")
        else:
            mp3_url = data.get(url_key)
            wav_url = data.get(wav_key)

        if mp3_url:
            ext = Path(mp3_url).suffix or ".mp3"
            dest = upload_dir / f"{v}{ext}"
            if _download_file(mp3_url, dest):
                apply_mp3_tags(str(dest), song_title)
                saved[field_mp3] = str(dest)
        if wav_url:
            dest_wav = upload_dir / f"{v}.wav"
            if _download_file(wav_url, dest_wav):
                saved[field_wav] = str(dest_wav)

    # Album cover
    cover_url = data.get("album_cover_path")
    if cover_url:
        saved["album_cover_url"] = cover_url

    if saved:
        update_request(request_id, **saved)
        # If both song versions have been downloaded, advance request status.
        from models import get_request_by_id
        req = get_request_by_id(request_id)
        if req and req.get('song_a_path') and req.get('song_b_path') and req.get('status') not in ('songs_uploaded', 'delivered', 'cancelled'):
            update_request(request_id, status='songs_uploaded')
    return saved


def format_musicgpt_cost(cost):
    """Return a human-readable cost string in USD."""
    if cost is None:
        return "—"
    return f"${float(cost):.4f} USD"


def download_album_cover(rid, cover_url, max_width=160):
    """
    Download a MusicGPT album cover to a temp file, resize it so it is not
    oversized in emails or pages, and return its path. The default max_width
    of 160px matches the inline album cover display size on the player/admin pages.
    """
    if not cover_url:
        return None
    try:
        import tempfile
        from PIL import Image
        resp = requests.get(cover_url, timeout=30)
        if resp.status_code == 200:
            ext = Path(cover_url).suffix or '.jpg'
            tmp = Path(tempfile.gettempdir()) / f"cover_{rid}{ext}"
            tmp.write_bytes(resp.content)
            # Resize if the image is wider than max_width to keep email/file size small.
            with Image.open(tmp) as img:
                if img.width > max_width:
                    ratio = max_width / img.width
                    new_height = int(img.height * ratio)
                    img = img.resize((max_width, new_height))
                    img.save(tmp)
            return str(tmp)
    except Exception:
        pass
    return None


def get_musicgpt_cost_totals():
    """Return aggregate MusicGPT cost metrics across all requests."""
    from models import get_db
    db = get_db()
    rows = db.execute(
        """
        SELECT
            COUNT(*) AS song_count,
            COALESCE(SUM(musicgpt_cost), 0) AS total_cost,
            COALESCE(SUM(
                CASE WHEN song_a_path IS NOT NULL AND song_b_path IS NOT NULL THEN 1 ELSE 0
            END), 0) AS completed_pairs,
            COALESCE(SUM(stems_cost), 0) AS stems_total
        FROM requests
        WHERE musicgpt_cost IS NOT NULL
        """
    ).fetchone()
    return {
        "song_count": rows["song_count"] or 0,
        "completed_pairs": rows["completed_pairs"] or 0,
        "total_cost": rows["total_cost"] or 0,
        "stems_total": rows["stems_total"] or 0,
    }


# ---------------------------------------------------------------------------
# Gokapi file-sharing integration (stems upload)
# ---------------------------------------------------------------------------

def parse_stems_urls(stems_url_field):
    """
    Parse the stems_url DB field into a list of (label, url) tuples.
    The field can be:
      - "label1: url1; label2: url2" (multiple individual stem files)
      - A single bundle/zip URL
      - A bare URL
    """
    if not stems_url_field:
        return []
    urls = []
    if ';' in stems_url_field or ': ' in stems_url_field:
        parts = stems_url_field.split(';')
        for part in parts:
            part = part.strip()
            if not part:
                continue
            if ': ' in part:
                label, url = part.split(': ', 1)
                urls.append((label.strip(), url.strip()))
            else:
                urls.append(('stem', part))
    else:
        urls.append(('stems', stems_url_field.strip()))
    return urls


def upload_to_gokapi(file_bytes, filename, expiry_days=30):
    """
    Upload a file to Gokapi and return the download URL.
    Returns (download_url, error_message).
    """
    gokapi_url = Config.GOKAPI_URL.rstrip('/')
    api_key = Config.GOKAPI_API_KEY
    if not gokapi_url or not api_key:
        return None, 'Gokapi URL or API key not configured'
    try:
        resp = requests.post(
            f"{gokapi_url}/api/files/add",
            headers={"apikey": api_key},
            files={"file": (filename, file_bytes, "application/octet-stream")},
            data={
                "allowedDownloads": "0",
                "expiryDays": str(expiry_days),
                "password": "",
            },
            timeout=120,
        )
        if resp.status_code != 200:
            return None, f'Gokapi HTTP {resp.status_code}: {resp.text[:200]}'
        data = resp.json()
        if data.get('Result') != 'OK':
            return None, f"Gokapi error: {data.get('Result')}"
        file_info = data.get('FileInfo', {})
        download_url = file_info.get('UrlDownload')
        if not download_url:
            return None, 'Gokapi returned no download URL'
        return download_url, None
    except Exception as e:
        return None, str(e)


def process_stems_to_gokapi(req):
    """
    Download stem files from MusicGPT CDN, zip them, upload the zip to Gokapi,
    and return (gokapi_url, error_message).

    Also updates the request row with the Gokapi download link in stems_link.
    """
    import io
    import zipfile

    stems_url = req.get('stems_url') or ''
    urls = parse_stems_urls(stems_url)
    if not urls:
        return None, 'No stems URL to process'

    # If there's only one URL and it's already a .zip, upload it directly.
    if len(urls) == 1 and urls[0][1].endswith('.zip'):
        try:
            r = requests.get(urls[0][1], timeout=120)
            r.raise_for_status()
            zip_bytes = r.content
        except Exception as e:
            return None, f'Failed to download stems zip: {e}'
    else:
        # Download each stem file and bundle into a zip in memory.
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
            for label, url in urls:
                try:
                    r = requests.get(url, timeout=60)
                    r.raise_for_status()
                    filename = url.split('/')[-1] if '/' in url else f"{label}.mp3"
                    if '.' not in filename:
                        filename = f"{label}.mp3"
                    zf.writestr(filename, r.content)
                except Exception:
                    pass
        zip_bytes = buf.getvalue()

    title = req.get('suno_title') or req.get('name') or f'request_{req["id"]}'
    zip_filename = f"stems_{title}.zip"

    download_url, error = upload_to_gokapi(zip_bytes, zip_filename, Config.GOKAPI_EXPIRY_DAYS)
    if error:
        return None, error

    # Save the Gokapi link to stems_link in the database.
    update_request(req['id'], stems_link=download_url)
    return download_url, None