theme-song-booth/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.
"""

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


# ---------------------------------------------------------------------------
# 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_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 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)