#!/usr/bin/env python3 """Atelier TTS : préparation locale par défaut, un appel explicite avec --generate. Python 3.10+, bibliothèque standard. Aucune relance automatique après un échec. """ import argparse import hashlib import json import os import re import tempfile from pathlib import Path from urllib.request import Request, urlopen API = 'https://api.elevenlabs.io' FORMAT = 'mp3_44100_128' MODELS = {'eleven_multilingual_v2', 'eleven_v4'} def canonical(value): return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode('utf-8') def prepare(text, voice_id, model, dictionary=None): if not text.strip() or len(text) > 10000: raise ValueError('Le texte doit contenir entre 1 et 10 000 caractères, espaces compris.') if model not in MODELS: raise ValueError('Cet atelier accepte uniquement Multilingual v2 ou Eleven v4.') if not re.fullmatch(r'[A-Za-z0-9_-]+', voice_id): raise ValueError('Identifiant de voix invalide.') body = {'text': text, 'model_id': model, 'language_code': 'fr', 'seed': 42, 'voice_settings': {'stability': 0.5, 'similarity_boost': 0.75, 'style': 0.0, 'use_speaker_boost': True, 'speed': 1.0}} if dictionary is not None: if not all(isinstance(dictionary.get(k), str) and dictionary[k] for k in ('id', 'version_id')): raise ValueError('Le dictionnaire doit posséder id et version_id non vides.') body['pronunciation_dictionary_locators'] = [{ 'pronunciation_dictionary_id': dictionary['id'], 'version_id': dictionary['version_id'], }] request = {'url': f'{API}/v1/text-to-speech/{voice_id}?output_format={FORMAT}', 'body': body} key = hashlib.sha256(canonical(request)).hexdigest() return {'key': key, 'characters': len(text), 'request': request, 'evaluation_status': 'not_listened'} def send(request): key = os.environ.get('ELEVENLABS_API_KEY') if not key: raise ValueError('Définir ELEVENLABS_API_KEY dans votre environnement local.') req = Request(request['url'], data=canonical(request['body']), method='POST', headers={ 'xi-api-key': key, 'Content-Type': 'application/json', 'Accept': 'audio/mpeg'}) with urlopen(req, timeout=90) as response: data = response.read() if response.status != 200 or 'audio/' not in response.headers.get('Content-Type', '') or not data: raise RuntimeError('Réponse audio inattendue ; inspecter la demande avant toute reprise.') return data def generate(plan, root, transport=send): directory = root / plan['key'] try: directory.mkdir(parents=True, exist_ok=False) except FileExistsError: meta = json.loads((directory / 'manifest.json').read_text()) if meta['generation_status'] != 'generated': raise RuntimeError('Demande déjà tentée ou en cours. Aucune nouvelle soumission automatique.') audio = (directory / 'audio.mp3').read_bytes() if hashlib.sha256(audio).hexdigest() != meta['audio_sha256']: raise RuntimeError('Audio local altéré ; restaurer la copie validée avant réutilisation.') return {'cache': True, 'directory': str(directory), 'evaluation_status': meta['evaluation_status']} meta = dict(plan, generation_status='attempted', audio_sha256=None) (directory / 'manifest.json').write_text(json.dumps(meta, ensure_ascii=False, indent=2)) # L'état attempted reste après une coupure : on ne sait pas si le fournisseur a facturé. audio = transport(plan['request']) if not audio: raise RuntimeError('Audio vide : demande conservée comme attempted.') partial = directory / 'audio.part' partial.write_bytes(audio) partial.replace(directory / 'audio.mp3') meta.update(generation_status='generated', audio_sha256=hashlib.sha256(audio).hexdigest()) temp = directory / 'manifest.part' temp.write_text(json.dumps(meta, ensure_ascii=False, indent=2)) temp.replace(directory / 'manifest.json') return {'cache': False, 'directory': str(directory), 'evaluation_status': 'not_listened'} def self_test(): base = prepare('Le chiffre d’affaires est disponible.', 'test_voice', 'eleven_multilingual_v2', {'id':'test_dict','version_id':'v1'}) assert base == prepare('Le chiffre d’affaires est disponible.', 'test_voice', 'eleven_multilingual_v2', {'id':'test_dict','version_id':'v1'}) changes = [ prepare('Le chiffre d’affaires a changé.', 'test_voice', 'eleven_multilingual_v2', {'id':'test_dict','version_id':'v1'}), prepare('Le chiffre d’affaires est disponible.', 'other_voice', 'eleven_multilingual_v2', {'id':'test_dict','version_id':'v1'}), prepare('Le chiffre d’affaires est disponible.', 'test_voice', 'eleven_v4', {'id':'test_dict','version_id':'v1'}), prepare('Le chiffre d’affaires est disponible.', 'test_voice', 'eleven_multilingual_v2', {'id':'test_dict','version_id':'v2'})] assert all(x['key'] != base['key'] for x in changes) for args in [('', 'voice', 'eleven_v4', None), ('a'*10001,'voice','eleven_v4',None), ('bonjour','voice','eleven_v4',{'id':'x'}), ('bonjour','bad/voice','eleven_v4',None)]: try: prepare(*args) except ValueError: pass else: raise AssertionError('Entrée invalide acceptée') calls = [] def fake(req): calls.append(req); return b'FAUX AUDIO POUR TEST DE STOCKAGE, NON ECOUTABLE' with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) assert not generate(base,root,fake)['cache'] assert generate(base,root,fake)['cache'] and len(calls)==1 assert generate(base,root,fake)['evaluation_status']=='not_listened' (root/base['key']/'audio.mp3').write_bytes(b'corrompu') try: generate(base,root,fake) except RuntimeError: pass else: raise AssertionError('Corruption non détectée') attempts = [] def failure(req): attempts.append(req); raise TimeoutError('Coupure simulée') try: generate(changes[0],root,failure) except TimeoutError: pass try: generate(changes[0],root,failure) except RuntimeError: pass else: raise AssertionError('Nouvel appel autorisé après résultat ambigu') assert len(attempts)==1 print('OK : empreintes, invalidation, validation des entrées, stockage, cache, corruption et coupure ambiguë. Zéro appel réseau ; faux octets audio.') def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--self-test', action='store_true') parser.add_argument('--text', type=Path) parser.add_argument('--voice-id') parser.add_argument('--model', default='eleven_multilingual_v2') parser.add_argument('--dictionary-json', type=Path) parser.add_argument('--output', type=Path, default=Path('sorties-voix')) parser.add_argument('--generate', action='store_true', help='Soumet une requête réelle, potentiellement facturée.') args = parser.parse_args() if args.self_test: self_test(); return if args.text is None or not args.voice_id: parser.error('--text et --voice-id sont requis') dictionary = json.loads(args.dictionary_json.read_text()) if args.dictionary_json else None plan = prepare(args.text.read_text(), args.voice_id, args.model, dictionary) if args.generate: # Vérification avant création du verrou local, sans afficher la clé. if not os.environ.get('ELEVENLABS_API_KEY'): parser.error('ELEVENLABS_API_KEY manquante') result = generate(plan, args.output) else: result = dict(plan, mode='preparation_locale_sans_reseau') print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()