Blame for app.py

import hashlib
import io
import logging
import os
import re
import secrets
import unicodedata
import uuid
import wave
from datetime import datetime, timedelta, timezone
from pathlib import Path

import httpx
from fastapi import Depends, FastAPI, File, Form, Header, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field

try:
    from .store import Conflict, Store
    from .layout import folder as audio_folder, manifests as audio_manifests
except ImportError:
    from store import Conflict, Store
    from layout import folder as audio_folder, manifests as audio_manifests

app = FastAPI(title='Archivo de Voz', version='0.2.7')
app.add_middleware(CORSMiddleware,
    allow_origins=[x.strip() for x in os.getenv('ALLOWED_ORIGINS', 'http://localhost:8080,http://127.0.0.1:8080').split(',') if x.strip()],
    allow_methods=['GET', 'POST'], allow_headers=['Content-Type', 'X-App-Key'])
if bool(os.getenv('HF_DATASET_REPO')) != bool(os.getenv('HF_TOKEN')):
    raise RuntimeError('Configura HF_DATASET_REPO y HF_TOKEN juntos, o elimina ambos para usar modo local.')
store = Store(os.getenv('HF_DATASET_REPO'), os.getenv('HF_TOKEN'), os.getenv('LOCAL_DATA_DIR', 'data'))
APP_KEY = os.getenv('APP_KEY', '')
KINDS = {'palabras', 'silabas', 'oraciones', 'cuentos'}
DICTIONARY = 'https://raw.githubusercontent.com/wooorm/dictionaries/main/dictionaries/es/index.dic'
LICENSE = 'https://github.com/wooorm/dictionaries/blob/main/dictionaries/es/license'
MAX_AUDIO = 16 * 1024 * 1024

def authorize(x_app_key: str = Header(default='')):
    if not APP_KEY:
        raise HTTPException(503, 'Configura APP_KEY en el servidor antes de conectar la app.')
    if not secrets.compare_digest(x_app_key, APP_KEY):
        raise HTTPException(401, 'La clave de acceso no es válida.')

@app.exception_handler(Conflict)
async def conflict_handler(request, exc):
    from fastapi.responses import JSONResponse
    return JSONResponse(status_code=409, content={'detail': str(exc) or 'El dataset cambió. Actualiza e intenta nuevamente.'})

@app.exception_handler(Exception)
async def storage_handler(request, exc):
    logging.exception('Error de almacenamiento')
    from fastapi.responses import JSONResponse
    return JSONResponse(status_code=502, content={'detail': 'No se pudo confirmar la operación en el almacenamiento. Puedes reintentar sin duplicar la toma.'})

def normalize(text):
    return unicodedata.normalize('NFC', re.sub(r'\s+', ' ', text.strip())).casefold()

def item_id(kind, text):
    return hashlib.sha256(f'{kind}:{normalize(text)}'.encode()).hexdigest()[:24]

def ensure_collection(data, collection):
    if not any(c['id'] == collection for c in data['collections']):
        raise HTTPException(404, 'La carpeta no existe.')

def is_test_collection(data, collection):
    return any(c['id'] == collection and (c.get('is_test') or normalize(c['name']) == 'prueba')
               for c in data['collections'])


def scoped_recordings(data, collection):
    testing = is_test_collection(data, collection)
    return [r for r in data['recordings'] if
            (r['collection'] == collection if testing else not is_test_collection(data, r['collection']))]


def view_item(data, item):
    recordings = [r for r in data['recordings'] if r['item_id'] == item['id']]
    return dict(item, count=len(recordings), recordings=recordings)

@app.get('/health')
def health():
    return {'mode': store.mode, 'configured': bool(APP_KEY), 'version': 2, 'app_version': '0.2.7', 'dataset': store.repo, 'release': os.getenv('APP_RELEASE', 'development')}

@app.get('/library', dependencies=[Depends(authorize)])
def library(collection: str = 'principal'):
    data, _ = store.read()
    ensure_collection(data, collection)
    recordings = {}
    for recording in scoped_recordings(data, collection):
        recordings.setdefault(recording['item_id'], []).append(recording)
    items = [dict(v, count=len(recordings.get(v['id'], [])), recordings=recordings.get(v['id'], []))
             for v in data['items'].values()]
    return dict(data, items=items, mode=store.mode)

