#!/usr/bin/env python3 from __future__ import annotations import base64 import datetime as dt import json import os import random import re import shutil import subprocess import sys import time import http.client import urllib.error import urllib.request from pathlib import Path ENV_FILE = Path('/Users/bot1/.hermes/profiles/it/.env') FFMPEG = Path('/Users/bot1/local-ai/video-edit-lite/bin/ffmpeg') FFPROBE = Path('/Users/bot1/local-ai/video-edit-lite/bin/ffprobe') BASE = 'https://api.elevenlabs.io/v1' VOICE_DESCRIPTION = ( "A refined young European male butler voice, late twenties to early thirties. " "Elegant, soft-spoken, aristocratic and deeply controlled, with a subtle British-influenced European accent. " "The voice is smooth, low, intimate, and slightly breathy, as if speaking close to the listener. " "On the surface he is polite, loyal, calm and impeccably trained; underneath there is hidden obsessive darkness, possessiveness, " "dangerous tenderness and restrained yandere energy. He should sound like a devoted servant whose true nature has just been discovered by his mistress. " "Emotional color: quiet menace, obsessive love, fragile tenderness, restrained panic, devotion. " "Pace: slow and deliberate, with soft pauses. Never loud, comedic, monstrous or cartoonish." ) PREVIEW_TEXTS = [ "My lady... please do not step back from me. I have served you so carefully, so beautifully, for all these years. Did you truly believe my devotion was only duty? Every locked door, every missing letter, every guest I sent away... it was all for you.", "My lady, the manor is quiet tonight. No servants in the hall, no guests at the gate, no one left to interrupt us. You found the letters, didn't you? I should apologize, as a proper butler would. But forgive me... I cannot regret loving you this much.", ] CHINESE_SAMPLE_TEXT = "小姐，请别后退。我侍奉您这么多年，您真的以为那只是职责吗？每一扇被我锁上的门，每一封没有送到您手里的信，都是为了保护您。您可以叫它偏执，但对我来说，这是忠诚。" def load_key() -> str: key = os.environ.get('ELEVENLABS_API_KEY', '').strip() if key: return key if ENV_FILE.exists(): for line in ENV_FILE.read_text(errors='ignore').splitlines(): if line.startswith('ELEVENLABS_API_KEY='): return line.split('=', 1)[1].strip().strip('"').strip("'") return '' def request_json(path: str, payload: dict, key: str, timeout: int = 180) -> dict: last_err: Exception | None = None for attempt in range(1, 4): req = urllib.request.Request( BASE + path, data=json.dumps(payload).encode('utf-8'), headers={ 'xi-api-key': key, 'Content-Type': 'application/json', 'Accept': 'application/json', 'Connection': 'close', }, method='POST', ) try: with urllib.request.urlopen(req, timeout=timeout) as r: raw = r.read() return json.loads(raw.decode('utf-8')) except urllib.error.HTTPError: raise except (http.client.IncompleteRead, json.JSONDecodeError, TimeoutError, urllib.error.URLError) as e: last_err = e time.sleep(1.5 * attempt) raise RuntimeError(f'request_json failed after retries: {type(last_err).__name__}: {last_err}') def post_tts(key: str, voice_id: str, text: str, out_path: Path, model: str = 'eleven_multilingual_v2') -> dict: payload = { 'text': text, 'model_id': model, 'voice_settings': { 'stability': 0.42, 'similarity_boost': 0.84, 'style': 0.18, 'use_speaker_boost': True, 'speed': 0.90, }, 'seed': random.randint(1, 2_000_000_000), 'apply_text_normalization': 'auto', } req = urllib.request.Request( f'{BASE}/text-to-speech/{voice_id}?output_format=mp3_44100_128', data=json.dumps(payload).encode('utf-8'), headers={'xi-api-key': key, 'Content-Type': 'application/json', 'Accept': 'audio/mpeg'}, method='POST', ) try: with urllib.request.urlopen(req, timeout=120) as r: out_path.write_bytes(r.read()) return {'ok': True, 'path': str(out_path)} except urllib.error.HTTPError as e: return {'ok': False, 'error': f'HTTP {e.code}: {e.read().decode(errors="ignore")[:600]}'} except Exception as e: return {'ok': False, 'error': f'{type(e).__name__}: {e}'} def probe_audio(path: Path) -> dict: meta = {'duration': 0.0, 'size': path.stat().st_size if path.exists() else 0} ffprobe = str(FFPROBE if FFPROBE.exists() else (shutil.which('ffprobe') or 'ffprobe')) ffmpeg = str(FFMPEG if FFMPEG.exists() else (shutil.which('ffmpeg') or 'ffmpeg')) try: p = subprocess.run([ffprobe, '-v', 'error', '-show_entries', 'format=duration,bit_rate,size', '-of', 'json', str(path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=20) if p.returncode == 0: d = json.loads(p.stdout).get('format', {}) meta['duration'] = float(d.get('duration') or 0) meta['bit_rate'] = int(float(d.get('bit_rate') or 0)) meta['size'] = int(d.get('size') or meta['size']) except Exception as e: meta['probe_error'] = f'{type(e).__name__}: {e}' try: p = subprocess.run([ffmpeg, '-hide_banner', '-nostats', '-i', str(path), '-af', 'volumedetect,silencedetect=noise=-42dB:d=0.55', '-f', 'null', '-'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=60) s = p.stderr m = re.search(r'mean_volume:\s*([-\d.]+) dB', s) meta['mean_db'] = float(m.group(1)) if m else None m = re.search(r'max_volume:\s*([-\d.]+) dB', s) meta['max_db'] = float(m.group(1)) if m else None silences = [] for line in s.splitlines(): me = re.search(r'silence_end:\s*([\d.]+) \| silence_duration:\s*([\d.]+)', line) if me: silences.append(float(me.group(2))) meta['silence_count'] = len(silences) meta['max_silence'] = max(silences) if silences else 0.0 meta['silence_total'] = sum(silences) except Exception as e: meta['ffmpeg_error'] = f'{type(e).__name__}: {e}' return meta def score(rec: dict) -> float: score = 50.0 dur = rec.get('duration') or 0 if 11 <= dur <= 30: score += 18 elif 7 <= dur < 11 or 30 < dur <= 38: score += 8 else: score -= 10 mean = rec.get('mean_db') maxdb = rec.get('max_db') if mean is not None: score += 10 if -27 <= mean <= -13 else -4 if maxdb is not None: if -4.5 <= maxdb <= -0.7: score += 8 elif maxdb > -0.3: score -= 10 if (rec.get('max_silence') or 0) <= 1.3: score += 6 if (rec.get('silence_total') or 0) > max(3.0, dur * 0.25): score -= 8 return round(score, 2) def create_voice_from_preview(key: str, generated_voice_id: str, voice_name: str) -> dict: attempts = [ {'voice_name': voice_name, 'voice_description': VOICE_DESCRIPTION, 'generated_voice_id': generated_voice_id, 'labels': {'character': 'white-cut-black yandere butler', 'accent': 'european', 'gender': 'male'}}, {'voice_name': voice_name, 'voice_description': VOICE_DESCRIPTION, 'generated_voice_id': generated_voice_id}, {'name': voice_name, 'voice_description': VOICE_DESCRIPTION, 'generated_voice_id': generated_voice_id}, ] errors = [] for payload in attempts: try: return request_json('/text-to-voice/create-voice-from-preview', payload, key) except urllib.error.HTTPError as e: errors.append(f'HTTP {e.code}: {e.read().decode(errors="ignore")[:600]}') except Exception as e: errors.append(f'{type(e).__name__}: {e}') return {'error': ' | '.join(errors)} def main() -> None: key = load_key() if not key: raise SystemExit('ELEVENLABS_API_KEY not found') stamp = dt.datetime.now().strftime('%Y%m%d_%H%M%S') out_dir = Path(f'/Users/bot1/Documents/aiwork/elevenlabs_butler_voice_{stamp}') out_dir.mkdir(parents=True, exist_ok=True) previews = [] for round_idx in range(2): payload = { 'voice_description': VOICE_DESCRIPTION, 'text': PREVIEW_TEXTS[round_idx % len(PREVIEW_TEXTS)], 'seed': random.randint(1, 2_000_000_000), } try: data = request_json('/text-to-voice/create-previews', payload, key) except urllib.error.HTTPError as e: print(json.dumps({'stage': 'create_previews_error', 'http': e.code, 'body': e.read().decode(errors='ignore')[:1000]}, ensure_ascii=False)) continue ps = data.get('previews', []) print(f'preview_round_{round_idx+1}: {len(ps)} previews') for j, p in enumerate(ps, 1): audio_b64 = p.get('audio_base_64') or p.get('audio_base64') or '' if not audio_b64: continue path = out_dir / f'preview_r{round_idx+1}_{j}.mp3' path.write_bytes(base64.b64decode(audio_b64)) rec = {k: v for k, v in p.items() if k not in {'audio_base_64', 'audio_base64'}} rec['path'] = str(path) rec['round'] = round_idx + 1 rec['preview_index'] = j rec.update(probe_audio(path)) rec['score'] = score(rec) previews.append(rec) if not previews: raise SystemExit('No previews generated') previews.sort(key=lambda r: r.get('score', 0), reverse=True) best = previews[0] selected_path = out_dir / 'SELECTED_voice_design_preview.mp3' shutil.copyfile(best['path'], selected_path) voice_name = '白切黑欧洲男管家_' + stamp create_result = {} generated_voice_id = best.get('generated_voice_id') or best.get('voice_id') if generated_voice_id: create_result = create_voice_from_preview(key, generated_voice_id, voice_name) voice_id = create_result.get('voice_id') or create_result.get('voice', {}).get('voice_id') or create_result.get('id') sample_result = None if voice_id: sample_result = post_tts(key, voice_id, CHINESE_SAMPLE_TEXT, out_dir / 'SELECTED_chinese_audition.mp3') if sample_result and sample_result.get('ok'): sample_result.update(probe_audio(Path(sample_result['path']))) summary = { 'out_dir': str(out_dir), 'selected_preview': str(selected_path), 'generated_voice_id_for_preview': generated_voice_id, 'created_voice_name': voice_name if voice_id else None, 'created_voice_id': voice_id, 'create_voice_response_safe': {k: v for k, v in create_result.items() if k not in {'settings'}}, 'chinese_sample': sample_result, 'top': previews[:5], } (out_dir / 'summary.json').write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding='utf-8') print(json.dumps({ 'out_dir': summary['out_dir'], 'selected_preview': summary['selected_preview'], 'created_voice_name': summary['created_voice_name'], 'created_voice_id': summary['created_voice_id'], 'chinese_sample': sample_result.get('path') if sample_result and sample_result.get('ok') else None, 'preview_count': len(previews), 'top_scores': [r.get('score') for r in previews[:3]], 'create_error': create_result.get('error') if create_result.get('error') else None, }, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()