import streamlit as st import requests import edge_tts import asyncio import os import subprocess import re import nest_asyncio import json import shlex import time import html from typing import List, Optional nest_asyncio.apply() # CONFIG st.set_page_config( page_title="Islamic Shorts Creator", page_icon="πŸ•Œ", layout="centered", initial_sidebar_state="collapsed" ) CHANNEL_NAME = "Abubakar Daily Islamic Shorts" LOGO_FILE = "logo.png" FREEMODEL_KEY = os.getenv("FREEMODEL_API_KEY", "").strip() PEXELS_KEY = os.getenv("PEXELS_API_KEY", "").strip() # freemodel.dev config FREEMODEL_BASE = "https://api.freemodel.dev/v1" FREEMODEL_MODEL = "gpt-5.5" if "final_video_path" not in st.session_state: st.session_state.final_video_path = None if "script_text" not in st.session_state: st.session_state.script_text = "" if "metadata" not in st.session_state: st.session_state.metadata = {} if "last_metadata_file" not in st.session_state: st.session_state.last_metadata_file = "" if "thumbnail_path" not in st.session_state: st.session_state.thumbnail_path = None if "topic_suggestions" not in st.session_state: st.session_state.topic_suggestions = [] if "nasheed_path" not in st.session_state: st.session_state.nasheed_path = None if "video_history" not in st.session_state: # load_history defined later β€” safe inline load here try: import json as _json _hf = "/tmp/video_history.json" if os.path.exists(_hf): with open(_hf, "r", encoding="utf-8") as _f: st.session_state.video_history = _json.load(_f) else: st.session_state.video_history = [] except: st.session_state.video_history = [] if "export_paths" not in st.session_state: st.session_state.export_paths = {} if "batch_results" not in st.session_state: st.session_state.batch_results = [] if "yt_status" not in st.session_state: st.session_state.yt_status = "" # UI styling st.markdown(""" """, unsafe_allow_html=True) # HERO st.markdown("""
بِسْمِ Ψ§Ω„Ω„ΩŽΩ‘Ω‡Ω Ψ§Ω„Ψ±ΩŽΩ‘Ψ­Ω’Ω…ΩŽΩ†Ω Ψ§Ω„Ψ±ΩŽΩ‘Ψ­ΩΩŠΩ…Ω
Islamic Shorts Creator
Abubakar Daily Islamic Channel
""", unsafe_allow_html=True) # CORE FUNCTIONS def call_freemodel(topic, lang): prompt = ( f"Write a 35-second spiritual Islamic script in {lang} about '{topic}'. " "Write like an imam giving a heartfelt short reminder. " "Use vivid imagery and short powerful sentences. " "Pause naturally between ideas. " "Output ONLY the spoken text. No titles, no hashtags, no stage directions. " "End with a short thought-provoking question to encourage comments." ) try: r = requests.post( f"{FREEMODEL_BASE}/chat/completions", headers={ "Authorization": f"Bearer {FREEMODEL_KEY}", "Content-Type": "application/json" }, json={ "model": FREEMODEL_MODEL, "messages": [{"role": "user", "content": prompt}], "max_tokens": 1024 }, timeout=60 ) return r.json()['choices'][0]['message']['content'].strip() except Exception as e: return f"Error: {e}" def call_freemodel_metadata(topic, lang, script_text): prompt = ( "You are a metadata assistant. Given the following short Islamic spoken script, " "produce a JSON object with keys: title (max 60 chars), description (50-150 words), " "hashtags (an array of 5 trending hashtags, include the # symbol). " "Output ONLY valid JSON and nothing else.\n\n" f"Language: {lang}\nTopic: {topic}\n\nScript:\n{script_text}\n" ) try: r = requests.post( f"{FREEMODEL_BASE}/chat/completions", headers={ "Authorization": f"Bearer {FREEMODEL_KEY}", "Content-Type": "application/json" }, json={ "model": FREEMODEL_MODEL, "messages": [{"role": "user", "content": prompt}], "max_tokens": 1024 }, timeout=60 ) raw = r.json()['choices'][0]['message']['content'].strip() try: return json.loads(raw) except Exception: jstart = raw.find("{") jend = raw.rfind("}") if jstart != -1 and jend != -1: try: return json.loads(raw[jstart:jend+1]) except Exception: return {"title": "", "description": raw, "hashtags": []} return {"title": "", "description": raw, "hashtags": []} except Exception as e: return {"title": "", "description": f"Error: {e}", "hashtags": []} # ════════════════════════════════════════════ # FEATURE 1: AI Topic Suggestions # ════════════════════════════════════════════ def get_topic_suggestions(lang: str) -> list: prompt = ( f"Give me 10 trending Islamic YouTube Shorts topic ideas in {lang}. " "Each topic should be short (3-6 words), spiritually engaging, and suitable for a 30-40 second video. " "Output ONLY a JSON array of 10 strings. No explanation, no numbering, no extra text." ) try: r = requests.post( f"{FREEMODEL_BASE}/chat/completions", headers={"Authorization": f"Bearer {FREEMODEL_KEY}", "Content-Type": "application/json"}, json={"model": FREEMODEL_MODEL, "messages": [{"role": "user", "content": prompt}], "max_tokens": 512}, timeout=30 ) raw = r.json()['choices'][0]['message']['content'].strip() raw = re.sub(r'^```json|^```|```$', '', raw, flags=re.MULTILINE).strip() return json.loads(raw) except Exception as e: print(f"Topic suggestion error: {e}") return [] # ════════════════════════════════════════════ # FEATURE 2: Thumbnail Generator # ════════════════════════════════════════════ def generate_thumbnail(title: str, aspect: str, out_path: str = "/tmp/thumbnail.jpg") -> bool: try: from PIL import Image, ImageDraw, ImageFont import textwrap if aspect == "9:16": w, h = 720, 1280 else: w, h = 1280, 720 # Background gradient (dark blue β†’ black) img = Image.new("RGB", (w, h), (5, 14, 24)) draw = ImageDraw.Draw(img) # Gold gradient overlay at top for y in range(h // 3): alpha = int(80 * (1 - y / (h / 3))) draw.line([(0, y), (w, y)], fill=(212, 175, 55, alpha)) # Decorative border border = 18 draw.rectangle([border, border, w - border, h - border], outline=(212, 175, 55), width=3) draw.rectangle([border + 8, border + 8, w - border - 8, h - border - 8], outline=(212, 175, 55, 80), width=1) # Bismillah Arabic text at top try: font_ar = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 48) font_title = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 64 if aspect == "9:16" else 54) font_sub = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 32) except: font_ar = ImageFont.load_default() font_title = font_ar font_sub = font_ar # Bismillah bism = "Abubakar Daily Islamic Shorts" bbox = draw.textbbox((0, 0), bism, font=font_sub) bw = bbox[2] - bbox[0] draw.text(((w - bw) // 2, border + 30), bism, font=font_sub, fill=(212, 175, 55)) # Gold divider line draw.line([(w // 4, border + 80), (3 * w // 4, border + 80)], fill=(212, 175, 55), width=2) # Main title β€” wrapped max_chars = 20 if aspect == "9:16" else 35 lines = textwrap.wrap(title.upper(), width=max_chars) total_h = len(lines) * 80 y_start = (h - total_h) // 2 - 40 for i, line in enumerate(lines): bbox = draw.textbbox((0, 0), line, font=font_title) lw = bbox[2] - bbox[0] x = (w - lw) // 2 y = y_start + i * 80 # Shadow draw.text((x + 3, y + 3), line, font=font_title, fill=(0, 0, 0)) # Gold text draw.text((x, y), line, font=font_title, fill=(212, 175, 55)) # Bottom gold bar draw.rectangle([0, h - 80, w, h], fill=(212, 175, 55, 40)) channel = "πŸ•Œ Islamic Shorts" bbox = draw.textbbox((0, 0), channel, font=font_sub) cw = bbox[2] - bbox[0] draw.text(((w - cw) // 2, h - 58), channel, font=font_sub, fill=(255, 255, 255)) img.save(out_path, "JPEG", quality=95) return os.path.exists(out_path) except Exception as e: print(f"Thumbnail error: {e}") return False # ════════════════════════════════════════════ # FEATURE 3: Real Nasheed from archive.org # ════════════════════════════════════════════ NASHEED_URLS = [ "https://archive.org/download/NasheedCollection/01-SubhanAllah.mp3", "https://archive.org/download/islamicnasheeds/nasheed_01.mp3", "https://archive.org/download/free-islamic-nasheeds/nasheed_background_01.mp3", ] def download_nasheed(out_path: str = "/tmp/nasheed.mp3") -> str: """Try to download a real nasheed; fallback to sine tone.""" if os.path.exists(out_path) and os.path.getsize(out_path) > 50000: return out_path # already cached for url in NASHEED_URLS: try: r = requests.get(url, timeout=20, stream=True) if r.status_code == 200: with open(out_path, "wb") as f: for chunk in r.iter_content(8192): if chunk: f.write(chunk) if os.path.getsize(out_path) > 50000: return out_path except Exception as e: print(f"Nasheed download failed ({url}): {e}") return "" # will fallback to sine def mix_background_nasheed(voice_path: str, nasheed_path: str, out_path: str = "/tmp/mixed_nasheed.mp3") -> str: """Mix voice with real nasheed at low volume, loop nasheed as needed.""" try: dur = get_tts_duration(voice_path) if not nasheed_path or not os.path.exists(nasheed_path): return voice_path mixed = out_path subprocess.run([ "ffmpeg", "-y", "-i", voice_path, "-stream_loop", "-1", "-i", nasheed_path, "-filter_complex", f"[1:a]volume=0.15,atrim=0:{dur:.3f}[bg];[0:a][bg]amix=inputs=2:duration=first[aout]", "-map", "[aout]", "-t", f"{dur:.3f}", mixed ], capture_output=True, text=True) return mixed if os.path.exists(mixed) and os.path.getsize(mixed) > 0 else voice_path except Exception as e: print(f"Nasheed mix error: {e}") return voice_path # ════════════════════════════════════════════ # FEATURE: Video Preview Card # ════════════════════════════════════════════ def render_preview_card(script: str, metadata: dict, aspect: str): """Show script + metadata preview before generating video.""" title = metadata.get("title", "") if metadata else "" desc = metadata.get("description", "") if metadata else "" tags = metadata.get("hashtags", []) if metadata else [] tags_str = " ".join(tags) if isinstance(tags, list) else str(tags) aspect_icon = "πŸ“±" if aspect == "9:16" else "πŸ–₯️" st.markdown(f"""
{aspect_icon} VIDEO PREVIEW β€” {aspect}
{html.escape(title)}
{html.escape(script[:200])}{'...' if len(script) > 200 else ''}
{html.escape(tags_str)}
""", unsafe_allow_html=True) # ════════════════════════════════════════════ # FEATURE: History / Archive # ════════════════════════════════════════════ HISTORY_FILE = "/tmp/video_history.json" def load_history() -> list: try: if os.path.exists(HISTORY_FILE): with open(HISTORY_FILE, "r", encoding="utf-8") as f: return json.load(f) except: pass return [] def save_to_history(topic: str, lang: str, metadata: dict, video_path: str, thumb_path: str): history = load_history() entry = { "id": int(time.time()), "date": time.strftime("%Y-%m-%d %H:%M"), "topic": topic, "lang": lang, "title": metadata.get("title", topic) if metadata else topic, "video_path": video_path, "thumb_path": thumb_path, } history.insert(0, entry) history = history[:20] # keep last 20 only try: with open(HISTORY_FILE, "w", encoding="utf-8") as f: json.dump(history, f, ensure_ascii=False, indent=2) except Exception as e: print(f"History save error: {e}") # ════════════════════════════════════════════ # FEATURE: Multi-Platform Export # ════════════════════════════════════════════ EXPORT_FORMATS = { "YouTube Shorts / TikTok / Reels (9:16 720x1280)": ("720", "1280", "9x16"), "Instagram Square (1:1 1080x1080)": ("1080", "1080", "1x1"), "Facebook / YouTube (16:9 1280x720)": ("1280", "720", "16x9"), "Instagram Story (9:16 1080x1920)": ("1080", "1920", "9x16_hd"), } def export_video_format(src_path: str, width: str, height: str, suffix: str) -> str: out = f"/tmp/export_{suffix}.mp4" subprocess.run([ "ffmpeg", "-y", "-i", src_path, "-vf", f"scale={width}:{height}:force_original_aspect_ratio=decrease," f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:black", "-c:v", "libx264", "-preset", "ultrafast", "-c:a", "copy", out ], capture_output=True, text=True) return out if os.path.exists(out) else "" async def tts(text, voice, path): clean = re.sub(r'[#*()<>]', '', text) # Slower rate + slightly lower pitch = more imam-like delivery comm = edge_tts.Communicate(clean, voice, rate="-10%", pitch="-5Hz") await comm.save(path) _tts = tts def mix_background_music(voice_path: str, out_path: str, music_volume: float = 0.18) -> str: try: dur = get_tts_duration(voice_path) bg_audio = "/tmp/bg_ambient.mp3" # Soft 432Hz ambient tone with echo β€” calming Islamic atmosphere subprocess.run([ "ffmpeg", "-y", "-f", "lavfi", "-i", f"sine=frequency=432:duration={dur:.3f}", "-af", "volume=0.06,aecho=0.8:0.88:60:0.4", bg_audio ], capture_output=True, text=True) if not os.path.exists(bg_audio): return voice_path mixed = "/tmp/mixed_audio.mp3" subprocess.run([ "ffmpeg", "-y", "-i", voice_path, "-i", bg_audio, "-filter_complex", f"[1:a]volume={music_volume}[bg];[0:a][bg]amix=inputs=2:duration=first[aout]", "-map", "[aout]", "-t", f"{dur:.3f}", mixed ], capture_output=True, text=True) return mixed if os.path.exists(mixed) else voice_path except Exception as e: print(f"Music mix error: {e}") return voice_path def get_tts_duration(path: str) -> float: try: result = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", path], capture_output=True, text=True ) return float(result.stdout.strip()) except: return 35.0 def escape_ffmpeg_text(text: str) -> str: """Escape text safely for ffmpeg drawtext filter""" # Remove characters that break ffmpeg filter syntax text = re.sub(r"[':=\\,\[\]@{}()]", " ", text) # Collapse multiple spaces text = re.sub(r" +", " ", text).strip() return text def build_professional_captions(words, audio_dur, tw, th, aspect): """ Professional caption style: - Bottom third positioning (like YouTube/Netflix) - Large bold white text with strong black outline (no semi-transparent box) - 4-5 words per line max - Two lines at a time - Smooth word-timed display """ if aspect == "9:16": fontsize = 62 y_pos = int(th * 0.72) # 72% from top = lower third for portrait chunk_size = 4 else: fontsize = 52 y_pos = int(th * 0.78) # lower third for landscape chunk_size = 5 total_words = len(words) sec_per_word = audio_dur / max(total_words, 1) drawtext_filters = [] for i in range(0, total_words, chunk_size): chunk = words[i: i + chunk_size] half = (len(chunk) + 1) // 2 line1 = escape_ffmpeg_text(" ".join(chunk[:half])) line2 = escape_ffmpeg_text(" ".join(chunk[half:])) if len(chunk) > half else "" t_start = i * sec_per_word t_end = t_start + (len(chunk) * sec_per_word) enable = f"between(t,{t_start:.3f},{t_end:.3f})" # LINE 1 β€” outline via borderw (professional look, no box) df1 = ( f"drawtext=text='{line1}':" f"fontcolor=white:fontsize={fontsize}:font=Arial:" f"borderw=4:bordercolor=black@0.95:" f"x=(w-text_w)/2:y={y_pos - fontsize - 10}:" f"enable='{enable}'" ) drawtext_filters.append(df1) # LINE 2 (if exists) if line2: df2 = ( f"drawtext=text='{line2}':" f"fontcolor=white:fontsize={fontsize}:font=Arial:" f"borderw=4:bordercolor=black@0.95:" f"x=(w-text_w)/2:y={y_pos}:" f"enable='{enable}'" ) drawtext_filters.append(df2) return drawtext_filters def apply_zoom_to_clip(src: str, dst: str, tw: int, th: int, zoom_dir: str, duration: float) -> bool: """ Apply Ken Burns zoom in or zoom out effect to a single clip. zoom_dir: 'in' or 'out' """ fps = 25 # zoompan filter: smooth zoom over the clip duration if zoom_dir == "in": # start at 1.0, zoom to 1.15 zoom_expr = "zoom+0.0006" z_start = 1.0 else: # start at 1.15, zoom back to 1.0 zoom_expr = "if(lte(zoom,1.0),1.15,zoom-0.0006)" z_start = 1.15 # zoompan needs source frames; scale up first then crop scale_w = int(tw * 1.2) scale_h = int(th * 1.2) vf = ( f"scale={scale_w}:{scale_h}:force_original_aspect_ratio=increase," f"crop={scale_w}:{scale_h}," f"zoompan=z='{zoom_expr}':x='iw/2-(iw/zoom/2)':y='ih/2-(ih/zoom/2)'" f":d={int(duration*fps)}:s={tw}x{th}:fps={fps}," f"crop={tw}:{th}" ) cmd = [ "ffmpeg", "-y", "-i", src, "-vf", vf, "-t", f"{duration:.3f}", "-r", str(fps), "-preset", "ultrafast", "-an", dst ] proc = subprocess.run(cmd, capture_output=True, text=True) return os.path.exists(dst) and os.path.getsize(dst) > 0 def create_video_centered(clip_paths: List[str], voice_path: str, out_path: str, script: str, aspect: str) -> bool: try: audio_dur = get_tts_duration(voice_path) if aspect == "9:16": tw, th = 720, 1280 watermark_fs = 30 else: tw, th = 1280, 720 watermark_fs = 24 n_clips = len(clip_paths) # Each clip gets equal share of audio duration dur_per_clip = audio_dur / n_clips # ── Step 1: Apply zoom effect to each clip individually, trim to equal duration zoom_dirs = ["in", "out", "in", "out"] # alternate zoom direction processed = [] for i, src in enumerate(clip_paths): dst = f"/tmp/zoom_{i}.mp4" zdir = zoom_dirs[i % len(zoom_dirs)] ok = apply_zoom_to_clip(src, dst, tw, th, zdir, dur_per_clip) if ok: processed.append(dst) else: # fallback: simple scale+crop without zoom fb = f"/tmp/fb_{i}.mp4" subprocess.run([ "ffmpeg", "-y", "-i", src, "-vf", f"scale={tw}:{th}:force_original_aspect_ratio=increase,crop={tw}:{th}", "-t", f"{dur_per_clip:.3f}", "-preset", "ultrafast", "-an", fb ], capture_output=True, text=True) if os.path.exists(fb): processed.append(fb) if not processed: return False # ── Step 2: Concatenate all processed clips into one background list_file = "/tmp/concat_list.txt" with open(list_file, "w") as lf: for p in processed: lf.write(f"file '{p}'\n") joined = "/tmp/joined_raw.mp4" subprocess.run([ "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", list_file, "-c", "copy", joined ], capture_output=True, text=True) if not os.path.exists(joined): return False # ── Step 3: Trim/extend to exactly audio_dur looped = "/tmp/looped_bg.mp4" subprocess.run([ "ffmpeg", "-y", "-stream_loop", "-1", "-i", joined, "-t", f"{audio_dur:.3f}", "-c", "copy", looped ], capture_output=True, text=True) bg = looped if os.path.exists(looped) else joined # ── Step 4: Build professional captions clean_script = re.sub(r'[^\w\s]', '', script) words = clean_script.split() caption_filters = build_professional_captions(words, audio_dur, tw, th, aspect) # ── Step 5: Watermark at bottom (smaller, elegant) watermark_text = "Abubakar Daily Islamic Shorts" wm_y = th - 36 watermark_draw = ( f"drawtext=text='{watermark_text}':" f"fontcolor=white@0.7:fontsize={watermark_fs}:font=Arial:" f"borderw=2:bordercolor=black@0.8:" f"x=(w-text_w)/2:y={wm_y}:" f"enable='between(t,0,{audio_dur:.3f})'" ) # Combine all drawtext filters all_text_filters = caption_filters + [watermark_draw] combined_text = ",".join(all_text_filters) # ── Step 6: Build final ffmpeg command with captions + watermark + logo if os.path.exists(LOGO_FILE): filter_complex = ( f"[2:v]scale=110:110[logo];" f"[0:v][logo]overlay=W-w-16:16[tmp];" f"[tmp]{combined_text}[vout]" ) ff_cmd = [ "ffmpeg", "-y", "-i", bg, "-i", voice_path, "-i", LOGO_FILE, "-filter_complex", filter_complex, "-map", "[vout]", "-map", "1:a", "-c:v", "libx264", "-preset", "ultrafast", "-t", f"{audio_dur:.3f}", out_path ] else: filter_complex = f"[0:v]{combined_text}[vout]" ff_cmd = [ "ffmpeg", "-y", "-i", bg, "-i", voice_path, "-filter_complex", filter_complex, "-map", "[vout]", "-map", "1:a", "-c:v", "libx264", "-preset", "ultrafast", "-t", f"{audio_dur:.3f}", out_path ] proc = subprocess.run(ff_cmd, capture_output=True, text=True) if proc.returncode != 0: print("ffmpeg stderr:", proc.stderr[-3000:]) return False # Trim to exact duration if needed if os.path.exists(out_path): try: pr = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", out_path], capture_output=True, text=True ) vdur = float(pr.stdout.strip()) except: vdur = None if vdur and abs(vdur - audio_dur) > 0.05: trimmed = out_path + ".trim.mp4" subprocess.run(["ffmpeg", "-y", "-i", out_path, "-t", f"{audio_dur:.3f}", "-c", "copy", trimmed], capture_output=True, text=True) if os.path.exists(trimmed): os.replace(trimmed, out_path) return os.path.exists(out_path) except Exception as e: print(f"Error in create_video_centered: {e}") return False # ════════════════════════════════════════════ # FEATURE: YouTube Upload (via YouTube Data API v3) # ════════════════════════════════════════════ YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY", "").strip() def upload_to_youtube(video_path: str, title: str, description: str, tags: list) -> dict: """ Upload video to YouTube using resumable upload. Requires YOUTUBE_API_KEY (OAuth token) in secrets. Returns {"success": True/False, "url": "...", "error": "..."} """ try: if not YOUTUBE_API_KEY: return {"success": False, "error": "YOUTUBE_API_KEY ba a saka ba a Secrets"} # Step 1: Initialize resumable upload metadata = { "snippet": { "title": title[:100], "description": description, "tags": tags[:10] if tags else [], "categoryId": "22" # People & Blogs }, "status": { "privacyStatus": "public", "selfDeclaredMadeForKids": False } } init_resp = requests.post( "https://www.googleapis.com/upload/youtube/v3/videos" "?uploadType=resumable&part=snippet,status", headers={ "Authorization": f"Bearer {YOUTUBE_API_KEY}", "Content-Type": "application/json", "X-Upload-Content-Type": "video/mp4", }, json=metadata, timeout=30 ) if init_resp.status_code not in (200, 201): return {"success": False, "error": f"Init failed: {init_resp.text[:200]}"} upload_url = init_resp.headers.get("Location", "") if not upload_url: return {"success": False, "error": "No upload URL returned"} # Step 2: Upload video bytes file_size = os.path.getsize(video_path) with open(video_path, "rb") as vf: upload_resp = requests.put( upload_url, data=vf, headers={ "Content-Type": "video/mp4", "Content-Length": str(file_size) }, timeout=300 ) if upload_resp.status_code in (200, 201): vid_id = upload_resp.json().get("id", "") return {"success": True, "url": f"https://youtube.com/shorts/{vid_id}", "id": vid_id} else: return {"success": False, "error": f"Upload failed: {upload_resp.text[:200]}"} except Exception as e: return {"success": False, "error": str(e)} # ════════════════════════════════════════════ # FEATURE: Batch Video Generation # ════════════════════════════════════════════ def generate_batch_video(topic: str, lang: str, voice: str, aspect: str, idx: int) -> dict: """Generate one complete video β€” used in batch mode.""" result = {"topic": topic, "status": "❌ Failed", "path": "", "metadata": {}} try: # Script script = call_freemodel(topic, lang) if script.startswith("Error"): result["status"] = f"❌ Script error" return result # Metadata metadata = call_freemodel_metadata(topic, lang, script) # TTS vpath = f"/tmp/batch_voice_{idx}.mp3" asyncio.run(_tts(script, voice, vpath)) # Nasheed mix nasheed = download_nasheed(f"/tmp/nasheed_{idx}.mp3") if nasheed: vpath = mix_background_nasheed(vpath, nasheed, f"/tmp/batch_mixed_{idx}.mp3") # Pexels clips headers = {"Authorization": PEXELS_KEY} queries = ["islamic architecture golden hour", "desert sunset sand dunes", "ocean waves peaceful nature", "green forest light rays"] clip_paths = [] for q in queries: if len(clip_paths) >= 4: break try: resp = requests.get( f"https://api.pexels.com/videos/search?query={requests.utils.quote(q)}&per_page=2&orientation=portrait", headers=headers, timeout=15 ) for v in resp.json().get("videos", []): for f in v.get("video_files", []): if f.get("file_type", "").lower() == "video/mp4" and f.get("width", 0) >= 720: rp = f"/tmp/batch_clip_{idx}_{len(clip_paths)}.mp4" dl = requests.get(f["link"], stream=True, timeout=60) with open(rp, "wb") as fh: for chunk in dl.iter_content(8192): if chunk: fh.write(chunk) clip_paths.append(rp) break if len(clip_paths) >= 4: break except: pass if not clip_paths: result["status"] = "❌ No clips" return result # Assemble video out = f"/tmp/batch_video_{idx}.mp4" ok = create_video_centered(clip_paths, vpath, out, script, aspect) if ok: result["status"] = "βœ… Done" result["path"] = out result["metadata"] = metadata result["script"] = script save_to_history(topic, lang, metadata, out, "") else: result["status"] = "❌ Assembly failed" except Exception as e: result["status"] = f"❌ {str(e)[:60]}" return result # ════════════════════════════════════════════ # FEATURE: Analytics Dashboard # ════════════════════════════════════════════ def compute_analytics(history: list) -> dict: if not history: return {} total = len(history) lang_count = {} topic_words = {} dates = [] for e in history: lg = e.get("lang", "Unknown") lang_count[lg] = lang_count.get(lg, 0) + 1 for w in e.get("topic", "").split(): topic_words[w] = topic_words.get(w, 0) + 1 dates.append(e.get("date", "")[:10]) top_lang = max(lang_count, key=lang_count.get) if lang_count else "-" top_topics = sorted(topic_words.items(), key=lambda x: x[1], reverse=True)[:5] unique_days = len(set(dates)) return { "total": total, "lang_count": lang_count, "top_lang": top_lang, "top_topics": top_topics, "unique_days": unique_days, "avg_per_day": round(total / max(unique_days, 1), 1) } # TABS tab1, tab2, tab3 = st.tabs(["🎬 Generate Video", "πŸ“¦ Batch Mode", "πŸ“ˆ Analytics"]) with tab1: # INPUT PANEL st.markdown('
', unsafe_allow_html=True) st.markdown('
πŸŒ™ ZaΙ“uΙ“Ι“uka β€” Settings
', unsafe_allow_html=True) col1, col2 = st.columns([1, 1]) with col1: lang = st.selectbox("Harshe / Language", ["Hausa πŸ‡³πŸ‡¬", "Larabci πŸ‡ΈπŸ‡¦", "English πŸ‡ΊπŸ‡Έ"], key="lang_select") with col2: voice_map = {"Hausa πŸ‡³πŸ‡¬": "ha-NG-AbdullahNeural", "Larabci πŸ‡ΈπŸ‡¦": "ar-SA-HamedNeural", "English πŸ‡ΊπŸ‡Έ": "en-US-AndrewNeural"} selected_voice = voice_map[lang] st.text_input("Muryar da za a yi amfani", value=selected_voice, disabled=True) topic = st.text_input("Maudu'i β€” Topic", placeholder="e.g. Tuba, Tawakkul, Ƙaunar Allah...", key="topic_input") # AI Topic Suggestions col_sug1, col_sug2 = st.columns([3, 1]) with col_sug2: if st.button("πŸ’‘ Get Topics", use_container_width=True): with st.spinner("AI na ba da shawarwari..."): st.session_state.topic_suggestions = get_topic_suggestions(lang.split()[0]) if st.session_state.topic_suggestions: st.markdown("
", unsafe_allow_html=True) st.markdown("πŸ‘† Danna topic Ι—aya don zaΙ“ar shi:", unsafe_allow_html=True) cols = st.columns(2) for i, sug in enumerate(st.session_state.topic_suggestions): with cols[i % 2]: if st.button(f"πŸ“Œ {sug}", key=f"sug_{i}", use_container_width=True): st.session_state.selected_topic = sug st.rerun() st.markdown("
", unsafe_allow_html=True) if "selected_topic" in st.session_state and st.session_state.selected_topic: st.info(f"βœ… Topic da aka zaΙ“a: **{st.session_state.selected_topic}**") st.markdown("
", unsafe_allow_html=True) manual_script = st.text_area( "Manual Script (paste here to bypass AI generation)", value="", placeholder="Paste your own spoken script here. If filled, AI generation will be skipped.", height=150, key="manual_script" ) video_size = st.selectbox("Video Size / Aspect Ratio", ["Shorts/Reels (9:16)", "Long Video (16:9)"], key="video_size_select") aspect_map = {"Shorts/Reels (9:16)": "9:16", "Long Video (16:9)": "16:9"} selected_aspect = aspect_map[video_size] st.markdown('
', unsafe_allow_html=True) # GENERATE BUTTON st.markdown('
', unsafe_allow_html=True) st.markdown('
βš™οΈ ƘirΖ™irar Bidiyo
', unsafe_allow_html=True) if st.button("πŸš€ GENERATE ISLAMIC VIDEO", use_container_width=True): # Use selected topic from suggestions if available final_topic = st.session_state.get("selected_topic", "") or topic.strip() if not final_topic: st.warning("⚠️ Rubuta maudu'i da farko ko zaΙ“i daga suggestions.") else: topic = final_topic st.session_state.selected_topic = "" # reset after use if final_topic: if True: if manual_script and manual_script.strip(): script = manual_script.strip() else: with st.spinner("✦ Ana rubuta rubutu..."): script = call_freemodel(topic, lang.split()[0]) st.session_state.script_text = script st.markdown("**πŸ“œ Rubutun da aka Ζ™irΖ™ira / Used Script:**") st.markdown(f'
{html.escape(script)}
', unsafe_allow_html=True) with st.spinner("✦ Ana Ζ™irΖ™irar metadata..."): metadata = call_freemodel_metadata(topic, lang.split()[0], script) st.session_state.metadata = metadata with st.spinner("✦ Ana Ζ™irΖ™irar murya (TTS)..."): voice_path = "/tmp/tts_audio.mp3" asyncio.run(_tts(script, selected_voice, voice_path)) # Download real nasheed and mix with st.spinner("✦ Ana nemo Nasheed..."): nasheed_file = download_nasheed("/tmp/nasheed.mp3") if nasheed_file: voice_path = mix_background_nasheed(voice_path, nasheed_file, "/tmp/mixed_nasheed.mp3") else: voice_path = mix_background_music(voice_path, "/tmp/mixed_audio.mp3") st.session_state.nasheed_path = nasheed_file # ── VIDEO PREVIEW CARD st.markdown('
πŸ‘οΈ Preview β€” Tabbatar kafin haΙ—a bidiyo
', unsafe_allow_html=True) render_preview_card(script, st.session_state.metadata, selected_aspect) with st.spinner("✦ Ana neman bidiyon bango masu dacewa..."): headers = {"Authorization": PEXELS_KEY} # Dynamic Islamic cinematic queries based on audio duration audio_dur_tmp = get_tts_duration(voice_path) n_clips_needed = max(4, int(audio_dur_tmp / 8)) preferred_queries = [ "islamic architecture golden hour", "desert sunset sand dunes", "ocean waves peaceful nature", "green forest light rays", "mountain landscape sunrise", "night sky stars peaceful", ][:n_clips_needed] clip_paths = [] for q in preferred_queries: if len(clip_paths) >= n_clips_needed: break try: resp = requests.get( f"https://api.pexels.com/videos/search?query={requests.utils.quote(q)}&per_page=3&orientation=portrait", headers=headers, timeout=20 ) r = resp.json() for v in r.get('videos', []): chosen = None for f in v.get('video_files', []): if f.get('file_type', '').lower() == 'video/mp4' and f.get('width', 0) >= 720: chosen = f break if chosen: raw_path = f"/tmp/r_{len(clip_paths)}.mp4" dl = requests.get(chosen['link'], stream=True, timeout=60) with open(raw_path, "wb") as fh: for chunk in dl.iter_content(chunk_size=8192): if chunk: fh.write(chunk) clip_paths.append(raw_path) break except Exception as e: print(f"Pexels fetch error for query '{q}': {e}") if not clip_paths: st.error("❌ Ba a sami bidiyo ba daga Pexels.") else: with st.spinner(f"✦ Ana haΙ—a bidiyo gaba daya..."): out_path = "final_islamic_video.mp4" success = create_video_centered(clip_paths, voice_path, out_path, script, selected_aspect) if success: st.session_state.final_video_path = out_path st.success("βœ… An gama! Bidiyon yana Ζ™asa.") # Save to history save_to_history( topic, lang.split()[0], st.session_state.metadata, out_path, "/tmp/thumbnail.jpg" ) st.session_state.video_history = load_history() # Generate thumbnail with st.spinner("✦ Ana Ζ™irΖ™irar thumbnail..."): thumb_title = st.session_state.metadata.get("title", topic) if st.session_state.metadata else topic thumb_ok = generate_thumbnail(thumb_title, selected_aspect, "/tmp/thumbnail.jpg") if thumb_ok: st.session_state.thumbnail_path = "/tmp/thumbnail.jpg" ts = int(time.time()) metadata_filename = f"/tmp/metadata_{ts}.json" try: with open(metadata_filename, "w", encoding="utf-8") as mf: json.dump(st.session_state.metadata, mf, ensure_ascii=False, indent=2) st.session_state.last_metadata_file = metadata_filename except Exception as e: st.session_state.last_metadata_file = "" md = st.session_state.metadata or {} uid = str(int(time.time() * 1000)) title_val = html.escape(md.get("title", "")) desc_val = html.escape(md.get("description", "")) tags_list = md.get("hashtags", []) tags_val = " ".join(tags_list) if isinstance(tags_list, list) else str(tags_list) tags_val = html.escape(tags_val) meta_html = f"""
""" st.markdown('
', unsafe_allow_html=True) st.markdown('
πŸ“ Generated Title, Description & Hashtags
', unsafe_allow_html=True) st.markdown(meta_html, unsafe_allow_html=True) try: st.download_button( label="πŸ“₯ Download metadata JSON", data=json.dumps(md, ensure_ascii=False, indent=2), file_name=f"metadata_{ts}.json", mime="application/json", use_container_width=True ) if st.session_state.last_metadata_file: st.markdown(f"Server copy: {st.session_state.last_metadata_file}", unsafe_allow_html=True) except Exception as e: st.write("Failed to create metadata download:", e) st.markdown('
', unsafe_allow_html=True) else: st.error("❌ Kuskure wajen haΙ—a bidiyo. (Sake gwadawa)") st.markdown('
', unsafe_allow_html=True) # VIDEO OUTPUT PANEL if st.session_state.final_video_path and os.path.exists(st.session_state.final_video_path): st.markdown('
', unsafe_allow_html=True) st.markdown('
🎬 Bidiyon da aka ƘirΖ™ira
', unsafe_allow_html=True) st.video(st.session_state.final_video_path) st.markdown("
", unsafe_allow_html=True) with open(st.session_state.final_video_path, "rb") as fh: st.download_button( label="πŸ“₯ SAUKE BIDIYO β€” Download Video", data=fh, file_name="Abubakar_Islamic_Short.mp4", mime="video/mp4", use_container_width=True ) st.markdown('
', unsafe_allow_html=True) # THUMBNAIL OUTPUT PANEL if st.session_state.thumbnail_path and os.path.exists(st.session_state.thumbnail_path): st.markdown('
', unsafe_allow_html=True) st.markdown('
πŸ–ΌοΈ YouTube Thumbnail
', unsafe_allow_html=True) st.image(st.session_state.thumbnail_path, use_column_width=True) st.markdown("
", unsafe_allow_html=True) with open(st.session_state.thumbnail_path, "rb") as fh: st.download_button( label="πŸ“₯ SAUKE THUMBNAIL β€” Download Thumbnail", data=fh, file_name="Abubakar_Islamic_Thumbnail.jpg", mime="image/jpeg", use_container_width=True ) st.markdown('
', unsafe_allow_html=True) # MULTI-PLATFORM EXPORT PANEL if st.session_state.final_video_path and os.path.exists(st.session_state.final_video_path): st.markdown('
', unsafe_allow_html=True) st.markdown('
πŸ“± Multi-Platform Export
', unsafe_allow_html=True) st.markdown("ZaΙ“i platform β†’ app zai canza girman bidiyo kai tsaye", unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) for fmt_name, (fw, fh, fsuffix) in EXPORT_FORMATS.items(): col_a, col_b = st.columns([3, 1]) with col_a: st.markdown(f"πŸ“ {fmt_name}", unsafe_allow_html=True) with col_b: btn_key = f"export_{fsuffix}" if st.button("Export", key=btn_key, use_container_width=True): with st.spinner(f"Ana export..."): ep = export_video_format(st.session_state.final_video_path, fw, fh, fsuffix) if ep: st.session_state.export_paths[fsuffix] = ep # Show download if exported if fsuffix in st.session_state.export_paths: ep = st.session_state.export_paths[fsuffix] if os.path.exists(ep): with open(ep, "rb") as ef: st.download_button( label=f"πŸ“₯ Download {fsuffix.upper()}", data=ef, file_name=f"Islamic_Short_{fsuffix}.mp4", mime="video/mp4", key=f"dl_{fsuffix}", use_container_width=True ) st.markdown("
", unsafe_allow_html=True) st.markdown('
', unsafe_allow_html=True) # HISTORY PANEL if st.session_state.video_history: st.markdown('
', unsafe_allow_html=True) st.markdown('
πŸ’Ύ Tarihin Videos β€” History
', unsafe_allow_html=True) for entry in st.session_state.video_history[:5]: col1, col2 = st.columns([4, 1]) with col1: st.markdown(f"""
{html.escape(entry.get('title',''))}
{entry.get('date','')} Β· {entry.get('lang','')} Β· {entry.get('topic','')}
""", unsafe_allow_html=True) with col2: vp = entry.get("video_path", "") if vp and os.path.exists(vp): with open(vp, "rb") as hf: st.download_button( "πŸ“₯", data=hf, file_name=f"Islamic_{entry['id']}.mp4", mime="video/mp4", key=f"hist_{entry['id']}", use_container_width=True ) st.markdown('
', unsafe_allow_html=True) # ══════════════════════════════════════════ # YOUTUBE UPLOAD PANEL (inside tab1 video output) # ══════════════════════════════════════════ if st.session_state.final_video_path and os.path.exists(st.session_state.final_video_path): if YOUTUBE_API_KEY: st.markdown('
', unsafe_allow_html=True) st.markdown('
πŸ“€ Loda YouTube kai tsaye
', unsafe_allow_html=True) md_yt = st.session_state.metadata or {} yt_title = md_yt.get("title", "Islamic Short") yt_desc = md_yt.get("description", "") yt_tags = md_yt.get("hashtags", []) col_yt1, col_yt2 = st.columns([3, 1]) with col_yt1: st.markdown(f"πŸ“Ή {html.escape(yt_title)}", unsafe_allow_html=True) with col_yt2: if st.button("πŸš€ Upload", key="yt_upload", use_container_width=True): with st.spinner("Ana loda YouTube..."): yt_result = upload_to_youtube( st.session_state.final_video_path, yt_title, yt_desc, [t.replace("#","") for t in yt_tags] ) if yt_result["success"]: st.session_state.yt_status = f"βœ… An loda! {yt_result.get('url','')}" else: st.session_state.yt_status = f"❌ {yt_result.get('error','')}" if st.session_state.yt_status: st.info(st.session_state.yt_status) st.markdown('
', unsafe_allow_html=True) # ══════════════════════════════════════════ # BATCH MODE TAB # ══════════════════════════════════════════ with tab2: st.markdown('
', unsafe_allow_html=True) st.markdown('
πŸ“¦ Batch Mode β€” ƘirΖ™irar Videos Da Yawa
', unsafe_allow_html=True) batch_lang = st.selectbox("Harshe", ["Hausa", "Arabic", "English"], key="batch_lang") batch_voice_map = {"Hausa": "ha-NG-AbdullahNeural", "Arabic": "ar-SA-HamedNeural", "English": "en-US-AndrewNeural"} batch_aspect = st.selectbox("Aspect Ratio", ["9:16", "16:9"], key="batch_aspect") st.markdown("Rubuta topics Ι—aya a kowace layi (max 10)", unsafe_allow_html=True) batch_topics_raw = st.text_area( "Topics (Ι—aya a kowace layi)", placeholder="Tuba, Tawakkul, Fadar Annabi, Sabr da lada, Kaunar Allah", height=180, key="batch_topics" ) col_b1, col_b2 = st.columns(2) with col_b1: if st.button("πŸ’‘ Auto-fill Topics (AI)", use_container_width=True, key="batch_autofill"): with st.spinner("AI na ba da topics..."): sugs = get_topic_suggestions(batch_lang) if sugs: st.session_state["batch_autofilled"] = "\n".join(sugs[:8]) st.rerun() with col_b2: run_batch = st.button("πŸš€ START BATCH", use_container_width=True, key="run_batch") if "batch_autofilled" in st.session_state: st.info("Topics sun shiga β€” kopiya su saka a box sama.") st.code(st.session_state["batch_autofilled"]) if run_batch: topics_list = [t.strip() for t in batch_topics_raw.strip().splitlines() if t.strip()][:10] if not topics_list: st.warning("⚠️ Rubuta topics da farko!") else: st.session_state.batch_results = [] progress = st.progress(0) status_box = st.empty() for i, tp in enumerate(topics_list): status_box.info(f"🎬 Ana Ζ™irΖ™irar {i+1}/{len(topics_list)}: **{tp}**") res = generate_batch_video(tp, batch_lang, batch_voice_map[batch_lang], batch_aspect, i) st.session_state.batch_results.append(res) progress.progress((i + 1) / len(topics_list)) status_box.success(f"βœ… An gama! Videos {len(topics_list)} sun shirya.") # Show batch results if st.session_state.batch_results: st.markdown('
πŸ“‹ Sakamakon Batch
', unsafe_allow_html=True) for res in st.session_state.batch_results: col_r1, col_r2 = st.columns([4, 1]) with col_r1: st.markdown(f"""
{res['status']} {html.escape(res['topic'])}
{html.escape(res.get('metadata',{}).get('title',''))}
""", unsafe_allow_html=True) with col_r2: vp = res.get("path", "") if vp and os.path.exists(vp): with open(vp, "rb") as bf: st.download_button( "πŸ“₯", data=bf, file_name=f"batch_{res['topic'][:20]}.mp4", mime="video/mp4", key=f"batch_dl_{res['topic'][:10]}_{id(res)}", use_container_width=True ) st.markdown('
', unsafe_allow_html=True) # ══════════════════════════════════════════ # ANALYTICS TAB # ══════════════════════════════════════════ with tab3: st.markdown('
', unsafe_allow_html=True) st.markdown('
πŸ“ˆ Analytics Dashboard
', unsafe_allow_html=True) history_data = st.session_state.video_history analytics = compute_analytics(history_data) if not analytics: st.markdown("
Babu bayanan tarihi tukuna.
ƘirΖ™iri videos da yawa sannan analytics zai bayyana.
", unsafe_allow_html=True) else: # KPI Cards col_k1, col_k2, col_k3 = st.columns(3) def kpi_card(label, value, icon): return f"""
{icon}
{value}
{label}
""" with col_k1: st.markdown(kpi_card("Jimillar Videos", analytics["total"], "🎬"), unsafe_allow_html=True) with col_k2: st.markdown(kpi_card("Yawan Harshe", analytics["top_lang"], "🌍"), unsafe_allow_html=True) with col_k3: st.markdown(kpi_card("Avg / Rana", analytics["avg_per_day"], "πŸ“…"), unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) # Language breakdown st.markdown("
🌍 Videos ta Harshe
", unsafe_allow_html=True) for lg, cnt in analytics["lang_count"].items(): pct = int(cnt / analytics["total"] * 100) st.markdown(f"""
{lg} {cnt} ({pct}%)
""", unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) # Top topics if analytics["top_topics"]: st.markdown("
πŸ”₯ Topics da suka fi
", unsafe_allow_html=True) for word, cnt in analytics["top_topics"]: st.markdown(f"πŸ“Œ {html.escape(word)} Γ—{cnt}", unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) # Recent history table st.markdown("
πŸ• Videos na Ƙarshe
", unsafe_allow_html=True) for e in history_data[:8]: st.markdown(f"""
{html.escape(e.get('title','')[:40])} {e.get('date','')}
""", unsafe_allow_html=True) st.markdown('
', unsafe_allow_html=True) # FOOTER st.markdown("""
ο·½    Abubakar Daily Islamic Shorts    ο·½
""", unsafe_allow_html=True)