class CollectionInput(BaseModel):
    name: str = Field(min_length=1, max_length=80)

@app.post('/collections', dependencies=[Depends(authorize)])
def create_collection(payload: CollectionInput):
    name = payload.name.strip()
    if not name:
        raise HTTPException(422, 'Escribe un nombre para la carpeta.')
    result = {'id': uuid.uuid4().hex[:16], 'name': name, 'is_test': normalize(name) == 'prueba'}
    def change(data):
        existing = next((c for c in data['collections'] if normalize(c['name']) == normalize(name)), None)
        if existing:
            if normalize(name) == 'prueba':
                existing['is_test'] = True
                result.update(existing)
                for recording in data['recordings']:
                    if recording['collection'] == existing['id']:
                        recording.update(is_test=True, use_for_training=False)
                files = {f"v1/colecciones/{existing['id']}/coleccion.json": json_bytes(existing),
                    'v1/registros/grabaciones.jsonl': b''.join(json_bytes(r).replace(b'\n', b'') + b'\n' for r in data['recordings'])}
                for c in data['collections']:
                    files[f"v1/colecciones/{c['id']}/practicas.json"] = json_bytes(practiced_words(data, c['id']))
                return data, files
            raise HTTPException(409, 'Ya existe una carpeta con ese nombre.')
        data['collections'].append(result)
        return data, {f"v1/colecciones/{result['id']}/coleccion.json": json_bytes(result)}
    store.mutate(change)
    return result

class ImportInput(BaseModel):
    collection: str
    kind: str
    text: str = Field(min_length=1, max_length=200000)
    title: str = Field(default='', max_length=160)
    single: bool = False

def add_items(data, kind, texts, collection, title='', source='manual'):
    ensure_collection(data, collection)
    added = 0
    for position, text in enumerate(texts):
        text = unicodedata.normalize('NFC', text.strip())
        if not text:
            continue
        if len(text) > 4000:
            raise HTTPException(422, 'Cada fragmento debe tener como máximo 4000 caracteres.')
        key = item_id(kind, text)
        if key not in data['items']:
            data['items'][key] = {'id': key, 'kind': kind, 'text': text, 'collections': [],
                'title': title, 'position': position + 1, 'source': source}
            added += 1
        if collection not in data['items'][key]['collections']:
            data['items'][key]['collections'].append(collection)
    return added

def catalog_files(data, kind):
    if kind not in {'palabras', 'silabas'}:
        return {}
    return {f'v1/catalogos/{kind}.json': json_bytes([{'id': i['id'], 'text': i['text']}
        for i in data['items'].values() if i['kind'] == kind])}


def practiced_words(data, collection):
    words = {}
    for recording in data['recordings']:
        if recording.get('collection') != collection:
            continue
        item = data['items'].get(recording['item_id'])
        if item is not None:
            words[item['id']] = item
    practiced = []
    for item in words.values():
        item_recordings = [r for r in scoped_recordings(data, collection)
            if r['item_id'] == item['id']]
        local_count = sum(r['item_id'] == item['id'] and r['collection'] == collection for r in data['recordings'])
        last = None
        for record in item_recordings:
            if last is None or str(record.get('recorded_at') or '') > str(last.get('recorded_at') or ''):
                last = record
        practiced.append({'id': item['id'], 'kind': item['kind'], 'text': item['text'],
            'count': len(item_recordings), 'collection_count': local_count, 'completed': len(item_recordings) >= 4,
            'last_practiced_at': (last or {}).get('recorded_at')})
    practiced.sort(key=lambda entry: (entry['kind'], entry['text'].casefold()))
    return practiced

