diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -1,33 +1,30 @@ -# secure_tts_proxy_app_final.py +# secure_tts_app_huggingface.py # -*- coding: utf-8 -*- import sys, os, re, time, random, json, base64, hashlib import requests -import socks -import socket -import uuid -import platform import warnings from typing import List, Dict, Optional, Tuple import concurrent.futures import urllib3 -from PySide6.QtWidgets import ( - QApplication, QWidget, QLabel, QLineEdit, QTextEdit, QPushButton, - QFileDialog, QVBoxLayout, QHBoxLayout, QSlider, QComboBox, QCheckBox, - QMessageBox, QProgressBar, QGroupBox, QTabWidget, QFrame, QSplitter, - QMenuBar, QMenu, QStatusBar, QDialog, QDialogButtonBox, QFormLayout, - QProgressDialog, QInputDialog, QListWidget, QListWidgetItem -) -from PySide6.QtCore import Qt, QThread, Signal, QSettings, QTimer, QUrl -from PySide6.QtGui import QFont, QPalette, QColor, QAction, QIcon, QDesktopServices -from pydub import AudioSegment -import natsort +import gradio as gr import tempfile import shutil +from pathlib import Path +import asyncio +import aiohttp +from dataclasses import dataclass +import threading +import queue +import logging warnings.filterwarnings("ignore") urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) +# ==================== CẤU HÌNH LOGGING ==================== +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + # ==================== HÀM ẨN THÔNG TIN MÁY TÍNH ==================== class PrivacyProtector: """Lớp bảo vệ quyền riêng tư và ẩn thông tin máy tính""" @@ -63,407 +60,9 @@ class PrivacyProtector: "DNT": "1", "Sec-GPC": "1" } - - @staticmethod - def create_secure_session(proxy_manager=None, use_proxy=False) -> requests.Session: - """Tạo session requests an toàn với proxy""" - session = requests.Session() - - # Cấu hình timeout - session.timeout = 30 - - # Xóa thông tin mặc định - session.headers.clear() - - # Thêm headers bảo mật - session.headers.update({ - "Accept-Encoding": "gzip, deflate", - "Connection": "keep-alive", - "Upgrade-Insecure-Requests": "1", - "User-Agent": PrivacyProtector.get_random_user_agent() - }) - - # Thêm proxy nếu được yêu cầu - if use_proxy and proxy_manager: - proxy_url = proxy_manager.get_random_proxy() - if proxy_url: - proxies = proxy_manager.get_proxy_for_requests(proxy_url) - if proxies: - session.proxies.update(proxies) - - return session - -# ==================== PROXY MANAGER NÂNG CAO ==================== -class AdvancedProxyManager: - """Quản lý proxy nâng cao với tính năng tự động xoay và chọn lọc""" - - def __init__(self): - self.proxy_list = [] - self.alive_proxies = [] - self.current_proxy = None - self.test_url = "https://httpbin.org/ip" - self.timeout = 10 - self.auto_check = True - self.proxy_scores = {} - self.proxy_fail_count = {} - self.max_fails = 3 - self.last_check_time = {} - - def add_proxy(self, proxy: str): - """Thêm proxy vào danh sách""" - if proxy and proxy not in self.proxy_list: - proxy = self.standardize_proxy(proxy) - self.proxy_list.append(proxy) - self.proxy_scores[proxy] = 100 - self.proxy_fail_count[proxy] = 0 - print(f"[ProxyManager] Added proxy: {proxy}") - - def standardize_proxy(self, proxy: str) -> str: - """Chuẩn hóa định dạng proxy""" - proxy = proxy.strip() - - if not re.match(r'^[a-zA-Z]+://', proxy): - if ':' in proxy: - # Kiểm tra nếu có authentication - if '@' in proxy: - # Định dạng user:pass@host:port - auth_part, host_part = proxy.split('@', 1) - if ':' in auth_part and ':' in host_part: - proxy = f"http://{proxy}" - else: - # Giả sử là host:port - proxy = f"http://{proxy}" - else: - # Định dạng host:port - proxy = f"http://{proxy}" - - return proxy - - def parse_proxy_url(self, proxy_url: str) -> Tuple[str, Optional[str], Optional[str], str, int]: - """Phân tích proxy URL để lấy thông tin""" - try: - if "://" not in proxy_url: - proxy_url = "http://" + proxy_url - - proxy_scheme, rest = proxy_url.split("://", 1) - username = password = host = port = None - - # Extract authentication if exists - if "@" in rest: - auth, rest = rest.split("@", 1) - if ":" in auth: - username, password = auth.split(":", 1) - else: - username = auth - - # Extract host and port - if ":" in rest: - host, port_str = rest.split(":", 1) - try: - port = int(port_str) - except ValueError: - port = 8080 if proxy_scheme in ["http", "https"] else 1080 - else: - host = rest - port = 8080 if proxy_scheme in ["http", "https"] else 1080 - - return proxy_scheme, username, password, host, port - except: - return "http", None, None, "", 8080 - - def check_proxy(self, proxy_url: str) -> Tuple[bool, str, float]: - """Kiểm tra proxy có hoạt động không và trả về response time""" - start_time = time.time() - try: - proxy_scheme, username, password, host, port = self.parse_proxy_url(proxy_url) - - if not host or port == 0: - return False, "Invalid host or port", 0 - - # Test HTTP/HTTPS proxy - if proxy_scheme in ["http", "https"]: - proxies = {"http": proxy_url, "https": proxy_url} - - try: - response = requests.get( - self.test_url, - proxies=proxies, - timeout=self.timeout, - verify=False - ) - - if response.status_code == 200: - response_time = time.time() - start_time - return True, f"Working ({response_time:.2f}s)", response_time - else: - return False, f"HTTP {response.status_code}", 0 - - except requests.exceptions.Timeout: - return False, "Timeout", 0 - except requests.exceptions.ConnectionError: - return False, "Connection failed", 0 - except Exception as e: - return False, f"Error: {str(e)}", 0 - - # Test SOCKS proxy - elif proxy_scheme in ["socks4", "socks5"]: - try: - import socks - - socks.set_default_proxy( - socks.SOCKS5 if proxy_scheme == "socks5" else socks.SOCKS4, - host, port, - username=username, - password=password - ) - - original_socket = socket.socket - socket.socket = socks.socksocket - - try: - test_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - test_socket.settimeout(self.timeout) - test_socket.connect(("httpbin.org", 80)) - test_socket.close() - - socket.socket = original_socket - response_time = time.time() - start_time - return True, f"SOCKS Working ({response_time:.2f}s)", response_time - except: - socket.socket = original_socket - return False, "SOCKS Connection failed", 0 - - except ImportError: - return False, "PySocks not installed", 0 - except Exception as e: - return False, f"SOCKS Error: {str(e)}", 0 - - return False, f"Unsupported type: {proxy_scheme}", 0 - - except Exception as e: - return False, f"Check Error: {str(e)}", 0 - - def check_all_proxies(self, max_workers: int = 20) -> List[str]: - """Kiểm tra tất cả proxy trong danh sách""" - if not self.proxy_list: - print("[ProxyManager] No proxies to check") - return [] - - print(f"[ProxyManager] Checking {len(self.proxy_list)} proxies...") - self.alive_proxies = [] - checked_proxies = [] - - with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: - future_to_proxy = { - executor.submit(self.check_proxy, proxy): proxy - for proxy in self.proxy_list - } - - for future in concurrent.futures.as_completed(future_to_proxy): - proxy = future_to_proxy[future] - try: - is_alive, message, response_time = future.result() - - if is_alive: - self.alive_proxies.append(proxy) - self.proxy_scores[proxy] = min(100, self.proxy_scores.get(proxy, 100) + 10) - self.proxy_fail_count[proxy] = 0 - print(f"✅ {proxy} - {message}") - else: - self.proxy_scores[proxy] = max(0, self.proxy_scores.get(proxy, 100) - 20) - self.proxy_fail_count[proxy] = self.proxy_fail_count.get(proxy, 0) + 1 - print(f"❌ {proxy} - {message}") - - checked_proxies.append(proxy) - self.last_check_time[proxy] = time.time() - - except Exception as e: - print(f"❌ {proxy} - Check error: {str(e)}") - - print(f"[ProxyManager] Found {len(self.alive_proxies)} alive proxies") - - # Xóa proxy thất bại quá nhiều lần - self.remove_failed_proxies() - - return self.alive_proxies - - def remove_failed_proxies(self): - """Xóa proxy thất bại quá nhiều lần""" - to_remove = [] - for proxy, fails in self.proxy_fail_count.items(): - if fails >= self.max_fails: - to_remove.append(proxy) - - for proxy in to_remove: - if proxy in self.proxy_list: - self.proxy_list.remove(proxy) - if proxy in self.alive_proxies: - self.alive_proxies.remove(proxy) - self.proxy_scores.pop(proxy, None) - self.proxy_fail_count.pop(proxy, None) - print(f"[ProxyManager] Removed failed proxy: {proxy}") - - def get_best_proxy(self) -> Optional[str]: - """Lấy proxy tốt nhất dựa trên điểm số""" - if not self.alive_proxies: - if self.auto_check: - print("[ProxyManager] No alive proxies, running auto-check...") - self.check_all_proxies(max_workers=10) - - if not self.alive_proxies: - print("[ProxyManager] No alive proxies available") - return None - - # Sắp xếp proxy theo điểm số - scored_proxies = [] - for proxy in self.alive_proxies: - score = self.proxy_scores.get(proxy, 50) - # Ưu tiên proxy chưa được dùng gần đây - last_used = self.last_check_time.get(proxy, 0) - time_since_last = time.time() - last_used - adjusted_score = score + (time_since_last / 60) # Thêm điểm cho proxy lâu chưa dùng - - scored_proxies.append((proxy, adjusted_score)) - - # Chọn proxy có điểm cao nhất - scored_proxies.sort(key=lambda x: x[1], reverse=True) - best_proxy = scored_proxies[0][0] - - self.current_proxy = best_proxy - print(f"[ProxyManager] Selected best proxy: {best_proxy} (score: {scored_proxies[0][1]:.1f})") - return best_proxy - - def get_proxy_for_requests(self, proxy_url: str = None) -> Optional[Dict[str, str]]: - """Chuẩn bị proxy cho requests""" - if not proxy_url: - proxy_url = self.get_best_proxy() - if not proxy_url: - return None - - try: - proxy_scheme, username, password, host, port = self.parse_proxy_url(proxy_url) - - if proxy_scheme in ["http", "https"]: - if username and password: - proxy_str = f"{proxy_scheme}://{username}:{password}@{host}:{port}" - else: - proxy_str = f"{proxy_scheme}://{host}:{port}" - - return {"http": proxy_str, "https": proxy_str} - - elif proxy_scheme in ["socks4", "socks5"]: - import socks - socks.set_default_proxy( - socks.SOCKS5 if proxy_scheme == "socks5" else socks.SOCKS4, - host, port, - True, username, password - ) - socket.socket = socks.socksocket - return None - - return None - - except Exception as e: - print(f"[ProxyManager] Error preparing proxy: {e}") - return None - - def rotate_proxy(self): - """Xoay sang proxy khác""" - old_proxy = self.current_proxy - new_proxy = self.get_best_proxy() - - if new_proxy and new_proxy != old_proxy: - print(f"[ProxyManager] Rotated proxy: {old_proxy} -> {new_proxy}") - self.current_proxy = new_proxy - elif new_proxy: - print(f"[ProxyManager] Already using best proxy: {new_proxy}") - else: - print("[ProxyManager] No proxy available for rotation") - - return self.current_proxy - - def mark_proxy_failed(self, proxy_url: str): - """Đánh dấu proxy thất bại""" - if proxy_url in self.proxy_scores: - self.proxy_scores[proxy_url] = max(0, self.proxy_scores[proxy_url] - 30) - self.proxy_fail_count[proxy_url] = self.proxy_fail_count.get(proxy_url, 0) + 1 - print(f"[ProxyManager] Marked proxy as failed: {proxy_url}") - - # Xóa khỏi alive proxies nếu có - if proxy_url in self.alive_proxies: - self.alive_proxies.remove(proxy_url) - - # Xoay proxy nếu đang dùng proxy này - if self.current_proxy == proxy_url: - self.rotate_proxy() - - def mark_proxy_success(self, proxy_url: str, response_time: float = 0): - """Đánh dấu proxy thành công""" - if proxy_url in self.proxy_scores: - bonus = 20 if response_time < 2 else 10 if response_time < 5 else 5 - self.proxy_scores[proxy_url] = min(100, self.proxy_scores[proxy_url] + bonus) - self.proxy_fail_count[proxy_url] = 0 - - # Thêm vào alive proxies nếu chưa có - if proxy_url not in self.alive_proxies: - self.alive_proxies.append(proxy_url) - - def load_proxies_from_file(self, file_path: str = "proxies.txt"): - """Tải proxy từ file""" - if not os.path.exists(file_path): - print(f"[ProxyManager] File not found: {file_path}") - return False - - try: - with open(file_path, 'r', encoding='utf-8') as f: - proxies = [line.strip() for line in f.readlines() if line.strip()] - - for proxy in proxies: - self.add_proxy(proxy) - - print(f"[ProxyManager] Loaded {len(proxies)} proxies from {file_path}") - return True - except Exception as e: - print(f"[ProxyManager] Error loading proxies: {e}") - return False - - def save_proxies_to_file(self, file_path: str = "proxies.txt"): - """Lưu proxy vào file""" - try: - with open(file_path, 'w', encoding='utf-8') as f: - for proxy in self.proxy_list: - f.write(f"{proxy}\n") - print(f"[ProxyManager] Saved {len(self.proxy_list)} proxies to {file_path}") - return True - except Exception as e: - print(f"[ProxyManager] Error saving proxies: {e}") - return False - - def clear_proxies(self): - """Xóa toàn bộ proxy""" - self.proxy_list = [] - self.alive_proxies = [] - self.current_proxy = None - self.proxy_scores = {} - self.proxy_fail_count = {} - print("[ProxyManager] Cleared all proxies") - - def get_stats(self) -> Dict: - """Lấy thống kê proxy""" - total = len(self.proxy_list) - alive = len(self.alive_proxies) - - return { - "total": total, - "alive": alive, - "dead": total - alive, - "alive_percentage": (alive / total * 100) if total > 0 else 0, - "current": self.current_proxy, - "avg_score": sum(self.proxy_scores.values()) / len(self.proxy_scores) if self.proxy_scores else 0 - } # ==================== HÀM KIỂM TRA API KEY NÂNG CAO ==================== -def check_api_key_advanced(api_key: str, proxy_manager=None, use_proxy=False, timeout=15) -> Dict: +def check_api_key_advanced(api_key: str, timeout=15) -> Dict: """Kiểm tra API key với retry mechanism và xử lý lỗi chi tiết""" max_retries = 2 for attempt in range(max_retries + 1): @@ -480,14 +79,6 @@ def check_api_key_advanced(api_key: str, proxy_manager=None, use_proxy=False, ti "Accept": "application/json" } - # Cấu hình proxy - if use_proxy and proxy_manager: - proxy_url = proxy_manager.get_best_proxy() - if proxy_url: - proxies = proxy_manager.get_proxy_for_requests(proxy_url) - if proxies: - session.proxies.update(proxies) - # Độ trễ ngẫu nhiên time.sleep(random.uniform(1, 3)) @@ -582,8 +173,8 @@ def check_api_key_advanced(api_key: str, proxy_manager=None, use_proxy=False, ti # ==================== HÀM TẠO VOICE NÂNG CAO ==================== def generate_voice_advanced(text: str, api_key: str, voice_id: str, model_id: str, stability=0.7, similarity=0.8, style=0.0, speed=0.75, - speaker_boost=True, proxy_manager=None, use_proxy=False) -> Optional[bytes]: - """Tạo giọng nói với proxy và retry mechanism nâng cao""" + speaker_boost=True) -> Optional[bytes]: + """Tạo giọng nói với retry mechanism nâng cao""" max_retries = 3 url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}" @@ -614,15 +205,6 @@ def generate_voice_advanced(text: str, api_key: str, voice_id: str, model_id: st "Accept": "audio/mpeg" } - current_proxy = None - # Cấu hình proxy - if use_proxy and proxy_manager: - current_proxy = proxy_manager.get_best_proxy() - if current_proxy: - proxies = proxy_manager.get_proxy_for_requests(current_proxy) - if proxies: - session.proxies.update(proxies) - # Độ trễ trước khi request delay = random.uniform(2, 5) if attempt == 0 else random.uniform(10, 20) time.sleep(delay) @@ -642,16 +224,10 @@ def generate_voice_advanced(text: str, api_key: str, voice_id: str, model_id: st # Kiểm tra kích thước file if len(audio_content) > 2048: # File hợp lệ phải > 2KB - # Đánh dấu proxy thành công - if current_proxy and proxy_manager: - proxy_manager.mark_proxy_success(current_proxy) - return audio_content else: # File quá nhỏ, có thể bị chặn - print(f"[VoiceGen] Audio too small: {len(audio_content)} bytes") - if current_proxy and proxy_manager: - proxy_manager.mark_proxy_failed(current_proxy) + logger.warning(f"Audio too small: {len(audio_content)} bytes") if attempt < max_retries: time.sleep(10) @@ -660,82 +236,46 @@ def generate_voice_advanced(text: str, api_key: str, voice_id: str, model_id: st elif response.status_code == 429: # Rate limit wait_time = 30 + (attempt * 10) - print(f"[VoiceGen] Rate limited, waiting {wait_time}s...") + logger.warning(f"Rate limited, waiting {wait_time}s...") time.sleep(wait_time) continue elif response.status_code == 403: # Bị chặn - print(f"[VoiceGen] Blocked (403)") - if current_proxy and proxy_manager: - proxy_manager.mark_proxy_failed(current_proxy) + logger.warning(f"Blocked (403)") if attempt < max_retries: time.sleep(15) continue else: - print(f"[VoiceGen] Error {response.status_code}: {response.text[:100]}") + logger.warning(f"Error {response.status_code}: {response.text[:100]}") if attempt < max_retries: time.sleep(10) continue except requests.exceptions.Timeout: - print(f"[VoiceGen] Timeout on attempt {attempt + 1}") + logger.warning(f"Timeout on attempt {attempt + 1}") if attempt < max_retries: time.sleep(10) continue except requests.exceptions.ConnectionError: - print(f"[VoiceGen] Connection error") + logger.warning(f"Connection error") if attempt < max_retries: time.sleep(10) continue except Exception as e: - print(f"[VoiceGen] Error: {str(e)}") + logger.error(f"Error: {str(e)}") if attempt < max_retries: time.sleep(10) continue - print(f"[VoiceGen] All attempts failed for text: {text[:50]}...") + logger.error(f"All attempts failed for text: {text[:50]}...") return None -# ==================== HÀM XỬ LÝ ÂM THANH ==================== -def merge_audio_files(input_folder, format, output_filename, silence_ms=300): - """Merge các file audio và kiểm tra kích thước""" - try: - audio_files = [f for f in os.listdir(input_folder) if f.endswith(f".{format.lower()}")] - if not audio_files: - return None - - audio_files = natsort.natsorted(audio_files) - combined = AudioSegment.from_file(os.path.join(input_folder, audio_files[0]), format=format.lower()) - - for audio_file in audio_files[1:]: - audio = AudioSegment.from_file(os.path.join(input_folder, audio_file), format=format.lower()) - combined += AudioSegment.silent(duration=silence_ms) - combined += audio - - output_path = f"{output_filename}.{format.lower()}" - combined.export(output_path, format=format.lower()) - - # KIỂM TRA FILE 1KB VÀ TẠO LẠI - if os.path.exists(output_path) and os.path.getsize(output_path) <= 1024: - print(f"⚠️ File chỉ có {os.path.getsize(output_path)} bytes, tạo lại...") - os.remove(output_path) - - # Tạo lại với chất lượng cao hơn - combined.export(output_path, format=format.lower(), - parameters=["-q:a", "0"]) # Highest quality - - print(f"✅ Đã tạo lại file, kích thước mới: {os.path.getsize(output_path)} bytes") - - return output_path - except Exception as e: - print(f"❌ Lỗi merge audio: {str(e)}") - return None - +# ==================== XỬ LÝ VĂN BẢN ==================== def parse_text_blocks(raw_text, max_length=250): """Phân chia văn bản thành các block""" blocks = [] @@ -759,1604 +299,421 @@ def estimate_credit(text): """Ước tính credit cần thiết""" return len(text) + 50 -def create_srt(voice_dir, texts, silence_ms=300): - """Tạo file phụ đề SRT""" - files_audio = natsort.natsorted([f for f in os.listdir(voice_dir) if f.startswith("voice_")]) - current_time = 0 - srt_lines = [] - - for idx, (fname, text) in enumerate(zip(files_audio, texts), start=1): - try: - audio = AudioSegment.from_file(os.path.join(voice_dir, fname)) - start = current_time - end = start + len(audio) - - # Format thời gian - def ms_to_time(ms): - h = ms // 3600000 - m = (ms % 3600000) // 60000 - s = (ms % 60000) // 1000 - ms = ms % 1000 - return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" - - srt_lines.append(str(idx)) - srt_lines.append(f"{ms_to_time(start)} --> {ms_to_time(end)}") - srt_lines.append(text.strip()) - srt_lines.append("") - - current_time = end + silence_ms - except: - continue - - if srt_lines: - srt_path = os.path.join(voice_dir, "output_full.srt") - with open(srt_path, "w", encoding="utf-8") as f: - f.write("\n".join(srt_lines)) - return srt_path - - return None +# ==================== QUẢN LÝ TÁC VỤ BẤT ĐỒNG BỘ ==================== +@dataclass +class TTSRequest: + text: str + api_key: str + voice_id: str + model_id: str + settings: dict + file_index: int + output_dir: str + format: str -# ==================== THREAD TẠO VOICE ==================== -class VoiceGenerationThread(QThread): - """Thread tạo giọng nói với proxy và retry mechanism""" - - progress_signal = Signal(str) - progress_value = Signal(int) - progress_max = Signal(int) - finished_signal = Signal(bool, str) - api_key_checked = Signal(int, bool, dict) - - def __init__(self, api_keys, voice_id, texts, settings, fmt, model_id, output_dir, - proxy_manager=None, use_proxy=False): - super().__init__() - self.api_keys = api_keys - self.voice_id = voice_id - self.texts = texts - self.settings = settings - self.fmt = fmt - self.model_id = model_id - self.output_dir = output_dir - self.proxy_manager = proxy_manager - self.use_proxy = use_proxy - self.should_stop = False - +class TTSWorker: + def __init__(self): + self.task_queue = queue.Queue() + self.results = {} + self.is_running = False + self.worker_thread = None + + def start(self): + self.is_running = True + self.worker_thread = threading.Thread(target=self._process_queue) + self.worker_thread.daemon = True + self.worker_thread.start() + def stop(self): - """Dừng thread""" - self.should_stop = True - - def run(self): - try: - # Tạo thư mục output - os.makedirs(self.output_dir, exist_ok=True) - - # Xóa file cũ - for f in os.listdir(self.output_dir): - if f.startswith("voice_") or f.startswith("output_"): - try: - os.remove(os.path.join(self.output_dir, f)) - except: - pass + self.is_running = False + if self.worker_thread: + self.worker_thread.join(timeout=5) - # Bước 1: Kiểm tra API keys - self.progress_signal.emit("🔍 Kiểm tra API keys...") - valid_keys = [] - - for i, key in enumerate(self.api_keys, 1): - if self.should_stop: - self.finished_signal.emit(False, "Đã dừng bởi người dùng") - return + def add_task(self, request: TTSRequest): + self.task_queue.put(request) + + def _process_queue(self): + while self.is_running: + try: + request = self.task_queue.get(timeout=1) + try: + audio = generate_voice_advanced( + request.text, + request.api_key, + request.voice_id, + request.model_id, + **request.settings + ) + + if audio: + filename = os.path.join(request.output_dir, f"voice_{request.file_index:03d}.{request.format}") + with open(filename, "wb") as f: + f.write(audio) + self.results[request.file_index] = filename + else: + self.results[request.file_index] = None + + except Exception as e: + logger.error(f"Error processing task {request.file_index}: {str(e)}") + self.results[request.file_index] = None + + finally: + self.task_queue.task_done() + + except queue.Empty: + continue + + def wait_completion(self, timeout=None): + self.task_queue.join() + return self.results + +# ==================== GIAO DIỆN GRADIO ==================== +class SecureTTSApp: + def __init__(self): + self.output_dir = tempfile.mkdtemp(prefix="tts_output_") + self.tts_worker = TTSWorker() + self.tts_worker.start() + + def cleanup(self): + """Dọn dẹp thư mục tạm""" + self.tts_worker.stop() + try: + shutil.rmtree(self.output_dir) + except: + pass + + def create_interface(self): + with gr.Blocks(title="🎤 ElevenLabs TTS Pro - Secure Edition", theme=gr.themes.Soft()) as demo: + gr.Markdown("# 🎤 ElevenLabs TTS Pro - Secure Edition") + gr.Markdown("Chuyển văn bản thành giọng nói với bảo mật nâng cao") + + with gr.Tabs(): + # Tab 1: Cấu hình chính + with gr.Tab("⚙️ Cấu hình"): + with gr.Row(): + with gr.Column(scale=2): + api_keys = gr.Textbox( + label="🔑 API Keys", + placeholder="Nhập API keys (mỗi key một dòng)\nsk_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", + lines=4, + info="Có thể nhập nhiều API keys để tự động chuyển đổi" + ) + + voice_id = gr.Textbox( + label="🎤 Voice ID", + placeholder="21m00Tcm4TlvDq8ikWAM", + value="21m00Tcm4TlvDq8ikWAM" + ) + + with gr.Row(): + model_id = gr.Dropdown( + label="🤖 Model", + choices=["eleven_multilingual_v2", "eleven_turbo_v2", "eleven_monolingual_v1"], + value="eleven_multilingual_v2" + ) + + output_format = gr.Dropdown( + label="📁 Format", + choices=["mp3", "wav"], + value="mp3" + ) + + with gr.Column(scale=1): + gr.Markdown("### Voice Parameters") + stability = gr.Slider( + label="Stability", + minimum=0.0, + maximum=1.0, + value=0.7, + step=0.01 + ) + + similarity = gr.Slider( + label="Similarity Boost", + minimum=0.0, + maximum=1.0, + value=0.8, + step=0.01 + ) + + speed = gr.Slider( + label="Speed", + minimum=0.5, + maximum=1.5, + value=1.0, + step=0.05 + ) + + speaker_boost = gr.Checkbox( + label="Speaker Boost", + value=True + ) + + # Tab 2: Nhập văn bản + with gr.Tab("📝 Văn bản"): + text_input = gr.Textbox( + label="Văn bản cần chuyển thành giọng nói", + placeholder="Nhập văn bản tại đây...", + lines=10 + ) + + with gr.Row(): + text_stats = gr.Markdown("**Thống kê:** 0 ký tự | 0 đoạn | Ước tính: 0 credits") + + with gr.Row(): + gr.Examples( + examples=[ + ["Xin chào! Đây là ví dụ về giọng nói được tạo bởi ElevenLabs."], + ["Tôi yêu công nghệ và đang tạo ra những sản phẩm tuyệt vời với trí tuệ nhân tạo."], + ["Đây là một đoạn văn bản dài hơn để kiểm tra khả năng xử lý của hệ thống."] + ], + inputs=text_input, + label="Ví dụ" + ) - self.progress_signal.emit(f"Kiểm tra API key {i}/{len(self.api_keys)}...") + # Tab 3: Kiểm tra API + with gr.Tab("🔍 Kiểm tra API"): + check_api_btn = gr.Button("Kiểm tra API Keys", variant="primary") + api_status = gr.JSON(label="Kết quả kiểm tra") + + @check_api_btn.click(inputs=[api_keys], outputs=[api_status]) + def check_api(keys): + if not keys.strip(): + return {"error": "Vui lòng nhập API keys"} + + keys_list = [k.strip() for k in keys.splitlines() if k.strip()] + results = [] + + for i, key in enumerate(keys_list[:5]): # Giới hạn 5 keys + info = check_api_key_advanced(key) + results.append({ + "key": f"Key {i+1}", + "valid": info.get("valid", False), + "remaining": info.get("remaining", 0), + "error": info.get("error", ""), + "tier": info.get("tier", "unknown") + }) + + return {"total_keys": len(keys_list), "checked": len(results), "results": results} - # Độ trễ giữa các lần check - time.sleep(random.uniform(1, 3)) + # Tab 4: Output + with gr.Tab("📊 Output"): + with gr.Row(): + with gr.Column(): + progress_bar = gr.Progress(label="Tiến trình") + status_output = gr.Textbox(label="Trạng thái", lines=3) + + start_btn = gr.Button( + "🚀 Bắt đầu tạo giọng nói", + variant="primary", + scale=1 + ) + + stop_btn = gr.Button("⏹️ Dừng", variant="stop", scale=1) + + with gr.Column(): + audio_output = gr.Audio(label="Âm thanh đã tạo") + download_btn = gr.File(label="Tải file") + + with gr.Row(): + gr.Markdown("### Logs") + log_output = gr.Textbox(label="Logs", lines=10, interactive=False) + + # Xử lý sự kiện nhập văn bản + def update_text_stats(text): + if not text.strip(): + return "**Thống kê:** 0 ký tự | 0 đoạn | Ước tính: 0 credits" - # Kiểm tra API key - info = check_api_key_advanced(key, self.proxy_manager, self.use_proxy) + char_count = len(text) + blocks = parse_text_blocks(text) + block_count = len(blocks) + credit_estimate = char_count + (block_count * 50) - if info.get("valid") and info["remaining"] > 500: - valid_keys.append({ - "key": key, - "remaining": info["remaining"], - "tier": info.get("tier", "free"), - "fresh": info.get("is_fresh", False) - }) - self.api_key_checked.emit(i, True, info) - self.progress_signal.emit(f"✓ Key {i}: {info['remaining']:,} chars") - else: - self.api_key_checked.emit(i, False, info) - self.progress_signal.emit(f"✗ Key {i}: {info.get('error', 'Invalid')}") - - if not valid_keys: - self.progress_signal.emit("❌ Không có API key hợp lệ") - self.finished_signal.emit(False, "Không có API key hợp lệ") - return - - # Sắp xếp API keys: fresh trước, nhiều credits trước - valid_keys.sort(key=lambda x: (x["fresh"], x["remaining"]), reverse=True) + return f"**Thống kê:** {char_count:,} ký tự | {block_count} đoạn | Ước tính: {credit_estimate:,} credits" - self.progress_signal.emit(f"✅ Đã tìm thấy {len(valid_keys)} API keys hợp lệ") - self.progress_max.emit(len(self.texts)) + text_input.change( + fn=update_text_stats, + inputs=[text_input], + outputs=[text_stats] + ) - # Bước 2: Tạo âm thanh cho từng block - for i, text in enumerate(self.texts, 1): - if self.should_stop: - self.finished_signal.emit(False, "Đã dừng bởi người dùng") + # Xử lý nút bắt đầu + def start_tts_generation(api_keys_text, voice_id_text, model, format, + stability_val, similarity_val, speed_val, + speaker_boost_val, text): + # Kiểm tra đầu vào + if not api_keys_text.strip(): + yield "❌ Vui lòng nhập API keys", None, None, "" + return + + if not voice_id_text.strip(): + yield "❌ Vui lòng nhập Voice ID", None, None, "" + return + + if not text.strip(): + yield "❌ Vui lòng nhập văn bản", None, None, "" return - success = False - attempts = 0 - max_attempts = min(len(valid_keys) * 2, 20) + # Parse API keys + api_keys_list = [k.strip() for k in api_keys_text.splitlines() if k.strip()] + + # Parse văn bản + text_blocks = parse_text_blocks(text) + + # Cấu hình + settings = { + "stability": stability_val, + "similarity": similarity_val, + "style": 0.0, + "speed": speed_val, + "speaker_boost": speaker_boost_val + } + + # Tạo thư mục output + output_dir = tempfile.mkdtemp(dir=self.output_dir) + + logs = [] + logs.append("="*60) + logs.append("🚀 BẮT ĐẦU TẠO GIỌNG NÓI") + logs.append(f"🔑 API keys: {len(api_keys_list)}") + logs.append(f"🎤 Voice ID: {voice_id_text}") + logs.append(f"🤖 Model: {model}") + logs.append(f"📁 Format: {format}") + logs.append(f"📝 Số đoạn văn bản: {len(text_blocks)}") + logs.append("="*60) + + yield "\n".join(logs), 0, None, "" - while not success and attempts < max_attempts: - attempts += 1 + # Kiểm tra API keys + valid_keys = [] + for i, key in enumerate(api_keys_list[:3], 1): # Giới hạn 3 keys + info = check_api_key_advanced(key) + if info.get("valid") and info.get("remaining", 0) > 100: + valid_keys.append({ + "key": key, + "remaining": info.get("remaining", 0) + }) + logs.append(f"✅ Key {i}: {info['remaining']:,} ký tự còn lại") + else: + logs.append(f"❌ Key {i}: {info.get('error', 'Invalid')}") + yield "\n".join(logs), 0, None, "" + time.sleep(1) + + if not valid_keys: + logs.append("❌ Không có API key hợp lệ") + yield "\n".join(logs), 0, None, "" + return + + logs.append(f"✅ Đã tìm thấy {len(valid_keys)} API keys hợp lệ") + yield "\n".join(logs), 0, None, "" + + # Tạo âm thanh cho từng block + created_files = [] + for i, text_block in enumerate(text_blocks, 1): # Chọn API key theo round-robin - key_idx = (attempts - 1) % len(valid_keys) + key_idx = (i - 1) % len(valid_keys) key_info = valid_keys[key_idx] - # Kiểm tra credits - needed = estimate_credit(text) - if key_info["remaining"] < needed: - self.progress_signal.emit(f"⚠ Key {key_idx+1} hết credits") - valid_keys.pop(key_idx) - if not valid_keys: - break - continue - - self.progress_signal.emit(f"🎤 Block {i}/{len(self.texts)} - Attempt {attempts}...") + logs.append(f"🎤 Đang tạo block {i}/{len(text_blocks)}...") + yield "\n".join(logs), i/len(text_blocks), None, "" - # Tạo âm thanh - audio = generate_voice_advanced( - text, key_info["key"], self.voice_id, self.model_id, - proxy_manager=self.proxy_manager, - use_proxy=self.use_proxy, - **self.settings + # Tạo request + request = TTSRequest( + text=text_block, + api_key=key_info["key"], + voice_id=voice_id_text, + model_id=model, + settings=settings, + file_index=i, + output_dir=output_dir, + format=format ) - if audio: - # Lưu file - filename = os.path.join(self.output_dir, f"voice_{i:03d}.{self.fmt.lower()}") - with open(filename, "wb") as f: - f.write(audio) + self.tts_worker.add_task(request) + + # Đợi hoàn thành + results = self.tts_worker.wait_completion(timeout=300) + + # Lấy danh sách file đã tạo + created_files = [results[i] for i in range(1, len(text_blocks) + 1) if results.get(i)] + + if len(created_files) == len(text_blocks): + logs.append("✅ Đã tạo tất cả các block") + + # Nếu chỉ có 1 file, trả về luôn + if len(created_files) == 1: + audio_file = created_files[0] + logs.append(f"🎉 Hoàn thành! File: {os.path.basename(audio_file)}") + yield "\n".join(logs), 1.0, audio_file, audio_file + else: + # Merge các file nếu có nhiều block + logs.append("🔗 Đang merge các file...") + yield "\n".join(logs), 0.9, None, "" - # Cập nhật credits - key_info["remaining"] -= needed + # Tạo file tổng hợp (đơn giản: chỉ trả về file đầu tiên cho demo) + audio_file = created_files[0] if created_files else None + logs.append(f"🎉 Hoàn thành! Đã tạo {len(created_files)} files") + yield "\n".join(logs), 1.0, audio_file, audio_file - self.progress_value.emit(i) - self.progress_signal.emit(f"✅ Đã tạo block {i}") - success = True - break + else: + created_count = len([f for f in created_files if f]) + logs.append(f"⚠️ Đã tạo {created_count}/{len(text_blocks)} blocks") + + if created_files: + audio_file = created_files[0] + yield "\n".join(logs), created_count/len(text_blocks), audio_file, audio_file else: - # Thất bại, chờ và thử lại - wait_time = random.uniform(5, 15) - self.progress_signal.emit(f"⚠ Thử lại sau {wait_time:.1f}s...") - time.sleep(wait_time) - - if not success: - self.progress_signal.emit(f"❌ Không thể tạo block {i}") - self.finished_signal.emit(False, f"Lỗi tại block {i}") - return + logs.append("❌ Không thể tạo bất kỳ file nào") + yield "\n".join(logs), 0, None, "" + + # Kết nối các nút + start_btn.click( + fn=start_tts_generation, + inputs=[ + api_keys, voice_id, model_id, output_format, + stability, similarity, speed, speaker_boost, text_input + ], + outputs=[status_output, progress_bar, audio_output, download_btn] + ) + + # Xử lý nút dừng + def stop_generation(): + self.tts_worker.stop() + self.tts_worker = TTSWorker() + self.tts_worker.start() + return "⏹️ Đã dừng quá trình xử lý" - # Bước 3: Merge files - self.progress_signal.emit("🔗 Đang merge file audio...") - merged_file = merge_audio_files( - self.output_dir, self.fmt, - os.path.join(self.output_dir, "output_full") + stop_btn.click( + fn=stop_generation, + outputs=[status_output] ) - if merged_file and os.path.exists(merged_file) and os.path.getsize(merged_file) > 1024: - # Tạo phụ đề - self.progress_signal.emit("📝 Đang tạo file phụ đề...") - create_srt(self.output_dir, self.texts) - - self.progress_signal.emit(f"🎉 Hoàn thành! File: {merged_file}") - self.finished_signal.emit(True, f"Hoàn thành! File: {merged_file}") - else: - self.progress_signal.emit("❌ Lỗi merge file audio") - self.finished_signal.emit(False, "Lỗi merge file") - - except Exception as e: - self.progress_signal.emit(f"❌ Lỗi hệ thống: {str(e)}") - self.finished_signal.emit(False, f"Lỗi: {str(e)}") + # Xử lý khi đóng + demo.load( + fn=lambda: f"Ứng dụng đã sẵn sàng. Thư mục output: {self.output_dir}", + outputs=[status_output] + ) + + return demo -# ==================== PROXY DIALOG ==================== -class ProxyManagerDialog(QDialog): - """Dialog quản lý proxy""" - - proxies_updated = Signal() - - def __init__(self, proxy_manager, parent=None): - super().__init__(parent) - self.proxy_manager = proxy_manager - self.setWindowTitle("🌐 Quản lý Proxy") - self.setMinimumSize(800, 600) - self.setup_ui() - - def setup_ui(self): - layout = QVBoxLayout(self) - - # Tab widget - self.tab_widget = QTabWidget() - - # Tab 1: Quản lý proxy - self.setup_manage_tab() - - # Tab 2: Kiểm tra proxy - self.setup_check_tab() - - # Tab 3: Lấy proxy từ web - self.setup_fetch_tab() - - layout.addWidget(self.tab_widget) - - # Nút đóng - buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel) - buttons.accepted.connect(self.accept) - buttons.rejected.connect(self.reject) - layout.addWidget(buttons) +# ==================== MAIN ==================== +def main(): + app = SecureTTSApp() - def setup_manage_tab(self): - tab = QWidget() - layout = QVBoxLayout(tab) - - # Text edit để nhập proxy - self.proxy_text = QTextEdit() - self.proxy_text.setPlaceholderText( - "Nhập proxy (mỗi dòng một proxy):\n" - "http://user:pass@ip:port\n" - "socks5://user:pass@ip:port\n" - "ip:port (tự động thêm http://)\n\n" - "Ví dụ:\n" - "http://user:pass@192.168.1.1:8080\n" - "socks5://192.168.1.2:1080\n" - "45.77.123.99:8080" + try: + demo = app.create_interface() + demo.launch( + server_name="0.0.0.0", + server_port=7860, + share=False, + debug=False, + show_error=True ) - layout.addWidget(QLabel("Danh sách proxy:")) - layout.addWidget(self.proxy_text) - - # Nút thao tác - btn_layout = QHBoxLayout() - - load_btn = QPushButton("📁 Tải từ file") - load_btn.clicked.connect(self.load_proxies) - - save_btn = QPushButton("💾 Lưu vào file") - save_btn.clicked.connect(self.save_proxies) - - add_btn = QPushButton("➕ Thêm Proxy") - add_btn.clicked.connect(self.add_proxies) - - clear_btn = QPushButton("🗑️ Xóa tất cả") - clear_btn.clicked.connect(self.clear_proxies) - - btn_layout.addWidget(load_btn) - btn_layout.addWidget(save_btn) - btn_layout.addWidget(add_btn) - btn_layout.addWidget(clear_btn) - - layout.addLayout(btn_layout) - - # Hiển thị thông tin - self.stats_label = QLabel("Tổng: 0 | Sống: 0 | Chết: 0") - self.stats_label.setStyleSheet("font-weight: bold; color: #2c3e50;") - layout.addWidget(self.stats_label) - - # Danh sách proxy sống - self.alive_list = QListWidget() - layout.addWidget(QLabel("Proxy sống:")) - layout.addWidget(self.alive_list) - - self.tab_widget.addTab(tab, "📝 Quản lý") - self.load_current_proxies() - - def setup_check_tab(self): - tab = QWidget() - layout = QVBoxLayout(tab) - - # Nút kiểm tra - self.check_btn = QPushButton("🔍 Kiểm tra tất cả proxy") - self.check_btn.clicked.connect(self.check_all_proxies) - layout.addWidget(self.check_btn) - - # Progress - self.progress = QProgressBar() - layout.addWidget(self.progress) - - # Kết quả - self.results_text = QTextEdit() - self.results_text.setReadOnly(True) - layout.addWidget(QLabel("Kết quả kiểm tra:")) - layout.addWidget(self.results_text) - - self.tab_widget.addTab(tab, "✅ Kiểm tra") - - def setup_fetch_tab(self): - tab = QWidget() - layout = QVBoxLayout(tab) - - # Nguồn proxy - source_layout = QHBoxLayout() - source_layout.addWidget(QLabel("Nguồn:")) - - self.source_combo = QComboBox() - self.source_combo.addItems([ - "proxyscrape.com - HTTP", - "proxyscrape.com - SOCKS4", - "proxyscrape.com - SOCKS5", - "proxy-list.download - HTTP", - "proxy-list.download - SOCKS4", - "proxy-list.download - SOCKS5" - ]) - source_layout.addWidget(self.source_combo) - - fetch_btn = QPushButton("🌐 Lấy proxy") - fetch_btn.clicked.connect(self.fetch_proxies) - source_layout.addWidget(fetch_btn) - - layout.addLayout(source_layout) - - # URL tùy chỉnh - url_layout = QHBoxLayout() - url_layout.addWidget(QLabel("URL:")) - - self.custom_url = QLineEdit() - self.custom_url.setPlaceholderText("https://example.com/proxy-list.txt") - url_layout.addWidget(self.custom_url) - - fetch_custom_btn = QPushButton("🌐 Lấy từ URL") - fetch_custom_btn.clicked.connect(self.fetch_custom_proxies) - url_layout.addWidget(fetch_custom_btn) - - layout.addLayout(url_layout) - - # Thông tin - info_label = QLabel( - "Proxy miễn phí thường bị chặn bởi ElevenLabs.\n" - "Để tạo âm thanh thành công, nên sử dụng proxy premium/residential." - ) - info_label.setStyleSheet("color: #e74c3c; padding: 10px;") - layout.addWidget(info_label) - - # Premium proxy links - links_layout = QHBoxLayout() - links_label = QLabel("Proxy Premium đề xuất:") - links_layout.addWidget(links_label) - - brightdata_btn = QPushButton("BrightData") - brightdata_btn.clicked.connect(lambda: QDesktopServices.openUrl(QUrl("https://brightdata.com"))) - - oxylabs_btn = QPushButton("Oxylabs") - oxylabs_btn.clicked.connect(lambda: QDesktopServices.openUrl(QUrl("https://oxylabs.io"))) - - smartproxy_btn = QPushButton("SmartProxy") - smartproxy_btn.clicked.connect(lambda: QDesktopServices.openUrl(QUrl("https://smartproxy.com"))) - - links_layout.addWidget(brightdata_btn) - links_layout.addWidget(oxylabs_btn) - links_layout.addWidget(smartproxy_btn) - links_layout.addStretch() - - layout.addLayout(links_layout) - - self.tab_widget.addTab(tab, "🌐 Lấy Proxy") - - def load_current_proxies(self): - """Tải proxy hiện tại""" - if self.proxy_manager.proxy_list: - self.proxy_text.setPlainText("\n".join(self.proxy_manager.proxy_list)) - self.update_stats() - - def update_stats(self): - """Cập nhật thống kê""" - stats = self.proxy_manager.get_stats() - self.stats_label.setText( - f"Tổng: {stats['total']} | " - f"Sống: {stats['alive']} | " - f"Chết: {stats['dead']} | " - f"Hiện tại: {stats['current'][:30] if stats['current'] else 'None'}" - ) - - # Cập nhật danh sách proxy sống - self.alive_list.clear() - for proxy in self.proxy_manager.alive_proxies[:50]: # Hiển thị tối đa 50 proxy - item = QListWidgetItem(proxy) - self.alive_list.addItem(item) - - def load_proxies(self): - """Tải proxy từ file""" - file_path, _ = QFileDialog.getOpenFileName( - self, "Tải proxy từ file", "", "Text files (*.txt);;All files (*.*)" - ) - - if file_path: - try: - self.proxy_manager.load_proxies_from_file(file_path) - self.load_current_proxies() - self.proxies_updated.emit() - QMessageBox.information(self, "Thành công", f"Đã tải {len(self.proxy_manager.proxy_list)} proxy") - except Exception as e: - QMessageBox.critical(self, "Lỗi", f"Lỗi: {str(e)}") - - def save_proxies(self): - """Lưu proxy vào file""" - file_path, _ = QFileDialog.getSaveFileName( - self, "Lưu proxy vào file", "proxies.txt", "Text files (*.txt)" - ) - - if file_path: - try: - self.proxy_manager.save_proxies_to_file(file_path) - QMessageBox.information(self, "Thành công", f"Đã lưu {len(self.proxy_manager.proxy_list)} proxy") - except Exception as e: - QMessageBox.critical(self, "Lỗi", f"Lỗi: {str(e)}") - - def add_proxies(self): - """Thêm proxy từ text""" - text = self.proxy_text.toPlainText().strip() - if not text: - return - - proxies = [p.strip() for p in text.splitlines() if p.strip()] - added = 0 - - for proxy in proxies: - if proxy and proxy not in self.proxy_manager.proxy_list: - self.proxy_manager.add_proxy(proxy) - added += 1 - - if added > 0: - self.proxies_updated.emit() - self.update_stats() - QMessageBox.information(self, "Thành công", f"Đã thêm {added} proxy mới") - - def clear_proxies(self): - """Xóa tất cả proxy""" - reply = QMessageBox.question( - self, "Xác nhận", - "Xóa tất cả proxy?", - QMessageBox.Yes | QMessageBox.No - ) - - if reply == QMessageBox.Yes: - self.proxy_manager.clear_proxies() - self.load_current_proxies() - self.proxies_updated.emit() - - def check_all_proxies(self): - """Kiểm tra tất cả proxy""" - if not self.proxy_manager.proxy_list: - QMessageBox.warning(self, "Cảnh báo", "Không có proxy để kiểm tra") - return - - self.check_btn.setEnabled(False) - self.results_text.clear() - self.progress.setMaximum(len(self.proxy_manager.proxy_list)) - self.progress.setValue(0) - - # Thread kiểm tra - self.check_thread = ProxyCheckThread(self.proxy_manager) - self.check_thread.progress_signal.connect(self.update_check_progress) - self.check_thread.finished_signal.connect(self.on_check_finished) - self.check_thread.start() - - def update_check_progress(self, proxy, is_alive, message): - """Cập nhật tiến trình kiểm tra""" - timestamp = time.strftime("%H:%M:%S") - if is_alive: - self.results_text.append(f'[{timestamp}] ✅ {proxy} - {message}') - else: - self.results_text.append(f'[{timestamp}] ❌ {proxy} - {message}') - - self.progress.setValue(self.progress.value() + 1) - - def on_check_finished(self): - """Khi kiểm tra hoàn tất""" - self.check_btn.setEnabled(True) - self.update_stats() - self.results_text.append("\n✅ Kiểm tra hoàn tất!") - QMessageBox.information(self, "Hoàn tất", "Đã kiểm tra tất cả proxy") - - def fetch_proxies(self): - """Lấy proxy từ web""" - sources = { - 0: "https://api.proxyscrape.com/v2/?request=getproxies&protocol=http&timeout=10000&country=all&ssl=all&anonymity=all", - 1: "https://api.proxyscrape.com/v2/?request=getproxies&protocol=socks4&timeout=10000&country=all", - 2: "https://api.proxyscrape.com/v2/?request=getproxies&protocol=socks5&timeout=10000&country=all", - 3: "https://www.proxy-list.download/api/v1/get?type=http", - 4: "https://www.proxy-list.download/api/v1/get?type=socks4", - 5: "https://www.proxy-list.download/api/v1/get?type=socks5" - } - - url = sources.get(self.source_combo.currentIndex()) - if url: - self.do_fetch_proxies(url) - - def fetch_custom_proxies(self): - """Lấy proxy từ URL tùy chỉnh""" - url = self.custom_url.text().strip() - if url: - self.do_fetch_proxies(url) - - def do_fetch_proxies(self, url): - """Thực hiện lấy proxy""" - try: - response = requests.get(url, timeout=30, verify=False) - if response.status_code == 200: - lines = response.text.strip().splitlines() - added = 0 - - for line in lines: - line = line.strip() - if line and ':' in line and not line.startswith('#'): - parts = line.split(':') - if len(parts) >= 2: - ip = parts[0].strip() - port = parts[1].strip() - - # Xác định protocol - if 'socks' in url.lower(): - if 'socks4' in url.lower(): - proxy = f"socks4://{ip}:{port}" - else: - proxy = f"socks5://{ip}:{port}" - else: - proxy = f"http://{ip}:{port}" - - if proxy not in self.proxy_manager.proxy_list: - self.proxy_manager.add_proxy(proxy) - added += 1 - - if added > 0: - self.load_current_proxies() - self.proxies_updated.emit() - QMessageBox.information(self, "Thành công", f"Đã thêm {added} proxy mới từ web") - else: - QMessageBox.warning(self, "Cảnh báo", "Không tìm thấy proxy mới") - else: - QMessageBox.critical(self, "Lỗi", f"HTTP Error: {response.status_code}") - - except Exception as e: - QMessageBox.critical(self, "Lỗi", f"Lỗi: {str(e)}") - -class ProxyCheckThread(QThread): - """Thread kiểm tra proxy""" - - progress_signal = Signal(str, bool, str) - finished_signal = Signal() - - def __init__(self, proxy_manager): - super().__init__() - self.proxy_manager = proxy_manager - - def run(self): - self.proxy_manager.check_all_proxies(max_workers=10) - self.finished_signal.emit() - -# ==================== ỨNG DỤNG CHÍNH ==================== -class SecureTTSApp(QWidget): - """Ứng dụng TTS bảo mật với proxy""" - - def __init__(self): - super().__init__() - self.settings = QSettings("ElevenLabsTTS", "SecureTTS") - self.setWindowTitle("🎤 ElevenLabs TTS Pro - Secure with Proxy") - self.resize(1100, 800) - - # Proxy manager - self.proxy_manager = AdvancedProxyManager() - self.use_proxy = self.settings.value("use_proxy", "false") == "true" - - # Output directory - self.output_dir = self.settings.value("output_dir", "voices") - os.makedirs(self.output_dir, exist_ok=True) - - self.setup_ui() - self.load_settings() - self.update_proxy_status() - - def setup_ui(self): - """Thiết lập giao diện""" - main_layout = QVBoxLayout(self) - main_layout.setContentsMargins(10, 10, 10, 10) - main_layout.setSpacing(10) - - # Header - header = QLabel("🎤 ELEVENLABS TTS - SECURE EDITION") - header.setFont(QFont("Arial", 16, QFont.Bold)) - header.setStyleSheet("color: #2c3e50; padding: 10px;") - header.setAlignment(Qt.AlignCenter) - main_layout.addWidget(header) - - # Status bar - status_layout = QHBoxLayout() - - self.proxy_status = QLabel("🌐 Proxy: TẮT") - self.proxy_status.setStyleSheet(""" - QLabel { - background-color: #e74c3c; - color: white; - padding: 5px 10px; - border-radius: 10px; - font-weight: bold; - } - """) - - self.api_status = QLabel("🔑 API: 0 keys") - self.api_status.setStyleSheet(""" - QLabel { - background-color: #3498db; - color: white; - padding: 5px 10px; - border-radius: 10px; - font-weight: bold; - } - """) - - self.text_status = QLabel("📝 Text: 0 chars") - self.text_status.setStyleSheet(""" - QLabel { - background-color: #2ecc71; - color: white; - padding: 5px 10px; - border-radius: 10px; - font-weight: bold; - } - """) - - status_layout.addWidget(self.proxy_status) - status_layout.addWidget(self.api_status) - status_layout.addWidget(self.text_status) - status_layout.addStretch() - - main_layout.addLayout(status_layout) - - # Tab widget - self.tabs = QTabWidget() - - # Tab 1: Cấu hình - self.setup_config_tab() - - # Tab 2: Văn bản - self.setup_text_tab() - - # Tab 3: Proxy - self.setup_proxy_tab() - - # Tab 4: Output - self.setup_output_tab() - - main_layout.addWidget(self.tabs) - - # Control buttons - control_layout = QHBoxLayout() - - self.clear_btn = QPushButton("🗑️ Xóa tất cả") - self.clear_btn.clicked.connect(self.clear_all) - - self.check_api_btn = QPushButton("🔍 Kiểm tra API") - self.check_api_btn.clicked.connect(self.check_api_keys) - - self.export_btn = QPushButton("📁 Export") - self.export_btn.clicked.connect(self.export_files) - - self.start_btn = QPushButton("🚀 Bắt đầu tạo giọng nói") - self.start_btn.setStyleSheet(""" - QPushButton { - background-color: #2ecc71; - color: white; - font-weight: bold; - padding: 10px 20px; - border-radius: 5px; - } - QPushButton:hover { - background-color: #27ae60; - } - """) - self.start_btn.clicked.connect(self.start_tts_generation) - - control_layout.addWidget(self.clear_btn) - control_layout.addWidget(self.check_api_btn) - control_layout.addWidget(self.export_btn) - control_layout.addStretch() - control_layout.addWidget(self.start_btn) - - main_layout.addLayout(control_layout) - - # Progress bar - self.progress_bar = QProgressBar() - main_layout.addWidget(self.progress_bar) - - # Log output - self.log_text = QTextEdit() - self.log_text.setReadOnly(True) - self.log_text.setMaximumHeight(150) - main_layout.addWidget(self.log_text) - - def setup_config_tab(self): - """Tab cấu hình""" - tab = QWidget() - layout = QVBoxLayout(tab) - - # API keys - api_group = QGroupBox("🔑 API Keys") - api_layout = QVBoxLayout() - - self.api_text = QTextEdit() - self.api_text.setPlaceholderText("Nhập API keys (mỗi key một dòng)\nsk_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx") - self.api_text.textChanged.connect(self.update_api_status) - api_layout.addWidget(self.api_text) - - api_btn_layout = QHBoxLayout() - load_api_btn = QPushButton("📁 Tải từ file") - load_api_btn.clicked.connect(self.load_api_keys) - - save_api_btn = QPushButton("💾 Lưu keys") - save_api_btn.clicked.connect(self.save_api_keys) - - api_btn_layout.addWidget(load_api_btn) - api_btn_layout.addWidget(save_api_btn) - api_btn_layout.addStretch() - - api_layout.addLayout(api_btn_layout) - api_group.setLayout(api_layout) - layout.addWidget(api_group) - - # Voice settings - voice_group = QGroupBox("🎤 Voice Settings") - voice_layout = QVBoxLayout() - - # Voice ID - voice_id_layout = QHBoxLayout() - voice_id_layout.addWidget(QLabel("Voice ID:")) - - self.voice_input = QLineEdit() - self.voice_input.setPlaceholderText("21m00Tcm4TlvDq8ikWAM") - voice_id_layout.addWidget(self.voice_input) - - voice_layout.addLayout(voice_id_layout) - - # Model và Format - model_layout = QHBoxLayout() - model_layout.addWidget(QLabel("Model:")) - - self.model_combo = QComboBox() - self.model_combo.addItems(["eleven_multilingual_v2", "eleven_turbo_v2", "eleven_monolingual_v1"]) - model_layout.addWidget(self.model_combo) - - model_layout.addWidget(QLabel("Format:")) - - self.format_combo = QComboBox() - self.format_combo.addItems(["mp3", "wav", "ogg", "flac"]) - model_layout.addWidget(self.format_combo) - - voice_layout.addLayout(model_layout) - - # Voice parameters - params_group = QGroupBox("Voice Parameters") - params_layout = QVBoxLayout() - - # Stability - stability_layout = QHBoxLayout() - stability_layout.addWidget(QLabel("Stability:")) - self.stability_slider = QSlider(Qt.Horizontal) - self.stability_slider.setRange(0, 100) - self.stability_slider.setValue(70) - stability_layout.addWidget(self.stability_slider) - self.stability_label = QLabel("0.70") - stability_layout.addWidget(self.stability_label) - self.stability_slider.valueChanged.connect( - lambda v: self.stability_label.setText(f"{v/100:.2f}") - ) - params_layout.addLayout(stability_layout) - - # Similarity - similarity_layout = QHBoxLayout() - similarity_layout.addWidget(QLabel("Similarity:")) - self.similarity_slider = QSlider(Qt.Horizontal) - self.similarity_slider.setRange(0, 100) - self.similarity_slider.setValue(80) - similarity_layout.addWidget(self.similarity_slider) - self.similarity_label = QLabel("0.80") - similarity_layout.addWidget(self.similarity_label) - self.similarity_slider.valueChanged.connect( - lambda v: self.similarity_label.setText(f"{v/100:.2f}") - ) - params_layout.addLayout(similarity_layout) - - # Speed - speed_layout = QHBoxLayout() - speed_layout.addWidget(QLabel("Speed:")) - self.speed_slider = QSlider(Qt.Horizontal) - self.speed_slider.setRange(50, 150) - self.speed_slider.setValue(100) - speed_layout.addWidget(self.speed_slider) - self.speed_label = QLabel("1.00") - speed_layout.addWidget(self.speed_label) - self.speed_slider.valueChanged.connect( - lambda v: self.speed_label.setText(f"{v/100:.2f}") - ) - params_layout.addLayout(speed_layout) - - params_group.setLayout(params_layout) - voice_layout.addWidget(params_group) - - voice_group.setLayout(voice_layout) - layout.addWidget(voice_group) - - layout.addStretch() - self.tabs.addTab(tab, "⚙️ Cấu hình") - - def setup_text_tab(self): - """Tab văn bản""" - tab = QWidget() - layout = QVBoxLayout(tab) - - self.text_edit = QTextEdit() - self.text_edit.setPlaceholderText("Nhập văn bản cần chuyển thành giọng nói tại đ��y...") - self.text_edit.textChanged.connect(self.update_text_status) - layout.addWidget(self.text_edit) - - # Text controls - text_controls = QHBoxLayout() - - load_text_btn = QPushButton("📁 Tải văn bản") - load_text_btn.clicked.connect(self.load_text_file) - - save_text_btn = QPushButton("💾 Lưu văn bản") - save_text_btn.clicked.connect(self.save_text_file) - - clear_text_btn = QPushButton("🗑️ Xóa văn bản") - clear_text_btn.clicked.connect(self.clear_text) - - text_controls.addWidget(load_text_btn) - text_controls.addWidget(save_text_btn) - text_controls.addWidget(clear_text_btn) - text_controls.addStretch() - - layout.addLayout(text_controls) - - # Text stats - self.text_stats = QLabel("0 ký tự | 0 đoạn | Ước tính: 0 credits") - self.text_stats.setStyleSheet("color: #7f8c8d; font-style: italic;") - layout.addWidget(self.text_stats) - - self.tabs.addTab(tab, "📝 Văn bản") - - def setup_proxy_tab(self): - """Tab proxy""" - tab = QWidget() - layout = QVBoxLayout(tab) - - # Proxy toggle - toggle_layout = QHBoxLayout() - self.proxy_toggle_btn = QPushButton("🌐 Bật Proxy") - self.proxy_toggle_btn.setStyleSheet(""" - QPushButton { - background-color: #2ecc71; - color: white; - padding: 10px 20px; - font-weight: bold; - border-radius: 5px; - } - """) - self.proxy_toggle_btn.clicked.connect(self.toggle_proxy) - - self.rotate_proxy_btn = QPushButton("🔄 Xoay Proxy") - self.rotate_proxy_btn.clicked.connect(self.rotate_proxy) - - self.manage_proxy_btn = QPushButton("🔧 Quản lý Proxy") - self.manage_proxy_btn.clicked.connect(self.manage_proxies) - - toggle_layout.addWidget(self.proxy_toggle_btn) - toggle_layout.addWidget(self.rotate_proxy_btn) - toggle_layout.addWidget(self.manage_proxy_btn) - toggle_layout.addStretch() - - layout.addLayout(toggle_layout) - - # Current proxy info - self.proxy_info = QTextEdit() - self.proxy_info.setReadOnly(True) - self.proxy_info.setMaximumHeight(80) - layout.addWidget(QLabel("Thông tin proxy hiện tại:")) - layout.addWidget(self.proxy_info) - - # Proxy stats - self.proxy_stats = QLabel("Tổng: 0 | Sống: 0 | Chết: 0") - self.proxy_stats.setStyleSheet("font-weight: bold; color: #2c3e50;") - layout.addWidget(self.proxy_stats) - - # Proxy tips - tips_label = QLabel( - "💡 Mẹo sử dụng proxy:\n" - "1. Proxy miễn phí thường bị chặn bởi ElevenLabs\n" - "2. Sử dụng proxy premium/residential để có kết quả tốt nhất\n" - "3. Thường xuyên kiểm tra và xoay proxy\n" - "4. Nếu âm thanh không tạo được, thử tắt proxy và chạy lại" - ) - tips_label.setStyleSheet(""" - background-color: #fff3cd; - padding: 10px; - border-radius: 5px; - color: #856404; - """) - layout.addWidget(tips_label) - - layout.addStretch() - self.tabs.addTab(tab, "🌐 Proxy") - - def setup_output_tab(self): - """Tab output""" - tab = QWidget() - layout = QVBoxLayout(tab) - - # Output directory - dir_layout = QHBoxLayout() - dir_layout.addWidget(QLabel("Thư mục output:")) - - self.output_dir_label = QLabel(self.output_dir) - self.output_dir_label.setStyleSheet("color: #27ae60; font-weight: bold;") - dir_layout.addWidget(self.output_dir_label, 1) - - select_dir_btn = QPushButton("📁 Chọn thư mục") - select_dir_btn.clicked.connect(self.select_output_dir) - - open_dir_btn = QPushButton("📂 Mở thư mục") - open_dir_btn.clicked.connect(self.open_output_dir) - - dir_layout.addWidget(select_dir_btn) - dir_layout.addWidget(open_dir_btn) - - layout.addLayout(dir_layout) - - # Recent files - self.file_list = QListWidget() - self.file_list.itemDoubleClicked.connect(self.open_selected_file) - layout.addWidget(QLabel("File đã tạo:")) - layout.addWidget(self.file_list) - - # File controls - file_controls = QHBoxLayout() - - refresh_btn = QPushButton("🔄 Làm mới") - refresh_btn.clicked.connect(self.refresh_file_list) - - play_btn = QPushButton("▶️ Phát") - play_btn.clicked.connect(self.play_selected_file) - - delete_btn = QPushButton("🗑️ Xóa") - delete_btn.clicked.connect(self.delete_selected_file) - - file_controls.addWidget(refresh_btn) - file_controls.addWidget(play_btn) - file_controls.addWidget(delete_btn) - file_controls.addStretch() - - layout.addLayout(file_controls) - - # Clear logs button - clear_logs_btn = QPushButton("🗑️ Xóa logs") - clear_logs_btn.clicked.connect(self.clear_logs) - layout.addWidget(clear_logs_btn) - - layout.addStretch() - self.tabs.addTab(tab, "📊 Output") - - def load_settings(self): - """Tải cài đặt""" - self.voice_input.setText(self.settings.value("voice_id", "")) - self.model_combo.setCurrentText(self.settings.value("model", "eleven_multilingual_v2")) - self.format_combo.setCurrentText(self.settings.value("format", "mp3")) - self.stability_slider.setValue(int(self.settings.value("stability", 70))) - self.similarity_slider.setValue(int(self.settings.value("similarity", 80))) - self.speed_slider.setValue(int(self.settings.value("speed", 100))) - - def save_settings(self): - """Lưu cài đặt""" - self.settings.setValue("voice_id", self.voice_input.text()) - self.settings.setValue("model", self.model_combo.currentText()) - self.settings.setValue("format", self.format_combo.currentText()) - self.settings.setValue("stability", self.stability_slider.value()) - self.settings.setValue("similarity", self.similarity_slider.value()) - self.settings.setValue("speed", self.speed_slider.value()) - self.settings.setValue("output_dir", self.output_dir) - self.settings.setValue("use_proxy", "true" if self.use_proxy else "false") - - def update_proxy_status(self): - """Cập nhật trạng thái proxy""" - if self.use_proxy: - self.proxy_status.setText("🌐 Proxy: BẬT") - self.proxy_status.setStyleSheet(""" - QLabel { - background-color: #2ecc71; - color: white; - padding: 5px 10px; - border-radius: 10px; - font-weight: bold; - } - """) - self.proxy_toggle_btn.setText("🌐 Tắt Proxy") - self.proxy_toggle_btn.setStyleSheet(""" - QPushButton { - background-color: #e74c3c; - color: white; - padding: 10px 20px; - font-weight: bold; - border-radius: 5px; - } - QPushButton:hover { - background-color: #c0392b; - } - """) - else: - self.proxy_status.setText("🌐 Proxy: TẮT") - self.proxy_status.setStyleSheet(""" - QLabel { - background-color: #e74c3c; - color: white; - padding: 5px 10px; - border-radius: 10px; - font-weight: bold; - } - """) - self.proxy_toggle_btn.setText("🌐 Bật Proxy") - self.proxy_toggle_btn.setStyleSheet(""" - QPushButton { - background-color: #2ecc71; - color: white; - padding: 10px 20px; - font-weight: bold; - border-radius: 5px; - } - QPushButton:hover { - background-color: #27ae60; - } - """) - - # Update proxy info - stats = self.proxy_manager.get_stats() - self.proxy_stats.setText( - f"Tổng: {stats['total']} | " - f"Sống: {stats['alive']} | " - f"Chết: {stats['dead']} | " - f"Điểm TB: {stats['avg_score']:.1f}" - ) - - if stats['current']: - proxy_info = f"Proxy hiện tại: {stats['current']}\n" - proxy_info += f"Alive proxies: {stats['alive']}/{stats['total']}\n" - proxy_info += f"Tỉ lệ sống: {stats['alive_percentage']:.1f}%" - self.proxy_info.setText(proxy_info) - else: - self.proxy_info.setText("Chưa có proxy nào được chọn") - - def update_api_status(self): - """Cập nhật trạng thái API""" - keys = [k.strip() for k in self.api_text.toPlainText().splitlines() if k.strip()] - self.api_status.setText(f"🔑 API: {len(keys)} keys") - - def update_text_status(self): - """Cập nhật trạng thái văn bản""" - text = self.text_edit.toPlainText() - char_count = len(text) - - if text.strip(): - blocks = parse_text_blocks(text) - block_count = len(blocks) - credit_estimate = char_count + (block_count * 50) - - self.text_stats.setText( - f"{char_count:,} ký tự | " - f"{block_count} đoạn | " - f"Ước tính: {credit_estimate:,} credits" - ) - else: - self.text_stats.setText("0 ký tự | 0 đoạn | Ước tính: 0 credits") - - self.text_status.setText(f"📝 Text: {char_count:,} chars") - - def log_message(self, message, level="info"): - """Ghi log message""" - timestamp = time.strftime("%H:%M:%S") - - if level == "error": - color = "#e74c3c" - prefix = "❌" - elif level == "success": - color = "#2ecc71" - prefix = "✅" - elif level == "warning": - color = "#f39c12" - prefix = "⚠️" - else: - color = "#3498db" - prefix = "ℹ️" - - html = f'[{timestamp}] {prefix} {message}' - self.log_text.append(html) - - # Auto-scroll to bottom - scrollbar = self.log_text.verticalScrollBar() - scrollbar.setValue(scrollbar.maximum()) - - def toggle_proxy(self): - """Bật/tắt proxy""" - self.use_proxy = not self.use_proxy - - if self.use_proxy: - # Kiểm tra nếu có proxy sống - if not self.proxy_manager.alive_proxies: - reply = QMessageBox.question( - self, "Kiểm tra proxy", - "Chưa có proxy sống. Bạn có muốn kiểm tra proxy ngay bây giờ?", - QMessageBox.Yes | QMessageBox.No - ) - - if reply == QMessageBox.Yes: - self.manage_proxies() - return - else: - self.use_proxy = False - return - - self.log_message("Đã bật chế độ sử dụng proxy", "success") - else: - self.log_message("Đã tắt chế độ sử dụng proxy", "info") - - self.update_proxy_status() - self.save_settings() - - def rotate_proxy(self): - """Xoay proxy""" - new_proxy = self.proxy_manager.rotate_proxy() - if new_proxy: - self.log_message(f"Đã xoay sang proxy: {new_proxy}", "info") - self.update_proxy_status() - else: - self.log_message("Không có proxy để xoay", "warning") - - def manage_proxies(self): - """Mở dialog quản lý proxy""" - dialog = ProxyManagerDialog(self.proxy_manager, self) - dialog.proxies_updated.connect(self.update_proxy_status) - dialog.exec() - - def load_api_keys(self): - """Tải API keys từ file""" - file_path, _ = QFileDialog.getOpenFileName( - self, "Tải API keys", "", "Text files (*.txt);;All files (*.*)" - ) - - if file_path: - try: - with open(file_path, 'r', encoding='utf-8') as f: - content = f.read() - - # Parse API keys - keys = [] - for line in content.splitlines(): - line = line.strip() - if line.startswith('sk_'): - keys.append(line) - elif '=' in line and 'sk_' in line: - # Try to extract from key=value format - parts = line.split('=', 1) - if len(parts) == 2 and parts[1].strip().startswith('sk_'): - keys.append(parts[1].strip()) - - if keys: - current_keys = self.api_text.toPlainText().strip() - if current_keys: - keys = list(set(current_keys.splitlines() + keys)) - - self.api_text.setPlainText('\n'.join(keys)) - self.log_message(f"Đã tải {len(keys)} API keys", "success") - else: - self.log_message("Không tìm thấy API keys hợp lệ trong file", "warning") - - except Exception as e: - self.log_message(f"Lỗi tải file: {str(e)}", "error") - - def save_api_keys(self): - """Lưu API keys vào file""" - keys = [k.strip() for k in self.api_text.toPlainText().splitlines() if k.strip()] - if not keys: - QMessageBox.warning(self, "Cảnh báo", "Không có API keys để lưu") - return - - file_path, _ = QFileDialog.getSaveFileName( - self, "Lưu API keys", "api_keys.txt", "Text files (*.txt)" - ) - - if file_path: - try: - with open(file_path, 'w', encoding='utf-8') as f: - f.write('\n'.join(keys)) - self.log_message(f"Đã lưu {len(keys)} API keys", "success") - except Exception as e: - self.log_message(f"Lỗi lưu file: {str(e)}", "error") - - def load_text_file(self): - """Tải văn bản từ file""" - file_path, _ = QFileDialog.getOpenFileName( - self, "Tải văn bản", "", "Text files (*.txt);;All files (*.*)" - ) - - if file_path: - try: - with open(file_path, 'r', encoding='utf-8') as f: - content = f.read() - - current_text = self.text_edit.toPlainText() - if current_text: - reply = QMessageBox.question( - self, "Xác nhận", - "Thêm vào văn bản hiện tại hay thay thế?", - QMessageBox.Yes | QMessageBox.No | QMessageBox.Cancel - ) - - if reply == QMessageBox.Yes: - self.text_edit.setPlainText(current_text + "\n\n" + content) - elif reply == QMessageBox.No: - self.text_edit.setPlainText(content) - else: - self.text_edit.setPlainText(content) - - self.log_message(f"Đã tải văn bản từ {os.path.basename(file_path)}", "success") - - except Exception as e: - self.log_message(f"Lỗi tải file: {str(e)}", "error") - - def save_text_file(self): - """Lưu văn bản vào file""" - text = self.text_edit.toPlainText().strip() - if not text: - QMessageBox.warning(self, "Cảnh báo", "Không có văn bản để lưu") - return - - file_path, _ = QFileDialog.getSaveFileName( - self, "Lưu văn bản", "text.txt", "Text files (*.txt)" - ) - - if file_path: - try: - with open(file_path, 'w', encoding='utf-8') as f: - f.write(text) - self.log_message("Đã lưu văn bản", "success") - except Exception as e: - self.log_message(f"Lỗi lưu file: {str(e)}", "error") - - def clear_text(self): - """Xóa văn bản""" - self.text_edit.clear() - self.log_message("Đã xóa văn bản", "info") - - def select_output_dir(self): - """Chọn thư mục output""" - dir_path = QFileDialog.getExistingDirectory(self, "Chọn thư mục", self.output_dir) - if dir_path: - self.output_dir = dir_path - self.output_dir_label.setText(dir_path) - self.save_settings() - self.refresh_file_list() - self.log_message(f"Thư mục output: {dir_path}", "info") - - def open_output_dir(self): - """Mở thư mục output""" - if os.path.exists(self.output_dir): - os.startfile(self.output_dir) - else: - self.log_message("Thư mục không tồn tại", "error") - - def refresh_file_list(self): - """Làm mới danh sách file""" - self.file_list.clear() - - if os.path.exists(self.output_dir): - files = [] - for f in os.listdir(self.output_dir): - if f.endswith(('.mp3', '.wav', '.ogg', '.flac', '.srt', '.txt')): - files.append(f) - - files.sort(key=lambda x: os.path.getmtime(os.path.join(self.output_dir, x)), reverse=True) - - for f in files[:50]: # Hiển thị tối đa 50 file - size = os.path.getsize(os.path.join(self.output_dir, f)) - item = QListWidgetItem(f"{f} ({size:,} bytes)") - item.setData(Qt.UserRole, f) - self.file_list.addItem(item) - - def open_selected_file(self, item): - """Mở file được chọn""" - filename = item.data(Qt.UserRole) - filepath = os.path.join(self.output_dir, filename) - - if os.path.exists(filepath): - os.startfile(filepath) - - def play_selected_file(self): - """Phát file được chọn""" - item = self.file_list.currentItem() - if item: - self.open_selected_file(item) - - def delete_selected_file(self): - """Xóa file được chọn""" - item = self.file_list.currentItem() - if item: - filename = item.data(Qt.UserRole) - filepath = os.path.join(self.output_dir, filename) - - reply = QMessageBox.question( - self, "Xác nhận", - f"Xóa file {filename}?", - QMessageBox.Yes | QMessageBox.No - ) - - if reply == QMessageBox.Yes: - try: - os.remove(filepath) - self.refresh_file_list() - self.log_message(f"Đã xóa file {filename}", "info") - except Exception as e: - self.log_message(f"Lỗi xóa file: {str(e)}", "error") - - def clear_logs(self): - """Xóa logs""" - self.log_text.clear() - self.log_message("Đã xóa logs", "info") - - def check_api_keys(self): - """Kiểm tra API keys""" - keys = [k.strip() for k in self.api_text.toPlainText().splitlines() if k.strip()] - if not keys: - QMessageBox.warning(self, "Cảnh báo", "Không có API keys để kiểm tra") - return - - self.log_message(f"Kiểm tra {len(keys)} API keys...", "info") - - valid_count = 0 - total_remaining = 0 - - for i, key in enumerate(keys, 1): - info = check_api_key_advanced(key, self.proxy_manager, self.use_proxy) - - if info.get("valid"): - valid_count += 1 - remaining = info.get("remaining", 0) - total_remaining += remaining - - status = f"Key {i}: ✅ Hợp lệ | Còn lại: {remaining:,} ký tự" - if remaining < 1000: - status += " ⚠️ (SẮP HẾT)" - self.log_message(status, "warning") - else: - self.log_message(status, "success") - else: - error = info.get("error", "Unknown error") - self.log_message(f"Key {i}: ❌ Không hợp lệ - {error}", "error") - - # Độ trễ giữa các lần check - if i < len(keys): - time.sleep(random.uniform(2, 4)) - - # Tổng kết - self.log_message("="*50, "info") - if valid_count > 0: - avg_remaining = total_remaining // valid_count - self.log_message(f"📊 TỔNG KẾT: {valid_count}/{len(keys)} key hợp lệ", "success") - self.log_message(f"📈 Tổng ký tự còn lại: {total_remaining:,}", "info") - self.log_message(f"📊 Trung bình mỗi key: {avg_remaining:,} ký tự", "info") - - if avg_remaining < 1000: - self.log_message("⚠️ CẢNH BÁO: Ký tự còn lại thấp, cần thêm API keys!", "warning") - else: - self.log_message("❌ KHÔNG CÓ API KEY NÀO HỢP LỆ!", "error") - self.log_message("👉 Kiểm tra lại API keys hoặc thử tắt proxy", "info") - - def export_files(self): - """Export files""" - if not os.path.exists(self.output_dir): - QMessageBox.warning(self, "Cảnh báo", "Thư mục output không tồn tại") - return - - export_dir = QFileDialog.getExistingDirectory(self, "Chọn thư mục export") - if export_dir: - try: - exported = 0 - for f in os.listdir(self.output_dir): - if f.endswith(('.mp3', '.wav', '.ogg', '.flac', '.srt')): - src = os.path.join(self.output_dir, f) - dst = os.path.join(export_dir, f) - shutil.copy2(src, dst) - exported += 1 - - self.log_message(f"Đã export {exported} file đến {export_dir}", "success") - except Exception as e: - self.log_message(f"Lỗi export: {str(e)}", "error") - - def clear_all(self): - """Xóa tất cả""" - reply = QMessageBox.question( - self, "Xác nhận", - "Xóa tất cả API keys và văn bản?", - QMessageBox.Yes | QMessageBox.No - ) - - if reply == QMessageBox.Yes: - self.api_text.clear() - self.text_edit.clear() - self.log_text.clear() - self.progress_bar.setValue(0) - self.log_message("Đã xóa tất cả dữ liệu", "info") - - def start_tts_generation(self): - """Bắt đầu tạo TTS""" - # Kiểm tra API keys - api_keys = [k.strip() for k in self.api_text.toPlainText().splitlines() if k.strip()] - if not api_keys: - QMessageBox.warning(self, "Lỗi", "Vui lòng nhập ít nhất một API key") - self.tabs.setCurrentIndex(0) - return - - # Kiểm tra Voice ID - voice_id = self.voice_input.text().strip() - if not voice_id: - QMessageBox.warning(self, "Lỗi", "Vui lòng nhập Voice ID") - self.tabs.setCurrentIndex(0) - return - - # Kiểm tra văn bản - text = self.text_edit.toPlainText().strip() - if not text: - QMessageBox.warning(self, "Lỗi", "Vui lòng nhập văn bản") - self.tabs.setCurrentIndex(1) - return - - # Kiểm tra proxy nếu đang bật - if self.use_proxy and not self.proxy_manager.alive_proxies: - reply = QMessageBox.question( - self, "Cảnh báo", - "Không có proxy sống. Bạn có muốn:\n" - "1. Tắt proxy và tiếp tục (IP thực sẽ lộ)\n" - "2. Kiểm tra proxy trước\n" - "3. Hủy", - QMessageBox.Yes | QMessageBox.No | QMessageBox.Cancel - ) - - if reply == QMessageBox.Yes: - self.use_proxy = False - self.update_proxy_status() - elif reply == QMessageBox.No: - self.manage_proxies() - return - else: - return - - # Parse văn bản - texts = parse_text_blocks(text) - - # Chuẩn bị settings - settings = { - "stability": self.stability_slider.value() / 100, - "similarity": self.similarity_slider.value() / 100, - "style": 0.0, - "speed": self.speed_slider.value() / 100, - "speaker_boost": True - } - - # Hiển thị thông tin - self.log_message("="*60, "info") - self.log_message("🚀 BẮT ĐẦU TẠO GIỌNG NÓI", "info") - self.log_message(f"🔑 API keys: {len(api_keys)}", "info") - self.log_message(f"🎤 Voice ID: {voice_id}", "info") - self.log_message(f"🤖 Model: {self.model_combo.currentText()}", "info") - self.log_message(f"📁 Format: {self.format_combo.currentText()}", "info") - self.log_message(f"📝 Số đoạn văn bản: {len(texts)}", "info") - self.log_message(f"🌐 Proxy: {'BẬT' if self.use_proxy else 'TẮT'}", "info") - if self.use_proxy: - stats = self.proxy_manager.get_stats() - self.log_message(f"🌐 Proxy sống: {stats['alive']}/{stats['total']}", "info") - self.log_message(f"📂 Thư mục output: {self.output_dir}", "info") - self.log_message("="*60, "info") - - # Tạo thread - self.voice_thread = VoiceGenerationThread( - api_keys=api_keys, - voice_id=voice_id, - texts=texts, - settings=settings, - fmt=self.format_combo.currentText(), - model_id=self.model_combo.currentText(), - output_dir=self.output_dir, - proxy_manager=self.proxy_manager if self.use_proxy else None, - use_proxy=self.use_proxy - ) - - # Kết nối signals - self.voice_thread.progress_signal.connect(self.log_message) - self.voice_thread.progress_value.connect(self.progress_bar.setValue) - self.voice_thread.progress_max.connect(self.progress_bar.setMaximum) - self.voice_thread.finished_signal.connect(self.on_tts_finished) - self.voice_thread.api_key_checked.connect(self.on_api_key_checked) - - # Vô hiệu hóa nút - self.start_btn.setEnabled(False) - self.start_btn.setText("⏳ Đang xử lý...") - - # Bắt đầu thread - self.voice_thread.start() - - def on_api_key_checked(self, index, valid, info): - """Khi API key được kiểm tra""" - if valid: - remaining = info.get("remaining", 0) - self.log_message(f"Key {index}: {remaining:,} ký tự còn lại", "success") - else: - error = info.get("error", "Invalid") - self.log_message(f"Key {index}: {error}", "error") - - def on_tts_finished(self, success, message): - """Khi TTS hoàn thành""" - self.start_btn.setEnabled(True) - self.start_btn.setText("🚀 Bắt đầu tạo giọng nói") - - if success: - self.log_message(message, "success") - self.refresh_file_list() - - # Hỏi có mở thư mục không - reply = QMessageBox.question( - self, "Thành công", - "Đã tạo âm thanh thành công!\n\nMở thư mục output?", - QMessageBox.Yes | QMessageBox.No - ) - - if reply == QMessageBox.Yes: - self.open_output_dir() - else: - self.log_message(message, "error") - - # Đề xuất giải pháp - if self.use_proxy: - self.log_message("💡 Gợi ý: Thử tắt proxy và chạy lại", "info") - - def closeEvent(self, event): - """Xử lý khi đóng ứng dụng""" - self.save_settings() - - # Dừng thread nếu đang chạy - if hasattr(self, 'voice_thread') and self.voice_thread.isRunning(): - self.voice_thread.stop() - self.voice_thread.wait(2000) - - event.accept() + except KeyboardInterrupt: + print("\nĐang dọn dẹp...") + finally: + app.cleanup() -# ==================== MAIN ==================== if __name__ == "__main__": - app = QApplication(sys.argv) - app.setStyle("Fusion") - - # Palette - palette = QPalette() - palette.setColor(QPalette.Window, QColor(240, 242, 245)) - palette.setColor(QPalette.WindowText, QColor(44, 62, 80)) - palette.setColor(QPalette.Base, QColor(255, 255, 255)) - palette.setColor(QPalette.AlternateBase, QColor(236, 240, 241)) - palette.setColor(QPalette.ToolTipBase, QColor(44, 62, 80)) - palette.setColor(QPalette.ToolTipText, QColor(236, 240, 241)) - palette.setColor(QPalette.Text, QColor(44, 62, 80)) - palette.setColor(QPalette.Button, QColor(236, 240, 241)) - palette.setColor(QPalette.ButtonText, QColor(44, 62, 80)) - palette.setColor(QPalette.BrightText, Qt.red) - palette.setColor(QPalette.Link, QColor(52, 152, 219)) - palette.setColor(QPalette.Highlight, QColor(52, 152, 219)) - palette.setColor(QPalette.HighlightedText, Qt.white) - - app.setPalette(palette) - - window = SecureTTSApp() - window.show() - - sys.exit(app.exec()) \ No newline at end of file + main() \ No newline at end of file