import os import shutil import subprocess import uuid import json import time import asyncio import random import re import importlib.util import inspect import requests from datetime import datetime from typing import List, Optional, Union, Dict from fastapi import FastAPI, UploadFile, File, Form, HTTPException, BackgroundTasks from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware import google.generativeai as genai from pydantic import BaseModel from PIL import Image, ImageDraw, ImageFont from pydub import AudioSegment from pydub.silence import detect_silence from groq import Groq app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) TEMP_DIR = "temp" STATIC_DIR = "static" STYLES_DIR = "styles" os.makedirs(TEMP_DIR, exist_ok=True) os.makedirs(STATIC_DIR, exist_ok=True) os.makedirs(STYLES_DIR, exist_ok=True) app.mount("/temp", StaticFiles(directory="temp"), name="temp") app.mount("/static", StaticFiles(directory="static"), name="static") MODEL_NAME = "gemini-2.5-flash" FONT_DIR = "static/fonts" FONT_FILES_MAP = { "vazir": "Vazirmatn.ttf", "lalezar": "Lalezar.ttf", "amiri": "Amiri-Bold.ttf", "sarbaz": "Sarbaz.ttf", "nastaliq": "IranNastaliq.ttf", "vazir-thin": "Vazirmatn-Thin.ttf", "mada-thin": "Mada-ExtraLight.ttf", "aref-bold": "ArefRuqaa-Bold.ttf", "dastnevis": "Dastnevis.ttf", "entazar": "Entazar.ttf", "kamran": "Kamran.ttf", "gharib": "Gharib.ttf", "pinar": "Pinar-Bold.ttf", "hasti": "Hasti.ttf", } # دریافت توکن Groq و Gemini از تنظیمات سکرت اسپیس GROQ_API_KEYS_ENV = os.getenv("GROQ_API_KEYS", "") GROQ_API_KEYS = [k.strip() for k in GROQ_API_KEYS_ENV.split(",") if k.strip()] groq_key_index = 0 API_KEYS = [os.getenv("GEMINI_API_KEY")] if os.getenv("GEMINI_API_KEY") else [] # --- سیستم صف و پایش موقت آپلود غیرهمگام --- class UploadJobStatus: QUEUED = "queued" PROCESSING = "processing" COMPLETED = "completed" FAILED = "failed" class UploadJob: def __init__(self, task_id: str, file_name: str): self.id = task_id self.file_name = file_name self.status = UploadJobStatus.QUEUED self.result = None self.error_message = None self.created_at = datetime.now() upload_jobs: Dict[str, UploadJob] = {} # --- Dynamic Style Loading --- loaded_styles = {} style_configs = {} style_templates = {} def load_all_styles(): print("--- Loading Styles from /styles ---") for filename in os.listdir(STYLES_DIR): if filename.endswith(".py") and filename != "__init__.py": module_name = filename[:-3] file_path = os.path.join(STYLES_DIR, filename) spec = importlib.util.spec_from_file_location(module_name, file_path) if spec and spec.loader: mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) if hasattr(mod, 'config'): ids = mod.config.get("ids", []) for style_id in ids: loaded_styles[style_id] = mod style_configs[style_id] = mod.config if hasattr(mod, 'frontend_template'): style_templates[style_id] = mod.frontend_template.strip() print(f"Loaded Style: {style_id}") load_all_styles() # تنظیم همزمانی پردازش CONCURRENCY_LIMIT = 20 GEMINI_SEMAPHORE = asyncio.Semaphore(CONCURRENCY_LIMIT) # --- Data Models --- class WordInfo(BaseModel): word: str; start: float; end: float highlight: Optional[bool] = False color: Optional[str] = None class SubtitleSegment(BaseModel): id: Optional[Union[str, int]] = None start: float end: float text: str words: Optional[List[WordInfo]] = [] class StyleConfig(BaseModel): font: str; fontSize: int; primaryColor: str; outlineColor: str backType: str; marginV: int x: Optional[int] = 0 name: Optional[str] = "classic" radius: Optional[int] = 16 paddingX: Optional[int] = 20 paddingY: Optional[int] = 10 total_video_duration: Optional[float] = None current_render_time: Optional[float] = None entry_anim_progress: Optional[float] = 1.0 styleBgColors: Dict[str, str] = {} styleColors: Dict[str, str] = {} styleActiveColors: Dict[str, str] = {} useActiveColor: Optional[bool] = True fadeUnread: Optional[bool] = True fadeSurrounding: Optional[bool] = False typewriter: Optional[bool] = False class ProcessRequest(BaseModel): file_id: str; segments: List[SubtitleSegment] video_width: int; video_height: int; style: StyleConfig class StylePrompt(BaseModel): description: str class JobStatus: QUEUED = "queued"; PROCESSING = "processing" COMPLETED = "completed"; FAILED = "failed" class Job: def __init__(self, job_id: str, request_data: ProcessRequest): self.id = job_id; self.data = request_data; self.status = JobStatus.QUEUED self.created_at = datetime.now(); self.result_url = None; self.error_message = None render_queue = asyncio.Queue() jobs_db: Dict[str, Job] = {} async def queue_worker(): print("--- Queue Worker Started ---") while True: job_id = await render_queue.get() job = jobs_db.get(job_id) if job: try: print(f"Processing job: {job_id}") job.status = JobStatus.PROCESSING output_url = process_render_logic(job.data) job.result_url = output_url job.status = JobStatus.COMPLETED # اصلاح باگ عدم تعریف نام متغیر COMPLETED print(f"Job {job_id} completed.") except Exception as e: print(f"Job {job_id} failed: {e}") job.status = JobStatus.FAILED job.error_message = str(e) render_queue.task_done() async def cleanup_old_files_loop(): print("--- File Cleanup Worker Started ---") while True: await asyncio.sleep(600) # بررسی وضعیت هر ۱۰ دقیقه یک‌بار try: now = time.time() for filename in os.listdir(TEMP_DIR): file_path = os.path.join(TEMP_DIR, filename) if filename.endswith(".mp4") and "_final_" in filename: try: mtime = os.path.getmtime(file_path) if now - mtime > 7200: os.remove(file_path) print(f"Cleanup: Removed expired final video: {filename}") except Exception as e: print(f"Cleanup error for {filename}: {e}") except Exception as e: print(f"Global cleanup loop error: {e}") @app.on_event("startup") async def startup_event(): print(f"تعداد {len(API_KEYS)} کلید جیمینای و تعداد {len(GROQ_API_KEYS)} کلید گراک شناسایی شد.") asyncio.create_task(queue_worker()) asyncio.create_task(cleanup_old_files_loop()) # --- Helper Functions --- def clean_json_response(text: str): text = text.strip() if text.startswith('{') and text.endswith('}'): return text match = re.search(r'```json\s*(\{.*?\})\s*```', text, re.DOTALL) if match: return match.group(1) match = re.search(r'\{.*\}', text, re.DOTALL) if match: return match.group(0) return text def get_video_info(path): try: cmd = ["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height,duration", "-of", "json", path] res = subprocess.run(cmd, capture_output=True, text=True) data = json.loads(res.stdout) stream = data['streams'][0] w = int(stream.get('width', 1080)); h = int(stream.get('height', 1920)); dur = stream.get('duration') if not dur: cmd_dur = ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "json", path] res_dur = subprocess.run(cmd_dur, capture_output=True, text=True) data_dur = json.loads(res_dur.stdout) dur = data_dur['format'].get('duration', 60) return w, h, float(dur) except: return 1080, 1920, 60.0 def get_font_object(style_font_name, size): target_filename = FONT_FILES_MAP.get(style_font_name, "Vazirmatn.ttf") target_path = os.path.join(FONT_DIR, target_filename) if not os.path.exists(target_path): target_path = os.path.join(FONT_DIR, "Vazirmatn.ttf") if os.path.exists(target_path): return ImageFont.truetype(target_path, size) return ImageFont.load_default() def get_color_tuple(color_str: str, default=(255, 255, 255, 255)): if not color_str or not isinstance(color_str, str): return default color_str = color_str.strip().lower() if color_str.startswith('#'): try: hex_val = color_str.lstrip('#') if len(hex_val) == 6: return tuple(int(hex_val[i:i+2], 16) for i in (0, 2, 4)) + (255,) elif len(hex_val) == 8: return tuple(int(hex_val[i:i+2], 16) for i in (0, 2, 4, 6)) except: pass elif color_str.startswith('rgba'): try: content = color_str[color_str.find('(')+1 : color_str.rfind(')')] parts = [x.strip() for x in content.split(',')] if len(parts) >= 4: r, g, b = int(parts[0]), int(parts[1]), int(parts[2]) a = int(float(parts[3]) * 255) return (r, g, b, a) except: pass elif color_str.startswith('rgb'): try: content = color_str[color_str.find('(')+1 : color_str.rfind(')')] parts = [x.strip() for x in content.split(',')] if len(parts) >= 3: return (int(parts[0]), int(parts[1]), int(parts[2]), 255) except: pass return default def create_subtitle_image(text_parts: list, active_idx: int, width: int, height: int, style: StyleConfig, word_infos: Optional[List[WordInfo]] = None): img = Image.new('RGBA', (width, height), (0, 0, 0, 0)) draw = ImageDraw.Draw(img) text_offset = 0 fs = style.fontSize if style.font == 'dastnevis': text_offset = int(fs * 0.15) if style.name == "music_player" else int(fs * 0.25) elif style.font == 'entazar': text_offset = int(fs * 0.03) if style.name == "music_player" else int(fs * 0.2) elif style.font == 'kamran': text_offset = int(fs * 0.03) if style.name == "music_player" else int(fs * 0.2) elif style.font == 'pinar': text_offset = 0 if style.name == "music_player" else int(fs * -0.2) stroke_val = 0 if style.font == 'entazar' and style.name != "music_player": stroke_val = 0.1 if text_offset != 0 or stroke_val > 0: original_text_method = draw.text def custom_draw_text(xy, text, **kwargs): if stroke_val > 0: current_fill = kwargs.get('fill', (255, 255, 255, 255)) kwargs['stroke_width'] = stroke_val kwargs['stroke_fill'] = current_fill # برای استایل‌های کادردار، انحراف دستی را صفر می‌کنیم تا تراز بصری کادر به هم نخورد actual_offset = text_offset if style.name in ["instagram_box", "alpha_gradient", "karaoke_static", "auto_director"]: actual_offset = 0 original_text_method((xy[0], xy[1] + actual_offset), text, **kwargs) draw.text = custom_draw_text font = get_font_object(style.font, style.fontSize) lines = [] if style.name == "music_player": lines.append(text_parts) else: MAX_WORDS_PER_LINE = 5 current_line = [] for i, word in enumerate(text_parts): current_line.append(word) if len(current_line) == MAX_WORDS_PER_LINE: lines.append(current_line) current_line = [] if current_line: lines.append(current_line) line_metrics = [] max_line_width = 0 for line_words in lines: w_widths = [] l_width = 0 full_line_text = " ".join(line_words) try: l_width = draw.textlength(full_line_text, font=font, direction='rtl', language='fa') except: l_width = font.getlength(full_line_text) if l_width > max_line_width: max_line_width = l_width for w in line_words: try: wl = draw.textlength(w, font=font, direction='rtl', language='fa') except: wl = font.getlength(w) w_widths.append(wl) line_metrics.append({"width": l_width, "words": line_words, "word_widths": w_widths}) safe_word_infos = word_infos if word_infos: safe_word_infos = [] self_handled = ["instagram_box", "alpha_gradient", "music_player", "falling_words"] should_fade_unread = getattr(style, 'fadeUnread', True) and (style.name not in self_handled) should_fade_surr = getattr(style, 'fadeSurrounding', False) and (style.name not in self_handled) curr_t = getattr(style, 'current_render_time', None) for i, w in enumerate(word_infos): w_copy = w.copy() if hasattr(w, 'copy') else w if (i == active_idx) and (style.name in ["auto_director", "karaoke_static", "instagram_box", "alpha_gradient"]) and not getattr(style, 'useActiveColor', True): w_copy.color = "#FFFFFF" is_future = (curr_t is not None and curr_t < w.start) or (curr_t is None and active_idx != -1 and i > active_idx) is_past = (curr_t is not None and curr_t >= w.end) or (curr_t is None and active_idx != -1 and i < active_idx) if (is_future and (should_fade_unread or should_fade_surr)) or (is_past and should_fade_surr): if style.name in ["classic", "progressive_write"]: default_txt_color = style.primaryColor else: default_txt_color = style.styleColors.get(style.name, "#FFFFFF") tr, tg, tb, _ = get_color_tuple(w_copy.color if w_copy.color else default_txt_color, (255, 255, 255, 255)) no_box_styles = ["karaoke_static", "auto_director", "plain_white", "white_outline", "dark_edges"] if style.name in no_box_styles: w_copy.color = f"rgba({tr},{tg},{tb},0.35)" else: bg_key = 'simple_bar_main_box' if style.name == "simple_bar" else style.name bgr, bgg, bgb, _ = get_color_tuple(style.styleBgColors.get(bg_key, style.outlineColor), (0, 0, 0, 255)) fr, fg, fb = int((tr * 0.35) + (bgr * 0.65)), int((tg * 0.35) + (bgg * 0.65)), int((tb * 0.35) + (bgb * 0.65)) w_copy.color = f"rgba({fr},{fg},{fb},255)" safe_word_infos.append(w_copy) style_module = loaded_styles.get(style.name) if style_module and hasattr(style_module, 'draw_frame'): style_module.draw_frame( draw=draw, img=img, width=width, height=height, style_config=style, lines=lines, line_metrics=line_metrics, active_idx=active_idx, font=font, color_parser=get_color_tuple, word_infos=safe_word_infos ) else: y = height - style.marginV draw.text((width/2, y), "Style Error", font=font, fill="red") return img def generate_subtitle_video(data: ProcessRequest, temp_dir: str): target_styles = [ "instagram_box", "dark_edges", "falling_words", "classic", "progressive_write" ] if data.style.name in target_styles: data.style.paddingX = (data.style.paddingX or 0) + 15 list_file = os.path.join(temp_dir, f"{data.file_id}_list.txt") empty_img_path = os.path.join(temp_dir, "empty.png") if not os.path.exists(empty_img_path): Image.new('RGBA', (data.video_width, data.video_height), (0, 0, 0, 0)).save(empty_img_path) sorted_segments = sorted(data.segments, key=lambda x: x.start) if sorted_segments: setattr(data.style, 'total_video_duration', sorted_segments[-1].end) else: setattr(data.style, 'total_video_duration', 1.0) with open(list_file, "w") as f: current_timeline = 0.0 last_generated_image = "empty.png" for idx, seg in enumerate(sorted_segments): start_time = round(max(seg.start, current_timeline), 3) end_time = round(max(seg.end, start_time + 0.1), 3) if end_time - start_time < 0.04: continue gap = round(start_time - current_timeline, 3) if gap > 0.005: # حذف کلمات چسبیده قبلی در زمان سکوت و نمایش ندادن هیچ متنی (نمایش فریم خالی) f.write(f"file 'empty.png'\nduration {gap:.3f}\n") current_timeline = start_time last_generated_image = "empty.png" current_timeline = start_time available_duration = round(end_time - current_timeline, 3) words = [w.word for w in seg.words] if seg.words else seg.text.split() if seg.words and len(words) > 0: seg.words.sort(key=lambda x: x.start) words = [w.word for w in seg.words] SUB_FRAME_DURATION = 0.025 if data.style.name == "falling_words" else 0.05 time_cursor = start_time ANIMATION_DURATION = 0.4 while time_cursor < end_time: active_word_index = -1 for i, w_info in enumerate(seg.words): if time_cursor >= w_info.start and time_cursor < w_info.end: active_word_index = i break setattr(data.style, 'current_render_time', time_cursor) time_into_segment = time_cursor - start_time anim_progress = min(1.0, time_into_segment / ANIMATION_DURATION) setattr(data.style, 'entry_anim_progress', anim_progress) name = f"sub_{data.file_id}_{idx}_{int(time_cursor*1000)}.png" img = create_subtitle_image(words, active_word_index, data.video_width, data.video_height, data.style, word_infos=seg.words) img.save(os.path.join(temp_dir, name)) last_generated_image = name f.write(f"file '{name}'\nduration {SUB_FRAME_DURATION:.3f}\n") time_cursor += SUB_FRAME_DURATION current_timeline = end_time else: name = f"sub_{data.file_id}_{idx}_full.png" img = create_subtitle_image(words, -1, data.video_width, data.video_height, data.style, word_infos=seg.words) img.save(os.path.join(temp_dir, name)) f.write(f"file '{name}'\nduration {available_duration:.3f}\n") last_generated_image = name current_timeline += available_duration f.write(f"file 'empty.png'\nduration 30.0\n") return list_file def process_render_logic(req: ProcessRequest) -> str: # اعمال زمان‌های مرزی دقیق برای هر سگمنت for s in req.segments: if s.words: s.words.sort(key=lambda x: x.start) s.start = s.words[0].start s.end = s.words[-1].end req.segments = [s for s in req.segments if s.end > s.start] req.segments.sort(key=lambda x: x.start) lst = generate_subtitle_video(req, TEMP_DIR) inp = f"{TEMP_DIR}/{req.file_id}.mp4" if not os.path.exists(inp): raise Exception("Input video not found") sub_video_path = f"{TEMP_DIR}/{req.file_id}_sub_render.mov" out = f"{TEMP_DIR}/{req.file_id}_final_{int(time.time())}.mp4" cmd_step1 = ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", lst, "-r", "30", "-s", f"{req.video_width}x{req.video_height}", "-c:v", "png", "-pix_fmt", "rgba", sub_video_path] res1 = subprocess.run(cmd_step1, capture_output=True, text=True) if res1.returncode != 0: raise Exception(f"Subtitle generation failed: {res1.stderr}") cmd_step2 = ["ffmpeg", "-y", "-i", inp, "-i", sub_video_path, "-filter_complex", "[0:v][1:v]overlay=0:0:eof_action=pass[outv]", "-map", "[outv]", "-map", "0:a", "-c:v", "libx264", "-r", "30", "-preset", "ultrafast", "-c:a", "aac", out] res2 = subprocess.run(cmd_step2, capture_output=True, text=True) if res2.returncode != 0: raise Exception(f"Merge failed: {res2.stderr}") if os.path.exists(sub_video_path): os.remove(sub_video_path) return f"/temp/{os.path.basename(out)}" @app.get("/") async def index(): return FileResponse("index.html") @app.get("/api/styles") def get_style_definitions(): return { "styles": style_configs, "templates": style_templates } @app.post("/api/generate-style") def generate_style_api(req: StylePrompt): if not API_KEYS: raise HTTPException(500, "API Keys Missing") for _ in range(3): try: genai.configure(api_key=random.choice(API_KEYS)) model = genai.GenerativeModel(MODEL_NAME) prompt = f"""You are a JSON generator. Create a subtitle style based on: "{req.description}". Return JSON only. Keys: primaryColor, outlineColor, backType (solid/transparent/outline), font (vazir/lalezar/bangers/roboto), fontSize (30-90).""" res = model.generate_content(prompt, generation_config={"response_mime_type": "application/json"}) data = json.loads(clean_json_response(res.text)) return {"primaryColor": data.get("primaryColor", "#FFFFFF"), "outlineColor": data.get("outlineColor", "#000000"), "backType": data.get("backType", "solid"), "font": data.get("font", "vazir"), "fontSize": int(data.get("fontSize", 60))} except: continue return {"primaryColor":"#FFFFFF", "outlineColor":"#000000", "font":"vazir", "fontSize":60, "backType":"solid"} # --- تابع ارسال فایل صوتی به Groq با پرامپت تقویت‌شده تکرارها --- async def transcribe_audio_via_groq(audio_path: str): global groq_key_index if not GROQ_API_KEYS: raise Exception("متغیر محیطی GROQ_API_KEYS در قسمت Secrets تنظیم نشده است.") # انتخاب چرخشی کلید current_key = GROQ_API_KEYS[groq_key_index % len(GROQ_API_KEYS)] groq_key_index += 1 client = Groq(api_key=current_key) def call_groq(): with open(audio_path, "rb") as file: return client.audio.transcriptions.create( file=(os.path.basename(audio_path), file.read()), model="whisper-large-v3", response_format="verbose_json", timestamp_granularities=["word", "segment"], temperature=0.0, language="fa", prompt="سلام سلام، چطوری؟ کلمات دقیقاً همان‌طور که تلفظ می‌شوند، با رعایت حروف اضافه، ساختار عامیانه، محاوره‌ای و شکسته نوشته شوند. حتماً تمامی کلمات تکراری و تکرار پشت سر هم کلمات (مثل سلام سلام، خوب خوب) دقیقاً و دونه به دونه ثبت شوند و به هیچ عنوان حذف یا خلاصه‌سازی نشوند." ) transcription = await asyncio.to_thread(call_groq) return transcription # --- تابع بازگشتی هوشمند برای برش فایل‌های صوتی سنگین از نقاط سکوت --- async def transcribe_audio_recursive(audio_path: str, start_time_offset: float = 0.0) -> list: file_size = os.path.getsize(audio_path) MAX_SIZE = 24 * 1024 * 1024 # 24 MB محدودیت امن if file_size <= MAX_SIZE: transcription = await transcribe_audio_via_groq(audio_path) # استخراج ساختار داده خروجی Groq if hasattr(transcription, "model_dump"): data = transcription.model_dump() elif hasattr(transcription, "dict"): data = transcription.dict() elif isinstance(transcription, dict): data = transcription else: try: data = dict(transcription) except: data = {} raw_word_items = [] segments = data.get("segments", []) words_list = data.get("words", []) if words_list: raw_word_items = words_list elif segments: for seg in segments: text_val = seg.get("text", "").strip() if not text_val: continue if "words" in seg and seg["words"]: raw_word_items.extend(seg["words"]) else: seg_start = float(seg.get("start", 0)) seg_end = float(seg.get("end", 0)) w_list = text_val.split() if w_list: duration = seg_end - seg_start word_duration = duration / max(1, len(w_list)) for i, w_txt in enumerate(w_list): w_start_time = seg_start + (i * word_duration) raw_word_items.append({ "word": w_txt, "start": round(w_start_time, 3), "end": round(w_start_time + word_duration, 3) }) # اعمال آفست زمانی برای برش‌های بعدی که در جای درست قرار بگیرند adjusted_words = [] for w in raw_word_items: if not isinstance(w, dict): word_text = getattr(w, "word", getattr(w, "text", "")) word_start = float(getattr(w, "start", getattr(w, "start_time", 0))) word_end = float(getattr(w, "end", getattr(w, "end_time", 0))) else: word_text = w.get("word", w.get("text", "")) word_start = float(w.get("start", w.get("start_time", 0))) word_end = float(w.get("end", w.get("end_time", 0))) adjusted_words.append({ "word": word_text, "start": round(word_start + start_time_offset, 3), "end": round(word_end + start_time_offset, 3) }) return adjusted_words else: # فایل سنگین‌تر از حد مجاز است، باید برش بخورد print(f"File size {file_size} > 24MB. Splitting audio...") audio = await asyncio.to_thread(AudioSegment.from_mp3, audio_path) mid_point_ms = len(audio) // 2 # جستجو برای پیدا کردن نقطه سکوت در محدوده 40% تا 60% فایل search_start = int(len(audio) * 0.4) search_end = int(len(audio) * 0.6) search_chunk = audio[search_start:search_end] def find_silences(): # تشخیص سکوت (سکوت بالاتر از 500 میلی‌ثانیه با آستانه -40dB) return detect_silence(search_chunk, min_silence_len=500, silence_thresh=-40) silences = await asyncio.to_thread(find_silences) if silences: # انتخاب سکوتی که نزدیک‌ترین فاصله را به وسط بخش مورد جستجو دارد best_silence = silences[len(silences)//2] split_point_ms = search_start + (best_silence[0] + best_silence[1]) // 2 else: # اگر هیچ سکوتی پیدا نشد، دقیقاً از وسط برش می‌زنیم split_point_ms = mid_point_ms part1_path = audio_path.replace(".mp3", f"_part1_{int(time.time()*1000)}_{random.randint(100,999)}.mp3") part2_path = audio_path.replace(".mp3", f"_part2_{int(time.time()*1000)}_{random.randint(100,999)}.mp3") def export_parts(): audio[:split_point_ms].export(part1_path, format="mp3", bitrate="64k") audio[split_point_ms:].export(part2_path, format="mp3", bitrate="64k") await asyncio.to_thread(export_parts) # فراخوانی بازگشتی هوش مصنوعی برای هر دو قسمت words1 = await transcribe_audio_recursive(part1_path, start_time_offset) words2 = await transcribe_audio_recursive(part2_path, start_time_offset + (split_point_ms / 1000.0)) # پاکسازی فایل‌های تکه‌تکه‌شده try: os.remove(part1_path) os.remove(part2_path) except Exception as e: print(f"Cleanup error for split files: {e}") # تجمیع نتایج return words1 + words2 # --- پردازش موقت در صف پس‌زمینه (نسخه اصلاح شده بدون فیلتر متنی و با پشتیبانی از فایل‌های سنگین) --- async def bg_process_upload(task_id: str, raw_path: str, fixed_path: str, audio_path: str): job = upload_jobs.get(task_id) if not job: return job.status = UploadJobStatus.PROCESSING try: # ۱. تبدیل ویدیوی خام به فرمت استاندارد 30fps h264 proc1 = await asyncio.create_subprocess_exec( "ffmpeg", "-y", "-i", raw_path, "-r", "30", "-c:v", "libx264", "-preset", "ultrafast", "-c:a", "copy", fixed_path, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) await proc1.communicate() w, h, total_duration = get_video_info(fixed_path) # ۲. استخراج مستقیم صدا به صورت MP3 با نرخ بیت 64k proc2 = await asyncio.create_subprocess_exec( "ffmpeg", "-y", "-i", fixed_path, "-vn", "-acodec", "libmp3lame", "-ar", "16000", "-ac", "1", "-b:a", "64k", audio_path, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) await proc2.communicate() print(f"--- [BG Task {task_id}] Transcribing audio with Recursive Groq Logic ---") # ۳. ارسال فایل صوتی به تابع هوشمند که فایل‌های سنگین را برش می‌دهد raw_word_items = await transcribe_audio_recursive(audio_path) # پاکسازی فایل‌های موقت پایه if os.path.exists(audio_path): try: os.remove(audio_path) except: pass if os.path.exists(raw_path): try: os.remove(raw_path) except: pass if not raw_word_items: job.status = UploadJobStatus.FAILED job.error_message = "تبدیل گفتار به متن ناموفق بود. هیچ متنی در فایل شناسایی نشد." return # ۶. گروه‌بندی کلمات به صورت دسته‌های ۵ تایی استاندارد final_segs = [] chunk_size = 5 for k in range(0, len(raw_word_items), chunk_size): sub = raw_word_items[k:k+chunk_size] if not sub: continue sc_words = [] for w in sub: sc_words.append({ "word": w["word"], "start": round(w["start"], 3), "end": round(w["end"], 3), "highlight": False }) if sc_words: final_segs.append({ "start": sc_words[0]["start"], "end": sc_words[-1]["end"], "text": " ".join([x["word"] for x in sc_words]), "words": sc_words }) suggested_style = {"primaryColor": "#FFFFFF", "outlineColor": "#000000", "font": "vazir", "fontSize": 60, "backType": "solid"} job.result = { "file_id": task_id, "url": f"/temp/{task_id}.mp4", "width": w, "height": h, "segments": final_segs, "suggested_style": suggested_style } job.status = UploadJobStatus.COMPLETED except Exception as e: if os.path.exists(audio_path): try: os.remove(audio_path) except: pass if os.path.exists(raw_path): try: os.remove(raw_path) except: pass job.status = UploadJobStatus.FAILED job.error_message = f"Processing Error: {str(e)}" @app.post("/api/upload") async def upload(background_tasks: BackgroundTasks, file: UploadFile = File(...)): task_id = str(uuid.uuid4())[:8] ext = file.filename.split('.')[-1] raw_path = f"{TEMP_DIR}/{task_id}_raw.{ext}" fixed_path = f"{TEMP_DIR}/{task_id}.mp4" audio_path = f"{TEMP_DIR}/{task_id}.mp3" try: with open(raw_path, "wb") as f: shutil.copyfileobj(file.file, f) upload_jobs[task_id] = UploadJob(task_id, file.filename) background_tasks.add_task( bg_process_upload, task_id, raw_path, fixed_path, audio_path ) return {"task_id": task_id, "status": UploadJobStatus.QUEUED} except Exception as e: if os.path.exists(raw_path): try: os.remove(raw_path) except: pass raise HTTPException(500, f"Failed to queue upload task: {str(e)}") @app.get("/api/upload-status/{task_id}") async def get_upload_status(task_id: str): job = upload_jobs.get(task_id) if not job: raise HTTPException(404, "Upload job not found") response = {"task_id": job.id, "status": job.status} if job.status == UploadJobStatus.COMPLETED: response["result"] = job.result elif job.status == UploadJobStatus.FAILED: response["error"] = job.error_message return response import glob @app.delete("/api/delete-project/{file_id}") async def delete_project_files(file_id: str): if "/" in file_id or "\\" in file_id or ".." in file_id: raise HTTPException(400, "Invalid file_id") raw_video_path = os.path.join(TEMP_DIR, f"{file_id}.mp4") if os.path.exists(raw_video_path): try: os.remove(raw_video_path) except Exception as e: print(f"Error removing raw video: {e}") final_patterns = os.path.join(TEMP_DIR, f"{file_id}_final_*.mp4") for f_path in glob.glob(final_patterns): try: os.remove(f_path) except Exception as e: print(f"Error removing final video {f_path}: {e}") return {"status": "success", "message": "Files deleted successfully"} @app.post("/api/reupload") async def reupload_video(file: UploadFile = File(...), file_id: str = Form(...)): if not file_id or '/' in file_id or '\\' in file_id: raise HTTPException(400, "Invalid file_id") target_path = os.path.join(TEMP_DIR, f"{file_id}.mp4") try: with open(target_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) except Exception as e: raise HTTPException(500, f"Could not save file: {e}") finally: await file.close() return {"status": "success", "message": f"File {file_id}.mp4 restored."} @app.post("/api/enqueue-render") async def enqueue_render(req: ProcessRequest): if not os.path.exists(os.path.join(TEMP_DIR, f"{req.file_id}.mp4")): return JSONResponse(status_code=200, content={"error": "Video not found", "error_code": "VIDEO_NOT_FOUND"}) job_id = str(uuid.uuid4()) jobs_db[job_id] = Job(job_id, req) await render_queue.put(job_id) return {"job_id": job_id, "status": JobStatus.QUEUED} @app.get("/api/job-status/{job_id}") async def get_job_status(job_id: str): job = jobs_db.get(job_id) if not job: raise HTTPException(404, "Job not found") response = {"job_id": job.id, "status": job.status} if job.status == JobStatus.QUEUED: response["queue_position"] = sum(1 for j in jobs_db.values() if j.status == JobStatus.QUEUED and j.created_at < job.created_at) + 1 elif job.status == JobStatus.COMPLETED: response["url"] = job.result_url # اصلاح نام متغیر COMPLETED به JobStatus.COMPLETED elif job.status == JobStatus.FAILED: response["error"] = job.error_message return response