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" ANTHROPIC_KEY = os.getenv("ANTHROPIC_API_KEY", "").strip() PEXELS_KEY = os.getenv("PEXELS_API_KEY", "").strip() 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 = "" # 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_claude(topic, lang): prompt = ( f"Write a 35-second spiritual Islamic script in {lang} about '{topic}'. " "Output ONLY the spoken text. No titles, no hashtags, no stage directions. " "Make sure the script ends with a short, thought-provoking question to encourage comments." ) try: r = requests.post( "https://api.anthropic.com/v1/messages", headers={ "x-api-key": ANTHROPIC_KEY, "anthropic-version": "2023-06-01", "content-type": "application/json" }, json={ "model": "claude-sonnet-4-5-20251001", "max_tokens": 1024, "messages": [{"role": "user", "content": prompt}] }, timeout=30 ) return r.json()['content'][0]['text'].strip() except Exception as e: return f"Error: {e}" def call_claude_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( "https://api.anthropic.com/v1/messages", headers={ "x-api-key": ANTHROPIC_KEY, "anthropic-version": "2023-06-01", "content-type": "application/json" }, json={ "model": "claude-sonnet-4-5-20251001", "max_tokens": 1024, "messages": [{"role": "user", "content": prompt}] }, timeout=30 ) raw = r.json()['content'][0]['text'].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": []} async def tts(text, voice, path): clean = re.sub(r'[#*()<>]', '', text) comm = edge_tts.Communicate(clean, voice) await comm.save(path) _tts = tts 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 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 # โ”€โ”€ Step 1: Scale & Crop clips scaled = [] for i, src in enumerate(clip_paths): dst = f"/tmp/scaled_{i}.mp4" scale_cmd = [ "ffmpeg", "-y", "-i", src, "-vf", f"scale={tw}:{th}:force_original_aspect_ratio=increase,crop={tw}:{th}", "-preset", "ultrafast", "-an", dst ] subprocess.run(scale_cmd, capture_output=True, text=True) if os.path.exists(dst): scaled.append(dst) if not scaled: return False # Concatenate list_file = "/tmp/concat_list.txt" with open(list_file, "w") as lf: for p in scaled: 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) # โ”€โ”€ Step 2: Loop/trim to audio duration looped = "/tmp/looped_bg.mp4" epsilon = 0.05 subprocess.run([ "ffmpeg", "-y", "-stream_loop", "-1", "-i", joined, "-t", f"{audio_dur + epsilon:.3f}", "-c", "copy", looped ], capture_output=True, text=True) if not os.path.exists(looped): looped = joined # โ”€โ”€ Step 3: 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 4: 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 5: Build ffmpeg command 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", looped, "-i", voice_path, "-i", LOGO_FILE, "-filter_complex", filter_complex, "-map", "[vout]", "-map", "1:a", "-c:v", "libx264", "-preset", "ultrafast", "-shortest", "-t", f"{audio_dur:.3f}", out_path ] else: filter_complex = f"[0:v]{combined_text}[vout]" ff_cmd = [ "ffmpeg", "-y", "-i", looped, "-i", voice_path, "-filter_complex", filter_complex, "-map", "[vout]", "-map", "1:a", "-c:v", "libx264", "-preset", "ultrafast", "-shortest", "-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 # 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") 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): if not topic.strip(): st.warning("โš ๏ธ Rubuta maudu'i da farko.") else: if manual_script and manual_script.strip(): script = manual_script.strip() else: with st.spinner("โœฆ Ana rubuta rubutu..."): script = call_claude(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_claude_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)) with st.spinner("โœฆ Ana neman bidiyon bango masu dacewa..."): headers = {"Authorization": PEXELS_KEY} preferred_queries = ["beautiful mosques", "flowing river water", "deep green forest", "majestic wildlife"] clip_paths = [] for q in preferred_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=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.") 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) # FOOTER st.markdown("""
๏ทฝ    Abubakar Daily Islamic Shorts    ๏ทฝ
""", unsafe_allow_html=True)