services.py

import glob
import os
import sqlite3
import time
import uuid
import wave
from datetime import datetime
from piper import PiperVoice

# Shared configuration
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
VOICE_MODELS_DIR = os.path.join(ROOT_DIR, 'voice_models')
OUTPUT_DIR = os.path.join(ROOT_DIR, 'static')
DB_NAME = os.path.join(ROOT_DIR, 'history.db')
MAX_TEXT_LENGTH = 5000
SAMPLE_TEXTS_DIR = os.path.join(ROOT_DIR, 'sample_texts')

# Internal cache for loaded models
_voice_cache = {}
_current_model_path = None


def ensure_directories():
    """Ensure required directories exist."""
    for path in (OUTPUT_DIR, VOICE_MODELS_DIR, SAMPLE_TEXTS_DIR):
        os.makedirs(path, exist_ok=True)


def init_db():
    """Initialize the SQLite database with enhanced schema."""
    conn = sqlite3.connect(DB_NAME)
    c = conn.cursor()

    c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='history'")
    table_exists = c.fetchone()

    if not table_exists:
        c.execute('''CREATE TABLE history
                     (id TEXT PRIMARY KEY,
                      user_ip TEXT,
                      text TEXT,
                      audio_file TEXT,
                      model_name TEXT,
                      generation_time REAL,
                      timestamp REAL,
                      char_count INTEGER,
                      language TEXT,
                      voice_quality TEXT,
                      project_name TEXT)''')
    else:
        c.execute("PRAGMA table_info(history)")
        columns = [info[1] for info in c.fetchall()]

        if 'char_count' not in columns:
            try:
                c.execute("ALTER TABLE history ADD COLUMN char_count INTEGER DEFAULT 0")
            except sqlite3.OperationalError:
                pass

        if 'language' not in columns:
            try:
                c.execute("ALTER TABLE history ADD COLUMN language TEXT DEFAULT 'Unknown'")
            except sqlite3.OperationalError:
                pass

        if 'voice_quality' not in columns:
            try:
                c.execute("ALTER TABLE history ADD COLUMN voice_quality TEXT DEFAULT 'Unknown'")
            except sqlite3.OperationalError:
                pass

        if 'project_name' not in columns:
            try:
                c.execute("ALTER TABLE history ADD COLUMN project_name TEXT DEFAULT 'Default'")
            except sqlite3.OperationalError:
                pass

    c.execute('''CREATE INDEX IF NOT EXISTS idx_user_ip ON history(user_ip)''')
    c.execute('''CREATE INDEX IF NOT EXISTS idx_timestamp ON history(timestamp)''')
    c.execute('''CREATE INDEX IF NOT EXISTS idx_project ON history(project_name)''')

    conn.commit()
    conn.close()


