import subprocess try: subprocess.run(["sed", "-i", 's/rights="none"/rights="read|write"/g', "/etc/ImageMagick-6/policy.xml"], capture_output=True) subprocess.run(["sed", "-i", 's/rights="none"/rights="read|write"/g', "/etc/ImageMagick-7/policy.xml"], capture_output=True) except: pass import os, asyncio, time, random, requests, threading import PIL.Image if not hasattr(PIL.Image, "ANTIALIAS"): PIL.Image.ANTIALIAS = PIL.Image.LANCZOS import gradio as gr from groq import Groq import edge_tts from moviepy.editor import VideoFileClip, AudioFileClip, TextClip, CompositeVideoClip NICHO = "NICHO_PLACEHOLDER" import json as _json try: CONFIG = _json.loads("""CONFIG_PLACEHOLDER""") except: CONFIG = {} # Mapeo de voces edge-tts VOCES = { "es": {"masculina": "es-ES-AlvaroNeural", "femenina": "es-ES-ElviraNeural"}, "en": {"masculina": "en-US-GuyNeural", "femenina": "en-US-JennyNeural"}, "pt": {"masculina": "pt-BR-AntonioNeural", "femenina": "pt-BR-FranciscaNeural"}, "fr": {"masculina": "fr-FR-HenriNeural", "femenina": "fr-FR-DeniseNeural"} } CLIENT_EMAIL = "EMAIL_PLACEHOLDER" HF_REPO = "REPO_PLACEHOLDER" # Descargar token.json del repo si existe try: import urllib.request urllib.request.urlretrieve(f"https://huggingface.co/spaces/{HF_REPO}/resolve/main/token.json", "token.json") print("Token descargado", flush=True) except Exception as e: print(f"Sin token aún: {e}", flush=True) GROQ_KEY = "GROQ_PLACEHOLDER" TIPO_VIDEO = "TIPO_PLACEHOLDER" PEXELS_KEY = "PEXELS_PLACEHOLDER" os.makedirs("media_cliente", exist_ok=True) LOGS = [] STATUS = "Iniciando..." def log(msg): t = time.strftime("%H:%M:%S") LOGS.append(f"[{t}] {msg}") if len(LOGS) > 40: LOGS.pop(0) print(msg, flush=True) def generar_guion(): _tono_actual = CONFIG.get("tono", "educativo") client = Groq(api_key=GROQ_KEY) _canal_url = CONFIG.get("canal_promocionar", "").strip() _contexto_canal = f" El video debe ser un teaser que invite a ver el canal {_canal_url}." if _canal_url else "" res = client.chat.completions.create( messages=[{"role":"user","content":f"Escribe SOLO el texto a narrar en un video de 60 segundos sobre {NICHO}. Tono: {_tono_actual}.{_contexto_canal} NO incluyas indicaciones como narrador dice, voz en off, intro, outro, ni nada que no sea el texto a leer. Directo al grano, sin simbolos, sin hashtags, sin emojis."}], model="llama-3.3-70b-versatile", temperature=0.82 ) guion_base = res.choices[0].message.content.strip() frase_i = CONFIG.get("frase_inicio", "").strip() frase_f = CONFIG.get("frase_final", "").strip() partes = [] if frase_i: partes.append(frase_i) partes.append(guion_base) if frase_f: partes.append(frase_f) return ". ".join(partes) async def pipeline(): global STATUS while True: try: STATUS = "Generando guion..." log(f"Nuevo video: {NICHO}") guion = generar_guion() log("Guion listo") STATUS = "Generando voz..." voz_ia_on = CONFIG.get("voz_ia", True) if voz_ia_on: idioma_v = CONFIG.get("idioma", "es") genero_v = CONFIG.get("voz_genero", "masculina") voz_seleccionada = VOCES.get(idioma_v, VOCES["es"]).get(genero_v, "es-ES-AlvaroNeural") vel = CONFIG.get("velocidad_voz", 1.0) rate_str = f"+{int((vel-1)*100)}%" if vel >= 1 else f"-{int((1-vel)*100)}%" await edge_tts.Communicate(guion, voz_seleccionada, rate=rate_str).save("voz.mp3") else: # Sin voz IA: crear audio en silencio import subprocess subprocess.run(["ffmpeg","-y","-f","lavfi","-i","anullsrc=r=44100:cl=mono","-t","60","voz.mp3"], capture_output=True) log("Voz lista") STATUS = "Preparando fondos..." fondos = [] # Revisar si hay media propia del cliente archivos_propios = [] if os.path.exists("media_cliente"): for f in sorted(os.listdir("media_cliente")): if f.lower().endswith((".mp4", ".mov", ".jpg", ".jpeg", ".png", ".webp")): archivos_propios.append(f"media_cliente/{f}") if archivos_propios: random.shuffle(archivos_propios) fondos = archivos_propios[:6] log(f"Usando {len(fondos)} archivos del cliente") else: STATUS = "Descargando fondo..." h = {"Authorization": PEXELS_KEY} r = requests.get(f"https://api.pexels.com/videos/search?query={NICHO}&per_page=15", headers=h) vids = r.json().get("videos", []) if vids: random.shuffle(vids) for i, v in enumerate(vids[:6]): try: u = v["video_files"][0]["link"] with open(f"fondo{i}.mp4","wb") as f: f.write(requests.get(u, timeout=30).content) fondos.append(f"fondo{i}.mp4") except: pass log(f"Fondos listos: {len(fondos)}") STATUS = "Renderizando video..." from moviepy.editor import AudioFileClip as AC, VideoFileClip as VC, concatenate_videoclips voz = AC("voz.mp3") if TIPO_VIDEO == "short": clips_lista = [] for fp in fondos: try: cc = VC(fp).without_audio() if cc.w/cc.h > 1080/1920: cc = cc.resize(height=1920) else: cc = cc.resize(width=1080) cc = cc.crop(x_center=cc.w/2, y_center=cc.h/2, width=1080, height=1920) cc = cc.loop(duration=voz.duration/max(len(fondos),1)) clips_lista.append(cc) except: pass clip = concatenate_videoclips(clips_lista, method="compose") clip = clip.crop(x_center=clip.w/2, width=1080, height=1920) else: clips_lista = [] for fp in fondos: try: cc = VC(fp).without_audio().resize(width=1280) cc = cc.loop(duration=voz.duration/max(len(fondos),1)) clips_lista.append(cc) except: pass clip = concatenate_videoclips(clips_lista, method="compose") palabras = guion.split() frases = [" ".join(palabras[i:i+6]) for i in range(0,len(palabras),6)] t_f = voz.duration / max(len(frases),1) if not CONFIG.get("subtitulos", True): txts = [] else: txts = [TextClip(t.upper(), fontsize={"tiktok":72,"clasico":50,"minimal":40}.get(CONFIG.get("estilo_subtitulos","tiktok"),72), color=CONFIG.get("color_subtitulos","#D4AF37"), font="DejaVu-Sans-Bold", stroke_color="black", stroke_width=2, method="caption", size=(clip.w*0.8,None)).set_start(i*t_f).set_duration(t_f).set_pos(("center", {"arriba":0.15,"centro":0.5,"abajo":0.75}.get(CONFIG.get("posicion_subtitulos","abajo"),0.75)), relative=True) for i,t in enumerate(frases)] final = CompositeVideoClip([clip]+txts).set_audio(voz) final.write_videofile("out.mp4", fps=15, codec="libx264", preset="ultrafast", logger=None, threads=2) log("Video renderizado") STATUS = "Esperando conexión YouTube..." while not os.path.exists("token.json"): await asyncio.sleep(30) STATUS = "Subiendo a YouTube..." try: import json, google.oauth2.credentials, googleapiclient.discovery, googleapiclient.http with open("token.json") as f: tok = json.load(f) creds = google.oauth2.credentials.Credentials( token=tok["access_token"], refresh_token=tok.get("refresh_token"), token_uri="https://oauth2.googleapis.com/token", client_id=tok.get("client_id",""), client_secret=tok.get("client_secret","") ) yt = googleapiclient.discovery.build("youtube","v3",credentials=creds) titulo = guion[:80].split(".")[0] req_yt = yt.videos().insert( part="snippet,status", body={ "snippet":{"title":titulo,"description":guion + ("\n\n" + CONFIG.get("link_descripcion","") if CONFIG.get("link_descripcion") else ""),"categoryId":"22"}, "status":{"privacyStatus":"public"} }, media_body=googleapiclient.http.MediaFileUpload("out.mp4",mimetype="video/mp4",resumable=True) ) resp = None while resp is None: _, resp = req_yt.next_chunk() log(f"Subido: {titulo[:40]}") except FileNotFoundError: log("Sin token.json, saltando subida") except Exception as ey: log(f"Error YouTube: {ey}") STATUS = "Esperando siguiente ciclo..." await asyncio.sleep(14400) except Exception as e: log(f"Error: {e}") STATUS = f"Error: {str(e)[:40]}" await asyncio.sleep(600) def start(): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) time.sleep(30); loop.run_until_complete(pipeline()) threading.Thread(target=start, daemon=True).start() CONTADOR_VIDEOS = 0 ULTIMO_VIDEO = "Ninguno aún" def youtube_conectado(): return os.path.exists("token.json") def get_yt_status(): if youtube_conectado(): return "✅ YouTube conectado — publicando automáticamente" return "❌ YouTube NO conectado — conecta tu canal para publicar" def get_stats(): return f"📊 Videos publicados: {CONTADOR_VIDEOS} | 🎬 Último: {ULTIMO_VIDEO} | ⚡ Estado: {STATUS}" from fastapi import FastAPI from fastapi.responses import HTMLResponse import uvicorn app = FastAPI() @app.get("/", response_class=HTMLResponse) async def panel(): yt_status = "✅ YouTube conectado" if os.path.exists("token.json") else "❌ YouTube NO conectado" logs_html = "
".join(LOGS[-20:]) if LOGS else "Iniciando..." return f"""TubeBot

🤖 TUBEBOT — {NICHO.upper()}

Tu canal publica solo cada 4 horas

📊 Videos publicados: {CONTADOR_VIDEOS}
🎬 Último video: {ULTIMO_VIDEO}
⚡ Estado: {STATUS}
📺 YouTube: {yt_status}
⏰ Próximo video en: --:--:--

📜 Logs en tiempo real

{logs_html}

💡 ¿Problemas? WhatsApp: +34656691085

""" uvicorn.run(app, host="0.0.0.0", port=7860)