import os import json import time import io import threading import uuid import requests import re import logging import random import base64 import atexit import concurrent.futures from datetime import datetime, timedelta from itertools import cycle from flask import Flask, request, jsonify, render_template, send_file from flask_cors import CORS from pydub import AudioSegment from huggingface_hub import HfApi, hf_hub_download from huggingface_hub.utils import RepositoryNotFoundError, EntryNotFoundError # --- CONFIGURATION & LOGGING --- CACHE_DIRECTORY = "/tmp/huggingface_cache_ezmary" os.makedirs(CACHE_DIRECTORY, exist_ok=True) logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S') app = Flask(__name__) CORS(app) # --- WORKER POOL SETUP --- WORKER_URLS = [ "https://opera8-ttspro.hf.space/generate", "https://hamed744-ttspro.hf.space/generate", "https://hamed744-ttspro2.hf.space/generate", ] # --- آدرس سرویس تغییر صدا (VC) --- VC_SPACE_URL = "https://ezmarynoori-sada.hf.space" # --- GLOBAL VARIABLES --- tasks = {} tasks_lock = threading.Lock() request_counter = 0 request_counter_lock = threading.Lock() DATASET_REPO = "opera8/Karbaran-rayegan-tedad" DATASET_FILENAME = "usage_data.json" USAGE_LIMIT = 5 HF_TOKEN = os.environ.get("HF_TOKEN") CLEANUP_INTERVAL_SECONDS = 6 * 30 * 24 * 60 * 60 last_cleanup_time = time.time() usage_data_cache = [] cache_lock = threading.Lock() data_changed = threading.Event() api = None # --- BROWSER HEADERS (Anti-Bot Bypass) --- STANDARD_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "application/json", "Content-Type": "application/json" } # --- DATABASE LOGIC --- if not HF_TOKEN: logging.error("CRITICAL: Secret 'HF_TOKEN' not found.") else: api = HfApi(token=HF_TOKEN) logging.info("HfApi initialized.") def load_initial_data(): global usage_data_cache with cache_lock: if not api: return try: local_path = hf_hub_download( repo_id=DATASET_REPO, filename=DATASET_FILENAME, repo_type="dataset", token=HF_TOKEN, force_download=True, cache_dir=CACHE_DIRECTORY ) with open(local_path, 'r', encoding='utf-8') as f: content = f.read() if content: usage_data_cache = json.loads(content) except (RepositoryNotFoundError, EntryNotFoundError): logging.warning("Dataset not found, creating new.") except Exception as e: logging.error(f"Failed to load data: {e}") def persist_data_to_hub(): global last_cleanup_time, usage_data_cache with cache_lock: now = time.time() if (now - last_cleanup_time) > CLEANUP_INTERVAL_SECONDS: six_months_ago = now - CLEANUP_INTERVAL_SECONDS usage_data_cache = [u for u in usage_data_cache if u.get('week_start', 0) > six_months_ago] last_cleanup_time = now data_changed.set() if not data_changed.is_set() or not api: return try: data_to_write = list(usage_data_cache) temp_filepath = os.path.join(CACHE_DIRECTORY, "temp_usage_data.json") with open(temp_filepath, 'w', encoding='utf-8') as f: json.dump(data_to_write, f, indent=2, ensure_ascii=False) api.upload_file(path_or_fileobj=temp_filepath, path_in_repo=DATASET_FILENAME, repo_id=DATASET_REPO, repo_type="dataset", commit_message="Update") os.remove(temp_filepath) data_changed.clear() except Exception as e: logging.error(f"Persist failed: {e}") def background_persister(): while True: time.sleep(10) persist_data_to_hub() def get_user_ip(): if request.headers.getlist("X-Forwarded-For"): return request.headers.getlist("X-Forwarded-For")[0].split(',')[0].strip() return request.remote_addr # --- TTS HELPER FUNCTIONS --- def call_worker(index, chunk_payload): raw_text = chunk_payload.get("text", "") # پاکسازی متن از براکت‌ها و کاراکترهای مخرب clean_text = re.sub(r'[*_`~#]', '', raw_text) clean_text = re.sub(r'\[.*?\]|\(.*?\)', '', clean_text) clean_text = clean_text.strip() if not clean_text: return index, AudioSegment.silent(duration=500) use_live = True if len(clean_text) <= 600 else False target_speaker = chunk_payload.get("speaker") is_custom = chunk_payload.get("is_custom", False) actual_speaker_request = "Charon" if is_custom else target_speaker worker_payload = { "text": clean_text, "speaker": actual_speaker_request, "temperature": chunk_payload.get("temperature", 0.9), "use_live_model": use_live, "retry_limit": 50, "fallback_to_live": True } max_attempts = 40 for attempt in range(max_attempts): workers = list(WORKER_URLS) random.shuffle(workers) for worker_url in workers: try: logging.info(f"▶️ [قطعه {index+1}] ارسال به {worker_url} (تلاش {attempt+1}/{max_attempts})") response = requests.post(worker_url, json=worker_payload, headers=STANDARD_HEADERS, timeout=120) if response.status_code == 200: logging.info(f"✅ [قطعه {index+1}] با موفقیت ساخته شد.") audio_data = io.BytesIO(response.content) audio_segment = AudioSegment.from_file(audio_data) return index, audio_segment else: logging.warning(f"⚠️ [قطعه {index+1}] ارور {response.status_code} از کارگر دریافت شد. تلاش بعدی...") continue except Exception as e: logging.error(f"❌ [قطعه {index+1}] قطعی ارتباط: {str(e)[:40]}") continue sleep_time = 3 + (attempt * 1.5) + random.uniform(0.5, 2.0) logging.warning(f"🔄 [قطعه {index+1}] کارگرها شلوغ/خطا دادند. استراحت به مدت {sleep_time:.1f} ثانیه...") time.sleep(sleep_time) raise ValueError(f"قطعه {index+1} پس از 40 بار تلاش ساخته نشد.") # --- AI PODCAST SCRIPT LOGIC (Cohere Labs Space) --- def generate_podcast_in_background(task_id, system_prompt, safety_settings): try: AYA_SPACE_URL = "https://coherelabs-aya-expanse.hf.space/gradio_api" MAX_ATTEMPTS = 50 for attempt in range(MAX_ATTEMPTS): try: session_hash = str(uuid.uuid4())[0:11] payload = { "data": [system_prompt, [], None, None], "event_data": None, "fn_index": 2, "session_hash": session_hash, "trigger_id": 37 } join_response = requests.post(f"{AYA_SPACE_URL}/queue/join", json=payload, headers=STANDARD_HEADERS, timeout=60) join_response.raise_for_status() stream_url = f"{AYA_SPACE_URL}/queue/data?session_hash={session_hash}" raw_text = None with requests.get(stream_url, stream=True, headers=STANDARD_HEADERS, timeout=120) as resp: for line in resp.iter_lines(): if line: decoded_line = line.decode('utf-8') if decoded_line.startswith('data: '): json_str = decoded_line[6:] try: msg_data = json.loads(json_str) msg_type = msg_data.get('msg') if msg_type == 'process_completed': output = msg_data.get('output', {}).get('data', []) if output and len(output) > 0: updated_history = output[0] if updated_history: raw_text = updated_history[-1][1] elif msg_type == 'queue_full': raise Exception("صف سرور شلوغ است") except json.JSONDecodeError: continue if not raw_text: raise ValueError("بدون پاسخ از مدل هوش مصنوعی") json_string = None json_pattern = r"`{3}json\s*(\{.*?\})\s*`{3}" match = re.search(json_pattern, raw_text, re.DOTALL) if match: json_string = match.group(1) else: s_idx = raw_text.find('{') e_idx = raw_text.rfind('}') if s_idx != -1 and e_idx != -1: json_string = raw_text[s_idx:e_idx+1] if not json_string: raise ValueError("No JSON found in response") data = json.loads(json_string) if "script" in data: for t in data["script"]: if "dialogue" in t: t["dialogue"] = re.sub(r'\[.*?\]|\(.*?\)', '', t["dialogue"]).strip() with tasks_lock: tasks[task_id].update({'status': 'completed', 'data': data}) return except Exception as e: logging.warning(f"AI Attempt {attempt} failed: {e}") time.sleep(2) with tasks_lock: tasks[task_id].update({'status': 'failed', 'error': 'Max retries reached'}) except Exception as e: with tasks_lock: tasks[task_id].update({'status': 'failed', 'error': str(e)}) # --- FULL AUTO PODCAST LOGIC --- def generate_full_podcast_audio_background(task_id, prompt, speakers): try: logging.info(f"🚀 [پروژه {task_id}] عملیات ساخت پادکست آغاز شد.") with tasks_lock: tasks[task_id] = {'status': 'writing_script', 'progress': 'در حال نگارش سناریو...'} valid_speaker_ids = [str(s['id']).strip() for s in speakers] default_speaker_id = valid_speaker_ids[0] if valid_speaker_ids else "Charon" # 1. ساخت سناریو spk_text = "\n".join([f"- {s['id']}: {s['name']}" for s in speakers]) system_prompt = f"""Act as a Professional Podcast Producer. Topic: "{prompt}" Speakers Available: {spk_text} CRITICAL INSTRUCTION: You must create a VERY LONG, in-depth, and highly detailed podcast script. - Do NOT write a short summary. - The conversation must deeply explore the topic. Make it a very long and detailed conversation, as long as necessary to fully cover the topic. - Make the dialogue engaging and informative. You must use two speakers for the podcast, or three if necessary, or whatever number of speakers the user specifies. The podcast should be an engaging exchange, typically between a man and a woman, with both speakers discussing the topic together. Output ONLY valid JSON. Format: {{"selected_speakers": ["id1", "id2"], "script": [{{"speaker_id": "id1", "dialogue": "..."}}]}} Dialogue rules: Do NOT avoid writing stage directions and emotional cues like laughing or sighing. They are allowed, but the mood/action MUST be placed inside parentheses, for example (laugh) or (sigh).""" AYA_SPACE_URL = "https://coherelabs-aya-expanse.hf.space/gradio_api" MAX_ATTEMPTS = 50 data = None logging.info("📝 در حال ارتباط با هوش مصنوعی برای نوشتن سناریو...") for attempt in range(MAX_ATTEMPTS): try: session_hash = str(uuid.uuid4())[0:11] payload = {"data": [system_prompt, [], None, None], "event_data": None, "fn_index": 2, "session_hash": session_hash, "trigger_id": 37} join_response = requests.post(f"{AYA_SPACE_URL}/queue/join", json=payload, headers=STANDARD_HEADERS, timeout=60) join_response.raise_for_status() stream_url = f"{AYA_SPACE_URL}/queue/data?session_hash={session_hash}" raw_text = None with requests.get(stream_url, stream=True, headers=STANDARD_HEADERS, timeout=120) as resp: for line in resp.iter_lines(): if line: decoded_line = line.decode('utf-8') if decoded_line.startswith('data: '): json_str = decoded_line[6:] try: msg_data = json.loads(json_str) if msg_data.get('msg') == 'process_completed': output = msg_data.get('output', {}).get('data', []) if output and len(output) > 0 and output[0]: raw_text = output[0][-1][1] except: pass if raw_text: json_string = None json_pattern = r"`{3}json\s*(\{.*?\})\s*`{3}" match = re.search(json_pattern, raw_text, re.DOTALL) if match: json_string = match.group(1) else: s_idx = raw_text.find('{') e_idx = raw_text.rfind('}') if s_idx != -1 and e_idx != -1: json_string = raw_text[s_idx:e_idx+1] if json_string: data = json.loads(json_string) break except Exception as e: time.sleep(2) if not data or "script" not in data: raise ValueError("هوش مصنوعی در نوشتن سناریو با خطا مواجه شد.") for t in data["script"]: if "dialogue" in t: t["dialogue"] = re.sub(r'\[.*?\]|\(.*?\)', '', t["dialogue"]).strip() script_turns = [t for t in data.get("script", []) if str(t.get("dialogue", "")).strip()] total_turns = len(script_turns) logging.info(f"✅ سناریو استخراج شد! شامل {total_turns} نوبت گفتگو.") with tasks_lock: tasks[task_id] = {'status': 'generating_audio', 'progress': f'در حال تولید صداها (0 از {total_turns} تکمیل شده)'} audio_results = [None] * total_turns completed_count = 0 def process_single_chunk(index, turn_data): # 🔴 سیستم هوشمند جایگزینی گوینده (شبیه‌سازی کامل رفتار سایت) raw_speaker_id = str(turn_data.get("speaker_id", "")).strip() final_speaker_id = raw_speaker_id if final_speaker_id not in valid_speaker_ids: found = False for v_id in valid_speaker_ids: if final_speaker_id.lower() == v_id.lower(): final_speaker_id = v_id found = True break if not found: logging.warning(f"⚠️ آیدی گوینده نامعتبر '{raw_speaker_id}' تشخیص داده شد. جایگزین شد با '{default_speaker_id}'.") final_speaker_id = default_speaker_id dialogue = turn_data.get("dialogue") payload = {"text": dialogue, "speaker": final_speaker_id, "temperature": 0.9, "is_custom": False} idx, audio_seg = call_worker(index, payload) if audio_seg is None: raise ValueError(f"خطا در تولید صدای نوبت {index+1}") return idx, audio_seg batch_size = 7 batches = [script_turns[i:i + batch_size] for i in range(0, total_turns, batch_size)] for batch_index, batch in enumerate(batches): logging.info(f"🔄 شروع پردازش دسته‌ی {batch_index+1} از {len(batches)} (شامل {len(batch)} دیالوگ)...") with concurrent.futures.ThreadPoolExecutor(max_workers=len(batch)) as executor: start_idx = batch_index * batch_size future_to_index = {executor.submit(process_single_chunk, start_idx + i, turn): start_idx + i for i, turn in enumerate(batch)} for future in concurrent.futures.as_completed(future_to_index): try: idx, audio_seg = future.result() audio_results[idx] = audio_seg completed_count += 1 with tasks_lock: tasks[task_id]['progress'] = f'در حال تولید صداها ({completed_count} از {total_turns} تکمیل شده)' except Exception as exc: logging.error(f"❌ خطای بحرانی: {str(exc)}") raise Exception(str(exc)) logging.info(f"✔️ دسته‌ی {batch_index+1} با موفقیت تمام شد.") logging.info("✂️ تمام قطعات ساخته شد. در حال ادغام (میکس) فایل‌ها...") combined_audio = AudioSegment.empty() for seg in audio_results: if seg: combined_audio += seg with tasks_lock: tasks[task_id] = {'status': 'mixing', 'progress': 'در حال میکس نهایی و ذخیره پادکست...'} filename = f"full_podcast_{task_id}.mp3" filepath = os.path.join(CACHE_DIRECTORY, filename) output_buffer = io.BytesIO() combined_audio.export(output_buffer, format="mp3") output_buffer.seek(0) with open(filepath, 'wb') as f: f.write(output_buffer.read()) logging.info(f"🎉 پادکست نهایی آماده شد! نام فایل: {filename}") with tasks_lock: tasks[task_id] = {'status': 'completed', 'filename': filename, 'progress': 'پادکست شما آماده است!'} except Exception as e: logging.error(f"❌ پردازش کل پادکست با شکست مواجه شد: {str(e)}") with tasks_lock: tasks[task_id] = {'status': 'failed', 'error': str(e)} # --- VC LOGIC --- def process_voice_conversion(tts_audio_io, ref_audio_base64): try: tts_audio_io.seek(0) if "," in ref_audio_base64: ref_audio_base64 = ref_audio_base64.split(",")[1] ref_bytes = base64.b64decode(ref_audio_base64) files = { 'source_audio': ('source.wav', tts_audio_io, 'audio/wav'), 'ref_audio': ('ref.wav', io.BytesIO(ref_bytes), 'audio/wav') } res = requests.post(f"{VC_SPACE_URL}/upload", files=files, headers=STANDARD_HEADERS, timeout=120) if res.status_code != 200: raise Exception(f"VC Upload Failed: {res.text}") job_data = res.json() for _ in range(120): time.sleep(4) chk = requests.post(f"{VC_SPACE_URL}/check_status", json=job_data, headers=STANDARD_HEADERS, timeout=30) if chk.status_code == 200: stat = chk.json() if stat.get("status") == "completed": filename = stat.get("filename") dl = requests.get(f"{VC_SPACE_URL}/download/{filename}", headers=STANDARD_HEADERS) if dl.status_code == 200: return io.BytesIO(dl.content) else: raise Exception("VC Download Failed") elif stat.get("status") == "failed": raise Exception(f"VC Remote Failed: {stat.get('detail', 'Unknown error')}") raise Exception("VC Timeout (Processing took too long)") except Exception as e: logging.error(f"VC Error: {e}") return None # --- ROUTES --- @app.route('/') def index(): return render_template('index.html') @app.route('/api/check-credit', methods=['POST']) def check_credit(): data = request.get_json() fingerprint = data.get('fingerprint') if not fingerprint: return jsonify({"status": "error"}), 400 with cache_lock: ip = get_user_ip() now = time.time() week_ago = now - (7*24*60*60) user = next((u for u in usage_data_cache if u.get('fingerprint') == fingerprint), None) user = user or next((u for u in usage_data_cache if ip in u.get('ips', [])), None) limit_reached = False remaining = USAGE_LIMIT reset_ts = 0 if user: if user.get('week_start', 0) < week_ago: user['count'] = 0 user['week_start'] = now data_changed.set() remaining = USAGE_LIMIT - user.get('count', 0) if remaining <= 0: limit_reached = True remaining = 0 reset_ts = user.get('week_start', now) + (7*24*60*60) return jsonify({"credits_remaining": remaining, "limit_reached": limit_reached, "reset_timestamp": reset_ts}) @app.route('/api/use-credit', methods=['POST']) def use_credit(): data = request.get_json() fingerprint = data.get('fingerprint') with cache_lock: ip = get_user_ip() now = time.time() week_ago = now - (7*24*60*60) user = next((u for u in usage_data_cache if u.get('fingerprint') == fingerprint), None) user = user or next((u for u in usage_data_cache if ip in u.get('ips', [])), None) if user: if user.get('week_start', 0) < week_ago: user['count'] = 0 user['week_start'] = now if user['count'] >= USAGE_LIMIT: return jsonify({"status": "limit"}), 429 user['count'] += 1 if ip not in user['ips']: user['ips'].append(ip) else: user = {"fingerprint": fingerprint, "ips": [ip], "count": 1, "week_start": now} usage_data_cache.append(user) data_changed.set() return jsonify({"status": "success", "credits_remaining": USAGE_LIMIT - user['count']}) @app.route('/api/create-full-podcast', methods=['POST']) def create_full_podcast(): try: data = request.get_json() prompt = data.get('prompt') speakers = data.get('available_speakers') if not prompt or not speakers: return jsonify({"error": "Bad request"}), 400 task_id = str(uuid.uuid4()) with tasks_lock: tasks[task_id] = {'status': 'pending'} safety = [{"category": c, "threshold": "BLOCK_NONE"} for c in ["HARM_CATEGORY_HARASSMENT", "HARM_CATEGORY_HATE_SPEECH", "HARM_CATEGORY_SEXUALLY_EXPLICIT", "HARM_CATEGORY_DANGEROUS_CONTENT"]] spk_text = "\n".join([f"- {s['id']}: {s['name']}" for s in speakers]) sys_prompt = f"""Act as a Professional Podcast Producer.\nTopic: "{prompt}"\nSpeakers Available:\n{spk_text}\nCRITICAL INSTRUCTIONS:\n1. Create a VERY LONG, in-depth podcast script.\n2. Keep EVERY dialogue line SHORT. Break long speeches into multiple consecutive turns for the same speaker.\n3. NO stage directions, NO emojis, NO brackets. Plain text ONLY.\n\nOutput ONLY valid JSON.\nFormat: {{"selected_speakers": ["id1", "id2"], "script": [{{"speaker_id": "id1", "dialogue": "..."}}]}}""" threading.Thread(target=generate_podcast_in_background, args=(task_id, sys_prompt, safety)).start() return jsonify({"task_id": task_id}), 202 except Exception as e: return jsonify({"error": str(e)}), 500 @app.route('/api/podcast-status/', methods=['GET']) def podcast_status(task_id): with tasks_lock: return jsonify(tasks.get(task_id, {'status': 'not_found'})), 200 @app.route('/api/auto-podcast', methods=['POST']) def auto_podcast(): try: data = request.get_json() prompt = data.get('prompt') speakers = data.get('available_speakers') if not prompt or not speakers: return jsonify({"error": "Bad request"}), 400 task_id = str(uuid.uuid4()) with tasks_lock: tasks[task_id] = {'status': 'pending', 'progress': 'در حال ورود به صف پردازش...'} threading.Thread(target=generate_full_podcast_audio_background, args=(task_id, prompt, speakers)).start() return jsonify({"task_id": task_id}), 202 except Exception as e: return jsonify({"error": str(e)}), 500 @app.route('/api/auto-podcast-status/', methods=['GET']) def auto_podcast_status(task_id): with tasks_lock: return jsonify(tasks.get(task_id, {'status': 'not_found'})), 200 @app.route('/api/download-podcast/', methods=['GET']) def download_podcast(filename): filepath = os.path.join(CACHE_DIRECTORY, filename) if os.path.exists(filepath): return send_file(filepath, mimetype="audio/mp3", as_attachment=True) return jsonify({"error": "File not found"}), 404 @app.route('/api/generate', methods=['POST']) def generate_audio_route(): try: data = request.get_json() if not data: return jsonify({"error": "No data"}), 400 text = data.get("text", "") speaker = data.get("speaker") temperature = data.get("temperature", 0.9) ref_base64 = data.get("ref_audio_base64") if not text: return jsonify({"error": "Text empty"}), 400 is_custom = bool(speaker.startswith("custom_") and ref_base64) payload = {"text": text, "speaker": speaker, "temperature": temperature, "is_custom": is_custom} idx, audio_seg = call_worker(0, payload) if audio_seg is None: return jsonify({"error": "Worker generation failed"}), 503 final_buffer = io.BytesIO() audio_seg.export(final_buffer, format="wav") final_buffer.seek(0) if is_custom: logging.info("Starting Custom VC...") vc_out = process_voice_conversion(final_buffer, ref_base64) if vc_out: return send_file(vc_out, mimetype="audio/wav", as_attachment=True, download_name=f"vc_{uuid.uuid4()}.wav") else: return jsonify({"error": "Voice Conversion failed"}), 500 return send_file(final_buffer, mimetype="audio/wav", as_attachment=True, download_name=f"gen_{uuid.uuid4()}.wav") except Exception as e: logging.error(f"Generate route error: {e}") return jsonify({"error": str(e)}), 500 # --- STARTUP --- load_initial_data() threading.Thread(target=background_persister, daemon=True).start() atexit.register(persist_data_to_hub) if __name__ == '__main__': port = int(os.environ.get('PORT', 7860)) app.run(host='0.0.0.0', port=port)