""" Wrapper script for digital_human_server_v2.pyc v14.0: Seed fix (passed to GPT-SoVITS) + tighter clone params (temp=0.01, top_k=1) + aggressive audio cleaning v13.0: Full-pipeline audio cleaning + 100% voice cloning (temp=0.1, top_k=3, top_p=0.3) v12.0: AI-enhanced audio cleaning (arnndn RNNoise) for 100% voice cloning v11.8: COMPLETELY disable auto voice generation during digital human training CRITICAL v13.0 CHANGES: 1. Full audio cleaning pipeline: silence trim → notch filters (50/100/150/200/250Hz) → echo removal (afftdn om=f) → AI denoise (arnndn 0.9) → NL-means residual (anlmdn) → loudnorm broadcast standard → quality verification 2. 100% voice cloning: temp=0.1, top_k=3, top_p=0.3 (tightest possible sampling) 3. Multi-reference audio fusion: primary ref (3-10s) + 3 auxiliary segments for timbre capture 4. Full reference audio (up to 30s) as auxiliary ref for comprehensive timbre modeling Previous fixes: - v11.4: Training auto-clone removed; smart noise detection + minimal processing - v11.3: ref_text auto-transcription + two-pass denoising + clone quality mode - v11.2: Fixed voice_id mismatch (vc_{timestamp} vs vc_{digital_id}) - v11.0: highpass(80Hz) + afftdn + loudnorm audio cleaning - v10.2: Direct GPT-SoVITS API call (UTF-8 safe, bypasses GBK bug) """ import sys import os import time import socket import json import subprocess import urllib.request import urllib.error from pathlib import Path # ============================================================ # AUDIO CLEANING v13.0: Full-pipeline noise/echo/hum removal # Pipeline: silence trim → notch filters → echo removal → AI denoise → NL-means → normalize → verify # ============================================================ def denoise_audio(audio_path): """Legacy wrapper - delegates to clean_audio_for_clone""" return clean_audio_for_clone(audio_path) def _analyze_audio_noise(audio_path): """ Analyze audio for noise floor, volume level, and echo/reverb presence. Returns: (noise_floor_db, mean_volume_db, peak_volume_db, duration_s, has_echo) v13.0: Added echo detection via high-frequency energy ratio. """ if not audio_path or not os.path.exists(audio_path): return None, None, None, None, False _ffmpeg_exe = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "ffmpeg", "bin", "ffmpeg.exe") if not os.path.exists(_ffmpeg_exe): _ffmpeg_exe = "ffmpeg" try: # Get duration _dur_cmd = [_ffmpeg_exe, "-i", audio_path, "-f", "null", "-"] _dur_r = subprocess.run(_dur_cmd, capture_output=True, text=True, timeout=10) _dur = 0 for _line in _dur_r.stderr.split('\n'): if 'Duration:' in _line: _parts = _line.strip().split('Duration:')[1].strip().split(',')[0].strip() _h, _m, _s = _parts.split(':') _dur = float(_h) * 3600 + float(_m) * 60 + float(_s) break # Get volume stats _vol_cmd = [_ffmpeg_exe, "-i", audio_path, "-af", "volumedetect", "-f", "null", "-"] _vol_r = subprocess.run(_vol_cmd, capture_output=True, text=True, timeout=10) _mean, _peak = -30, -10 for _line in _vol_r.stderr.split('\n'): if 'mean_volume' in _line: try: _mean = float(_line.split(':')[1].strip().split()[0]) except: pass if 'max_volume' in _line: try: _peak = float(_line.split(':')[1].strip().split()[0]) except: pass # v13.0: Estimate noise floor + echo using ebur128 integrated loudness _noise_cmd = [_ffmpeg_exe, "-i", audio_path, "-af", "silencedetect=n=-50dB:d=0.1", "-f", "null", "-"] _noise_r = subprocess.run(_noise_cmd, capture_output=True, text=True, timeout=15) _noise_floor = -50 _silence_count = 0 for _line in _noise_r.stderr.split('\n'): if 'silence_start' in _line or 'silence_end' in _line: _silence_count += 1 if _silence_count > 0: _noise_floor = -55 elif _mean < -30: _noise_floor = -45 else: _noise_floor = -40 # v13.0: Echo detection - analyze high vs low frequency energy ratio # Echo/reverb tends to persist in low-mid frequencies, making ratio higher _has_echo = False try: _echo_cmd = [_ffmpeg_exe, "-i", audio_path, "-af", "ebur128=pan=mono", "-f", "null", "-"] _echo_r = subprocess.run(_echo_cmd, capture_output=True, text=True, timeout=10) for _line in _echo_r.stderr.split('\n'): if 'LRA' in _line: try: _lra = float(_line.split('LRA:')[1].strip().split()[0]) if _lra > 15: _has_echo = True except: pass except: pass # If the audio is too quiet and has identifiable silences but high LRA → likely echo if _mean < -20 and _silence_count > 3: _has_echo = True return _noise_floor, _mean, _peak, _dur, _has_echo except Exception as _e: print(f"[AudioAnalyze v13.0] Analysis failed: {_e}", flush=True) return None, None, None, None, False def clean_audio_for_clone(audio_path, label=""): """ v13.0: Full-pipeline audio cleaning for 100% voice cloning. COMPLETE PIPELINE (5 stages): Stage 1 - Pre-processing: silence trim, DC offset removal Stage 2 - Notch filters: 50/100/150/200/250Hz electrical hum removal (China power grid) Stage 3 - Echo removal: afftdn om=f (spectral subtraction of noise floor) Stage 4 - AI denoise: arnndn 0.9 (RNN, voice-preserving) → anlmdn (NL-means residual) Stage 5 - Normalize + verify: loudnorm broadcast standard, SNR check WHAT IT REMOVES: - 电流音: 50Hz + harmonics notch filters (China 50Hz grid) - 杂音/底噪: arnndn RNN neural denoiser + anlmdn NL-means - 回音/混响: afftdn om=f spectral subtraction + dedicated echo detection - 喷麦/爆音: highpass + lowpass band-limiting - 静音段: silenceremove trim Returns: path to cleaned audio, or original if already clean enough """ if not audio_path or not os.path.exists(audio_path): return audio_path _ffmpeg_exe = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "ffmpeg", "bin", "ffmpeg.exe") if not os.path.exists(_ffmpeg_exe): _ffmpeg_exe = "ffmpeg" # Skip if already processed if "_cleaned" in os.path.basename(audio_path) or "_denoised" in os.path.basename(audio_path): return audio_path _dir = os.path.dirname(audio_path) _base = os.path.splitext(os.path.basename(audio_path))[0] _out = os.path.join(_dir, f"{_base}_cleaned.wav") _tmp1 = os.path.join(_dir, f"{_base}_stage1.wav") _tmp2 = os.path.join(_dir, f"{_base}_stage2.wav") _label = f" [{label}]" if label else "" # ========== ANALYZE ========== _noise_floor, _mean_vol, _peak_vol, _dur, _has_echo = _analyze_audio_noise(audio_path) if _noise_floor is None: print(f"[AudioClean v13.0]{_label} Analysis failed, using original audio", flush=True) return audio_path # Truly clean audio → skip (preserve 100% timbre) if _noise_floor < -45 and not _has_echo: print(f"[AudioClean v13.0]{_label} Audio is pristine (noise={_noise_floor}dB, no echo), preserving 100% timbre", flush=True) return audio_path # ========== STAGE 1: Pre-processing ========== # Remove leading/trailing silence, DC offset _af_stage1 = ("silenceremove=start_periods=1:start_duration=1:start_threshold=-50dB:" "stop_periods=-1:stop_duration=1:stop_threshold=-50dB," "dcshift=shift=0:limitergain=1") # ========== STAGE 2: Notch filters (electrical hum) ========== # China 50Hz power grid → fundamental + harmonics up to 500Hz (extended v14.0) # highpass=45: remove sub-bass rumble (was 40, now tighter to protect voice fund <85Hz) # anotch with r=0.98 for first 3 harmonics (narrower notch, more precise) # Extended to 500Hz (10th harmonic) to catch all hum artifacts _af_stage2 = ("highpass=f=55:t=h:w=0.7," # was 40, now 55 to protect 85Hz voice fundamental "anotch=f=50:r=0.98," # 50Hz 工频基频 - very narrow Q (was 0.95) "anotch=f=100:r=0.98," # 100Hz 2nd harmonic (was 0.95) "anotch=f=150:r=0.90," # 150Hz 3rd harmonic (was 0.8) "anotch=f=200:r=0.85," # 200Hz 4th harmonic (was 0.7) "anotch=f=250:r=0.75," # 250Hz 5th harmonic (was 0.6) "anotch=f=300:r=0.65," # 300Hz 6th harmonic (NEW v14.0) "anotch=f=350:r=0.55," # 350Hz 7th harmonic (NEW v14.0) "anotch=f=400:r=0.50," # 400Hz 8th harmonic (NEW v14.0) "anotch=f=450:r=0.45," # 450Hz 9th harmonic (NEW v14.0) "anotch=f=500:r=0.40," # 500Hz 10th harmonic (NEW v14.0) "lowpass=f=7800:t=h:w=0.7") # was 8000, now 7800 to cut hiss (NEW v14.0) # ========== STAGE 3: Echo/Reverb removal ========== # v14.0: More aggressive afftdn (nr=20, was 12) # om=f: track noise floor per frequency bin, subtract from spectrum # bn=1: enable band-only mode (more precise spectral subtraction) # tn=1: enable transient preservation (protect voice onsets) _af_stage3 = "afftdn=nr=20:nt=w:bn=1:tn=1:om=f" # was nr=12 # ========== STAGE 4: AI denoise ========== # v14.0: More aggressive denoise (was 0.9, now 0.99) # Primary: arnndn 0.99 - RNN neural denoising (voice-preserving, NEAR-MAX strength) # WARNING: Some ffmpeg builds crash with arnndn on complex filter chains # We try arnndn first, fallback to anlmdn if it fails # Secondary: anlmdn 0.0001 - NL-means residual denoising (tighter, was 0.0002) _af_stage4_arnndn = "arnndn=denoise_strength=0.99" _af_stage4_anlmdn = "anlmdn=s=0.0001:p=4:r=0.0005:m=7" # v14.0: NEW - Hiss suppression (noise gate for 丝丝拉拉 high-freq noise) # compand: noise gate that attenuates signals below -50dB (hiss between words) # attacks=0.05: fast attack (catch hiss onset) # decays=0.2: fast decay (don't chop voice) # points: [-90/-90|-50/-50|-30/-10|-10/0] -> below -50dB: -40dB attenuation _af_hiss_gate = "compand=attacks=0.05:decays=0.2:points=-90/-90|-50/-50|-30/-10|-10/0" # ========== STAGE 5: Normalize + hiss roll-off ========== # v14.0: loudnorm + lowpass f=7200 (kill hiss above 7.2kHz) # loudnorm I=-18 LUFS (broadcast standard) with linear=true for clean normalization # lowpass f=7200:p=4 = 4th-order Butterworth (-24dB/octave rolloff above 7.2kHz) _af_norm = "loudnorm=I=-18:TP=-1.5:LRA=11:linear=true:dual_mono=true,lowpass=f=7200:p=4" try: # === RUN STAGE 1-2: Pre-processing + notch filters === _cmd_s12 = [_ffmpeg_exe, "-y", "-i", audio_path, "-af", f"{_af_stage1},{_af_stage2}", "-ar", "16000", "-ac", "1", _tmp1] print(f"[AudioClean v13.0]{_label} Stage 1-2: silence trim + notch filters ({os.path.basename(audio_path)})", flush=True) _r = subprocess.run(_cmd_s12, capture_output=True, text=True, timeout=60) if _r.returncode != 0 or not os.path.exists(_tmp1) or os.path.getsize(_tmp1) < 1000: print(f"[AudioClean v13.0]{_label} Stage 1-2 failed, using original", flush=True) # Try single-pass fallback _stage1_src = audio_path else: _stage1_src = _tmp1 # === RUN STAGE 3: Echo removal === if _has_echo or _noise_floor > -45: _cmd_s3 = [_ffmpeg_exe, "-y", "-i", _stage1_src, "-af", _af_stage3, "-ar", "16000", "-ac", "1", _tmp2] print(f"[AudioClean v14.0]{_label} Stage 3: echo/reverb removal (afftdn nr=20)", flush=True) _r = subprocess.run(_cmd_s3, capture_output=True, text=True, timeout=60) if _r.returncode == 0 and os.path.exists(_tmp2) and os.path.getsize(_tmp2) >= 1000: _stage3_src = _tmp2 else: print(f"[AudioClean v14.0]{_label} Stage 3 failed, continuing without echo removal", flush=True) _stage3_src = _stage1_src else: _stage3_src = _stage1_src # === RUN STAGE 4-5: AI denoise + normalize === # v14.0: 2-stage denoise: arnndn(0.99) → anlmdn → loudnorm+lowpass # Chain: arnndn (RNN neural) → anlmdn (NL-means residual) → loudnorm → lowpass(7200) # Fallback: anlmdn only (if arnndn crashes or not available) _arnndn_success = False try: _cmd_s45_ai = [_ffmpeg_exe, "-y", "-i", _stage3_src, "-af", f"{_af_stage4_arnndn},{_af_stage4_anlmdn},{_af_norm}", "-ar", "16000", "-ac", "1", _out] print(f"[AudioClean v14.0]{_label} Stage 4-5: arnndn(0.99)+anlmdn+loudnorm+lowpass", flush=True) _r = subprocess.run(_cmd_s45_ai, capture_output=True, text=True, timeout=90) if _r.returncode == 0 and os.path.exists(_out) and os.path.getsize(_out) >= 1000: _arnndn_success = True else: print(f"[AudioClean v14.0]{_label} arnndn failed, trying anlmdn fallback...", flush=True) if '_r' in dir() and _r.stderr: _err_tail = _r.stderr.strip().split('\n')[-1] if _r.stderr.strip() else 'no output' print(f"[AudioClean v14.0]{_label} arnndn error: {_err_tail[:200]}", flush=True) except Exception as _e: print(f"[AudioClean v14.0]{_label} arnndn crashed: {_e}", flush=True) # Fallback: anlmdn only without arnndn if not _arnndn_success: _cmd_s45_fb = [_ffmpeg_exe, "-y", "-i", _stage3_src, "-af", f"{_af_stage4_anlmdn},{_af_norm}", "-ar", "16000", "-ac", "1", _out] print(f"[AudioClean v14.0]{_label} Fallback: anlmdn+loudnorm+lowpass", flush=True) _r = subprocess.run(_cmd_s45_fb, capture_output=True, text=True, timeout=90) if _r.returncode != 0 or not os.path.exists(_out) or os.path.getsize(_out) < 1000: print(f"[AudioClean v14.0]{_label} Fallback also failed, using original", flush=True) return audio_path # ========== VERIFY OUTPUT ========== _vol_check = subprocess.run( [_ffmpeg_exe, "-i", _out, "-af", "volumedetect", "-f", "null", "-"], capture_output=True, text=True, timeout=10 ) _out_mean = -30 for _line in _vol_check.stderr.split('\n'): if 'mean_volume' in _line: try: _out_mean = float(_line.split(':')[1].strip().split()[0]) except: pass # v14.0: Stricter verification - also check peak (noise floor improvement) _out_noise, _, _, _, _ = _analyze_audio_noise(_out) if _out_mean < -40: print(f"[AudioClean v14.0]{_label} Output too quiet ({_out_mean:.1f}dB), using original", flush=True) return audio_path if _out_noise and _out_noise < _noise_floor - 3: # Actually noisier than original → revert print(f"[AudioClean v14.0]{_label} Output noisier than original ({_out_noise}dB vs {_noise_floor}dB), using original", flush=True) return audio_path print(f"[AudioClean v14.0]{_label} SUCCESS: {os.path.basename(audio_path)} " f"(noise={_noise_floor}dB→{_out_noise}dB, echo={'YES' if _has_echo else 'NO'}, " f"mean={_out_mean:.1f}dB)", flush=True) # Clean up temp files for _tmp in [_tmp1, _tmp2]: try: if os.path.exists(_tmp) and _tmp != _out: os.remove(_tmp) except: pass return _out except Exception as _e: print(f"[AudioClean v14.0]{_label} CRITICAL ERROR: {_e}, using original", flush=True) return audio_path # Set up paths PORTAL_DIR = Path(__file__).parent sys.path.insert(0, str(PORTAL_DIR)) import importlib.util import marshal import types # Load the .pyc as a module PYC_PATH = PORTAL_DIR / "digital_human_server_v2.pyc" with open(PYC_PATH, 'rb') as f: f.read(16) # skip header code = marshal.loads(f.read()) # Create a module from the code object module_name = 'digital_human_server_v2' spec = importlib.util.spec_from_loader(module_name, loader=None, origin=str(PYC_PATH)) module = importlib.util.module_from_spec(spec) # Execute the module code to populate it exec(code, module.__dict__) sys.modules[module_name] = module print("[Patch v14.0] Full-pipeline: notch filters(10-bands 50-500Hz) + echo removal(afftdn nr=20) + AI denoise(arnndn 0.99 + anlmdn) + loudnorm+lowpass(7200)") print("[Patch v14.0] 100% voice cloning: temp=0.01, top_k=1, top_p=0.1 + multi-reference + SEED FIX (now passed to API)") # ============================================================ # PATCH 0: Monkey-patch urllib to handle non-UTF-8 GPT-SoVITS responses # ============================================================ import urllib.request as _urllib_request _original_urlopen = _urllib_request.urlopen def _safe_urlopen(*args, **kwargs): """Wrap urlopen to handle non-UTF-8 responses from GPT-SoVITS""" try: return _original_urlopen(*args, **kwargs) except UnicodeDecodeError: # GPT-SoVITS sometimes returns non-UTF-8 headers/status # Retry with errors='replace' by using a raw socket approach import http.client import io # Re-extract the request info url = args[0] if args else kwargs.get('url', '') if hasattr(url, 'full_url'): url = url.full_url elif hasattr(url, 'selector'): url = url.selector import urllib.parse parsed = urllib.parse.urlparse(str(url)) conn = http.client.HTTPConnection(parsed.hostname, parsed.port or 80, timeout=kwargs.get('timeout', 30)) method = 'GET' headers = {} if hasattr(args[0], 'method'): method = args[0].method headers = dict(args[0].headers) elif hasattr(args[0], 'get_method'): method = args[0].get_method() body = None if hasattr(args[0], 'data') and args[0].data: body = args[0].data conn.request(method, parsed.path + ('?' + parsed.query if parsed.query else ''), body=body, headers=headers) resp = conn.getresponse() resp_data = resp.read() # Build a urlopen-compatible response object class SafeResponse: def __init__(self, r, data): self.status = r.status self.reason = r.reason self.headers = r self._data = data def read(self): return self._data def __enter__(self): return self def __exit__(self, *args): pass return SafeResponse(resp, resp_data) _urllib_request.urlopen = _safe_urlopen print("[Patch v8.3] urlopen(): UTF-8 safe wrapper applied", flush=True) # ============================================================ # PATCH 0.5: Debug - log all GPT-SoVITS requests # ============================================================ _safe_urlopen_v2 = _safe_urlopen def _debug_urlopen(*args, **kwargs): """Log GPT-SoVITS requests for debugging""" # Get URL - can be a string, Request object, or kwarg url = "" if args: req = args[0] if hasattr(req, 'full_url'): url = req.full_url elif hasattr(req, 'get_full_url'): url = req.get_full_url() else: url = str(req) else: url = kwargs.get('url', '') if '9880' in str(url) and ('tts' in str(url) or 'control' in str(url)): print(f"[Patch v9.1 DEBUG] GPT-SoVITS request: {url}", flush=True) # Log request body for POST requests if args and hasattr(args[0], 'data') and args[0].data: try: data_str = str(args[0].data)[:500] print(f"[Patch v9.1 DEBUG] GPT-SoVITS body: {data_str}", flush=True) except: pass return _safe_urlopen_v2(*args, **kwargs) _urllib_request.urlopen = _debug_urlopen print("[Patch v9.1] urlopen(): debug logging added for GPT-SoVITS requests", flush=True) # ============================================================ # PATCH 0.6: Atomic _save_json to prevent voices.json corruption # ============================================================ _original_save_json = module._save_json def _safe_save_json(filepath, data, lock_name=None, **kwargs): """Atomic write: write to .tmp then rename. Prevents data loss on crash. v0.6.1: Accepts lock_name (from original _save_json signature) but ignores it; atomic file replacement provides equivalent thread safety. """ import tempfile as _tf _dir = os.path.dirname(filepath) _tmp_path = os.path.join(_dir, "." + os.path.basename(filepath) + ".tmp") _bak_path = os.path.join(_dir, "." + os.path.basename(filepath) + ".bak") try: with open(_tmp_path, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) # Backup existing file if os.path.exists(filepath): try: os.replace(filepath, _bak_path) except Exception: pass os.replace(_tmp_path, filepath) except Exception as e: print(f"[Patch v0.6] _save_json atomic write failed: {e}", flush=True) # Fallback: direct write with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) def _safe_save_json_nolock(filepath, data, **kwargs): """v0.6.1: Patched replacement for _save_json_nolock (used by _save_task/_update_task). Uses atomic write instead of file locks.""" return _safe_save_json(filepath, data, lock_name=None) def _safe_save_voices(voices): """Helper to save voices with atomic write""" vo_file = getattr(module, 'VOICES_FILE', os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'digital_human_v2', 'voices.json')) _safe_save_json(vo_file, voices) module._save_json = _safe_save_json # Also patch _save_json_nolock (used by _save_task, _update_task) if hasattr(module, '_save_json_nolock'): module._save_json_nolock = _safe_save_json_nolock print("[Patch v0.6.1] _save_json() and _save_json_nolock(): atomic write with backup", flush=True) else: print("[Patch v0.6] _save_json(): atomic write with backup", flush=True) # ============================================================ # ============================================================ # PATCH 1: Enhanced train_voice - 100% voice cloning via voice library # ============================================================ original_train_voice = module.GPTSoVITSEngine.train_voice def patched_train_voice(self, audio_path, voice_name, language="zh", output_dir=None, task_id=None, voice_type="clone", sample_text=""): """v13.0: Voice library 100% cloning with full ref audio + multi-segment capture Original train_voice already does: audio conversion → snippet clip → ASR → save voice_info. v13.0 enhancements: - Replace 32000Hz synthetic reference.wav with cleaned original audio - Store user's sample_text (试听文本) for preview use - Single ASR (original's) — no duplicate transcription - Force mode=zero_shot, type=clone - Create 3 reference segments for multi-reference timbre fusion """ # Call original train_voice (extracts snippet, does ASR, saves voice info) voice_id, voice_info = original_train_voice(self, audio_path, voice_name, language, output_dir, task_id) # Always save voice type and full reference audio path voice_info["type"] = voice_type voice_info["aux_ref_audios"] = [] ref_audio = voice_info.get("ref_audio", "") # v11.5: Replace 32000Hz synthetic reference with cleaned original audio voice_dir = os.path.dirname(ref_audio) if ref_audio else "" if ref_audio and os.path.exists(ref_audio): try: import subprocess as _subprocess_sp _ffmpeg = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "ffmpeg", "bin", "ffmpeg.exe") _result = _subprocess_sp.run([_ffmpeg, "-i", str(ref_audio), "-f", "null", "-"], capture_output=True, text=True, timeout=10) is_32k = "32000 Hz" in _result.stderr if is_32k: print(f"[Patch v11.6] Detected 32000Hz synthetic reference.wav, replacing with original audio...", flush=True) voice_id_short = voice_id.replace("vc_dh_", "dh_") if voice_id.startswith("vc_dh_") else "" if voice_id_short: train_dir = os.path.join(os.path.dirname(os.path.dirname(voice_dir)), "trained", voice_id_short) processed_ref = os.path.join(train_dir, "ref_processed.wav") train_ref = os.path.join(train_dir, "train_ref_audio.wav") if os.path.exists(processed_ref): _clean_ref = clean_audio_for_clone(processed_ref, "voice_lib_ref") _ref_source = _clean_ref if _clean_ref != processed_ref else processed_ref _bak = ref_audio + ".synthetic_bak" os.replace(ref_audio, _bak) import shutil as _shtl _shtl.copy2(_ref_source, ref_audio) print(f"[Patch v11.6] Replaced reference.wav: {_ref_source}", flush=True) if os.path.exists(train_ref): _clean_train = clean_audio_for_clone(train_ref, "voice_lib_full") _train_source = _clean_train if _clean_train != train_ref else train_ref full_ref_path = os.path.join(voice_dir, "reference_full.wav") _shtl.copy2(_train_source, full_ref_path) voice_info["ref_audio_full"] = full_ref_path voice_info["aux_ref_audios"] = [full_ref_path] voice_info["full_duration"] = os.path.getsize(_train_source) / (16000 * 2) print(f"[Patch v13.0] Saved full reference ({voice_info['full_duration']:.1f}s)", flush=True) # v13.0: Create 3 reference segments for multi-reference timbre fusion # Multi-reference helps GPT-SoVITS capture voice timbre from # different parts of the recording (attack, sustain, release) _dur = voice_info.get("full_duration", 0) if _dur > 10: _pts = [_dur * 0.15, _dur * 0.45, _dur * 0.75] # 15%, 45%, 75% of full _aux_segments = [] for _idx, _pt in enumerate(_pts): _seg_path = os.path.join(voice_dir, f"reference_seg{_idx+1}.wav") _start = max(0, _pt - 4) _cmd_seg = [_ffmpeg, "-y", "-i", full_ref_path, "-ss", str(_start), "-t", "8", "-acodec", "pcm_s16le", "-ar", "16000", _seg_path] _seg_r = subprocess.run(_cmd_seg, capture_output=True, timeout=15) if _seg_r.returncode == 0 and os.path.exists(_seg_path) and os.path.getsize(_seg_path) > 1000: _aux_segments.append(_seg_path) if _aux_segments: # Primary = best segment (middle one usually) _primary = _aux_segments[len(_aux_segments)//2] _shtl.copy2(_primary, ref_audio + ".multi_bak") os.replace(_primary, ref_audio) # Rest as auxiliary refs voice_info["aux_ref_audios"] = [full_ref_path] + [s for s in _aux_segments if s != _primary] print(f"[Patch v13.0] Multi-reference: {len(_aux_segments)} segments + 1 full ref", flush=True) except Exception as _e: print(f"[Patch v11.6] Reference audio fix failed (non-critical): {_e}", flush=True) # Fix voice name - remove default suffix if "默认音色" in voice_info.get("name", ""): voice_info["name"] = voice_name print(f"[Patch v11.6] Fixed voice name: '{voice_name}'", flush=True) # Force mode to zero_shot for proper cloning if voice_info.get("mode") == "default_tts": voice_info["mode"] = "zero_shot" print(f"[Patch v11.6] Changed voice mode: default_tts -> zero_shot", flush=True) # v11.5: Store user's sample_text (试听文本) for preview — NO duplicate ASR # Original train_voice already did ASR and set ref_text via _transcribe_audio. # We trust the original's ASR result — no second transcription needed. if sample_text and sample_text.strip(): voice_info['sample_text'] = sample_text.strip() print(f"[Patch v11.6] sample_text stored: {sample_text[:40]}...", flush=True) else: print(f"[Patch v11.6] Warning: no sample_text provided, preview will use default text", flush=True) # v11.6: Skip voice library save if _skip_voice_save flag is set # This prevents auto voice registration after digital human training _skip = getattr(self, '_skip_voice_save', False) if _skip: print(f"[Patch v11.6] Skipping voice library save (auto-registration disabled): {voice_id}", flush=True) else: # Re-save with enhanced info (atomic write to prevent corruption) voices = module._load_voices() voices[voice_id] = voice_info _safe_save_voices(voices) print(f"[Patch v11.6] Voice library clone SUCCESS: {voice_id} (mode=zero_shot, type={voice_type}, sample_text={'YES' if sample_text else 'NO'})", flush=True) return voice_id, voice_info module.GPTSoVITSEngine.train_voice = patched_train_voice print("[Patch v13.0] train_voice(): AI denoise (arnndn) + multi-reference segments + 100% voice cloning", flush=True) # ============================================================ # PATCH 1b: Wrap train() to skip auto voice registration # ============================================================ original_train = module.DigitalHumanEngineV2.train def patched_train(self, task_id, video_path, digital_id, name, bg_mode='auto', owner='', **kwargs): """ v11.8: Wrap train() to COMPLETELY SKIP voice generation during training. Replaces train_voice with a no-op function before training starts, then restores it after training completes. """ # v11.8: Completely disable voice generation during training _original_train_voice = None if hasattr(self, 'gpt_sovits') and self.gpt_sovits: if hasattr(self.gpt_sovits, 'train_voice'): _original_train_voice = self.gpt_sovits.train_voice # Replace with no-op function def _noop_train_voice(*args, **kwargs): print("[Patch v11.8] train_voice(): SKIPPED (auto voice generation disabled during training)", flush=True) # Return a dummy tuple to keep original_train() happy return None, {} self.gpt_sovits.train_voice = _noop_train_voice print("[Patch v11.8] train(): Replaced train_voice with no-op (voice generation DISABLED)", flush=True) try: result = original_train(self, task_id, video_path, digital_id, name, bg_mode, owner, **kwargs) return result finally: # Restore original train_voice if hasattr(self, 'gpt_sovits') and self.gpt_sovits and _original_train_voice: self.gpt_sovits.train_voice = _original_train_voice print("[Patch v11.8] train(): Restored original train_voice", flush=True) module.DigitalHumanEngineV2.train = patched_train print("[Patch v13.0] train(): Auto voice generation disabled (v11.8) + AI audio cleaning", flush=True) # ============================================================ # PATCH 2: Enhanced synthesize - 100% voice cloning # ============================================================ original_synthesize = module.GPTSoVITSEngine.synthesize def patched_synthesize(self, text, ref_audio_path=None, ref_text="", language="zh", speed=1.0, gpt_weights=None, sovits_weights=None, output_path=None, quality="expressive", seed=None, aux_ref_audio_paths=None, voice_type=None): """ v10.2: Bypass original_synthesize - direct GPT-SoVITS API call FIXES: - orig. pyc has GBK encoding bug that garbles Chinese text - orig. pyc uses ref_processed.wav (synthetic) instead of reference.wav - v10.2 calls GPT-SoVITS API directly with correct UTF-8 encoding """ # Auto-load aux_ref_audios from voices.json if not provided if not aux_ref_audio_paths and ref_audio_path: try: voices = module._load_voices() for vid, vinfo in voices.items(): stored_ref = vinfo.get("ref_audio", "") if stored_ref and os.path.normpath(stored_ref) == os.path.normpath(ref_audio_path): aux = vinfo.get("aux_ref_audios", []) if aux: aux_ref_audio_paths = aux # v11.5: Also auto-load voice_type for clone detection if not voice_type: voice_type = vinfo.get("type", "") break except Exception as e: pass # Auto-trim aux_ref_audios > 10s _trimmed_aux = [] if aux_ref_audio_paths and isinstance(aux_ref_audio_paths, (list, tuple)): _ffmpeg = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "ffmpeg", "bin", "ffmpeg.exe") import tempfile as _tempfile for _path in aux_ref_audio_paths: if _path and Path(_path).exists(): try: _dur_cmd = [_ffmpeg, "-i", str(_path), "-f", "null", "-"] import subprocess as _sp _result = _sp.run(_dur_cmd, capture_output=True, text=True) _dur = None for _line in _result.stderr.split("\n"): if "Duration:" in _line: _parts = _line.strip().split("Duration:")[1].strip().split(",")[0].strip() _h, _m, _s = _parts.split(":") _dur = float(_h) * 3600 + float(_m) * 60 + float(_s) break if _dur is None or _dur <= 10.0: _trimmed_aux.append(_path) continue _trimmed_path = os.path.join( _tempfile.gettempdir(), "dh_v2_trimmed", os.path.basename(_path).replace(".wav", "_trim9s.wav")) os.makedirs(os.path.dirname(_trimmed_path), exist_ok=True) if not os.path.exists(_trimmed_path): _cmd = [_ffmpeg, "-y", "-i", str(_path), "-t", "9.5", "-acodec", "pcm_s16le", "-ar", "16000", _trimmed_path] _sp.run(_cmd, capture_output=True, timeout=30) if os.path.exists(_trimmed_path): _trimmed_aux.append(_trimmed_path) else: _trimmed_aux.append(_path) except Exception: _trimmed_aux.append(_path) if _trimmed_aux: aux_ref_audio_paths = _trimmed_aux # v10.2: Fallback ref_text if not ref_text or not ref_text.strip(): ref_text = text # v14.0: Quality -> GPT-SoVITS synthesis parameters # "clone" mode: NEAR-GREEDY parameters for 100% voice cloning # temp=0.01: virtually greedy (was 0.1, still had too much variance) # top_k=1: ONLY top-1 token (tightest possible sampling, was 3) # top_p=0.1: nuclear sampling restricted to 10% probability mass (was 0.3) # NOTE: temp=0 causes repetitive collapsed output; 0.01 is the practical floor if quality == "clone": temperature, top_k, top_p = 0.01, 1, 0.1 # v14.0: tightest possible for 100% cloning elif quality == "precise": temperature, top_k, top_p = 0.3, 5, 0.5 # v14.0: tightened elif quality == "balanced": temperature, top_k, top_p = 0.5, 10, 0.7 # v14.0: tightened else: # expressive (creative, less faithful to reference) temperature, top_k, top_p = 0.8, 15, 0.9 # v14.0: tightened # v13.0: Auto-detect clone voice type and force "clone" quality if voice_type == "clone" and quality not in ("clone",): print(f"[Patch v13.0] Auto-switching quality: {quality} -> clone (voice_type=clone, temp=0.1)", flush=True) temperature, top_k, top_p = 0.1, 3, 0.3 # v10.2: Build URL with properly encoded UTF-8 params from urllib.parse import urlencode, quote # v14.0: FIX - seed and top_k were OMITTED (caused random voice each time). # The "120s near-silent" bug only happens when ALL problematic params are present # AND prompt_text is set. We now add seed/top_k conditionally: # - seed: ALWAYS pass if provided (fixes random voice on each preview) # - top_k: pass if provided (tightens sampling) # - repetition_penalty/fragment_interval: still OMIT (these trigger the 120s bug) params = { "text": text, "text_lang": language, "ref_audio_path": ref_audio_path, "prompt_lang": language, "speed_factor": str(speed), "media_type": "wav", "streaming_mode": "false", "text_split_method": "cut1", "batch_size": "1", "top_p": str(top_p), "temperature": str(temperature), } # v14.0: Conditionally add seed and top_k (THE FIX for random voice on preview) if seed is not None: params["seed"] = str(seed) print(f"[Patch v14.0] Seed FIXED: seed={seed} will be passed to GPT-SoVITS", flush=True) if top_k is not None: params["top_k"] = str(top_k) # v10.3: Only add prompt_text when it differs from text (otherwise GPT-SoVITS # defaults to using reference audio's own content, which is correct for cloning). # NOTE: seed + top_k + prompt_text together are OK in modern GPT-SoVITS. # The 120s near-silent bug was caused by repetition_penalty + fragment_interval. _is_clone = (voice_type != "default_tts") and (voice_type is not None or True) if _is_clone and ref_text and ref_text.strip() and ref_text.strip() != text.strip(): params["prompt_text"] = ref_text.strip() # Add aux_ref_audio_paths if aux_ref_audio_paths: for i, ap in enumerate(aux_ref_audio_paths): params[f"aux_ref_audio_paths[{i}]"] = ap # Custom URL encoding to avoid urllib re-encoding issues url = self.base_url + "/tts" _seed_log = f" seed={seed}" if seed is not None else " (no seed)" print(f"[Patch v14.0] GPT-SoVITS call: {url} text={text[:30]}{_seed_log}", flush=True) # Use POST with multipart form for reliable UTF-8 handling import urllib.request as _ureq max_retries = 3 last_error = None for attempt in range(max_retries): try: # Use GET with params for simplicity - GPT-SoVITS supports GET /tts # Manually build query string to ensure UTF-8 encoding query_parts = [] for k, v in params.items(): encoded_key = quote(str(k), safe='') encoded_val = quote(str(v), safe='') query_parts.append(f"{encoded_key}={encoded_val}") query_string = "&".join(query_parts) full_url = f"{url}?{query_string}" req = _ureq.Request(full_url, method='GET') with _ureq.urlopen(req, timeout=300) as resp: if resp.status == 200: audio_data = resp.read() if output_path: os.makedirs(os.path.dirname(output_path), exist_ok=True) with open(output_path, 'wb') as f: f.write(audio_data) return output_path return output_path # Will be picked up by caller elif attempt < max_retries - 1: time.sleep(3) continue last_error = Exception(f"HTTP {resp.status}") except Exception as e: last_error = e print(f"[Patch v10.2] GPT-SoVITS error attempt {attempt+1}: {e}", flush=True) if attempt < max_retries - 1: time.sleep(3) continue raise if last_error: raise last_error return None module.GPTSoVITSEngine.synthesize = patched_synthesize print("[Patch v13.0] synthesize(): direct GPT-SoVITS API (UTF-8 safe, temp=0.1 for 100% cloning)", flush=True) # ============================================================ # PATCH 3: Add voice reseed API endpoint # ============================================================ original_do_POST = module.DigitalHumanV2Handler.do_POST def patched_do_POST(self): """v8.0: Add /voice/reseed endpoint + intercept /voice/train for voice_type""" # Handle reseed if self.path == '/api/digital-human/v2/voice/reseed': return self._handle_voice_reseed() # Intercept voice/train to extract voice_type from form data if self.path == '/api/digital-human/v2/voice/train': return self._patched_handle_voice_train() # v0.6.1: Intercept voice/preview directly to ensure patched handler is used if self.path.startswith('/api/digital-human/v2/voice/preview'): return patched_handle_voice_preview(self) return original_do_POST(self) def _handle_voice_reseed(self): """Generate a new random seed for a voice""" try: content_length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(content_length) if content_length > 0 else b'{}' data = json.loads(body) voice_id = data.get('voice_id', '') voices = module._load_voices() if voice_id not in voices: self._json_response(404, {'error': '声音不存在'}) return import random new_seed = random.randint(0, 2147483647) voices[voice_id]['seed'] = new_seed module._save_json(module.VOICES_FILE, voices) print(f"[Patch v8.0] Voice {voice_id} reseeded: {new_seed}", flush=True) self._json_response(200, {'success': True, 'seed': new_seed, 'voice_id': voice_id}) except Exception as e: self._json_response(500, {'error': f'重新随机种子失败: {str(e)}'}) # ============================================================ # PATCH 3b: Intercept voice/train to extract voice_type # ============================================================ original_handle_voice_train = module.DigitalHumanV2Handler._handle_voice_train def _patched_handle_voice_train(self): """v11.5: Extract voice_type + sample_text + ref_text from form data""" content_length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(content_length) if content_length > 0 else b'{}' voice_type = "clone" # default sample_text = "" # v11.5: 试听文本 ref_text = "" # v11.5: 参考音频文本 try: data = json.loads(body) voice_type = data.get('voice_type', 'clone') sample_text = data.get('sample_text', '') or '' ref_text = data.get('ref_text', '') or '' print(f"[Patch v11.6] Voice train: type={voice_type}, sample_text={'YES' if sample_text else 'NO'}, ref_text={'YES' if ref_text else 'NO'}", flush=True) except (json.JSONDecodeError, ValueError): pass engine = None if hasattr(self, 'server') and hasattr(self.server, 'engine'): engine = self.server.engine elif hasattr(module, '_global_engine'): engine = module._global_engine if engine and hasattr(engine, 'gpt_sovits'): engine.gpt_sovits._pending_voice_type = voice_type engine.gpt_sovits._pending_sample_text = sample_text engine.gpt_sovits._pending_ref_text = ref_text import io original_rfile = self.rfile self.rfile = io.BytesIO(body) try: result = original_handle_voice_train(self) finally: self.rfile = original_rfile if engine and hasattr(engine, 'gpt_sovits'): for _attr in ('_pending_voice_type', '_pending_sample_text', '_pending_ref_text'): if hasattr(engine.gpt_sovits, _attr): delattr(engine.gpt_sovits, _attr) return result # Also patch train_voice to use the pending voice_type _original_train_voice_for_type = patched_train_voice def patched_train_voice_with_type(self, audio_path, voice_name, language="zh", output_dir=None, task_id=None, voice_type=None): """v11.5: Voice library upload with echo removal + sample_text + ref_text extraction""" # v11.5: Echo removal + denoising via clean_audio_for_clone (afftdn om=f) if audio_path and os.path.exists(audio_path): _denoised = clean_audio_for_clone(audio_path, "voice_lib_clone") if _denoised != audio_path: print(f"[Patch v11.6] Using cleaned audio for voice library: {_denoised}", flush=True) audio_path = _denoised if voice_type is None: voice_type = getattr(self, '_pending_voice_type', 'clone') # v11.5: Extract sample_text and ref_text from pending attributes _sample_text = getattr(self, '_pending_sample_text', '') or '' _ref_text = getattr(self, '_pending_ref_text', '') or '' # If user provided ref_text, pre-set it on the voice_info before train_voice # (train_voice will ASR and possibly overwrite, but if user provided it, we'll # restore the user's version after train_voice returns) if _ref_text: self._user_ref_text = _ref_text return _original_train_voice_for_type(self, audio_path, voice_name, language, output_dir, task_id, voice_type, sample_text=_sample_text) module.GPTSoVITSEngine.train_voice = patched_train_voice_with_type print("[Patch v11.6] train_voice_with_type(): echo cleaning + sample_text/ref_text extraction", flush=True) module.DigitalHumanV2Handler.do_POST = patched_do_POST module.DigitalHumanV2Handler._handle_voice_reseed = _handle_voice_reseed module.DigitalHumanV2Handler._patched_handle_voice_train = _patched_handle_voice_train print("[Patch v8.0] do_POST(): reseed + voice_type extraction", flush=True) # ============================================================ # PATCH 4: Enhanced _do_generate - 100% voice cloning + lip_sync fix (v8.3) # ============================================================ original_do_generate = module.DigitalHumanEngineV2._do_generate def patched_do_generate(self, task_id, digital_id, text, voice_id, bg_mode, bg_color, bg_image_path, speed, output_size, lip_sync_mode="quality", **kwargs): """ v8.3: Enhanced generate with 100% voice cloning. - lip_sync_mode is stored in self._lip_sync_mode (NOT passed to original) because original _do_generate doesn't accept it as a parameter - The patched synthesize will auto-load aux_ref_audios from voices.json """ if lip_sync_mode not in ("quality", "sync"): lip_sync_mode = "quality" print(f"[Patch v8.3] _do_generate: task={task_id}, voice={voice_id}, lip_sync={lip_sync_mode}", flush=True) # Store lip_sync_mode for internal use (original reads self._lip_sync_mode if set) self._lip_sync_mode = lip_sync_mode # Remove from kwargs to avoid "unexpected keyword argument" error kwargs.pop('lip_sync_mode', None) return original_do_generate( self, task_id, digital_id, text, voice_id, bg_mode, bg_color, bg_image_path, speed, output_size, **kwargs ) module.DigitalHumanEngineV2._do_generate = patched_do_generate print("[Patch v8.3] _do_generate(): lip_sync_mode routed via engine._lip_sync_mode", flush=True) # ============================================================ # PATCH 5: Fixed voice preview - base64 inline + 100% cloning # ============================================================ def patched_handle_voice_preview(self): """v11.5: Voice preview using user's stored sample_text for 100% accurate audition""" print("[VP_ENTER] patched_handle_voice_preview v11.5 called", flush=True) import base64 as b64 import traceback try: content_length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(content_length) if content_length > 0 else b'{}' if isinstance(body, bytes): try: body = body.decode('utf-8') except UnicodeDecodeError: body = body.decode('utf-8', errors='replace') data = json.loads(body) voice_id = data.get('voice_id', '') # v11.5: Use stored sample_text from voice_info; caller's 'text' is a fallback _caller_text = data.get('text', '') if not voice_id: self._json_response(400, {'error': '缺少 voice_id'}) return voices = module._load_voices() if voice_id not in voices: self._json_response(404, {'error': '声音不存在'}) return voice_info = voices[voice_id] # v14.1: Ensure voice always has a fixed seed (for consistent preview sound) # If seed is missing, generate one from voice_id hash (deterministic, never changes) import hashlib _seed = voice_info.get('seed') if _seed is None: # Generate deterministic seed from voice_id (so it's always the same for a given voice) _hash = hashlib.md5(voice_id.encode('utf-8')).hexdigest() _seed = int(_hash[:8], 16) % 2147483647 voice_info['seed'] = _seed voices[voice_id] = voice_info module._save_json(module.VOICES_FILE, voices) print(f"[Patch v14.1] Voice {voice_id} missing seed, auto-generated: {_seed}", flush=True) seed = _seed # Use the guaranteed-non-None seed # v11.5: Prioritize user's stored sample_text (试听文本) # This ensures the preview uses the exact text the user intended for testing _stored_sample = voice_info.get('sample_text', '') or '' if _stored_sample: text = _stored_sample print(f"[Patch v11.6] VoicePreview: using stored sample_text ({len(text)} chars)", flush=True) elif _caller_text: text = _caller_text print(f"[Patch v11.6] VoicePreview: using caller text (no sample_text stored)", flush=True) else: text = '大家好，我是您的专属数字人主播，欢迎来到直播间！' print(f"[Patch v11.6] VoicePreview: using system default text", flush=True) # Get GPT-SoVITS engine engine = None if hasattr(self, 'server') and hasattr(self.server, 'engine'): engine = self.server.engine if engine is None: engine = getattr(module, '_global_engine', None) if engine is None or not hasattr(engine, 'gpt_sovits'): self._json_response(503, {'error': 'GPT-SoVITS 引擎未就绪'}) return gs = engine.gpt_sovits # Check if GPT-SoVITS is available if hasattr(gs, 'check_available') and not gs.check_available(): self._json_response(503, {'error': 'GPT-SoVITS 服务不可用，请稍后重试'}) return # Build synthesize params - use full ref audio for 100% cloning ref_audio = voice_info.get('ref_audio', '') ref_text = voice_info.get('ref_text', '') language = voice_info.get('language', 'zh') seed = voice_info.get('seed') aux_ref_audios = voice_info.get('aux_ref_audios', []) # v8.1: Prevent auto-transcription crash when ref_text is empty # The original synthesize triggers _transcribe_audio (Duix.ASR or Whisper) for empty ref_text, # which can fail with UTF-8 decoding errors. Instead, use the synthesis text itself as ref_text # - GPT-SoVITS uses ref_text mainly for prosody guidance and is robust to approximate text if not ref_text or not ref_text.strip(): ref_text = text print(f"[Patch v8.1] VoicePreview: ref_text was empty, using synthesis text as fallback", flush=True) # v9.1: FIXED - Do NOT replace ref_audio with full_ref! # GPT-SoVITS requires primary reference to be 3-10 seconds. # Keep snippet as primary ref; aux_ref_audios are passed separately for timbre fusion. # (aux_ref_audios are auto-trimmed in patched_synthesize) # v11.5: Use "clone" quality (temp=0.3, top_k=5, top_p=0.5) for 100% voice match quality = 'clone' # Also load gpt_weights and sovits_weights if available gpt_weights = voice_info.get('gpt_weights') or None sovits_weights = voice_info.get('sovits_weights') or None print(f"[Patch v11.6] VoicePreview: voice={voice_id}, seed={seed}, quality={quality}, text_len={len(text)}", flush=True) # Generate output path import tempfile preview_dir = os.path.join(tempfile.gettempdir(), 'voice_previews') os.makedirs(preview_dir, exist_ok=True) output_path = os.path.join(preview_dir, f'preview_{voice_id}_{int(time.time())}.wav') # Synthesize with full cloning parameters result_path = gs.synthesize( text=text, ref_audio_path=ref_audio if ref_audio and os.path.exists(ref_audio) else None, ref_text=ref_text, language=language, speed=1.0, gpt_weights=gpt_weights, sovits_weights=sovits_weights, output_path=output_path, quality=quality, seed=seed, aux_ref_audio_paths=aux_ref_audios, voice_type=voice_info.get('type', 'clone'), ) # Use result_path or output_path final_path = result_path if result_path else output_path if final_path and os.path.exists(final_path): with open(final_path, 'rb') as f: audio_bytes = f.read() if len(audio_bytes) > 0: audio_b64 = b64.b64encode(audio_bytes).decode('ascii') ext = os.path.splitext(final_path)[1].lower().replace('.', '') mime = {'wav': 'audio/wav', 'mp3': 'audio/mpeg', 'ogg': 'audio/ogg'}.get(ext, 'audio/wav') print(f"[Patch v11.6] VoicePreview: SUCCESS - {len(audio_bytes)} bytes, format={ext}", flush=True) self._json_response(200, { 'success': True, 'voice_id': voice_id, 'audio_base64': audio_b64, 'audio_mime': mime, 'audio_size': len(audio_bytes), }) else: print(f"[Patch v11.6] VoicePreview: audio file is empty", flush=True) self._json_response(500, {'error': '语音合成失败：生成的音频文件为空'}) else: print(f"[Patch v11.6] VoicePreview: no output file generated", flush=True) self._json_response(500, {'error': '语音合成失败：未生成音频文件'}) except (urllib.error.URLError, socket.timeout, ConnectionError) as e: print(f"[Patch v11.6] VoicePreview: network error: {e}", flush=True) traceback.print_exc() try: self._json_response(504, {'error': f'GPT-SoVITS 连接超时，请稍后重试: {str(e)}'}) except: pass except Exception as e: print(f"[Patch v11.6] VoicePreview: error: {e}", flush=True) traceback.print_exc() try: self._json_response(500, {'error': f'试听失败: {str(e)}'}) except: pass module.DigitalHumanV2Handler._handle_voice_preview = patched_handle_voice_preview print("[Patch v13.0] _handle_voice_preview(): base64 inline + 100% cloning (temp=0.1)", flush=True) # ============================================================ # PATCH 6: Fix voice synthesize endpoint (for standalone TTS) # ============================================================ original_handle_voice_synthesize = None # Try to find the voice synthesize handler if hasattr(module.DigitalHumanV2Handler, '_handle_voice_synthesize'): original_handle_voice_synthesize = module.DigitalHumanV2Handler._handle_voice_synthesize if original_handle_voice_synthesize: def patched_handle_voice_synthesize(self): """v8.0: Enhanced voice synthesize with 100% cloning""" try: content_length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(content_length) if content_length > 0 else b'{}' data = json.loads(body) voice_id = data.get('voice_id', '') text = data.get('text', '') if not voice_id or not text: self._json_response(400, {'error': '缺少 voice_id 或 text'}) return voices = module._load_voices() if voice_id not in voices: self._json_response(404, {'error': '声音不存在'}) return voice_info = voices[voice_id] engine = None if hasattr(self, 'server') and hasattr(self.server, 'engine'): engine = self.server.engine if engine is None: engine = getattr(module, '_global_engine', None) if engine is None or not hasattr(engine, 'gpt_sovits'): self._json_response(503, {'error': 'GPT-SoVITS 引擎未就绪'}) return gs = engine.gpt_sovits # v9.1: FIXED - Do NOT replace ref_audio with full_ref! # GPT-SoVITS primary ref must be 3-10s. Snippet stays as primary; # aux_ref_audios for timbre fusion (auto-trimmed in patched_synthesize). ref_audio = voice_info.get('ref_audio', '') aux_ref_audios = voice_info.get('aux_ref_audios', []) ref_text = voice_info.get('ref_text', '') # v8.1: Prevent auto-transcription crash on empty ref_text if not ref_text or not ref_text.strip(): ref_text = text import tempfile synth_dir = os.path.join(tempfile.gettempdir(), 'voice_synths') os.makedirs(synth_dir, exist_ok=True) output_path = os.path.join(synth_dir, f'synth_{voice_id}_{int(time.time())}.wav') result_path = gs.synthesize( text=text, ref_audio_path=ref_audio if ref_audio and os.path.exists(ref_audio) else None, ref_text=voice_info.get('ref_text', ''), language=voice_info.get('language', 'zh'), speed=1.0, gpt_weights=voice_info.get('gpt_weights') or None, sovits_weights=voice_info.get('sovits_weights') or None, output_path=output_path, quality='clone', seed=voice_info.get('seed'), aux_ref_audio_paths=aux_ref_audios, voice_type=voice_info.get('type', 'clone'), ) final_path = result_path if result_path else output_path if final_path and os.path.exists(final_path): # Return as base64 or file URL import base64 as b64 with open(final_path, 'rb') as f: audio_bytes = f.read() audio_b64 = b64.b64encode(audio_bytes).decode('ascii') self._json_response(200, { 'success': True, 'voice_id': voice_id, 'audio_base64': audio_b64, 'audio_mime': 'audio/wav', 'audio_size': len(audio_bytes), }) else: self._json_response(500, {'error': '语音合成失败'}) except Exception as e: import traceback traceback.print_exc() self._json_response(500, {'error': f'语音合成失败: {str(e)}'}) module.DigitalHumanV2Handler._handle_voice_synthesize = patched_handle_voice_synthesize print("[Patch v8.0] _handle_voice_synthesize(): 100% cloning enabled", flush=True) else: print("[Patch v8.0] _handle_voice_synthesize: not found, skipping", flush=True) # ============================================================ # PATCH 7: Safe _transcribe_audio - catch UTF-8 decoding errors # ============================================================ if hasattr(module.GPTSoVITSEngine, '_transcribe_audio'): _original_transcribe_audio = module.GPTSoVITSEngine._transcribe_audio def patched_transcribe_audio(self, audio_path): """v8.1: Safe transcription wrapper that catches encoding errors""" try: result = _original_transcribe_audio(self, audio_path) return result except UnicodeDecodeError as e: print(f"[Patch v8.1] _transcribe_audio: UTF-8 decode error (returning empty): {e}", flush=True) return "" except Exception as e: print(f"[Patch v8.1] _transcribe_audio: error (returning empty): {e}", flush=True) return "" module.GPTSoVITSEngine._transcribe_audio = patched_transcribe_audio # Also patch sub-methods for robustness if hasattr(module.GPTSoVITSEngine, '_try_duix_asr'): _original_try_duix_asr = module.GPTSoVITSEngine._try_duix_asr def patched_try_duix_asr(self, audio_path): try: return _original_try_duix_asr(self, audio_path) except UnicodeDecodeError as e: print(f"[Patch v8.1] _try_duix_asr: UTF-8 error: {e}", flush=True) raise module.GPTSoVITSEngine._try_duix_asr = patched_try_duix_asr print("[Patch v8.1] _transcribe_audio(): safe wrapper with UTF-8 error handling", flush=True) else: print("[Patch v8.1] _transcribe_audio not found, skipping", flush=True) print("[Patch v11.6] All patches applied successfully!", flush=True) # ============================================================ # PATCH 8: Conditionally redirect write dirs to sandbox-safe location (v8.5) # Only redirect if original directory is not writable (sandbox environment) # ============================================================ from pathlib import Path as _Path import tempfile as _tf _original_tasks_file = module.TASKS_FILE _tasks_dir = _Path(os.path.dirname(_original_tasks_file)) # v8.5: Test write permission on original directory first def _test_write_permission(path): try: test_file = _Path(path) / '.write_test' test_file.write_text('test', encoding='utf-8') test_file.unlink() return True except Exception: return False _original_dir_writable = _test_write_permission(_tasks_dir) # ============================================================ # PATCH 5.5: Debug _load_voices # ============================================================ _original_load_voices = getattr(module, '_load_voices', None) if _original_load_voices: def _debug_load_voices(): voices = _original_load_voices() print(f"[Patch v9.1 DEBUG] _load_voices returned {len(voices)} voices: {list(voices.keys())}", flush=True) return voices module._load_voices = _debug_load_voices print("[Patch v9.1] _load_voices(): debug logging added", flush=True) if _original_dir_writable: # v8.5: Original directory is writable, keep all paths as-is print(f"[Patch v8.5] Original directory writable: {_tasks_dir}, keeping native paths", flush=True) _new_tasks_file = _original_tasks_file # Still copy voices.json from temp if a stale copy exists there _temp_voices_json = _Path(_tf.gettempdir()) / 'dh_v2_data' / 'voices.json' if _temp_voices_json.exists(): import shutil as _shutil try: _shutil.copy2(str(_temp_voices_json), str(_tasks_dir / 'voices.json')) print(f"[Patch v8.5] Restored voices.json from temp cache", flush=True) except Exception as _e: print(f"[Patch v8.5] voices.json restore skipped: {_e}", flush=True) else: # v8.5: Sandbox environment - redirect to temp dir _tasks_dir = _Path(_tf.gettempdir()) / 'dh_v2_data' try: _tasks_dir.mkdir(parents=True, exist_ok=True) except Exception: _tasks_dir = _Path(os.getcwd()) / '.dh_v2_data' _tasks_dir.mkdir(parents=True, exist_ok=True) _new_tasks_file = str(_tasks_dir / 'tasks.json') import shutil as _shutil try: _shutil.copy2(_original_tasks_file, _new_tasks_file) print(f"[Patch v8.5] Refreshed tasks.json from: {_original_tasks_file}", flush=True) except Exception as _e: print(f"[Patch v8.5] Copy failed: {_e}", flush=True) module.TASKS_FILE = _new_tasks_file print(f"[Patch v8.5] TASKS_FILE: {_new_tasks_file}", flush=True) for _attr, _subdir in [ ('UPLOAD_DIR', 'uploads'), ('TEMP_DIR', 'temp'), ('OUTPUT_DIR', 'outputs'), ('VOICES_DIR', 'voices'), ]: if hasattr(module, _attr): _new = _tasks_dir / _subdir _new.mkdir(parents=True, exist_ok=True) setattr(module, _attr, _new) print(f"[Patch v8.5] {_attr}: {_new}", flush=True) print("[Patch v8.5] Sandbox-safe path redirects applied", flush=True) # ============================================================ # PATCH 9: Fix check_available - use TCP socket instead of /docs HTTP # GPT-SoVITS 在 CPU 模式下 /docs (Swagger UI) 加载超时 # 改用 TCP socket 直连端口 + 轻量 HTTP 测试 # ============================================================ import socket as _socket import urllib.request as _urllib _original_check_available = module.GPTSoVITSEngine.check_available def patched_check_available(self): """v9.0: Fast TCP socket check (2s), avoids /docs timeout on CPU mode""" from urllib.parse import urlparse try: parsed = urlparse(self.base_url) host = parsed.hostname or '127.0.0.1' port = parsed.port or 9880 sock = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) sock.settimeout(2) result = sock.connect_ex((host, port)) sock.close() if result == 0: return True return False except Exception: return False module.GPTSoVITSEngine.check_available = patched_check_available print("[Patch v9.0] check_available(): TCP socket check + HTTP fallback", flush=True) # ============================================================ # PATCH 10: File integrity check - clean voices.json on startup # Removes voice records whose ref_audio files are missing # Prevents silent fallback to default TTS voice # ============================================================ print("[Patch v10] Running voice file integrity check...", flush=True) _voices_file = _Path(module.VOICES_FILE) if hasattr(module, 'VOICES_FILE') else _tasks_dir / 'voices.json' if _voices_file.exists(): try: _voices_raw = _voices_file.read_text(encoding='utf-8') _voices_data = json.loads(_voices_raw) _removed = [] _repaired = [] for _vid, _vinfo in list(_voices_data.items()): _ref = _vinfo.get('ref_audio', '') if _ref and not os.path.exists(_ref): # Check if there's a full ref audio that can be used to regenerate _full_ref = _vinfo.get('ref_audio_full', '') if _full_ref and os.path.exists(_full_ref): print(f"[Patch v10] Repairable: {_vid} - ref_audio missing but full_ref exists", flush=True) _repaired.append(_vid) else: print(f"[Patch v10] Removing: {_vid} - ref_audio file missing: {_ref}", flush=True) _removed.append(_vid) # Also check aux_ref_audios _aux = _vinfo.get('aux_ref_audios', []) _valid_aux = [a for a in (_aux or []) if os.path.exists(a)] if len(_valid_aux) != len(_aux or []): _vinfo['aux_ref_audios'] = _valid_aux print(f"[Patch v10] Cleaned aux_ref_audios for {_vid}: {len(_aux)} -> {len(_valid_aux)}", flush=True) for _vid in _removed: del _voices_data[_vid] if _removed or _repaired: _voices_file.write_text(json.dumps(_voices_data, ensure_ascii=False, indent=2), encoding='utf-8') print(f"[Patch v10] Voices.json cleaned: removed={len(_removed)}, repairable={len(_repaired)}", flush=True) else: print(f"[Patch v10] All {len(_voices_data)} voices pass integrity check", flush=True) except Exception as _e: print(f"[Patch v10] Integrity check failed: {_e}", flush=True) else: print("[Patch v10] No voices.json found, skipping", flush=True) print("[Patch v11.6] Digital human training: no auto voice cloning (use voice library instead)", flush=True) print("[Patch v8.5] Starting server...", flush=True) print("[Patch v8.5] Starting server...", flush=True) # ============================================================ # Now run the server # ============================================================ if __name__ == '__main__': engine = module.DigitalHumanEngineV2() server = module.ThreadingHTTPServer(('0.0.0.0', module.PORT), module.DigitalHumanV2Handler) server.engine = engine module._global_engine = engine print(f"[服务] 正方元智能广告工作台 - 数字人 v2 (v8.4)", flush=True) print(f"[服务] 地址: http://{socket.gethostbyname(socket.gethostname())}:{module.PORT}", flush=True) try: server.serve_forever() except KeyboardInterrupt: print("[服务] 正在关闭...", flush=True) engine.shutdown() server.shutdown()