Spaces:
Sleeping
Sleeping
| 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.') | |
| 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.'}) | |
| 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) | |
| 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')} | |
| 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) | |
| 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 | |
| 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 | |
| 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)} | |
| 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.') | |
| 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 | |
| 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) | |
| 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') | |