| import gradio as gr |
| import os, requests, json, subprocess |
| import boto3 |
| from groq import Groq |
| import google.generativeai as genai |
|
|
| |
| GROQ_KEY = os.getenv("GROQ_API_KEY") |
| GEMINI_KEY = os.getenv("GEMINI_API_KEY") |
| R2_ACCESS = os.getenv("R2_ACCESS") |
| R2_SECRET = os.getenv("R2_SECRET") |
| R2_BUCKET = os.getenv("R2_BUCKET") |
| R2_BUCKET_URL = os.getenv("R2_BUCKET_URL") |
| R2_PUBLIC = os.getenv("R2_PUBLIC_URL") |
| LWS_WEBHOOK = "https://opus.dostodgroup.com/webhook.php?token=change_moi_12345_opus" |
|
|
| genai.configure(api_key=GEMINI_KEY) |
| groq_client = Groq(api_key=GROQ_KEY) |
|
|
| def upload_r2(local_path, remote_name): |
| s3 = boto3.client('s3', |
| endpoint_url=R2_BUCKET_URL, |
| aws_access_key_id=R2_ACCESS, |
| aws_secret_access_key=R2_SECRET |
| ) |
| s3.upload_file(local_path, R2_BUCKET, remote_name) |
| return f"{R2_PUBLIC}/{remote_name}" |
|
|
| def factory(job_id, video_url): |
| try: |
| tmp_in = f"/tmp/{job_id}.mp4" |
| gr.Info(f"Download {video_url}...") |
| with requests.get(video_url, stream=True, timeout=120) as r: |
| r.raise_for_status() |
| with open(tmp_in, 'wb') as f: |
| for chunk in r.iter_content(chunk_size=1024*1024): |
| if chunk: f.write(chunk) |
|
|
| gr.Info("Transcription Whisper...") |
| from faster_whisper import WhisperModel |
| model = WhisperModel("small", device="cpu", compute_type="int8") |
| segments, _ = model.transcribe(tmp_in, language="fr") |
| full_text = " ".join([s.text for s in segments])[:8000] |
|
|
| if not full_text.strip(): |
| return "❌ Transcription vide" |
|
|
| gr.Info("Analyse virale Groq...") |
| prompt = f"""Tu es Opus.pro. Trouve 10 clips viraux 30-60s max score viral. |
| Format JSON STRICT sans texte autour, juste le tableau: [{{"start":12.5,"end":45.2,"title":"Hook choc","score":92}}] |
| Transcription: {full_text}""" |
|
|
| comp = groq_client.chat.completions.create( |
| model="llama-3.3-70b-versatile", |
| messages=[{"role":"user","content":prompt}], |
| temperature=0.7 |
| ) |
| txt = comp.choices[0].message.content.replace("```json","").replace("```","").strip() |
| clips = json.loads(txt) |
|
|
| final_for_lws = [] |
| gr.Info(f"{len(clips)} clips detectes, rendu 1080p...") |
| for i, c in enumerate(clips[:10]): |
| out = f"/tmp/{job_id}_clip{i+1}_1080p.mp4" |
| cmd = [ |
| "ffmpeg","-ss",str(c['start']),"-to",str(c['end']),"-i",tmp_in, |
| "-vf","crop=1080:1920:(in_w-1080)/2:0,scale=1080:1920:flags=lanczos", |
| "-c:v","libx264","-crf","18","-preset","ultrafast","-c:a","aac","-b:a","128k","-y",out |
| ] |
| subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
|
|
| if os.path.exists(out) and os.path.getsize(out) > 10000: |
| remote = f"finales/{job_id}_clip{i+1}_1080p.mp4" |
| r2_url = upload_r2(out, remote) |
| final_for_lws.append({ |
| "url": r2_url, |
| "title": c.get('title',''), |
| "score": c.get('score',0) |
| }) |
|
|
| |
| gr.Info(f"Envoi {len(final_for_lws)} clips vers LWS...") |
| requests.post(LWS_WEBHOOK, json={"job_id": job_id, "clips": final_for_lws}, timeout=60) |
|
|
| return f"✅ {len(final_for_lws)} clips 1080p envoyés sur opus.dostodgroup.com/videos/finales/\n\n" + "\n".join([f"{x['score']} - {x['title']} -> {x['url']}" for x in final_for_lws]) |
|
|
| except Exception as e: |
| import traceback |
| return f"❌ Erreur: {str(e)}\n{traceback.format_exc()}" |
|
|
| with gr.Blocks(title="OPUS 1080p Factory") as demo: |
| gr.Markdown("# 🏭 OPUS 1080p - Usine Gratuite ZeroGPU\nR2 -> 10 clips viraux 1080p -> LWS opus.dostodgroup.com") |
| job_id = gr.Textbox(label="job_id (ex: job_123)", value="job_test_1") |
| video_url = gr.Textbox(label="URL video longue R2 (https://pub-.../video.mp4)", placeholder="https://pub-xxx.r2.dev/uploads/...") |
| btn = gr.Button("🚀 RENDRE 10 CLIPS 1080p", variant="primary") |
| output = gr.Textbox(label="Logs", lines=15) |
| btn.click(factory, [job_id, video_url], output) |
|
|
| demo.launch() |