def add_history_item(user_ip, item):
    """Add a history item to the database with additional metadata."""
    conn = sqlite3.connect(DB_NAME)
    c = conn.cursor()
    c.execute("""INSERT INTO history VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
              (item['id'], user_ip, item['text'], item['audio_file'],
               item['model_name'], item['generation_time'], item['timestamp'],
               item.get('char_count', len(item['text'])),
               item.get('language', 'Unknown'),
               item.get('voice_quality', 'Unknown'),
               item.get('project_name', 'Default')))
    conn.commit()
    conn.close()


def get_user_history(user_ip, limit=50, project_name=None):
    """Get history for a specific user IP with limit and optional project filter."""
    conn = sqlite3.connect(DB_NAME)
    conn.row_factory = sqlite3.Row
    c = conn.cursor()

    if project_name:
        c.execute("""SELECT * FROM history WHERE user_ip = ? AND project_name = ?
                     ORDER BY timestamp DESC LIMIT ?""", (user_ip, project_name, limit))
    else:
        c.execute("""SELECT * FROM history WHERE user_ip = ?
                     ORDER BY timestamp DESC LIMIT ?""", (user_ip, limit))

    rows = c.fetchall()
    conn.close()
    return [dict(row) for row in rows]


def get_user_projects(user_ip):
    """Get list of unique project names for a user."""
    conn = sqlite3.connect(DB_NAME)
    c = conn.cursor()
    c.execute("SELECT DISTINCT project_name FROM history WHERE user_ip = ? ORDER BY project_name", (user_ip,))
    rows = c.fetchall()
    conn.close()
    projects = [row[0] for row in rows]
    if 'Default' not in projects:
        projects.insert(0, 'Default')
    return projects


def get_user_statistics(user_ip):
    """Get usage statistics for a user."""
    conn = sqlite3.connect(DB_NAME)
    c = conn.cursor()

    c.execute("SELECT COUNT(*) FROM history WHERE user_ip = ?", (user_ip,))
    total_generations = c.fetchone()[0]

    c.execute("SELECT SUM(char_count) FROM history WHERE user_ip = ?", (user_ip,))
    total_chars = c.fetchone()[0] or 0

    c.execute("SELECT AVG(generation_time) FROM history WHERE user_ip = ?", (user_ip,))
    avg_time = c.fetchone()[0] or 0

    c.execute("""SELECT language, COUNT(*) as count FROM history
                 WHERE user_ip = ? GROUP BY language
                 ORDER BY count DESC LIMIT 1""", (user_ip,))
    result = c.fetchone()
    most_used_lang = result[0] if result else 'None'

    conn.close()

    return {
        'total_generations': total_generations,
        'total_characters': total_chars,
        'avg_generation_time': round(avg_time, 2),
        'most_used_language': most_used_lang
    }


def clear_user_history(user_ip):
    """Clear history for a specific user IP and delete audio files."""
    conn = sqlite3.connect(DB_NAME)
    conn.row_factory = sqlite3.Row
    c = conn.cursor()

    c.execute("SELECT audio_file FROM history WHERE user_ip = ?", (user_ip,))
    audio_files = [row['audio_file'] for row in c.fetchall()]

    c.execute("DELETE FROM history WHERE user_ip = ?", (user_ip,))
    conn.commit()
    conn.close()

    for audio_file in audio_files:
        file_path = os.path.join(OUTPUT_DIR, audio_file)
        if os.path.exists(file_path):
            try:
                os.remove(file_path)
            except Exception as e:
                print(f"Error deleting file {file_path}: {e}")


def parse_model_info(filename):
    """Parse model filename into structured info with enhanced metadata."""
    name_no_ext = filename.replace('.onnx', '')
    parts = name_no_ext.split('-')

    info = {
        'id': filename,
        'display_name': name_no_ext,
        'lang': 'Unknown',
        'name': name_no_ext,
        'quality': 'Unknown',
        'gender': 'Neutral',
        'description': ''
    }

    if len(parts) >= 3:
        info['lang'] = parts[0]
        info['name'] = parts[1].title()
        info['quality'] = parts[2].title()

        quality_emoji = {'Low': '⚡', 'Medium': '🎵', 'High': '💎', 'Unknown': '🔊'}
        emoji = quality_emoji.get(info['quality'], '🔊')
        info['display_name'] = f"{emoji} {info['name']} - {info['quality']}"

        female_names = ['amy', 'emma', 'lisa', 'sarah', 'jenny']
        male_names = ['john', 'ryan', 'danny', 'brian', 'joe']

        name_lower = info['name'].lower()
        if any(fn in name_lower for fn in female_names):
            info['gender'] = 'Female'
        elif any(mn in name_lower for mn in male_names):
            info['gender'] = 'Male'

    return info


def get_available_models():
    """Get a list of available voice models with full metadata."""
    models = glob.glob(os.path.join(VOICE_MODELS_DIR, '*.onnx'))
    parsed_models = []
    for model_path in models:
        filename = os.path.basename(model_path)
        model_info = parse_model_info(filename)
        file_size = os.path.getsize(model_path)
        model_info['file_size_mb'] = round(file_size / (1024 * 1024), 2)
        parsed_models.append(model_info)

    quality_order = {'High': 0, 'Medium': 1, 'Low': 2, 'Unknown': 3}
    parsed_models.sort(key=lambda x: (
        x['lang'],
        quality_order.get(x['quality'], 99),
        x['name']
    ))

    return parsed_models


def get_model_stats(models):
    """Calculate comprehensive statistics from the model list."""
    stats = {
        'total': len(models),
        'languages': {},
        'qualities': {},
        'genders': {},
        'total_size_mb': 0
    }

    for model in models:
        lang = model['lang']
        stats['languages'][lang] = stats['languages'].get(lang, 0) + 1

        quality = model['quality']
        stats['qualities'][quality] = stats['qualities'].get(quality, 0) + 1

        gender = model['gender']
        stats['genders'][gender] = stats['genders'].get(gender, 0) + 1

        stats['total_size_mb'] += model.get('file_size_mb', 0)

    stats['total_size_mb'] = round(stats['total_size_mb'], 2)
    return stats


def load_sample_texts():
    """Load curated sample texts from the filesystem grouped by topic and language."""
    samples = {}

    if not os.path.isdir(SAMPLE_TEXTS_DIR):
        return samples

    for topic_dir in sorted(os.listdir(SAMPLE_TEXTS_DIR)):
        topic_path = os.path.join(SAMPLE_TEXTS_DIR, topic_dir)
        if not os.path.isdir(topic_path):
            continue

        label = topic_dir.replace('_', ' ').title()
        language_items = {}

        default_texts = []
        for txt_path in sorted(glob.glob(os.path.join(topic_path, '*.txt'))):
            try:
                with open(txt_path, 'r', encoding='utf-8') as file:
                    content = file.read().strip()
                    if content:
                        default_texts.append(content)
            except Exception as e:
                print(f"Error reading sample text {txt_path}: {e}")
        if default_texts:
            language_items['default'] = default_texts

        for lang_dir in sorted(os.listdir(topic_path)):
            lang_path = os.path.join(topic_path, lang_dir)
            if not os.path.isdir(lang_path):
                continue

            lang_texts = []
            for txt_path in sorted(glob.glob(os.path.join(lang_path, '*.txt'))):
                try:
                    with open(txt_path, 'r', encoding='utf-8') as file:
                        content = file.read().strip()
                        if content:
                            lang_texts.append(content)
                except Exception as e:
                    print(f"Error reading sample text {txt_path}: {e}")

            if lang_texts:
                language_items[lang_dir.lower()] = lang_texts

        if language_items:
            samples[topic_dir] = {
                'label': label,
                'items': language_items
            }

    return samples


def load_voice_model(model_path):
    """Load voice model with caching."""
    global _current_model_path

    if model_path not in _voice_cache:
        if len(_voice_cache) >= 3:
            oldest_key = next(iter(_voice_cache))
            del _voice_cache[oldest_key]

        _voice_cache[model_path] = PiperVoice.load(model_path)

    _current_model_path = model_path
    return _voice_cache[model_path]


def generate_speech(text, model_name, project_name='Default'):
    """Generate speech audio and return history payload."""
    model_path = os.path.join(VOICE_MODELS_DIR, model_name)
    if not os.path.exists(model_path):
        raise FileNotFoundError('Selected model not found.')

    filename = f"neurovox_{uuid.uuid4().hex}.wav"
    output_path = os.path.join(OUTPUT_DIR, filename)

    voice = load_voice_model(model_path)

    start_time = time.time()
    with wave.open(output_path, "wb") as wav_file:
        voice.synthesize_wav(text, wav_file)
    generation_time = time.time() - start_time

    model_info = parse_model_info(model_name)

    history_item = {
        'id': uuid.uuid4().hex,
        'text': text,
        'audio_file': filename,
        'model_name': model_name,
        'generation_time': generation_time,
        'timestamp': time.time(),
        'char_count': len(text),
        'language': model_info['lang'],
        'voice_quality': model_info['quality'],
        'project_name': project_name
    }

    return history_item, output_path


def export_history(user_ip, limit=1000):
    """Export user history as a JSON-friendly dict."""
    history = get_user_history(user_ip, limit=limit)
    return {
        'export_date': datetime.now().isoformat(),
        'total_items': len(history),
        'history': history
    }