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
# โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
# SOUND LIBRARY โ Nasheeds + Quran (free, archive.org)
# โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
SOUND_LIBRARY = {
"๐ Quran โ Al-Fatihah (Mishary)": "https://archive.org/download/mishary-al-afasy-quran/001.mp3",
"๐ Quran โ Al-Baqarah opener (Mishary)": "https://archive.org/download/mishary-al-afasy-quran/002.mp3",
"๐ต Nasheed โ Tala Al Badru (Instrumental)": "https://archive.org/download/NasheedCollection_201904/Tala_Al_Badru.mp3",
"๐ต Nasheed โ Ya Taiba (Instrumental)": "https://archive.org/download/NasheedCollection_201904/Ya_Taiba.mp3",
"๐ต Nasheed โ Allahu Allah (Soft)": "https://archive.org/download/NasheedCollection_201904/Allahu_Allah.mp3",
"๐ต Islamic Instrumental โ Oud Calm": "https://archive.org/download/free-islamic-music/oud_calm_islamic.mp3",
"๐ต Islamic Instrumental โ Flute Peaceful": "https://archive.org/download/free-islamic-music/flute_peaceful.mp3",
"๐ Babu Music (murya kawai)": "",
}
# Fallback URLs idan ษaya ya kasa
FALLBACK_URLS = [
"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/quran-recitation-mishary/001.mp3",
]
def download_sound(url: str, out_path: str) -> str:
"""Download sound from URL. Returns path if success, empty string if fail."""
if not url:
return ""
if os.path.exists(out_path) and os.path.getsize(out_path) > 10000:
return out_path
try:
r = requests.get(url, timeout=30, 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.exists(out_path) and os.path.getsize(out_path) > 10000:
return out_path
except Exception as e:
print(f"Sound download failed ({url}): {e}")
# Try fallbacks
for fb_url in FALLBACK_URLS:
try:
r = requests.get(fb_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.exists(out_path) and os.path.getsize(out_path) > 10000:
return out_path
except:
pass
return ""
def download_nasheed(out_path: str = "/tmp/nasheed.mp3") -> str:
"""Legacy wrapper โ download default Quran recitation."""
return download_sound(list(SOUND_LIBRARY.values())[0], out_path)
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):
# Circle logo: scale โ geq mask โ alphaextract โ overlay
filter_complex = (
f"[2:v]scale=110:110[logo_sq];"
# Create circle mask via geq
f"[logo_sq]format=rgba,"
f"geq=r='r(X,Y)':g='g(X,Y)':b='b(X,Y)'"
f":a='if(lte(pow(X-55,2)+pow(Y-55,2),pow(54,2)),255,0)'[logo_circ];"
f"[0:v][logo_circ]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)
# Sound Library selector
st.markdown(" ", unsafe_allow_html=True)
selected_sound_name = st.selectbox(
"๐ต Background Sound โ Zaษi Music",
options=list(SOUND_LIBRARY.keys()),
key="sound_library_select"
)
selected_sound_url = SOUND_LIBRARY[selected_sound_name]
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 selected background sound and mix
if selected_sound_url:
with st.spinner(f"โฆ Ana sauke {selected_sound_name}..."):
sound_file = download_sound(selected_sound_url, "/tmp/selected_sound.mp3")
if sound_file:
voice_path = mix_background_nasheed(voice_path, sound_file, "/tmp/mixed_nasheed.mp3")
st.success(f"โ {selected_sound_name} an haษa da murya!")
else:
st.warning("โ ๏ธ Ba a sami sound ba โ murya kawai.")
st.session_state.nasheed_path = sound_file
else:
st.info("๐ Murya kawai โ babu background music.")
st.session_state.nasheed_path = ""
# โโ 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}
audio_dur_tmp = get_tts_duration(voice_path)
n_clips_needed = max(4, int(audio_dur_tmp / 8))
# Use topic hash for random page โ different clips every video even same topic
import hashlib
topic_seed = int(hashlib.md5((topic + str(time.time())).encode()).hexdigest()[:4], 16) % 8 + 1
all_queries = [
"islamic architecture golden hour",
"mosque minaret sky",
"desert sunset sand dunes",
"ocean waves peaceful nature",
"green forest light rays sunbeam",
"mountain landscape sunrise fog",
"night sky stars milky way",
"waterfall nature peaceful",
"flowers bloom nature spring",
"river stream forest calm",
"aerial city lights night",
"clouds sky time lapse",
]
import random
random.seed(topic_seed)
random.shuffle(all_queries)
preferred_queries = all_queries[:n_clips_needed]
clip_paths = []
for q in preferred_queries:
if len(clip_paths) >= n_clips_needed:
break
try:
page = random.randint(1, 4)
resp = requests.get(
f"https://api.pexels.com/videos/search?query={requests.utils.quote(q)}&per_page=5&page={page}&orientation=portrait",
headers=headers, timeout=20
)
r = resp.json()
videos = r.get('videos', [])
random.shuffle(videos)
for v in 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"""