File size: 4,320 Bytes
3c66a95 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 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() |