@app.post('/items/import', dependencies=[Depends(authorize)])
def import_items(payload: ImportInput):
    if payload.kind not in KINDS:
        raise HTTPException(422, 'Tipo de audio no válido.')
    # Stories split by blank lines; sentences/words/syllables one per line.
    texts = [payload.text] if payload.single else (re.split(r'\n\s*\n', payload.text) if payload.kind == 'cuentos' else payload.text.splitlines())
    if not any(t.strip() for t in texts):
        raise HTTPException(422, 'El texto está vacío.')
    if len(texts) > 10000:
        raise HTTPException(422, 'Importa como máximo 10000 elementos por lote.')
    result = {}
    def change(data):
        result['added'] = add_items(data, payload.kind, texts, payload.collection, payload.title)
        result['item_ids'] = [item_id(payload.kind, unicodedata.normalize('NFC', t.strip())) for t in texts if t.strip()]
        files = catalog_files(data, payload.kind)
        if payload.kind == 'cuentos':
            story = item_id('cuento', payload.title + '\n' + payload.text)
            files[f"{audio_folder(data, payload.collection, 'cuentos')}/{story}/texto.json"] = json_bytes(
                {'id': story, 'title': payload.title, 'text': payload.text,
                 'fragments': [item_id('cuentos', t) for t in texts if t.strip()]})
            for text in texts:
                if text.strip():
                    data['items'][item_id('cuentos', text)]['story_id'] = story
        return data, files
    store.mutate(change)
    return result

class CatalogInput(BaseModel):
    collection: str

@app.post('/catalogs/syllables', dependencies=[Depends(authorize)])
def syllables(payload: CatalogInput):
    # Explicit pedagogical catalog; no automatic linguistic claims.
    syllables = ['a', 'e', 'i', 'o', 'u'] + [c+v for c in ['b','c','d','f','g','j','l','m','n','ñ','p','r','s','t','v','y','z','ch','ll','br','bl','cr','cl','dr','fr','fl','gr','gl','pr','pl','tr'] for v in 'aeiou']
    def change(data):
        add_items(data, 'silabas', syllables, payload.collection, source='catalogo_inicial_v1')
        return data, catalog_files(data, 'silabas')
    store.mutate(change)
    return {'total': len(syllables)}

@app.post('/catalogs/dictionary', dependencies=[Depends(authorize)])
def dictionary(payload: CatalogInput):
    with httpx.Client(timeout=45, follow_redirects=True) as client:
        response = client.get(DICTIONARY)
        response.raise_for_status()
        # Hunspell headwords only: remove affix flags; never expose them as words.
        words = sorted({unicodedata.normalize('NFC', line.split('/')[0].strip())
            for line in response.text.splitlines()
            if re.fullmatch(r'[a-záéíóúüñÁÉÍÓÚÜÑ]+(?:/[^\s]+)?', line.strip())}, key=str.casefold)
        license_response = client.get('https://raw.githubusercontent.com/wooorm/dictionaries/main/dictionaries/es/license')
        license_response.raise_for_status()
    if len(words) < 1000 or len(response.content) > 10_000_000:
        raise HTTPException(502, 'La fuente del diccionario no contiene un listado válido.')
    source = {'id': 'hunspell-es', 'url': DICTIONARY, 'license_url': LICENSE,
              'sha256': hashlib.sha256(response.content).hexdigest(),
              'imported_at': datetime.now(timezone.utc).isoformat(), 'total': len(words),
              'note': 'Palabras base de Hunspell; no incluye expansión de reglas de afijos.'}
    def change(data):
        add_items(data, 'palabras', words, payload.collection, source='hunspell-es')
        data['sources'] = [s for s in data['sources'] if s['id'] != source['id']] + [source]
        return data, {'v1/catalogos/fuentes.json': json_bytes(data['sources']),
            'v1/catalogos/licencia-diccionario-es.txt': license_response.content,
            'v1/catalogos/palabras.json': json_bytes([{'id': i['id'], 'text': i['text']} for i in data['items'].values() if i['kind'] == 'palabras'])}
    store.mutate(change)
    return source

def json_bytes(data):
    import json
    return json.dumps(data, ensure_ascii=False, indent=2).encode()

