let5see/AI-Video
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化本地调用edge-tts的证书逻辑

Browse Source
This commit is contained in:
puke
2025-11-12 20:57:29 +08:00
parent 0be64009a6
commit 4785d1ddb5
1 changed files with 21 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
pixelle_video/utils/tts_util.py
View File

@@ -20,13 +20,14 @@ Currently, TTS service uses ComfyUI workflows only.
import asyncio import asyncio
import ssl import ssl
import random import random
import certifi
import edge_tts as edge_tts_sdk import edge_tts as edge_tts_sdk
from loguru import logger from loguru import logger
from aiohttp import WSServerHandshakeError, ClientResponseError from aiohttp import WSServerHandshakeError, ClientResponseError
# Global flag for SSL verification (set to False for development only) # Use certifi bundle for SSL verification instead of disabling it
_SSL_VERIFY_ENABLED = False _USE_CERTIFI_SSL = True
# Retry configuration for Edge TTS (to handle 401 errors) # Retry configuration for Edge TTS (to handle 401 errors)
_RETRY_COUNT = 5 # Default retry count (increased from 3 to 5) _RETRY_COUNT = 5 # Default retry count (increased from 3 to 5)
@@ -117,20 +118,18 @@ async def edge_tts(
logger.info(f"🔄 Retrying Edge TTS (attempt {attempt + 1}/{retry_count + 1}) after {retry_delay:.2f}s delay...") logger.info(f"🔄 Retrying Edge TTS (attempt {attempt + 1}/{retry_count + 1}) after {retry_delay:.2f}s delay...")
await asyncio.sleep(retry_delay) await asyncio.sleep(retry_delay)
# Monkey patch ssl.create_default_context if SSL verification is disabled # Use certifi SSL context for proper certificate verification
if not _SSL_VERIFY_ENABLED: if _USE_CERTIFI_SSL:
if attempt == 0: # Only log warning once if attempt == 0: # Only log info once
logger.warning("SSL verification is disabled for development. This is NOT recommended for production!") logger.debug("Using certifi SSL certificates for secure Edge TTS connection")
original_create_default_context = ssl.create_default_context original_create_default_context = ssl.create_default_context
def create_unverified_context(*args, **kwargs): def create_certifi_context(*args, **kwargs):
ctx = original_create_default_context(*args, **kwargs) # Build SSL context that uses certifi bundle (resolves Windows / missing CA issues)
ctx.check_hostname = False return original_create_default_context(cafile=certifi.where())
ctx.verify_mode = ssl.CERT_NONE
return ctx
# Temporarily replace the function # Temporarily replace the function
ssl.create_default_context = create_unverified_context ssl.create_default_context = create_certifi_context
try: try:
# Create communicate instance # Create communicate instance
@@ -190,7 +189,7 @@ async def edge_tts(
finally: finally:
# Restore original function if we patched it # Restore original function if we patched it
if not _SSL_VERIFY_ENABLED: if _USE_CERTIFI_SSL:
ssl.create_default_context = original_create_default_context ssl.create_default_context = original_create_default_context
# Should not reach here, but just in case # Should not reach here, but just in case
@@ -275,19 +274,17 @@ async def list_voices(locale: str = None, retry_count: int = _RETRY_COUNT, retry
logger.info(f"🔄 Retrying list voices (attempt {attempt + 1}/{retry_count + 1}) after {retry_delay:.2f}s delay...") logger.info(f"🔄 Retrying list voices (attempt {attempt + 1}/{retry_count + 1}) after {retry_delay:.2f}s delay...")
await asyncio.sleep(retry_delay) await asyncio.sleep(retry_delay)
# Monkey patch SSL if verification is disabled # Use certifi SSL context for proper certificate verification
if not _SSL_VERIFY_ENABLED: if _USE_CERTIFI_SSL:
if attempt == 0: # Only log warning once if attempt == 0: # Only log info once
logger.warning("SSL verification is disabled for development. This is NOT recommended for production!") logger.debug("Using certifi SSL certificates for secure Edge TTS connection")
original_create_default_context = ssl.create_default_context original_create_default_context = ssl.create_default_context
def create_unverified_context(*args, **kwargs): def create_certifi_context(*args, **kwargs):
ctx = original_create_default_context(*args, **kwargs) # Build SSL context that uses certifi bundle (resolves Windows / missing CA issues)
ctx.check_hostname = False return original_create_default_context(cafile=certifi.where())
ctx.verify_mode = ssl.CERT_NONE
return ctx
ssl.create_default_context = create_unverified_context ssl.create_default_context = create_certifi_context
try: try:
# Get all voices # Get all voices
@@ -331,7 +328,7 @@ async def list_voices(locale: str = None, retry_count: int = _RETRY_COUNT, retry
finally: finally:
# Restore original function if we patched it # Restore original function if we patched it
if not _SSL_VERIFY_ENABLED: if _USE_CERTIFI_SSL:
ssl.create_default_context = original_create_default_context ssl.create_default_context = original_create_default_context
# Should not reach here, but just in case # Should not reach here, but just in case