Create app.py

app.py

@@ -0,0 +1,99 @@
+import gradio as gr
+import os, requests, json, subprocess
+import boto3
+from groq import Groq
+import google.generativeai as genai
+
+# SECRETS HF - Settings > Variables and Secrets
+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")  # https://xxx.r2.cloudflarestorage.com
+R2_PUBLIC = os.getenv("R2_PUBLIC_URL")  # https://pub-xxx.r2.dev
+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)
+                })
+
+        # Envoie à LWS
+        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()