def validate_wav(content):
    try:
        with wave.open(io.BytesIO(content), 'rb') as wav:
            frames, rate = wav.getnframes(), wav.getframerate()
            if wav.getnchannels() != 1 or wav.getsampwidth() != 2 or rate != 16000 or wav.getcomptype() != 'NONE':
                raise ValueError()
            if frames < 1600 or frames > 16000 * 300 or len(wav.readframes(frames)) != frames * 2:
                raise ValueError()
            return round(frames / rate, 3)
    except (wave.Error, EOFError, ValueError):
        raise HTTPException(422, 'Se requiere WAV PCM de 16 bits, mono, 16 kHz, de 0,1 a 300 segundos.')

@app.post('/recordings', dependencies=[Depends(authorize)])
def upload_audio(collection: str = Form(...), item: str = Form(...),
                 request_id: str = Form(...), file: UploadFile = File(...), recorded_at: str | None = Form(None)):
    if not re.fullmatch(r'[a-zA-Z0-9_-]{16,80}', request_id):
        raise HTTPException(422, 'Identificador de toma no válido.')
    captured_at = datetime.now(timezone.utc)
    if recorded_at is not None:
        try:
            captured_at = datetime.fromisoformat(recorded_at.replace('Z', '+00:00'))
            if captured_at.tzinfo is None or captured_at > datetime.now(timezone.utc) + timedelta(minutes=5):
                raise ValueError()
            captured_at = captured_at.astimezone(timezone.utc)
        except (ValueError, TypeError):
            raise HTTPException(422, 'La fecha de grabación no es válida.')
    content = file.file.read(MAX_AUDIO + 1)
    if len(content) > MAX_AUDIO:
        raise HTTPException(413, 'El audio supera el tamaño permitido.')
    duration = validate_wav(content)
    digest = hashlib.sha256(content).hexdigest()
    result = {}
    def change(data):
        ensure_collection(data, collection)
        if item not in data['items']:
            raise HTTPException(404, 'El elemento no existe.')
        existing = next((r for r in data['recordings'] if r['request_id'] == request_id), None)
        if existing:
            if existing['item_id'] != item or existing['sha256'] != digest or existing['collection'] != collection:
                raise HTTPException(409, 'El identificador de toma ya fue utilizado.')
            result.update(existing)
            return data, {}
        recordings = [r for r in scoped_recordings(data, collection) if r['item_id'] == item]
        if len(recordings) >= 4:
            raise HTTPException(409, 'Este elemento ya tiene sus cuatro repeticiones.')
        rep = len(recordings) + 1
        target = data['items'][item]
        if collection not in target['collections']:
            target['collections'].append(collection)
        folder = audio_folder(data, collection, target['kind'])
        if target['kind'] == 'cuentos':
            folder += f"/{target.get('story_id', item)}/fragmentos"
        path = f"{folder}/{item}/rep_{rep:02d}.wav"
        entry = {'request_id': request_id, 'item_id': item, 'collection': collection,
            'is_test': is_test_collection(data, collection), 'use_for_training': not is_test_collection(data, collection),
            'repetition': rep, 'path': path, 'text': target['text'], 'duration_sec': duration,
            'sample_rate': 16000, 'sha256': digest, 'recorded_at': captured_at.isoformat(),
            'uploaded_at': datetime.now(timezone.utc).isoformat()}
        data['recordings'].append(entry)
        result.update(entry)
        files = {path: content,
            'v1/registros/grabaciones.jsonl': b''.join(json_bytes(r).replace(b'\n', b'') + b'\n' for r in data['recordings']),
            f'v1/colecciones/{collection}/coleccion.json': json_bytes(next(c for c in data['collections'] if c['id'] == collection))}
        for c in data['collections']:
            files[f"v1/colecciones/{c['id']}/practicas.json"] = json_bytes(practiced_words(data, c['id']))
        files.update({name: json_bytes(records) for name, records in audio_manifests(data).items()})
        if data.get('storage_layout') == 'types': files['registros/grabaciones.json'] = json_bytes(data['recordings'])
        return data, files
    store.mutate(change)
    return dict(result, mode=store.mode, confirmed=True)


class ClearTestInput(BaseModel):
    collection: str
    item: str | None = None


