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("""
""", unsafe_allow_html=True)
# CORE FUNCTIONS
def call_freemodel(topic, lang, duration_sec: int = 35):
lang_instruction = {
"Hausa": "Write ONLY in Hausa language.",
"Larabci": "ุงูุชุจ ุงููุต ุจุงููุบุฉ ุงูุนุฑุจูุฉ ุงููุตุญู ููุท. ูุง ุชุณุชุฎุฏู ุฃู ูุบุฉ ุฃุฎุฑู.",
"Arabic": "ุงูุชุจ ุงููุต ุจุงููุบุฉ ุงูุนุฑุจูุฉ ุงููุตุญู ููุท. ูุง ุชุณุชุฎุฏู ุฃู ูุบุฉ ุฃุฎุฑู.",
"English": "Write ONLY in English language.",
}.get(lang, f"Write ONLY in {lang} language.")
prompt = (
f"{lang_instruction} "
f"Write a {duration_sec}-second spiritual Islamic script 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, no translation. "
"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/quran-recitation-mishary/001.mp3",
"https://archive.org/download/Mishari_Rashid_Al-Afasy_Quran/001.mp3",
"https://download.quranicaudio.com/quran/mishaari_raashid_al_3afaasee/001.mp3",
"https://archive.org/download/SurahAl-Fatihah/Al-Fatihah.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 nasheed at low volume, loop nasheed to match voice length."""
try:
dur = get_tts_duration(voice_path)
if not nasheed_path or not os.path.exists(nasheed_path):
# Fallback: soft sine ambient
return mix_background_music(voice_path, out_path)
proc = subprocess.run([
"ffmpeg", "-y",
"-i", voice_path,
"-stream_loop", "-1", "-i", nasheed_path,
"-filter_complex",
"[0:a]volume=1.0[voice];[1:a]volume=0.12,atrim=0:{dur}[bg];[voice][bg]amix=inputs=2:duration=first:dropout_transition=2[aout]".format(dur=f"{dur:.3f}"),
"-map", "[aout]",
"-ar", "44100",
"-t", f"{dur:.3f}",
out_path
], capture_output=True, text=True)
if proc.returncode == 0 and os.path.exists(out_path) and os.path.getsize(out_path) > 1000:
return out_path
else:
print("Nasheed mix stderr:", proc.stderr[-500:])
return mix_background_music(voice_path, out_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:
"""
Reliable Ken Burns effect using scale expression.
zoom_dir: 'in' = slowly zoom in, 'out' = slowly zoom out
"""
fps = 25
n_frames = int(duration * fps)
# Scale slightly larger than target so zoom has room
sw = int(tw * 1.3)
sh = int(th * 1.3)
cx = sw // 2
cy = sh // 2
if zoom_dir == "in":
# Pan from slightly zoomed-out to center โ zooming in
vf = (
f"scale={sw}:{sh}:force_original_aspect_ratio=increase,"
f"crop={sw}:{sh},"
f"crop=w='if(lte(t,0),{tw},{tw}+({sw}-{tw})*max(0,1-t/{duration:.3f}))'"
f":h='if(lte(t,0),{th},{th}+({sh}-{th})*max(0,1-t/{duration:.3f}))'"
f":x='({sw}-out_w)/2':y='({sh}-out_h)/2',"
f"scale={tw}:{th}"
)
else:
# Zoom out: start tight, slowly reveal more
vf = (
f"scale={sw}:{sh}:force_original_aspect_ratio=increase,"
f"crop={sw}:{sh},"
f"crop=w='if(lte(t,0),{tw},{tw}+({sw}-{tw})*min(1,t/{duration:.3f}))'"
f":h='if(lte(t,0),{th},{th}+({sh}-{th})*min(1,t/{duration:.3f}))'"
f":x='({sw}-out_w)/2':y='({sh}-out_h)/2',"
f"scale={tw}:{th}"
)
cmd = [
"ffmpeg", "-y", "-i", src,
"-vf", vf,
"-t", f"{duration:.3f}",
"-r", str(fps),
"-c:v", "libx264",
"-preset", "ultrafast",
"-an", dst
]
proc = subprocess.run(cmd, capture_output=True, text=True)
if not (os.path.exists(dst) and os.path.getsize(dst) > 0):
# Fallback: simple scale crop no zoom
fb_cmd = [
"ffmpeg", "-y", "-i", src,
"-vf", f"scale={tw}:{th}:force_original_aspect_ratio=increase,crop={tw}:{th}",
"-t", f"{duration:.3f}",
"-r", str(fps),
"-c:v", "libx264", "-preset", "ultrafast", "-an", dst
]
subprocess.run(fb_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]
video_duration = st.slider(
"โฑ๏ธ Video Duration (seconds)",
min_value=20, max_value=90, value=35, step=5,
key="video_duration_slider"
)
st.markdown(f"Script zai kasance kusan {video_duration} seconds", unsafe_allow_html=True)
st.markdown('
', 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], duration_sec=video_duration)
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"""
', 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)
# 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"""
"""
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"""