@app.post('/test-recordings/clear', dependencies=[Depends(authorize)])
def clear_test_recordings(payload: ClearTestInput):
    result = {}
    def change(data):
        ensure_collection(data, payload.collection)
        if not is_test_collection(data, payload.collection):
            raise HTTPException(403, 'Solo se pueden borrar datos de la carpeta de prueba.')
        removed = [r for r in data['recordings'] if r['collection'] == payload.collection
                   and (payload.item is None or r['item_id'] == payload.item)]
        paths = {r['path'] for r in removed}
        data['recordings'] = [r for r in data['recordings'] if r not in removed]
        result['deleted'] = len(removed)
        # Keep shared catalog words; remove only membership in this test folder.
        for item in data['items'].values():
            if payload.item is None or item['id'] == payload.item:
                item['collections'] = [c for c in item['collections'] if c != payload.collection]
        files = {path: None for path in paths}
        files['v1/registros/grabaciones.jsonl'] = b''.join(
            json_bytes(r).replace(b'\n', b'') + b'\n' for r in data['recordings'])
        files[f'v1/colecciones/{payload.collection}/practicas.json'] = json_bytes(practiced_words(data, payload.collection))
        files.update({name: json_bytes(records) for name, records in audio_manifests(data).items()})
        if data.get('storage_layout') == 'types': files['registros/grabaciones.json'] = json_bytes(data['recordings'])
        return data, files
    store.mutate(change)
    return dict(result, confirmed=True)


@app.post('/migration/legacy', dependencies=[Depends(authorize)])
def migrate_legacy():
    current, _ = store.read()
    if current.get('legacy_imported'):
        return {'imported': 0, 'already_imported': True}
    manifests, existing_files = store.legacy_manifests()
    result = {'imported': 0, 'skipped': 0}
    def change(data):
        if data.get('legacy_imported'):
            return data, {}
        imported_paths = {r['path'] for r in data['recordings']}
        for manifest_path, entries in manifests.items():
            if not isinstance(entries, list):
                raise HTTPException(422, 'Un manifiesto anterior tiene un formato no válido. Revísalo antes de continuar.')
            folder = manifest_path.rsplit('/', 1)[0]
            collection = 'anterior_' + hashlib.sha256(folder.encode()).hexdigest()[:12]
            if not any(c['id'] == collection for c in data['collections']):
                data['collections'].append({'id': collection, 'name': folder + ' (anterior)'})
            for entry in entries:
                if not isinstance(entry, dict):
                    result['skipped'] += 1
                    continue
                text = entry.get('text', '')
                path = entry.get('path', '')
                if not path and isinstance(entry.get('file_name'), str):
                    path = folder + '/' + entry['file_name']
                if not isinstance(text, str) or not text.strip() or not isinstance(path, str) or path not in existing_files or not path.endswith('.wav'):
                    result['skipped'] += 1
                    continue
                if path in imported_paths:
                    continue
                add_items(data, 'palabras', [text], collection, source='manifiesto_anterior')
                key = item_id('palabras', text)
                count = sum(r['item_id'] == key for r in data['recordings'])
                if count >= 4:
                    result['skipped'] += 1
                    continue
                data['recordings'].append({'request_id': 'legacy_' + hashlib.sha256(path.encode()).hexdigest()[:32],
                    'item_id': key, 'collection': collection, 'repetition': count + 1, 'path': path,
                    'text': text, 'duration_sec': entry.get('duration_sec'), 'sample_rate': entry.get('sample_rate'),
                    'recorded_at': entry.get('recorded_at'), 'source': 'legacy'})
                imported_paths.add(path)
                result['imported'] += 1
        data['legacy_imported'] = True
        return data, catalog_files(data, 'palabras')
    store.mutate(change)
    return result


# API routes are registered first, then the optional Flutter web client.
def mount_web_client(application, directory):
    from fastapi.staticfiles import StaticFiles
    if directory.is_dir():
        application.mount('/', StaticFiles(directory=directory, html=True), name='web')

mount_web_client(app, Path(__file__).parent / 'site')