Upload 41 files

app.py

@@ -0,0 +1,89 @@
+import streamlit as st
+from dotenv import load_dotenv
+
+from app_pages.script_library import render_script_library
+from app_pages.video_analyser import render_video_analyser_page
+from authen.authentication import login_gate, logout
+
+
+from app_pages.text_image_generation import render_multimodel_ad_generator
+from app_pages.image_generation import render_bulk_image_generator
+
+from app_pages.video_library import render_video_library
+from app_pages.image_generation_library import render_image_variations_library
+from app_pages.text_image_generation_library import render_text_image_library
+
+from app_pages.script_generator import generator_page
+
+
+load_dotenv()
+st.set_page_config(page_title="Gen AI + AI Library", page_icon="✨", layout="wide")
+
+
+# ---------- Auth ----------
+uid = login_gate()
+with st.sidebar:
+    st.markdown(f"**Signed in as:** `{uid}`")
+    if st.button("Log out", use_container_width=True, key="btn_logout"):
+        logout()
+        st.rerun()
+
+    section = st.radio(" ", ["Gen AI", "AI Library"], index=0, horizontal=True, key="main_section")
+
+# ---------------------------- Gen AI ---------------------------------
+def render_genai():
+    _, main = st.columns([0.1, 1], gap="small")
+    with main:
+        mod_tabs = st.tabs(["Image Gen", "Video Analyser", "Script Generator"])
+
+        # -------- Image Generation --------
+        with mod_tabs[0]:
+            sub = st.tabs(["Text", "Img"])
+            with sub[0]:
+                render_multimodel_ad_generator(prefix="ig_text")
+            with sub[1]:
+                render_bulk_image_generator(prefix="ig_img")
+
+        # -------- Video Analyser --------
+        with mod_tabs[1]:
+            render_video_analyser_page()
+
+        # -------- Script Generator --------
+        with mod_tabs[2]:
+            generator_page()
+
+
+# ---------------------------- AI Library ---------------------------------
+def render_ai_library():
+    _, main = st.columns([0.1, 1], gap="small")
+    with main:
+        st.subheader("Library")
+        lib_tabs = st.tabs(["Image", "Analysed Video", "Generated Scripts"])
+
+
+
+    # ---------- Image Tab ----------
+    with lib_tabs[0]:
+            sub = st.tabs(["New generated", "Variations"])
+
+            with sub[0]:
+                render_text_image_library()
+
+
+            with sub[1]:
+                render_image_variations_library()
+
+    # ---------- Video Library Tab ----------
+    with lib_tabs[1]:
+        render_video_library()
+
+    # ---------- Generated Scripts Tab ----------
+    with lib_tabs[2]:
+        render_script_library()
+
+
+
+if section == "Gen AI":
+    render_genai()
+else:
+    render_ai_library()

app_pages/image_generation.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+import zipfile, requests, logging, tempfile, shutil, os
+import streamlit as st
+from generator_function.image_processor import process_zip_and_generate_images
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger(__name__)
+
+def _zip_gallery_images(gallery_items):
+    if not gallery_items: return None
+    image_urls = [item[0] if isinstance(item, tuple) else item for item in gallery_items]
+    temp_dir = tempfile.mkdtemp()
+    try:
+        zip_path = tempfile.NamedTemporaryFile(delete=False, suffix=".zip").name
+        for i, url in enumerate(image_urls):
+            try:
+                ext = url.split('?')[0].split('.')[-1]; ext = ext if ext and len(ext) <= 5 else "png"
+                file_path = os.path.join(temp_dir, f"image_{i}.{ext}")
+                if url.startswith(("http://", "https://")):
+                    resp = requests.get(url, timeout=10); resp.raise_for_status()
+                    with open(file_path, "wb") as f: f.write(resp.content)
+                elif os.path.exists(url):
+                    shutil.copy(url, file_path)
+            except Exception as e:
+                logger.error(f"Error processing image {url}: {e}")
+        with zipfile.ZipFile(zip_path, "w") as zipf:
+            for file_name in os.listdir(temp_dir):
+                zipf.write(os.path.join(temp_dir, file_name), arcname=file_name)
+        return zip_path
+    except Exception as e:
+        logger.critical(f"Failed to create zip: {e}"); return None
+    finally:
+        shutil.rmtree(temp_dir, ignore_errors=True)
+
+def render_bulk_image_generator(prefix: str = "ig_img"):
+    zip_file = st.file_uploader("Upload Zip or Single File", type=["zip", "png", "jpg", "jpeg"], key=f"{prefix}_upload")
+    category = st.text_input("Category", key=f"{prefix}_category")
+    col1, col2, col3, col4, col5, col6 = st.columns([2,2.5,2.5,2.5,2.5,2])
+    with col1:  blur = st.checkbox("Blur Image", key=f"{prefix}_blur")
+    with col2:  size = st.selectbox("Image Size", ["auto","1024x1024","1536x1024","1024x1536"], key=f"{prefix}_size")
+    with col3:  quality = st.selectbox("Quality", ["auto","low","medium"], key=f"{prefix}_quality")
+    with col4:  sentiment = st.selectbox("Sentiment", ["as original image","positive","negative"], key=f"{prefix}_sentiment")
+    with col5:  platform = st.selectbox("Platform", ["Facebook","Native","Newsbreak","Google Display Network"], key=f"{prefix}_platform")
+    with col6:  num_images = st.slider("No. of Images to be generated:", 1, 10, value=2, step=1, key=f"{prefix}_num_images")
+    user_prompt = st.text_area("User Prompt", height=100, key=f"{prefix}_user_prompt")
+
+    colA, colB, colC = st.columns([1,1,1])
+    with colA: demo_btn = st.button("Generate Demo Image", key=f"{prefix}_demo")
+    with colB: gen_all_btn = st.button("Generate All Images", key=f"{prefix}_gen_all")
+    with colC: download_btn = st.button("Download All", key=f"{prefix}_download_all")
+
+    gallery_key = f"{prefix}_gallery"
+    gallery = st.session_state.setdefault(gallery_key, [])
+
+    if demo_btn or gen_all_btn:
+        if zip_file and category and user_prompt:
+            temp_dir = tempfile.mkdtemp()
+            try:
+                temp_path = os.path.join(temp_dir, zip_file.name if zip_file.name.lower().endswith(".zip") else f"input.{zip_file.name.split('.')[-1]}")
+                with open(temp_path, "wb") as f: f.write(zip_file.read())
+                images = process_zip_and_generate_images(temp_path, category, size, quality, user_prompt, sentiment, platform, num_images, demo_btn, gallery, blur)
+                st.session_state[gallery_key] = images or []; images = st.session_state[gallery_key]
+                if images:
+                    cols = st.columns(4)
+                    for idx, img_path in enumerate(images):
+                        with cols[idx % 4]: st.image(img_path, use_container_width=True)
+                else:
+                    st.info("No images generated.")
+            except Exception as e:
+                st.error(f"Error: {e}"); logger.exception("Generation failed (embedded).")
+            finally:
+                shutil.rmtree(temp_dir, ignore_errors=True)
+        else:
+            st.warning("Please upload a file and fill all required fields.")
+
+    if download_btn:
+        zip_path = _zip_gallery_images(st.session_state.get(gallery_key, []))
+        if zip_path:
+            with open(zip_path, "rb") as f:
+                st.download_button("Download ZIP", data=f.read(), file_name="generated_images.zip", mime="application/zip", key=f"{prefix}_zip_dl_btn", use_container_width=True)
+            try: os.remove(zip_path)
+            except Exception: pass
+        else:
+            st.warning("No images to zip and download.")

app_pages/image_generation_library.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, date
+from typing import Any, Dict, List
+
+import streamlit as st
+
+from database.connections import get_results_collection
+
+def _distinct_options(col, field: str, source: str) -> List[str]:
+    try:
+        vals = col.distinct(field, {"source": source})
+        vals = [v for v in vals if v not in (None, "", [])]
+        vals = sorted(set(vals))
+        return ["All"] + vals if vals else ["All"]
+    except Exception:
+        return ["All"]
+
+
+def _query_docs(
+    col,
+    source: str,
+    start_d: date,
+    end_d: date,
+    category: str,
+    platform: str,
+    limit: int = 200,
+) -> List[Dict[str, Any]]:
+    try:
+        start_dt = datetime.combine(start_d, datetime.min.time())
+        end_dt = datetime.combine(end_d + timedelta(days=1), datetime.min.time())
+        q: Dict[str, Any] = {"source": source, "created_at": {"$gte": start_dt, "$lt": end_dt}}
+        if category and category != "All":
+            q["category"] = category
+        if platform and platform != "All":
+            q["platform"] = platform
+        cursor = col.find(q).sort("created_at", -1).limit(limit)
+        return list(cursor)
+    except Exception:
+        return []
+
+
+def _render_grid(docs: List[Dict[str, Any]]) -> None:
+    seen = set()
+    items = []
+    for d in docs:
+        urls = d.get("urls") or []
+        for u in urls:
+            if u and u not in seen:
+                seen.add(u)
+                items.append((u, d))
+
+    if not items:
+        st.info("No images for the current filters.")
+        return
+
+    cols = st.columns(4)
+    for i, (url, meta) in enumerate(items):
+        with cols[i % 4]:
+            try:
+                st.image(url, use_container_width=True)
+                cat = meta.get("category") or "—"
+                
+                plat = (
+                    meta.get("platform")
+                    or (meta.get("settings") or {}).get("platform")
+                    or (meta.get("provider_settings") or {}).get("platform")
+                    or (meta.get("metadata") or {}).get("platform")
+                    or "—"
+                )
+                ts = meta.get("created_at")
+                ts_str = ts.strftime("%Y-%m-%d %H:%M UTC") if isinstance(ts, datetime) else "—"
+                st.caption(f"{cat} • {plat} • {ts_str}")
+            except Exception:
+                st.error("Failed to load image")
+
+
+def render_image_variations_library() -> None:
+
+    col = get_results_collection()
+    if col is None:
+        st.warning("Database not available. Set MONGO_URI and restart.")
+        return
+
+    today = datetime.utcnow().date()
+    default_start = today - timedelta(days=30)
+
+    c1, c2 = st.columns(2)
+    with c1:
+        start_date = st.date_input("Start date", value=default_start, key="img_var_start")
+    with c2:
+        end_date = st.date_input("End date", value=today, key="img_var_end")
+
+    cat_opts = _distinct_options(col, "category", source="variation")
+    plat_opts = _distinct_options(col, "platform", source="variation")
+
+    r1, r2 = st.columns(2)
+    with r1:
+        category = st.selectbox("Category", options=cat_opts, key="img_var_cat")
+    with r2:
+        platform = st.selectbox("Platform", options=plat_opts, key="img_var_plat")
+
+    st.markdown("---")
+    docs = _query_docs(col, "variation", start_date, end_date, category, platform)
+    _render_grid(docs)

app_pages/script_generator.py

@@ -0,0 +1,143 @@
+import os
+import tempfile
+import pandas as pd
+import streamlit as st
+
+from generator_function.script_generator import generate_scripts
+from helpers_function.helpers import get_video_thumbnail_base64
+from database.operations import insert_script_result
+from authen.authentication import login_gate
+
+uid = login_gate()
+
+def _fmt_ts(ts) -> str:
+    try:
+        return ts.strftime("%Y-%m-%d %H:%M")
+    except Exception:
+        return "Unknown time"
+
+
+def generator_page():
+    st.subheader("Script Generator")
+
+    uploaded_video = st.file_uploader(
+        "Upload Video or ZIP (max 3 videos)", type=["mp4", "mov", "avi", "mkv", "zip"], key="sg_upload"
+    )
+    script_duration = st.slider("Script Duration (seconds)", 0, 180, 60, 5, key="sg_dur")
+    num_scripts = st.slider("Number of Scripts", 1, 5, 3, key="sg_count")
+
+    st.markdown("Additional Information")
+    offer_details = st.text_area(
+        "Offer Details",
+        placeholder="e.g., Solar installation with $0 down payment...",
+        key="sg_offer",
+    )
+    target_audience = st.text_area(
+        "Target Audience",
+        placeholder="e.g., 40+ homeowners with high electricity bills...",
+        key="sg_aud",
+    )
+    specific_hooks = st.text_area(
+        "Specific Hooks to Test",
+        placeholder="e.g., Government rebate angle...",
+        key="sg_hooks",
+    )
+    additional_context = st.text_area(
+        "Additional Context",
+        placeholder="Compliance requirements, brand guidelines...",
+        key="sg_ctx",
+    )
+
+    script_button = st.button("Generate Scripts", use_container_width=True, key="sg_go")
+    if script_button and uploaded_video:
+        with tempfile.NamedTemporaryFile(
+            delete=False, suffix=os.path.splitext(uploaded_video.name)[1]
+        ) as tmp:
+            src_bytes = uploaded_video.read()
+            tmp.write(src_bytes)
+            video_path = tmp.name
+
+        with st.spinner("Generating scripts..."):
+            st.session_state.setdefault("scripts", [])
+            result = generate_scripts(
+                video_path,
+                offer_details,
+                target_audience,
+                specific_hooks,
+                additional_context,
+                num_scripts=num_scripts,
+                duration=script_duration,
+            )
+            if result and "script_variations" in result:
+                st.session_state["scripts"].append(
+                    {"prompt_used": "Initial Generation", "variations": result["script_variations"]}
+                )
+                st.session_state["video_name"] = uploaded_video.name
+                st.session_state["video_path"] = video_path
+                try:
+                    st.session_state["thumbnail"] = get_video_thumbnail_base64(video_path)
+                except Exception:
+                    st.session_state["thumbnail"] = ""
+                st.session_state["meta"] = {
+                    "offer_details": offer_details,
+                    "target_audience": target_audience,
+                    "specific_hook": specific_hooks,
+                    "additional_context": additional_context,
+                }
+
+    if "scripts" in st.session_state and st.session_state["scripts"]:
+        for round_idx, round_data in enumerate(st.session_state["scripts"], 1):
+            st.markdown(f"### Generation Round {round_idx}")
+            st.text_input(
+                "Prompt used:", round_data["prompt_used"], disabled=True, key=f"prompt_{round_idx}"
+            )
+            for i, variation in enumerate(round_data["variations"], 1):
+                st.markdown(f"#### Variation {i}: {variation.get('variation_name', 'Var')}")
+                df = pd.DataFrame(variation.get("script_table", []))
+                st.table(df)
+
+        st.divider()
+        save_button = st.button("Save to DB", use_container_width=True, key="sg_save")
+        if save_button:
+            try:
+                insert_script_result(
+                    video_name=st.session_state.get("video_name", "unknown"),
+                    offer_details=st.session_state["meta"].get("offer_details", ""),
+                    target_audience=st.session_state["meta"].get("target_audience", ""),
+                    specific_hook=st.session_state["meta"].get("specific_hook", ""),
+                    additional_context=st.session_state["meta"].get("additional_context", ""),
+                    response=st.session_state["scripts"],
+                    thumbnail=st.session_state.get("thumbnail", ""),
+                    created_by=uid,
+                    num_scripts=len(st.session_state["scripts"][-1]["variations"]),
+                    category="general",
+                )
+                st.success("Scripts saved to database!")
+            except Exception as e:
+                st.error(f"Failed to save scripts: {e}")
+
+        st.subheader("Generate More Scripts")
+        more_num = st.slider("How many more scripts?", 1, 5, 1, key="sg_more_count")
+        more_prompt = st.text_area("Required Prompt", placeholder="Add specific guidance", key="sg_more_prompt")
+        if st.button("Generate More Scripts", use_container_width=True, key="sg_more_go"):
+            if not more_prompt.strip():
+                st.error("Please provide a prompt before generating more scripts.")
+            else:
+                video_path = st.session_state.get("video_path")
+                if not video_path:
+                    st.error("No video available. Please upload again.")
+                else:
+                    with st.spinner("Generating more scripts..."):
+                        extra_result = generate_scripts(
+                            video_path,
+                            st.session_state["meta"]["offer_details"],
+                            st.session_state["meta"]["target_audience"],
+                            st.session_state["meta"]["specific_hook"],
+                            st.session_state["meta"]["additional_context"] + "\n\n" + more_prompt,
+                            num_scripts=more_num,
+                            duration=script_duration,
+                        )
+                        if extra_result and "script_variations" in extra_result:
+                            st.session_state["scripts"].append(
+                                {"prompt_used": more_prompt, "variations": extra_result["script_variations"]}
+                            )

app_pages/script_library.py

@@ -0,0 +1,90 @@
+from datetime import datetime, timedelta
+from typing import Any, Dict
+
+from database.operations import find_script_results
+import streamlit as st
+import pandas as pd
+
+def render_script_library() ->None:
+    st.subheader("Generated Scripts")
+
+    today = datetime.utcnow().date()
+    default_start = today - timedelta(days=30)
+
+    f1, f2, _ = st.columns([1, 1, 1.2])
+    with f1:
+        start_date_scripts = st.date_input("Start date", value=default_start, key="scripts_start")
+    with f2:
+        end_date_scripts = st.date_input("End date", value=today, key="scripts_end")
+
+    if "scripts_page" not in st.session_state:
+        st.session_state.scripts_page = 0
+
+    start_dt = datetime.combine(start_date_scripts, datetime.min.time())
+    end_dt = datetime.combine(end_date_scripts + timedelta(days=1), datetime.min.time())
+
+    records, total_count = find_script_results(
+        start_date=start_dt,
+        end_date=end_dt,
+        page=st.session_state.scripts_page,
+        page_size=20,
+    )
+
+    if total_count == 0 or not records:
+        st.info("No script generations for the selected filters.")
+        st.stop()
+
+    start_idx = st.session_state.scripts_page * 20 + 1
+    end_idx = min(start_idx + len(records) - 1, total_count)
+    st.caption(f"Showing {start_idx}-{end_idx} of {total_count} items")
+
+    def _label(d: Dict[str, Any]) -> str:
+        ts = d.get("created_at")
+        ts_s = ts.strftime("%Y-%m-%d %H:%M") if hasattr(ts, "strftime") else "Unknown time"
+        vn = d.get("video_name") or d.get("file_name") or "Untitled"
+        return f"{ts_s} · {vn}"
+
+    options = [_label(d) for d in records]
+    selected_label = st.selectbox("Select generated script", options=options, index=0, key="scripts_sel_dropdown")
+    sel_idx = options.index(selected_label) if selected_label in options else 0
+    doc = records[sel_idx]
+
+    if doc.get("thumbnail"):
+        try:
+            st.image("data:image/jpeg;base64," + doc["thumbnail"], width=160, caption="Thumbnail")
+        except Exception:
+            pass
+
+    json_response = doc.get("response")
+    if not json_response:
+        st.info("No variations saved.")
+    else:
+        all_tables = []
+        if isinstance(json_response, list):
+            for round_idx, round_data in enumerate(json_response, 1):
+                st.markdown(f"#### Generation Round {round_idx}")
+                st.text_input(
+                    "Prompt used:",
+                    round_data.get("prompt_used", "N/A"),
+                    disabled=True,
+                    key=f"hist_prompt_{doc.get('_id')}_{round_idx}",
+                )
+                for i, variation in enumerate(round_data.get("variations", []), 1):
+                    st.markdown(f"**Variation {i}: {variation.get('variation_name', 'Var')}**")
+                    df = pd.DataFrame(variation.get("script_table", []))
+                    st.table(df)
+                    if not df.empty:
+                        df["Variation"] = variation.get("variation_name", f"Var{i}")
+                        df["Round"] = round_idx
+                        all_tables.append(df)
+
+        if all_tables:
+            csv_scripts = pd.concat(all_tables, ignore_index=True).to_csv(index=False)
+            st.download_button(
+                "Download CSV",
+                data=csv_scripts,
+                file_name=f"{(doc.get('video_name') or doc.get('file_name') or 'scripts')}_scripts.csv",
+                mime="text/csv",
+                use_container_width=True,
+                key="scripts_download_csv",
+            )

app_pages/text_image_generation.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+import streamlit as st
+
+from generator_function.image_function import (
+    handle_image_generation_optimized,
+    MODEL_REGISTRY,
+    get_model_config,
+
+)
+
+def render_multimodel_ad_generator(prefix: str = "img_text"):
+    colA, colB = st.columns([1, 1])
+
+    with colA:
+        model_key = st.selectbox("Model", list(MODEL_REGISTRY.keys()), index=0, key=f"{prefix}_model")
+        aspect_options = get_model_config(model_key)["aspect_ratios"] if get_model_config(model_key) else ["1:1"]
+        aspect_ratio = st.selectbox("Aspect Ratio", aspect_options, index=0, key=f"{prefix}_ar")
+        num_images = st.slider("Number of images", 1, 10, 1, key=f"{prefix}_count")
+
+        category = st.text_input("Category", key=f"{prefix}_cat", placeholder="e.g. skincare, supplements")
+        platform = st.selectbox(
+            "Platform",
+            ["Facebook", "Instagram", "TikTok", "YouTube", "Google Display Network"],
+            index=0,
+            key=f"{prefix}_plat",
+        )
+
+    with colB:
+        prompt = st.text_area("Prompt", placeholder="Describe the image you want to generate...", height=160, key=f"{prefix}_prompt")
+        debug_mode = st.checkbox("Debug mode", key=f"{prefix}_debug", value=False)
+
+    if st.button("Generate Images", type="primary", key=f"{prefix}_go"):
+        handle_image_generation_optimized(
+            model_key=model_key,
+            aspect_ratio=aspect_ratio,
+            prompt=prompt,
+            num_images=num_images,
+            debug_mode=debug_mode,
+            category=(category.strip() or None),
+            platform=platform or None,
+        )

app_pages/text_image_generation_library.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, date
+from typing import Any, Dict, List
+
+import streamlit as st
+
+from database.connections import get_results_collection
+
+
+def _distinct_options(col, field: str, source: str) -> List[str]:
+    try:
+        vals = col.distinct(field, {"source": source})
+        vals = [v for v in vals if v not in (None, "", [])]
+        vals = sorted(set(vals))
+        return ["All"] + vals if vals else ["All"]
+    except Exception:
+        return ["All"]
+
+
+def _query_docs(
+    col,
+    source: str,
+    start_d: date,
+    end_d: date,
+    category: str,
+    platform: str,
+    limit: int = 200,
+) -> List[Dict[str, Any]]:
+    try:
+        start_dt = datetime.combine(start_d, datetime.min.time())
+        end_dt = datetime.combine(end_d + timedelta(days=1), datetime.min.time())
+        q: Dict[str, Any] = {"source": source, "created_at": {"$gte": start_dt, "$lt": end_dt}}
+        if category and category != "All":
+            q["category"] = category
+        if platform and platform != "All":
+            q["platform"] = platform
+        cursor = col.find(q).sort("created_at", -1).limit(limit)
+        return list(cursor)
+    except Exception:
+        return []
+
+
+def _render_grid(docs: List[Dict[str, Any]], key_prefix: str) -> None:
+    seen = set()
+    items = []
+    for d in docs:
+        urls = d.get("urls") or []
+        for u in urls:
+            if u and u not in seen:
+                seen.add(u)
+                items.append((u, d))
+
+    if not items:
+        st.info("No images for the current filters.")
+        return
+
+    cols = st.columns(4)
+    for i, (url, meta) in enumerate(items):
+        with cols[i % 4]:
+            try:
+                st.image(url, use_container_width=True)
+                cat = meta.get("category") or "—"
+                plat = (
+                    meta.get("platform")
+                    or (meta.get("settings") or {}).get("platform")
+                    or (meta.get("provider_settings") or {}).get("platform")
+                    or (meta.get("metadata") or {}).get("platform")
+                    or "—"
+                )
+                ts = meta.get("created_at")
+                ts_str = ts.strftime("%Y-%m-%d %H:%M UTC") if isinstance(ts, datetime) else "—"
+                st.caption(f"{cat} • {plat} • {ts_str}")
+            except Exception:
+                st.error("Failed to load image")
+
+
+def render_text_image_library() -> None:
+
+    col = get_results_collection()
+    if col is None:
+        st.warning("Database not available. Set MONGO_URI and restart.")
+        return
+
+    today = datetime.utcnow().date()
+    default_start = today - timedelta(days=30)
+
+    c1, c2 = st.columns(2)
+    with c1:
+        start_date = st.date_input("Start date", value=default_start, key="img_text_start")
+    with c2:
+        end_date = st.date_input("End date", value=today, key="img_text_end")
+
+    cat_opts = _distinct_options(col, "category", source="text")
+    plat_opts = _distinct_options(col, "platform", source="text")
+
+    r1, r2 = st.columns(2)
+    with r1:
+        category = st.selectbox("Category", options=cat_opts, key="img_text_cat")
+    with r2:
+        platform = st.selectbox("Platform", options=plat_opts, key="img_text_plat")
+
+    st.markdown("---")
+    docs = _query_docs(col, "text", start_date, end_date, category, platform)
+    _render_grid(docs, key_prefix="text")

app_pages/video_analyser.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+import os, tempfile
+import streamlit as st
+import base64
+from typing import Any, Dict
+from background_task.generation_tasks import run_and_store_video_analysis
+from app_pages.video_library import render_analyzer_results
+
+
+def render_video_library_page() -> None:
+    from app_pages.video_library import render_video_library
+    render_video_library()
+
+
+def render_video_analyser_page(task_manager=None, uid: str = "anonymous") -> None:
+    st.markdown("### Video Analyser")
+
+    category_va = st.text_input("Category", key="va_category")
+    uploaded = st.file_uploader(
+        "Upload video",
+        type=["mp4", "mov", "mkv", "webm", "avi"],
+        key="va_upload",
+    )
+
+    video_bytes = None
+    video_ext = "mp4"
+
+    if uploaded is not None:
+
+        video_bytes = uploaded.read()
+        if uploaded.name and "." in uploaded.name:
+            video_ext = uploaded.name.rsplit(".", 1)[-1].lower() or "mp4"
+
+
+        b64 = base64.b64encode(video_bytes).decode("utf-8")
+        st.caption("Preview")
+        st.markdown(
+            f"""
+                        <video controls width="720">
+                            <source src="data:video/{video_ext};base64,{b64}">
+                            Your browser does not support the video tag.
+                        </video>
+                        """,
+            unsafe_allow_html=True,
+        )
+
+    go = st.button("Analyze", key="va_analyse")
+
+    if go:
+        if not video_bytes or not category_va.strip():
+            st.warning("Please provide a category and upload a video.")
+        else:
+
+            suffix = f".{video_ext}" if video_ext else ".mp4"
+            with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+                tmp.write(video_bytes)
+                tmp_path = tmp.name
+
+            try:
+                with st.spinner("Analyzing video..."):
+                    result: Dict[str, Any] = run_and_store_video_analysis(
+                        category=category_va.strip(),
+                        uploaded_file_path=tmp_path,
+                        created_by=uid,
+                    )
+            finally:
+                try:
+                    os.remove(tmp_path)
+                except Exception:
+                    pass
+
+            if isinstance(result, dict) and result.get("results"):
+                st.success(f"Saved analysis (_id: {result.get('_id', 'N/A')})")
+                render_analyzer_results(result["results"])
+            else:
+                st.error("Analysis failed. Check GEMINI_API_KEY and try again.")
+

app_pages/video_library.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, date
+from typing import Any, Dict, List, Optional
+
+import streamlit as st
+
+from database.operations import find_video_analyses, list_video_categories
+from components.render_analysis import render_analyzer_results
+
+
+def _coerce_dt(val: Any) -> Optional[datetime]:
+    if isinstance(val, datetime):
+        return val
+    try:
+        return datetime.fromisoformat(str(val))
+    except Exception:
+        return None
+
+
+def _label_for_item(doc: Dict[str, Any]) -> str:
+    """Shown in the wide dropdown."""
+    ts = _coerce_dt(doc.get("created_at"))
+    ts_s = ts.strftime("%Y-%m-%d %H:%M") if ts else "Unknown time"
+    cat = doc.get("category") or "—"
+    model = doc.get("analyzer_model") or "—"
+    return f"{ts_s} · {cat} · {model}"
+
+
+def render_video_library() -> None:
+    st.subheader("Video Library")
+
+    # --- Filters (unchanged layout) ---
+    today = datetime.utcnow().date()
+    default_start = today - timedelta(days=30)
+
+    c1, c2, c3 = st.columns([1, 1, 1])
+    with c1:
+        start_date: date = st.date_input("Start date", value=default_start, key="vid_lib_start")
+    with c2:
+        end_date: date = st.date_input("End date", value=today, key="vid_lib_end")
+    with c3:
+        try:
+            cats: List[str] = list_video_categories()
+            if "All" not in cats:
+                cats = ["All"] + [c for c in cats if c != "All"]
+        except Exception:
+            cats = ["All"]
+        category = st.selectbox("Category", options=cats, index=0, key="vid_lib_cat")
+
+    st.markdown("---")
+
+
+    try:
+        start_dt = datetime.combine(start_date, datetime.min.time())
+        end_dt = datetime.combine(end_date + timedelta(days=1), datetime.min.time())
+        query_cat = None if (not category or category == "All") else category
+        docs: List[Dict[str, Any]] = find_video_analyses(
+            category=query_cat, start_date=start_dt, end_date=end_dt, limit=200
+        )
+    except Exception as e:
+        st.error(f"Failed to load video analyses: {e}")
+        return
+
+    if not docs:
+        st.info("No video analyses for the selected filters.")
+        return
+
+
+    labels = [_label_for_item(d) for d in docs]
+    selected_label = st.selectbox(
+        "Select an analysis",
+        options=labels,
+        index=0,
+        key="vid_lib_sel",
+    )
+    sel_idx = labels.index(selected_label) if selected_label in labels else 0
+    doc = docs[sel_idx]
+
+    thumb_b64 = doc.get("thumbnail")
+    if thumb_b64:
+        try:
+            st.image(
+                f"data:image/jpeg;base64,{thumb_b64}",
+                caption="Thumbnail",
+                width=220,
+            )
+        except Exception:
+            pass
+
+
+    analysis = doc.get("results") or {}
+    if analysis:
+        render_analyzer_results(analysis)
+        try:
+            pretty = json.dumps(analysis, indent=2, ensure_ascii=False)
+            st.download_button(
+                "Download JSON",
+                data=pretty.encode("utf-8"),
+                file_name=f"video_analysis_{doc.get('_id','item')}.json",
+                mime="application/json",
+                use_container_width=True,
+                key="vid_lib_json_dl",
+            )
+        except Exception:
+            pass
+    else:
+        st.info("No analysis  stored for this item.")

authen/__init__.py

@@ -0,0 +1 @@
+

authen/authentication.py

@@ -0,0 +1,100 @@
+import re
+from datetime import datetime
+
+import streamlit as st
+import os
+from pymongo import MongoClient, ASCENDING
+from pymongo.errors import OperationFailure
+from dotenv import load_dotenv
+
+from .password_utils import hash_password, verify_password, verify_legacy_password
+
+load_dotenv()
+EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
+
+def _users_col():
+
+    uri = os.getenv("MONGO_URI")
+    dbname = os.getenv("MONGO_DB", "Gen_Ai")
+    coll = os.getenv("USERS_COLLECTION", "users")
+    if not uri:
+        raise RuntimeError("MONGO_URI is not set")
+    client = MongoClient(uri)
+    col = client[dbname][coll]
+    _ensure_email_unique_index(col)
+    return col
+
+def _ensure_email_unique_index(col):
+    try:
+        unique_ok, to_drop = False, None
+        for idx in col.list_indexes():
+            keys = list(idx.get("key", {}).items())
+            if keys == [("email", 1)]:
+                if idx.get("unique", False): unique_ok = True
+                else: to_drop = idx["name"]
+                break
+        if to_drop: col.drop_index(to_drop)
+        if not unique_ok: col.create_index([("email", ASCENDING)], unique=True)
+    except OperationFailure:
+        pass
+
+def login_or_create(email: str, password: str):
+    if not email or not password:
+        return False, "Enter both email and password."
+    email_norm = email.strip().lower()
+    if not EMAIL_RE.match(email_norm):
+        return False, "Enter a valid email address."
+
+    col = _users_col()
+    user = col.find_one({"email": email_norm})
+
+    if user:
+        pwd = user.get("password")
+        if isinstance(pwd, str):
+            if not verify_password(password, pwd):
+                return False, "Invalid credentials."
+            col.update_one({"_id": user["_id"]}, {"$set": {"last_login": datetime.utcnow()}})
+            st.session_state["uid"] = email_norm
+            return True, email_norm
+
+        if isinstance(pwd, dict):
+            if verify_legacy_password(password, pwd.get("salt"), pwd.get("hash")):
+                new_hash = hash_password(password)
+                col.update_one(
+                    {"_id": user["_id"]},
+                    {"$set": {"password": new_hash, "last_login": datetime.utcnow()}}
+                )
+                st.session_state["uid"] = email_norm
+                return True, email_norm
+            return False, "Invalid credentials."
+    else:
+        st.warning("Invalid Credentials")
+
+
+
+def logout():
+    st.session_state.pop("uid", None)
+    st.session_state.pop("login_ts", None)
+
+def login_gate() -> str:
+    if st.session_state.get("uid"):
+        return st.session_state["uid"]
+
+    st.markdown("<div style='height: 6vh'></div>", unsafe_allow_html=True)
+    c1, c2, c3 = st.columns([1, 1, 1])
+    with c2:
+        st.markdown("### Log in")
+        with st.form("login_form", clear_on_submit=False):
+            email = st.text_input("Email", key="login_email", placeholder="you@example.com")
+            password = st.text_input("Password", type="password", key="login_pass")
+            submit = st.form_submit_button("Continue", use_container_width=True)
+        if submit:
+            ok, msg = login_or_create(email, password)
+            if ok:
+                st.session_state["uid"] = msg
+                st.session_state["login_ts"] = datetime.utcnow().isoformat()
+                st.rerun()
+            else:
+                st.error(msg)
+    st.stop()
+

authen/password_utils.py

@@ -0,0 +1,18 @@
+from typing import Optional
+import bcrypt, binascii, hashlib, secrets
+
+def hash_password(password: str) -> str:
+    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt(rounds=12)).decode("utf-8")
+
+def verify_password(password: str, hashed: str) -> bool:
+    try:
+        return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8"))
+    except Exception:
+        return False
+
+def verify_legacy_password(password: str, salt_hex: Optional[str], hash_hex: Optional[str]) -> bool:
+    if not salt_hex or not hash_hex:
+        return False
+    salt = binascii.unhexlify(salt_hex.encode())
+    dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 200_000)
+    return secrets.compare_digest(binascii.hexlify(dk).decode(), hash_hex)

background_task/__init__.py

@@ -0,0 +1,18 @@
+from .task_manager import TaskManager
+from .generation_tasks import (
+    register_video_analyzer_tasks,
+    enqueue_video_analyzer,
+    register_script_generator_tasks,
+    enqueue_script_generation,
+    register_text_to_image_tasks,
+    enqueue_text_to_image,
+    register_image_gen_tasks,
+    enqueue_image_gen,
+)
+
+def register_all_background_tasks(manager: TaskManager) -> None:
+    """Register all background tasks with a given TaskManager instance."""
+    register_video_analyzer_tasks(manager)
+    register_script_generator_tasks(manager)
+    register_text_to_image_tasks(manager)
+    register_image_gen_tasks(manager)

background_task/generation_tasks.py

@@ -0,0 +1,288 @@
+from __future__ import annotations
+
+import os
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+
+from generator_function.video_analyzer_services import run_and_store_video_analysis
+from generator_function.script_generator import generate_scripts
+from database.operations import insert_script_result
+from helpers_function.helpers import get_video_thumbnail_base64
+from database.connections import get_results_collection
+from generator_function.image_function import generate_images_parallel
+
+from core.task_enum import TaskType
+from .utils import safe_copy_temp, safe_unlink
+
+from database.operations import start_job, finish_job
+
+# --- TEXT -> IMAGE
+
+def background_text_to_image(
+    job_id: str,
+    progress_cb,
+    *,
+    model_key: str,
+    aspect_ratio: str,
+    prompt: str,
+    num_images: int,
+    category: Optional[str],
+    platform: Optional[str],
+    created_by: Optional[str],
+) -> Dict[str, Any]:
+    """
+    Runs multi-model text-to-image generation and logs a job to the results collection.
+    """
+    progress_cb(job_id, 10, "Starting text→image generation...")
+
+
+    results_col = get_results_collection()
+    db_job_id: Optional[str] = None
+
+    if results_col is not None:
+        db_job_id = start_job(
+            results_col,
+            type="generation",
+            created_by=created_by or "anonymous",
+            category=(category or "general"),
+            inputs={"model_key": model_key, "aspect_ratio": aspect_ratio, "num_images": num_images},
+            settings={"platform": platform},
+            user_prompt=prompt,
+        )
+
+    r2_urls, source_urls, errors = generate_images_parallel(model_key, aspect_ratio, prompt, num_images)
+    urls = r2_urls or source_urls
+
+    if results_col is not None and db_job_id:
+        finish_job(
+            results_col,
+            db_job_id,
+            status="completed" if urls else "failed",
+            outputs_urls=urls or [],
+            provider_update={"errors": errors} if errors else None,
+        )
+
+    progress_cb(job_id, 100, f"Generated {len(urls)} image(s).")
+    return {"success": True, "type": TaskType.TEXT_TO_IMAGE, "urls": urls, "errors": errors or []}
+
+
+def register_text_to_image_tasks(task_manager) -> None:
+    task_manager.register(TaskType.TEXT_TO_IMAGE, background_text_to_image)
+
+
+def enqueue_text_to_image(task_manager, **kwargs) -> str:
+    job_id = task_manager.create_job(TaskType.TEXT_TO_IMAGE)
+    task_manager.submit(job_id, background_text_to_image, **kwargs)
+    return job_id
+
+
+# --- IMAGE GEN
+
+def background_image_gen(
+    job_id: str,
+    progress_cb,
+    *,
+    upload_path: str,
+    category: str,
+    size: str,
+    quality: str,
+    user_prompt: str,
+    sentiment: str,
+    platform: str,
+    num_images: int,
+    blur: bool,
+) -> Dict[str, Any]:
+    """
+    Runs image generation from uploaded asset(s) (zip or single); returns paths/urls.
+    """
+    progress_cb(job_id, 10, "Processing input archive...")
+
+    from generator_function.image_processor import process_zip_and_generate_images
+
+
+    images = process_zip_and_generate_images(
+        upload_path, category, size, quality, user_prompt, sentiment, platform, num_images, False, [], blur
+    )
+
+    progress_cb(job_id, 100, f"Generated {len(images)} images.")
+    return {"success": True, "type": TaskType.IMAGE_GEN, "images": images or []}
+
+
+def register_image_gen_tasks(task_manager) -> None:
+    task_manager.register(TaskType.IMAGE_GEN, background_image_gen)
+
+
+def enqueue_image_gen(task_manager, **kwargs) -> str:
+    job_id = task_manager.create_job(TaskType.IMAGE_GEN)
+    task_manager.submit(job_id, background_image_gen, **kwargs)
+    return job_id
+
+
+# --- VIDEO ANALYZER
+
+def background_video_analyzer(
+    job_id: str,
+    progress_cb,
+    *,
+    uploaded_file_path: str,
+    uploaded_file_name: str,
+    category: str,
+    created_by: Optional[str],
+    script_num: int = 3,
+    script_duration: int = 60,
+    offer_details: str = "",
+    target_audience: str = "",
+    specific_hooks: str = "",
+    additional_context: str = "",
+    task_manager=None,
+) -> Dict[str, Any]:
+    """
+    Analyzes the uploaded video and (optionally) chains a background script-gen job.
+    """
+    progress_cb(job_id, 5, "Preparing video...")
+
+    if not os.path.exists(uploaded_file_path):
+        raise FileNotFoundError(uploaded_file_path)
+
+
+
+    tmp = safe_copy_temp(
+        uploaded_file_path,
+        suffix=(os.path.splitext(uploaded_file_name or ".mp4")[1] or ".mp4"),
+    )
+    try:
+        progress_cb(job_id, 20, "Analyzing video...")
+        result = run_and_store_video_analysis(
+            category=category,
+            uploaded_file_path=tmp,
+            created_by=created_by,
+        )
+
+        if not isinstance(result, dict) or not result.get("results"):
+            raise RuntimeError("Video analysis failed.")
+
+        progress_cb(job_id, 95, "Analysis saved.")
+
+        chained_job_id = None
+        if task_manager:
+            chained_job_id = enqueue_script_generation(
+                task_manager,
+                video_path=tmp,
+                video_name=uploaded_file_name,
+                offer_details=offer_details,
+                target_audience=target_audience,
+                specific_hooks=specific_hooks,
+                additional_context=additional_context,
+                num_scripts=script_num,
+                duration=script_duration,
+                created_by=created_by,
+                category=category,
+            )
+
+        progress_cb(job_id, 100, "Video analysis complete.")
+        return {
+            "success": True,
+            "type": TaskType.VIDEO_ANALYZER,
+            "video_analysis_id": result.get("_id"),
+            "chained_script_job_id": chained_job_id,
+        }
+    finally:
+        #
+        safe_unlink(tmp)
+
+
+def register_video_analyzer_tasks(task_manager) -> None:
+    task_manager.register(TaskType.VIDEO_ANALYZER, background_video_analyzer)
+
+
+def enqueue_video_analyzer(task_manager, **kwargs) -> str:
+    job_id = task_manager.create_job(TaskType.VIDEO_ANALYZER)
+    task_manager.submit(job_id, background_video_analyzer, task_manager=task_manager, **kwargs)
+    return job_id
+
+
+# --- SCRIPT GENERATOR --------------------------------------------------------
+
+def background_script_generation(
+    job_id: str,
+    progress_cb,
+    *,
+    video_path: str,
+    video_name: str,
+    offer_details: str,
+    target_audience: str,
+    specific_hooks: str,
+    additional_context: str,
+    num_scripts: int,
+    duration: int,
+    created_by: Optional[str],
+    category: Optional[str] = None,
+) -> Dict[str, Any]:
+    """
+    Generates script variations from the provided video and persists the run.
+    """
+    progress_cb(job_id, 5, "Preparing inputs...")
+
+    if not os.path.exists(video_path):
+        raise FileNotFoundError(video_path)
+
+
+
+    tmp = safe_copy_temp(video_path, suffix=(os.path.splitext(video_name or ".mp4")[1] or ".mp4"))
+    try:
+        progress_cb(job_id, 20, "Generating scripts...")
+        gen = generate_scripts(
+            tmp,
+            offer_details,
+            target_audience,
+            specific_hooks,
+            additional_context,
+            num_scripts=max(1, int(num_scripts)),
+            duration=max(0, int(duration)),
+        )
+        if not gen or "script_variations" not in gen or not gen["script_variations"]:
+            raise RuntimeError("Script generation returned no variations.")
+
+        packed_round = [{"prompt_used": "Auto after analysis", "variations": gen["script_variations"]}]
+
+        progress_cb(job_id, 75, "Creating thumbnail...")
+        thumb = ""
+        try:
+            thumb = get_video_thumbnail_base64(tmp) or ""
+        except Exception:
+            pass
+
+        progress_cb(job_id, 90, "Saving run...")
+        insert_script_result(
+            video_name=video_name or os.path.basename(tmp),
+            offer_details=offer_details or "",
+            target_audience=target_audience or "",
+            specific_hook=specific_hooks or "",
+            additional_context=additional_context or "",
+            response=packed_round,
+            thumbnail=thumb,
+            created_by=created_by,
+            num_scripts=len(gen["script_variations"]),
+            category=category or "general",
+        )
+
+        progress_cb(job_id, 100, "Scripts saved.")
+        return {
+            "success": True,
+            "type": TaskType.SCRIPT_GENERATION,
+            "num_variations": len(gen["script_variations"]),
+            "created_at": datetime.utcnow().isoformat(),
+        }
+    finally:
+        safe_unlink(tmp)
+
+
+def register_script_generator_tasks(task_manager) -> None:
+    task_manager.register(TaskType.SCRIPT_GENERATION, background_script_generation)
+
+
+def enqueue_script_generation(task_manager, **kwargs) -> str:
+    job_id = task_manager.create_job(TaskType.SCRIPT_GENERATION)
+    task_manager.submit(job_id, background_script_generation, **kwargs)
+    return job_id

background_task/task_manager.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+import threading, uuid, traceback
+from concurrent.futures import ThreadPoolExecutor
+from typing import Callable, Dict, Any, Optional
+
+ProgressCB = Callable[[str, int, str], None]
+
+class TaskManager:
+    def __init__(self, max_workers: int = 8):
+        self._exec = ThreadPoolExecutor(max_workers=max_workers)
+        self._lock = threading.Lock()
+        self._jobs: Dict[str, dict] = {}
+        self._registry: Dict[str, Callable[..., Any]] = {}
+
+    # ---- job bookkeeping ----
+    def create_job(self, name: str) -> str:
+        job_id = uuid.uuid4().hex
+        with self._lock:
+            self._jobs[job_id] = {"name": name, "status": "queued", "progress": 0, "message": "", "result": None, "error": None}
+        return job_id
+
+    def register(self, name: str, worker: Callable[..., Any]) -> None:
+        self._registry[name] = worker
+
+    def progress_update(self, job_id: str, percent: int, message: str = "") -> None:
+        with self._lock:
+            job = self._jobs.get(job_id)
+            if job:
+                job["progress"] = max(0, min(100, int(percent)))
+                if message:
+                    job["message"] = message
+
+    def _set_status(self, job_id: str, status: str) -> None:
+        with self._lock:
+            job = self._jobs.get(job_id)
+            if job:
+                job["status"] = status
+
+    def submit(self, job_id: str, worker: Callable[..., Any], **kwargs) -> None:
+        self._set_status(job_id, "running")
+
+        def _runner():
+            try:
+                result = worker(job_id, self.progress_update, **kwargs)
+                with self._lock:
+                    self._jobs[job_id]["result"] = result
+                    self._jobs[job_id]["status"] = "completed"
+                    self._jobs[job_id]["progress"] = 100
+            except Exception:
+                err = traceback.format_exc()
+                with self._lock:
+                    self._jobs[job_id]["status"] = "failed"
+                    self._jobs[job_id]["error"] = err
+
+        self._exec.submit(_runner)
+
+    # ---- accessors for UI ----
+    def get(self, job_id: str) -> Optional[dict]:
+        with self._lock:
+            return dict(self._jobs.get(job_id) or {})
+
+    def list_jobs(self) -> Dict[str, dict]:
+        with self._lock:
+            return {k: dict(v) for k, v in self._jobs.items()}

background_task/utils.py

@@ -0,0 +1,12 @@
+import os, shutil, tempfile
+
+def safe_copy_temp(src_path: str, suffix: str) -> str:
+    """Copy a file to a managed temp path so the caller can delete their tmp safely."""
+    out = tempfile.NamedTemporaryFile(delete=False, suffix=suffix)
+    out.close()
+    shutil.copyfile(src_path, out.name)
+    return out.name
+
+def safe_unlink(path: str) -> None:
+    try: os.unlink(path)
+    except Exception: pass

components/display_variations.py

@@ -0,0 +1,19 @@
+import pandas as pd
+import streamlit as st
+
+def display_script_variations(json_data: dict):
+    if not json_data or "script_variations" not in json_data:
+        st.error("No script variations found")
+        return
+    for i, variation in enumerate(json_data["script_variations"], 1):
+        st.markdown(f"### Variation {i}: {variation.get('variation_name','Var')}")
+        df = pd.DataFrame(variation.get("script_table", []))
+        st.table(df)
+    csv_content = pd.concat(
+        [pd.DataFrame(v.get("script_table", []))
+         .assign(Variation=v.get("variation_name", f"Var{i+1}"))
+         for i, v in enumerate(json_data["script_variations"])],
+        ignore_index=True
+    ).to_csv(index=False)
+    st.download_button("Download CSV", data=csv_content,
+                       file_name="scripts.csv", mime="text/csv")

components/render_analysis.py

@@ -0,0 +1,141 @@
+import json, pandas as pd, streamlit as st
+from typing import Dict, Any
+
+def _normalize_list(v):
+    if v is None: return []
+    if isinstance(v, list): return [str(x) for x in v]
+    return [s for s in str(v).splitlines() if s.strip()]
+
+def _to_dataframe(items, columns_map):
+    import pandas as pd
+    if not isinstance(items, list) or not items:
+        return pd.DataFrame(columns=list(columns_map.values()))
+    df = pd.DataFrame(items).rename(columns=columns_map)
+    ordered_cols = [columns_map[k] for k in columns_map.keys() if columns_map[k] in df.columns]
+    return df.reindex(columns=ordered_cols)
+
+def _mean_effectiveness(metrics):
+    if not metrics: return 0.0
+    scores = []
+    for m in metrics:
+        s = str(m.get("effectiveness_score", "0/10")).split("/")[0]
+        try: scores.append(int(s))
+        except Exception: pass
+    return round(sum(scores) / len(scores), 2) if scores else 0.0
+
+def _search_dataframe(df, query):
+    if not query or df.empty: return df
+    mask = df.apply(lambda col: col.astype(str).str.contains(query, case=False, na=False))
+    return df[mask.any(axis=1)]
+
+def render_analyzer_results(analysis: Dict[str, Any]) -> None:
+    if not isinstance(analysis, dict) or not analysis:
+        st.warning("No analysis available."); return
+
+    st.markdown(
+        """<style>
+        .metric-card{background:#0f172a;padding:14px 16px;border-radius:14px;border:1px solid #1f2937}
+        .label{font-size:12px;color:#94a3b8;margin-bottom:6px}
+        .value{font-size:16px;color:#e2e8f0}
+        </style>""",
+        unsafe_allow_html=True,
+    )
+    va = analysis.get("video_analysis", {}) or {}
+    storyboard = analysis.get("storyboard", []) or []
+    script = analysis.get("script", []) or []
+    metrics = va.get("video_metrics", []) or []
+    mean_score = _mean_effectiveness(metrics)
+
+    m1,m2,m3,m4 = st.columns([1,1,1,1])
+    m1.markdown(f'<div class="metric-card"><div class="label">Scenes</div><div class="value">{len(storyboard)}</div></div>', unsafe_allow_html=True)
+    m2.markdown(f'<div class="metric-card"><div class="label">Dialogue Lines</div><div class="value">{len(script)}</div></div>', unsafe_allow_html=True)
+    m3.markdown(f'<div class="metric-card"><div class="label">Avg Effectiveness</div><div class="value">{mean_score}/10</div></div>', unsafe_allow_html=True)
+    m4.markdown(f'<div class="metric-card"><div class="label">Improvements</div><div class="value">{len(analysis.get("timestamp_improvements", []) or [])}</div></div>', unsafe_allow_html=True)
+
+    colA, colB = st.columns([1.3,1])
+    with colA:
+        st.markdown("### Executive Summary")
+        c1,c2 = st.columns(2)
+        with c1:
+            with st.expander("Brief", expanded=True): st.write(analysis.get("brief", "N/A"))
+            with st.expander("Caption Details", expanded=False): st.write(analysis.get("caption_details", "N/A"))
+        with c2:
+            hook = analysis.get("hook", {}) or {}
+            with st.expander("Hook", expanded=True):
+                st.markdown(f"**Opening:** {hook.get('hook_text','N/A')}")
+                st.markdown(f"**Principle:** {hook.get('principle','N/A')}")
+                adv = _normalize_list(hook.get("advantages"))
+                if adv:
+                    st.markdown("**Advantages:**")
+                    st.markdown("\n".join([f"- {a}" for a in adv]))
+        st.divider()
+        st.markdown("### Narrative & Copy Frameworks")
+        with st.expander("Framework Analysis", expanded=True): st.write(analysis.get("framework_analysis", "N/A"))
+    with colB:
+        st.markdown("### Snapshot")
+        st.caption("Top Drivers"); st.markdown(f'{va.get("effectiveness_factors","N/A")}</div>', unsafe_allow_html=True)
+        st.markdown(""); st.caption("Psychological Triggers"); st.markdown(f'{va.get("psychological_triggers","N/A")}</div>', unsafe_allow_html=True)
+        st.markdown(""); st.caption("Target Audience"); st.markdown(f'{va.get("target_audience","N/A")}</div>', unsafe_allow_html=True)
+
+    st.divider()
+    tabs = st.tabs(["Storyboard","Script","Scored Metrics","Improvements","Raw JSON"])
+
+    with tabs[0]:
+        q = st.text_input("Search storyboard")
+        if storyboard:
+            df = _to_dataframe(
+                storyboard,
+                {"timeline":"Timeline","scene":"Scene","visuals":"Visuals","dialogue":"Dialogue","camera":"Camera","sound_effects":"Sound Effects"}
+            )
+            st.dataframe(_search_dataframe(df, q), use_container_width=True, height=480)
+        else:
+            st.info("No storyboard available.")
+
+    with tabs[1]:
+        q2 = st.text_input("Search script")
+        if script:
+            df = _to_dataframe(script, {"timeline":"Timeline","dialogue":"Dialogue"})
+            st.dataframe(_search_dataframe(df, q2), use_container_width=True, height=480)
+        else:
+            st.info("No script breakdown available.")
+
+    with tabs[2]:
+        q3 = st.text_input("Search metrics")
+        if metrics:
+            dfm = _to_dataframe(
+                metrics,
+                {"timestamp":"Timestamp","element":"Element","current_approach":"Current Approach","effectiveness_score":"Effectiveness Score","notes":"Notes"}
+            )
+            st.dataframe(_search_dataframe(dfm, q3), use_container_width=True, height=480)
+        else:
+            st.info("No video metrics available.")
+
+    with tabs[3]:
+        improvements = analysis.get("timestamp_improvements", []) or []
+        q4 = st.text_input("Search improvements")
+        if improvements:
+            imp_df = _to_dataframe(
+                improvements,
+                {"timestamp":"Timestamp","current_element":"Current Element","improvement_type":"Improvement Type",
+                 "recommended_change":"Recommended Change","expected_impact":"Expected Impact","priority":"Priority"}
+            )
+            if "Priority" in imp_df.columns:
+                order = pd.CategoricalDtype(["High","Medium","Low"], ordered=True)
+                imp_df["Priority"] = imp_df["Priority"].astype(order)
+                if "Timestamp" in imp_df.columns:
+                    imp_df = imp_df.sort_values(["Priority","Timestamp"])
+            st.dataframe(_search_dataframe(imp_df, q4), use_container_width=True, height=480)
+        else:
+            st.info("No timestamp-based improvements available.")
+
+
+    with tabs[4]:
+        pretty = json.dumps(analysis, indent=2, ensure_ascii=False)
+        st.code(pretty, language="json")
+        st.download_button(
+            "Download JSON",
+            data=pretty.encode("utf-8"),
+            file_name="ad_analysis.json",
+            mime="application/json",
+            use_container_width=True
+        )

core/logger.py

@@ -0,0 +1,8 @@
+import logging
+
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[logging.StreamHandler()]
+)
+logger = logging.getLogger(__name__)

core/settings.py

@@ -0,0 +1,91 @@
+from dataclasses import dataclass
+import os
+import os
+from typing import Dict, Any, Optional
+
+from dotenv import dotenv_values
+from pydantic import BaseModel, ConfigDict
+from pydantic_settings import BaseSettings, JsonConfigSettingsSource
+
+from core.logger import logger
+
+
+def load_json_config(json_file: str) -> Dict[str, Any]:
+    try:
+        file_path = os.path.join('/mnt/secrets-store', json_file)
+        if os.path.exists(file_path):
+            logger.info(f"The file is located at: {file_path}")
+        else:
+            logger.error("JSON file not found.")
+    except FileNotFoundError:
+        logger.error("Secrets file not found.")
+        return {}
+    return JsonConfigSettingsSource(settings_cls=GlobalConfig, json_file=file_path).json_data
+
+
+def load_env_config():
+    if GlobalConfig().APP_ENV == "dev":
+        return dotenv_values()
+    else:
+        return dict(os.environ)
+
+
+def get_config() -> Dict[str, Any]:
+    json_config = load_json_config('config.json')
+    env_config = load_env_config()
+
+    config_model = GlobalConfig.model_construct()
+
+
+    final_config = {**config_model.model_dump(), **env_config, **json_config}
+
+    return final_config
+
+class AppConfig(BaseModel):
+    """Application configurations."""
+
+    app_name: str = "Creative AdGenesis"
+    app_version: str = "0.1.0"
+    app_description: str = (
+        "Creative AdGenesis"
+    )
+    debug_mode: bool = False
+
+
+class GlobalConfig(BaseSettings):
+    """Typed global core object."""
+    APP_CONFIG: AppConfig = AppConfig()
+    model_config = ConfigDict(extra="ignore")
+
+    # Environment
+    APP_ENV: Optional[str] = "dev"
+    DEBUG: Optional[bool] = False
+
+    MONGO_HOST: Optional[str] = None
+    MONGO_USERNAME: Optional[str] = None
+    MONGO_PASSWORD: Optional[str] = None
+    MONGO_DB: Optional[str] = None
+
+
+    # Cloudflare R2
+    R2_BUCKET_NAME: Optional[str] = None
+    R2_ACCESS_KEY: Optional[str] = None
+    R2_SECRET_KEY: Optional[str] = None
+    R2_ENDPOINT: Optional[str] = None
+
+
+    # OpenAI
+    OPENAI_API_KEY: Optional[str] = None
+
+    # Replicate
+    REPLICATE_API_KEY: Optional[str] = None
+    GEMINI_API_KEY:  Optional[str] = None
+
+
+    # Custom
+    NEW_BASE: Optional[str] = None
+
+# Load configurations
+config_data = get_config()
+cnf = GlobalConfig(**config_data)
+app_cnf = cnf.APP_CONFIG

core/task_enum.py

@@ -0,0 +1,5 @@
+class TaskType:
+    VIDEO_ANALYZER       = "video_analyzer"
+    SCRIPT_GENERATION    = "script_generation"
+    TEXT_TO_IMAGE        = "text_to_image"
+    IMAGE_GEN            = "image_gen"

database/__init__.py

@@ -0,0 +1,5 @@
+from database.connections import get_mongo_client
+from database.operations import (
+    start_job, finish_job,
+    insert_video_analysis, list_video_categories, find_video_analyses
+)

database/connections.py

@@ -0,0 +1,46 @@
+import streamlit as st
+from pymongo import MongoClient
+
+
+from core.logger import logger
+from core.settings import cnf
+
+
+
+
+def get_mongo_client():
+    """Initialize MongoDB connection with enhanced error handling"""
+    try:
+        username = cnf.MONGO_USERNAME
+        password = cnf.MONGO_PASSWORD
+        db_name = cnf.MONGO_DB
+        host = cnf.MONGO_HOST
+
+        connection_string = f"mongodb+srv://{username}:{password}@{host}/{db_name}?retryWrites=true&w=majority"
+        client = MongoClient(connection_string, serverSelectionTimeoutMS=5000)
+        db = client[db_name]
+        db.command('ping')
+        logger.info("MongoDB connection established successfully")
+        return db
+    except Exception as e:
+        print("Database connection failed")
+        return None
+
+
+def get_results_collection():
+    db = get_mongo_client()
+    if db is None:
+        return None
+    return db["results"]
+
+def get_video_collection():
+    db = get_mongo_client()
+    if db is None:
+        return None
+    return db["video_analyses"]
+
+def get_script_collection():
+    db = get_mongo_client()
+    if db is None:
+        return None
+    return db["script_generator"]

database/operations.py

@@ -0,0 +1,334 @@
+from __future__ import annotations
+import os, logging
+from typing import Optional, List, Dict, Any, Tuple
+from datetime import datetime
+from bson import ObjectId
+from pymongo import ASCENDING, DESCENDING
+from pymongo import MongoClient
+
+
+from database.connections import get_video_collection, get_script_collection, get_results_collection
+
+
+log = logging.getLogger(__name__)
+
+
+def _ensure_results_indexes(col):
+    try:
+        col.create_index([("created_at", DESCENDING)])
+        col.create_index([("type", ASCENDING), ("created_at", DESCENDING)])
+        col.create_index([("category", ASCENDING), ("created_at", DESCENDING)])
+        col.create_index([("created_by", ASCENDING), ("created_at", DESCENDING)])
+
+    except Exception:
+        pass
+
+def _ensure_video_indexes(col):
+    try:
+        col.create_index([("created_at", DESCENDING)])
+        col.create_index([("category", ASCENDING), ("created_at", DESCENDING)])
+        col.create_index([("created_by", ASCENDING), ("created_at", DESCENDING)])
+    except Exception:
+        pass
+
+
+_rc = get_results_collection()
+if _rc is not None:
+    _ensure_results_indexes(_rc)
+
+_vc = get_video_collection()
+if _vc is not None:
+    _ensure_video_indexes(_vc)
+
+_sg = get_script_collection()
+if _sg is not None:
+    _ensure_video_indexes(_sg)
+
+
+
+# ---------- Image jobs (generation / variation) ----------
+def start_job(
+    col,
+    *,
+    type: str,
+    created_by: str,
+    category: str,
+    inputs: Dict[str, Any],
+    settings: Dict[str, Any],
+    user_prompt: Optional[str] = None,
+) -> str:
+    """
+    Variation schema (AI Library compatible):
+      {
+        type: "variation",
+        source: "variation",
+        created_by, category, file_name, prompt, lob,
+        status: "in_progress",
+        urls: [],
+        created_at,
+        settings, inputs
+      }
+
+    Generation schema:
+      {
+        type: "generation",
+        source: "text",
+        category, prompt,
+        status: "in_progress",
+        urls: [],
+        settings,
+        inputs,
+        created_by,
+        created_at,
+      }
+    """
+    now = datetime.utcnow()
+    doc: Dict[str, Any] = {
+        "type": type,
+        "source": "variation" if type == "variation" else "text",
+        "category": category or "general",
+        "prompt": user_prompt,
+        "status": "in_progress",
+        "urls": [],
+        "inputs": inputs or {},
+        "settings": settings or {},
+        "created_by": created_by,
+        "created_at": now,
+    }
+
+    if "file_name" in (inputs or {}):
+        doc["file_name"] = inputs["file_name"]
+
+    res = col.insert_one(doc)
+    return str(res.inserted_id)
+
+
+def finish_job(
+    col,
+    job_id: str,
+    *,
+    status: str = "completed",
+    outputs_urls: Optional[List[str]] = None,
+    provider_update: Optional[Dict[str, Any]] = None,
+) -> None:
+    """
+    Finalizes a job. Writes top-level `urls` and `status`.
+    """
+    set_fields: Dict[str, Any] = {"status": status}
+    if outputs_urls is not None:
+        set_fields["urls"] = list(dict.fromkeys(outputs_urls or []))
+    if provider_update:
+        for k, v in provider_update.items():
+            set_fields[f"provider.{k}"] = v
+    try:
+        col.update_one({"_id": ObjectId(job_id)}, {"$set": set_fields})
+    except Exception as e:
+        log.error(f"finish_job update failed: {e}")
+
+
+# ---------- Video analyses ----------
+def insert_video_analysis(
+    *,
+    video_name: str,
+    response: Dict[str, Any],
+    category: Optional[str] = None,
+    created_by: Optional[str] = None,
+    analyzer_model: Optional[str] = None,
+    video_meta: Optional[Dict[str, Any]] = None,
+    thumbnail: str = "",
+) -> Optional[str]:
+    col = get_video_collection()
+    if col is None:
+        return None
+    doc: Dict[str, Any] = {
+        "category": (category or "general"),
+        "video": {"name": video_name, **(video_meta or {})},
+        "analyzer_model": analyzer_model or os.getenv("VIDEO_ANALYZER_MODEL", "gemini-2.0-flash"),
+        "results": response or {},
+        "created_by": created_by,
+        "created_at": datetime.utcnow(),
+        "thumbnail": thumbnail or "",
+    }
+    res = col.insert_one(doc)
+    return str(res.inserted_id)
+
+
+def list_video_categories() -> List[str]:
+    col = get_video_collection()
+    if col is None:
+        return []
+    try:
+        vals = col.distinct("category")
+        return sorted({v for v in vals if v not in (None, "", [])})
+    except Exception:
+        return []
+
+
+def find_video_analyses(
+    *,
+    category: Optional[str] = None,
+    start_date: Optional[datetime] = None,
+    end_date: Optional[datetime] = None,
+    limit: int = 200,
+) -> List[Dict[str, Any]]:
+    col = get_video_collection()
+    if col is None:
+        return []
+    q: Dict[str, Any] = {}
+    if category: q["category"] = category
+    if start_date or end_date:
+        rng: Dict[str, Any] = {}
+        if start_date: rng["$gte"] = start_date
+        if end_date:   rng["$lt"]  = end_date
+        q["created_at"] = rng
+    cur = col.find(q).sort("created_at", DESCENDING).limit(max(1, int(limit)))
+    out: List[Dict[str, Any]] = []
+    for d in cur:
+        d["_id"] = str(d.get("_id"))
+        out.append(d)
+    return out
+
+
+
+def find_generation_jobs(
+    *,
+    category: Optional[str] = None,
+    start_date: Optional[datetime] = None,
+    end_date: Optional[datetime] = None,
+    page: int = 0,
+    page_size: int = 20,
+) -> Tuple[List[Dict[str, Any]], int]:
+    col = get_results_collection()
+    if col is None:
+        return [], 0
+
+    q: Dict[str, Any] = {"type": "generation", "source": "text"}
+    if category:
+        q["category"] = category
+    if start_date or end_date:
+        rng: Dict[str, Any] = {}
+        if start_date: rng["$gte"] = start_date
+        if end_date:   rng["$lt"] = end_date
+        q["created_at"] = rng
+
+    total = col.count_documents(q)
+    cur = (
+        col.find(q)
+        .sort("created_at", DESCENDING)
+        .skip(page * page_size)
+        .limit(page_size)
+    )
+
+    out: List[Dict[str, Any]] = []
+    for d in cur:
+        d["_id"] = str(d.get("_id"))
+        out.append(d)
+    return out, int(total)
+
+
+script_collection = get_script_collection()
+
+
+
+def insert_script_result(
+    *,
+    video_name: str,
+    offer_details: str,
+    target_audience: str,
+    specific_hook: str,
+    additional_context: str,
+    response: List[Dict[str, Any]],
+    thumbnail: str = "",
+    created_by: Optional[str] = None,
+    num_scripts: Optional[int] = None,
+    category: Optional[str] = None,
+) -> None:
+    """
+    Save a script generation run into the results collection.
+    Stored with type/source so the library can query it.
+    """
+    col = get_script_collection()
+    if col is None:
+        raise ValueError("Results collection is not available.")
+
+    doc: Dict[str, Any] = {
+        "type": "script",
+        "source": "script_generator",
+        "video_name": video_name,
+        "category": category or "general",
+        "offer_details": offer_details,
+        "target_audience": target_audience,
+        "specific_hook": specific_hook,
+        "additional_context": additional_context,
+        "response": response,
+        "thumbnail": thumbnail,
+        "created_by": created_by,
+        "num_scripts": num_scripts,
+        "created_at": datetime.utcnow(),
+    }
+
+    try:
+        col.insert_one(doc)
+    except Exception as e:
+        raise ValueError(f"Failed to insert script result: {e}")
+
+
+def find_script_results(
+    *,
+    start_date: Optional[datetime] = None,
+    end_date: Optional[datetime] = None,
+    page: int = 0,
+    page_size: int = 20,
+    created_by: Optional[str] = None,
+    video_name_query: Optional[str] = None,
+) -> Tuple[List[Dict[str, Any]], int]:
+    col = get_script_collection()
+    if col is None:
+        return [], 0
+
+    q: Dict[str, Any] = {
+        "$or": [
+            {"type": "script", "source": "script_generator"},
+            {"type": {"$exists": False}, "response": {"$exists": True}},
+        ]
+    }
+
+    if start_date or end_date:
+        rng: Dict[str, Any] = {}
+        if start_date: rng["$gte"] = start_date
+        if end_date:   rng["$lt"]  = end_date
+        q["created_at"] = rng
+
+    if created_by:
+        q["created_by"] = created_by
+
+    if video_name_query:
+        q["video_name"] = {"$regex": video_name_query, "$options": "i"}
+
+    total = col.count_documents(q)
+    cur = (col.find(q)
+             .sort("created_at", DESCENDING)
+             .skip(page * page_size)
+             .limit(page_size))
+
+    out: List[Dict[str, Any]] = []
+    for d in cur:
+        d["_id"] = str(d.get("_id"))
+        out.append(d)
+    return out, int(total)
+
+
+def get_all_scripts(start_date: Optional[datetime] = None, end_date: Optional[datetime] = None, limit: int = 20) -> \
+List[Dict[str, Any]]:
+    col = get_script_collection()
+    if col is None:
+        return []
+    query = {"type": "generation", "source": "text"}
+
+    if start_date:
+        query["created_at"] = {"$gte": start_date}
+    if end_date:
+        query["created_at"]["$lt"] = end_date
+
+    cursor = col.find(query).sort("created_at", DESCENDING).limit(limit)
+    return list(cursor)

generator_function/image_function.py

@@ -0,0 +1,339 @@
+import os, io, zipfile, replicate, time, logging, requests, streamlit as st, boto3, threading
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import Dict, Any, List, Tuple, Optional, Union
+from uuid import uuid4
+from urllib.parse import urlparse
+from functools import lru_cache
+import os, base64, logging
+from openai import OpenAI
+
+from database.operations import start_job, finish_job
+
+from database.connections import get_results_collection
+
+
+from dotenv import load_dotenv
+
+
+load_dotenv()
+logger = logging.getLogger(__name__)
+
+def _encode_image_to_base64(image_path):
+    try:
+        with open(image_path, "rb") as f:
+            return base64.b64encode(f.read()).decode("utf-8")
+    except Exception:
+        logger.exception(f"Failed to base64 encode image: {image_path}")
+        return ""
+
+REPLICATE_API_TOKEN = os.getenv("REPLICATE_API_TOKEN")
+MAX_WORKERS = min(32, (os.cpu_count() or 1) + 4)
+REQUEST_TIMEOUT = 30
+RETRY_ATTEMPTS = 3
+
+MODEL_REGISTRY: Dict[str, Dict[str, Any]] = {
+    "imagegen-4-ultra": {"id": "google/imagen-4-ultra","aspect_ratios": ["1:1","16:9","9:16","3:4","4:3"],"param_name":"aspect_ratio"},
+    "imagen-4": {"id": "google/imagen-4","aspect_ratios": ["1:1","16:9","9:16","3:4","4:3"],"param_name":"aspect_ratio"},
+    "nano-banana": {"id": "google/nano-banana","aspect_ratios": ["1:1","16:9","9:16","3:4","4:3"],"param_name":"aspect_ratio"},
+    "qwen": {"id": "qwen/qwen-image","aspect_ratios": ["1:1","16:9","9:16","3:4","4:3","3:2","2:3"],"param_name":"aspect_ratio"},
+    "seedream-3": {"id": "bytedance/seedream-3","aspect_ratios": ["1:1","16:9","9:16","3:4","4:3","3:2","2:3","21:9"],"param_name":"aspect_ratio"},
+    "recraft-v3": {"id": "recraft-ai/recraft-v3","aspect_ratios": ["1:1","4:3","3:4","3:2","2:3","16:9","9:16","1:2","2:1","7:5","5:7","4:5","5:4","3:5","5:3"],"param_name":"aspect_ratio"},
+    "photon": {"id": "luma/photon","aspect_ratios": ["1:1","3:4","4:3","9:16","16:9","9:21","21:9"],"param_name":"aspect_ratio"},
+    "ideogram-v3-quality": {"id": "ideogram-ai/ideogram-v3-quality","aspect_ratios": ["1:3","3:1","1:2","2:1","9:16","16:9","10:16","16:10","2:3","3:2","3:4","4:3","4:5","5:4","1:1"],"param_name":"aspect_ratio"},
+}
+
+_thread_local = threading.local()
+
+
+
+def get_model_config(model_key: str) -> Optional[Dict[str, Any]]:
+    return MODEL_REGISTRY.get(model_key)
+
+@lru_cache(maxsize=128)
+def _get_model_config_cached(model_key: str) -> Optional[Dict[str, Any]]:
+    return MODEL_REGISTRY.get(model_key)
+
+def _s3():
+    if not hasattr(_thread_local, "s3"):
+        needed = ["R2_ENDPOINT","R2_ACCESS_KEY","R2_SECRET_KEY","R2_BUCKET_NAME","NEW_BASE"]
+        if any(not os.getenv(k) for k in needed):
+            _thread_local.s3 = None
+            return None
+        try:
+            _thread_local.s3 = boto3.client(
+                "s3",
+                endpoint_url=os.getenv("R2_ENDPOINT"),
+                aws_access_key_id=os.getenv("R2_ACCESS_KEY"),
+                aws_secret_access_key=os.getenv("R2_SECRET_KEY"),
+                region_name="auto",
+            )
+        except Exception as e:
+            logger.error(f"S3 init failed: {e}")
+            _thread_local.s3 = None
+    return _thread_local.s3
+
+def _upload_to_r2(image_bytes: bytes) -> Optional[str]:
+    s3 = _s3()
+    if not s3:
+        return None
+    for attempt in range(RETRY_ATTEMPTS):
+        try:
+            filename = f"{uuid4().hex}.png"
+            key = f"adgenesis_image_text/creative_adgenesis/images/{filename}"
+            s3.put_object(
+                Bucket=os.getenv("R2_BUCKET_NAME"),
+                Key=key,
+                Body=image_bytes,
+                ContentType="image/png",
+            )
+            return f"{os.getenv('NEW_BASE').rstrip('/')}/{key}"
+        except Exception as e:
+            if attempt == RETRY_ATTEMPTS - 1:
+                logger.error(f"R2 upload failed: {e}")
+                return None
+            time.sleep(2 ** attempt)
+    return None
+
+def _generate_one(model_key: str, prompt: str, aspect_ratio: str) -> List[str]:
+    if not REPLICATE_API_TOKEN:
+        return []
+    cfg = _get_model_config_cached(model_key)
+    if not cfg:
+        return []
+    for attempt in range(RETRY_ATTEMPTS):
+        try:
+            output = replicate.run(cfg["id"], input={"prompt": prompt, cfg["param_name"]: aspect_ratio})
+            urls: List[str] = []
+            if isinstance(output, list) and output:
+                first = output[0]
+                url = getattr(first, "url", str(first))
+                urls = [url]
+            elif isinstance(output, str):
+                urls = [output]
+            elif hasattr(output, "url"):
+                urls = [getattr(output, "url")]
+            if urls:
+                return urls
+        except Exception as e:
+            if attempt == RETRY_ATTEMPTS - 1:
+                logger.error(f"replicate run failed: {e}")
+                return []
+            time.sleep(1)
+    return []
+
+def _fetch(url: Union[str, Any]) -> Optional[bytes]:
+    url_str = getattr(url, "url", str(url))
+    for attempt in range(RETRY_ATTEMPTS):
+        try:
+            r = requests.get(
+                url_str, timeout=REQUEST_TIMEOUT, stream=True,
+                headers={"Cache-Control":"no-cache","Pragma":"no-cache","User-Agent":"ImageBot/1.0"}
+            )
+            r.raise_for_status()
+            buf = b""
+            for chunk in r.iter_content(8192):
+                buf += chunk
+            return buf
+        except Exception:
+            if attempt == RETRY_ATTEMPTS - 1:
+                return None
+            time.sleep(1)
+    return None
+
+def _process_one(args: Tuple[str, str, str, int]) -> Dict[str, Any]:
+    model_key, prompt, aspect_ratio, idx = args
+    out = {"index": idx, "success": False, "source_url": None, "r2_url": None, "error": None}
+    try:
+        urls = _generate_one(model_key, prompt, aspect_ratio)
+        if not urls:
+            out["error"] = "No URLs returned"; return out
+        src = urls[0]
+        out["source_url"] = getattr(src, "url", str(src))
+        b = _fetch(src)
+        if not b:
+            out["error"] = "Fetch failed"; return out
+        r2 = _upload_to_r2(b)
+        if r2:
+            out["r2_url"] = r2; out["success"] = True
+        else:
+            out["error"] = "Upload to R2 failed"
+    except Exception as e:
+        out["error"] = str(e)
+    return out
+
+def _generate_images_parallel(model_key: str, aspect_ratio: str, prompt: str, num_images: int) -> Tuple[List[str], List[str], List[str]]:
+    if num_images == 1:
+        res = _process_one((model_key, prompt, aspect_ratio, 0))
+        if res["success"]:
+            return [res["r2_url"]], [res["source_url"]], []
+        return [], [], [res["error"] or "Generation failed"]
+    args = [(model_key, prompt, aspect_ratio, i) for i in range(num_images)]
+    r2, src, errs = [], [], []
+    with ThreadPoolExecutor(max_workers=min(MAX_WORKERS, num_images)) as ex:
+        for fut in as_completed({ex.submit(_process_one, a): a[3] for a in args}):
+            try:
+                res = fut.result()
+                if res["success"]:
+                    if res["r2_url"]: r2.append(res["r2_url"])
+                    if res["source_url"]: src.append(res["source_url"])
+                else:
+                    errs.append(res["error"] or "Generation failed")
+            except Exception as e:
+                errs.append(f"Future err: {e}")
+    # de-dup
+    r2 = list(dict.fromkeys(r2)); src = list(dict.fromkeys(src))
+    return r2, src, errs
+
+
+
+def generate_images_parallel(model_key: str, aspect_ratio: str, prompt: str, num_images: int) -> Tuple[List[str], List[str], List[str]]:
+    """Back-compat public export used by background tasks."""
+    return _generate_images_parallel(model_key, aspect_ratio, prompt, num_images)
+
+def handle_image_generation_optimized(
+    *,
+    model_key: str,
+    aspect_ratio: str,
+    prompt: str,
+    num_images: int,
+    debug_mode: bool = False,
+    category: Optional[str] = None,
+    platform: Optional[str] = None,
+):
+    """
+    Streamlit-friendly wrapper: kicks off parallel gen, persists a job row,
+    and renders results in-place (no return value).
+    """
+    if not REPLICATE_API_TOKEN:
+        st.error("Missing REPLICATE_API_TOKEN. Set it as an environment variable.")
+        return
+    if not prompt.strip():
+        st.warning("Please enter a prompt.")
+        return
+
+    created_by = (
+        st.session_state.get("uid")
+        or os.getenv("USER_ID")
+        or os.getenv("CREATED_BY")
+        or "anonymous"
+    )
+
+    results_col = get_results_collection()
+    db_job_id = None
+    if results_col is not None:
+        try:
+            db_job_id = start_job(
+                results_col,
+                type="generation",
+                created_by=str(created_by),
+                category=(category or "general"),
+                inputs={"model_key": model_key, "aspect_ratio": aspect_ratio, "num_images": num_images},
+                settings={"platform": platform},
+                user_prompt=prompt.strip(),
+            )
+        except Exception as e:
+            logger.error(f"start_job failed: {e}")
+
+    progress = st.progress(0, text="Starting generation...")
+    status = st.empty()
+    start = time.time()
+
+    try:
+        with status.container():
+            st.info(f"Generating {num_images} image(s) in parallel...")
+        progress.progress(10, text="Running...")
+
+        r2_urls, source_urls, errors = _generate_images_parallel(model_key, aspect_ratio, prompt.strip(), num_images)
+        urls = r2_urls or source_urls
+
+        if results_col is not None and db_job_id:
+            try:
+                finish_job(
+                    results_col,
+                    db_job_id,
+                    status="completed" if urls else "failed",
+                    outputs_urls=urls or [],
+                    provider_update={"errors": errors} if errors else None,
+                )
+            except Exception as e:
+                logger.error(f"finish_job failed: {e}")
+
+        progress.progress(100, text="Complete!")
+        took = time.time() - start
+
+        if urls:
+            with status.container():
+                st.success(f"Generated {len(urls)} image(s) in {took:.1f}s. Job ID: {db_job_id or 'N/A'}")
+
+            cols = st.columns(min(4, len(urls)) or 1)
+            for i, u in enumerate(urls):
+                with cols[i % len(cols)]:
+                    try:
+                        b = _fetch(u)
+                        if b is None:
+                            st.error("Failed to load image")
+                            continue
+                        st.image(b, use_container_width=True)
+                        path = urlparse(str(u)).path
+                        base = os.path.basename(path) or "image.png"
+                        if not os.path.splitext(base)[1]:
+                            base = f"{base}.png"
+                        st.download_button("Download image", b, base, mime="image/png", use_container_width=True)
+                    except Exception as e:
+                        st.error(f"Display failed: {e}")
+        else:
+            with status.container():
+                st.error("No images were generated.")
+            if errors and debug_mode:
+                with st.expander("Generation Errors", expanded=True):
+                    for e in errors:
+                        st.error(e)
+    except Exception as e:
+        if results_col is not None and db_job_id:
+            try:
+                finish_job(results_col, db_job_id, status="failed")
+            except Exception:
+                pass
+        with status.container():
+            st.error(f"Generation failed: {e}")
+
+
+
+def generate_image(file_path, size, quality, category, sentiment, user_prompt, platform, blur, i=None):
+    try:
+        api_key = os.getenv("OPENAI_API_KEY")
+        if not api_key:
+            logger.critical("OPENAI_API_KEY is not set.")
+            raise RuntimeError("OPENAI_API_KEY is missing")
+        client = OpenAI(api_key=api_key)
+
+        if platform == "Google Display Network":
+            size_messages = {0: "The aspect ratio of the image should be 1024x1024",
+                             1: "The aspect ratio of the image should be 1536x1024.",
+                             2: "The aspect ratio of the image should be 1024x1536."}
+            img_size = size_messages.get(i, "")
+        else:
+            img_size = ""
+
+        with open(file_path, "rb") as img_file:
+            background = "blurred background." if blur else " not blurred background."
+            result = client.images.edit(
+                model="gpt-image-1",
+                prompt=(
+                    f"You are a top-tier performance digital marketer and creative strategist with 15+ years of expertise in affiliate marketing.\n"
+                    f"Your objective is to analyze the provided winning ad image, deconstruct its concept, visual composition, and color scheme, and generate a fresh, conversion-focused ad visual tailored for the {category} niche.\n"
+                    f"The new design should convey a {sentiment} sentiment and incorporate the user instruction: \n {user_prompt}.\n If user has given multple choices or options to be include in the image so choose randomly relevant to the reference image."
+                    f"Create a visually compelling ad optimized for {platform} Ads that is scroll-stopping, pattern-interrupting, and designed to drive high CTR and Conversion Rate. Utilize striking color combinations, dynamic contrast levels, and strategic layout compositions to command attention while aligning with the target audience avatar.\n"
+                    f"Make sure the images should be realistic, not be stocky at all and raw which should look like they are shot from an iPhone with {background}.{img_size}"
+                ),
+                image=img_file,
+                size=size,
+                quality=quality,
+            )
+        image_base64 = result.data[0].b64_json
+        image_bytes = base64.b64decode(image_base64)
+        logger.info(f"Successfully generated image for {file_path}")
+        return image_bytes
+    except Exception as e:
+        logger.exception(f"Failed to generate image for {file_path}: {e}")
+        raise

generator_function/image_processor.py

@@ -0,0 +1,176 @@
+import os, zipfile, tempfile, logging
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import List, Tuple, Optional
+
+from generator_function.image_function import generate_image
+from helpers_function.helper_meta_data import meta_data_helper_function
+from helpers_function.helpers import upload_image_to_r2
+from helpers_function.helpers import is_valid_image
+from database.connections import get_results_collection as get_collection
+from database.operations import start_job, finish_job
+from util.session_state import current_uid
+
+logger = logging.getLogger(__name__)
+COL = get_collection()
+
+def _resolve_user_id() -> str:
+    return current_uid() or os.getenv("DEFAULT_USER_ID") or "anonymous"
+
+uid = _resolve_user_id()
+
+def process_zip_and_generate_images(
+    zip_path: str,
+    category: str,
+    size: str,
+    quality: str,
+    user_prompt: str,
+    sentiment: str,
+    platform: str,
+    num_images: int,
+    demo_mode: bool,
+    existing_images: Optional[List[str]],
+    blur: bool,
+) -> List[str]:
+    num_images = 1 if demo_mode else num_images
+    try:
+        if zip_path.endswith(".zip"):
+            temp_dir = extract_zip_file(zip_path)
+            image_files = get_valid_image_files(temp_dir)
+        else:
+            image_files = [(os.path.basename(zip_path), zip_path)]
+
+        results = process_image_files(
+            image_files, category, size, quality, user_prompt, sentiment, platform, num_images, blur
+        )
+        all_urls = [url for entry in results for url in entry["urls"]]
+        seen, deduped = set(), []
+        for u in all_urls:
+            if u not in seen:
+                seen.add(u); deduped.append(u)
+        return (existing_images or []) + deduped
+    except Exception:
+        logger.exception(f"Global error during processing file: {zip_path}")
+        return existing_images or []
+
+def extract_zip_file(zip_path: str) -> tempfile.TemporaryDirectory:
+    temp_dir = tempfile.TemporaryDirectory()
+    with zipfile.ZipFile(zip_path, "r") as zip_ref:
+        zip_ref.extractall(temp_dir.name)
+    logger.info(f"Extracted ZIP file: {zip_path}")
+    return temp_dir
+
+def get_valid_image_files(temp_dir: tempfile.TemporaryDirectory) -> List[Tuple[str, str]]:
+    valid_files: List[Tuple[str, str]] = []
+    for file in os.listdir(temp_dir.name):
+        if "__MACOSX" in file: continue
+        file_path = os.path.join(temp_dir.name, file)
+        if is_valid_image(file):
+            valid_files.append((file, file_path))
+        else:
+            logger.warning(f"Ignored non-image file: {file}")
+    logger.info(f"Found {len(valid_files)} valid images.")
+    return valid_files
+
+def process_image_files(
+    image_files: List[Tuple[str, str]],
+    category: str,
+    size: str,
+    quality: str,
+    user_prompt: str,
+    sentiment: str,
+    platform: str,
+    num_images: int,
+    blur: bool,
+) -> List[dict]:
+    final_results: List[dict] = []
+    with ThreadPoolExecutor(max_workers=5) as executor:
+        futures = []
+        for file_name, file_path in image_files:
+            job_id: Optional[str] = None
+            if COL is not None:
+                try:
+                    settings = {"size": size,"quality": quality,"sentiment": sentiment,"platform": platform,"num_images": num_images,"blur": bool(blur)}
+                    inputs = {"file_name": file_name, "mode": "img_or_zip"}
+                    job_id = start_job(
+                        COL,
+                        type="variation",
+                        created_by=uid,
+                        category=category or "general",
+                        inputs=inputs,
+                        settings=settings,
+                        user_prompt=user_prompt
+
+                    )
+                except Exception:
+                    logger.exception("Failed to start DB job; continuing without DB logging.")
+            futures.append(
+                executor.submit(
+                    process_single_image,
+                    file_name, file_path, category, size, quality, user_prompt, sentiment, platform, num_images, blur, job_id,
+                )
+            )
+        for future in as_completed(futures):
+            try:
+                result = future.result()
+                if result: final_results.append(result)
+            except Exception:
+                logger.exception("Unhandled exception during image processing thread.")
+    return final_results
+
+def process_single_image(
+    file_name: str,
+    file_path: str,
+    category: str,
+    size: str,
+    quality: str,
+    user_prompt: str,
+    sentiment: str,
+    platform: str,
+    num_images: int,
+    blur: bool,
+    job_id: Optional[str],
+) -> Optional[dict]:
+    try:
+        image_urls = generate_images_from_prompts(
+            file_path, size, quality, category, sentiment, user_prompt, platform, num_images, blur
+        )
+        if COL is not None and job_id:
+            try:
+                finish_job(COL, job_id, status=("completed" if image_urls else "failed"), outputs_urls=image_urls)
+            except Exception:
+                logger.exception("Failed to finish DB job.")
+        if image_urls:
+            return {"file_name": file_name, "urls": image_urls}
+        return None
+    except Exception as e:
+        logger.error(f"Processing failed for {file_name}: {e}")
+        if COL is not None and job_id:
+            try:
+                finish_job(COL, job_id, status="failed", outputs_urls=[])
+            except Exception:
+                logger.exception("Also failed to mark DB job as failed.")
+        return None
+
+def generate_images_from_prompts(
+    file_path: str, size: str, quality: str, category: str, sentiment: str, user_prompt: str,
+    platform: str, num_images: int, blur: bool,
+) -> List[str]:
+    image_urls: List[str] = []
+
+    def worker(i: int) -> Optional[str]:
+        try:
+            image_bytes = generate_image(file_path, size, quality, category, sentiment, user_prompt, platform, blur, i)
+            if not image_bytes: return None
+            image_with_metadata = meta_data_helper_function(image_bytes)
+            s3_url = upload_image_to_r2(image_with_metadata)
+            return s3_url
+        except Exception as e:
+            logger.error(f"Image generation failed: {e}")
+            return None
+
+    with ThreadPoolExecutor(max_workers=min(10, num_images)) as executor:
+        futures = [executor.submit(worker, i) for i in range(num_images)]
+        for future in as_completed(futures):
+            result = future.result()
+            if result: image_urls.append(result)
+    return image_urls

generator_function/script_generator.py

@@ -0,0 +1,105 @@
+import json
+import os
+import time
+from core.logger import logger
+from typing import Dict, Any
+
+from prompt.system_prompt_script import SYSTEM_PROMPT
+from schema.pydantic_schema_script import ScriptResponse
+from google import genai
+from dotenv import load_dotenv
+
+load_dotenv()
+GEMINI_API_KEY = os.getenv("GEMINI_KEY")
+
+def configure_gemini():
+    return genai.Client(api_key=GEMINI_API_KEY)
+
+
+def generate_scripts(
+        video_path: str,
+        offer_details: str,
+        target_audience: str,
+        specific_hooks: str,
+        additional_context: str,
+        num_scripts: int = 3,
+        duration: int = 60
+) -> Dict[str, Any]:
+    client = configure_gemini()
+
+    try:
+        user_prompt = f"""
+                    Generate {num_scripts} high-converting direct response script variations,
+                    each about {duration} seconds long.
+                      Each variation MUST be designed for a total runtime ≤ {duration} seconds.
+                    - Do NOT include any timestamp beyond {duration}s.
+                    - Keep all beats within 0–{duration}s.
+                    - Use timestamps as "M:SS" (e.g., "0:03", "0:12").
+
+
+                    CONTEXT TO FOLLOW:
+                    - Offer Details: {offer_details}
+                    - Target Audience: {target_audience}
+                    - Specific Hooks: {specific_hooks}
+
+                     ADDITIONAL CONTEXT:
+                    {additional_context}
+                    You must reflect this additional context in:
+                    - The script tone, CTA, visuals
+                    - Compliance or branding constraints
+                    - Any assumptions about audience or product
+                    Failure to include this will be considered incomplete.
+                    Please provide a comprehensive analysis including:
+                    1. DETAILED VIDEO ANALYSIS with timestamp-based metrics:
+                       - Break down the video into 5-10 second segments
+                       - Rate each segment's effectiveness (1-10 scale)
+                       - Identify specific elements (hook, transition, proof, CTA, etc.)
+                    2. TIMESTAMP-BASED IMPROVEMENTS:
+                       - Specific recommendations for each time segment
+                       - Priority level for each improvement
+                       - Expected impact of implementing changes
+                    3. SCRIPT VARIATIONS:
+                       - Create complete script variations
+                        - Each with timestamp-by-timestamp breakdown
+                        - Different psychological triggers and approaches
+         IMPORTANT: Return only valid JSON in the exact format specified in the system prompt. Analyze the video second-by-second for maximum detail."""
+
+        video_file = client.files.upload(file=video_path)
+
+        while getattr(video_file.state, "name", "") == "PROCESSING":
+            time.sleep(1.0)
+            video_file = client.files.get(name=video_file.name)
+        if getattr(video_file.state, "name", "") == "FAILED":
+            logger.error("Video processing FAILED.")
+            return {}
+
+        resp = client.models.generate_content(
+            model="gemini-2.0-flash",
+            contents=[SYSTEM_PROMPT, user_prompt, video_file],
+            config={
+                "response_mime_type": "application/json",
+                "response_schema": ScriptResponse,
+
+            },
+        )
+
+        parsed = getattr(resp, "parsed", None)
+        if parsed is None:
+
+            raw_text = getattr(resp, "text", "") or ""
+            if not raw_text:
+                parts = None
+                if getattr(resp, "candidates", None):
+                    parts = getattr(resp.candidates[0].content, "parts", None)
+                raise RuntimeError(f"Model returned no JSON text. parts={parts}")
+            data = json.loads(raw_text)
+            return data
+
+        out = parsed.model_dump()
+        logger.info("Generated %d variations.", len(out.get("script_variations", [])))
+        return out
+
+    except Exception as e:
+        logger.exception("generate_scripts failed: %s", e)
+        return {}
+

generator_function/video_analyzer_services.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+import os, time, json, mimetypes, tempfile, logging
+from typing import Dict, Any,  Optional
+from schema.pydantic_schema_video import AdAnalysis
+from prompt.analyser_prompt import ANALYSER_PROMPT
+from google import genai
+from dotenv import load_dotenv
+from database.operations import insert_video_analysis
+from helpers_function.helpers import get_video_thumbnail_base64
+
+load_dotenv()
+logger = logging.getLogger(__name__)
+GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
+
+def _configure_gemini() -> genai.Client:
+    if not GEMINI_API_KEY:
+        raise RuntimeError("GEMINI_API_KEY is not set")
+    return genai.Client(api_key=GEMINI_API_KEY)
+
+
+
+def analyze_video_only(video_path: str) -> Dict[str, Any]:
+    try:
+        client = _configure_gemini()
+    except Exception as e:
+        logger.exception("Gemini configuration error")
+        return {"__error__": str(e)}
+
+    try:
+        f = client.files.upload(file=video_path)
+        while getattr(f.state, "name", "") == "PROCESSING":
+            time.sleep(2); f = client.files.get(name=f.name)
+        if getattr(f.state, "name", "") == "FAILED":
+            return {"__error__": "Video indexing failed."}
+
+        resp = client.models.generate_content(
+            model=os.getenv("VIDEO_ANALYZER_MODEL", "gemini-2.0-flash"),
+            contents=[ANALYSER_PROMPT, f],
+            config={"response_mime_type": "application/json"},
+        )
+        raw = getattr(resp, "text", "") or ""
+        if not raw.strip():
+            return {"__error__": "Empty response from model."}
+
+        try:
+            return AdAnalysis.model_validate_json(raw).model_dump()
+        except Exception:
+            try:
+                return json.loads(raw)
+            except Exception:
+                return {"__error__": "Model response not valid JSON."}
+    except Exception as e:
+        logger.exception("Video analysis failed")
+        return {"__error__": str(e)}
+
+
+def run_and_store_video_analysis(
+    *,
+    category: str,
+    created_by: str,
+    uploaded_file=None,
+    uploaded_file_path: Optional[str] = None,
+    analyzer_model: Optional[str] = None,
+) -> Dict[str, Any]:
+
+    if not uploaded_file and not uploaded_file_path:
+        return {"_id": None, "results": {"__error__": "No video provided."}, "video_meta": {}, "thumbnail": ""}
+
+    tmp_created: Optional[str] = None
+    try:
+
+        if uploaded_file_path:
+            video_path = uploaded_file_path
+            orig_name = os.path.basename(uploaded_file_path)
+            mime = mimetypes.guess_type(orig_name)[0] or ""
+            size_bytes = os.path.getsize(uploaded_file_path) if os.path.exists(uploaded_file_path) else None
+        else:
+            suffix = os.path.splitext(getattr(uploaded_file, "name", "upload.mp4"))[1] or ".mp4"
+            with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+                tmp.write(uploaded_file.read())
+                video_path = tmp.name
+                tmp_created = video_path
+            orig_name = getattr(uploaded_file, "name", os.path.basename(video_path))
+            mime = getattr(uploaded_file, "type", None) or mimetypes.guess_type(orig_name)[0] or ""
+            size_bytes = os.path.getsize(video_path) if os.path.exists(video_path) else None
+
+
+        results = analyze_video_only(video_path)
+
+        try:
+            thumbnail_b64 = get_video_thumbnail_base64(video_path) or ""
+        except Exception:
+            thumbnail_b64 = ""
+
+        video_meta = {"name": orig_name, "mimetype": mime, "size_bytes": size_bytes}
+        model_label = analyzer_model or os.getenv("VIDEO_ANALYZER_MODEL", "gemini-2.0-flash")
+
+        inserted_id = insert_video_analysis(
+            video_name=orig_name,
+            response=results,
+            category=(category or "general"),
+            created_by=(created_by or "anonymous"),
+            analyzer_model=model_label,
+            video_meta=video_meta,
+            thumbnail=thumbnail_b64,
+        )
+
+        return {
+            "_id": inserted_id,
+            "results": results,
+            "video_meta": video_meta,
+            "thumbnail": thumbnail_b64,
+        }
+
+    finally:
+        if tmp_created and os.path.exists(tmp_created):
+            try:
+                os.remove(tmp_created)
+            except Exception:
+                pass

helpers_function/helper_meta_data.py

@@ -0,0 +1,166 @@
+import random
+import piexif
+from PIL import Image
+from io import BytesIO
+
+def generate_metadata(model):
+    # Define different values based on iPhone model
+    exposure_times = {
+        "iPhone 11": "1/60",
+        "iPhone 11 Pro": "1/70",
+        "iPhone 12": "1/100",
+        "iPhone 12 Pro": "1/110",
+        "iPhone 13": "1/120",
+        "iPhone 13 Pro": "1/130",
+        "iPhone 13 Pro Max": "1/140",
+        "iPhone 14": "1/200",
+        "iPhone 14 Pro": "1/220",
+        "iPhone 14 Pro Max": "1/240",
+        "iPhone 15": "1/300",
+        "iPhone 15 Plus": "1/320",
+        "iPhone 15 Pro": "1/400",
+        "iPhone 15 Pro Max": "1/500",
+        "iPhone 16": "1/600",
+        "iPhone 16 Pro": "1/700",
+        "iPhone 16 Pro Max": "1/1000"
+    }
+    f_numbers = {
+        "iPhone 11": "f/1.8",
+        "iPhone 11 Pro": "f/1.8",
+        "iPhone 12": "f/1.6",
+        "iPhone 12 Pro": "f/1.6",
+        "iPhone 13": "f/1.5",
+        "iPhone 13 Pro": "f/1.5",
+        "iPhone 13 Pro Max": "f/1.5",
+        "iPhone 14": "f/1.4",
+        "iPhone 14 Pro": "f/1.4",
+        "iPhone 14 Pro Max": "f/1.4",
+        "iPhone 15": "f/1.4",
+        "iPhone 15 Plus": "f/1.4",
+        "iPhone 15 Pro": "f/1.4",
+        "iPhone 15 Pro Max": "f/1.3",
+        "iPhone 16": "f/1.3",
+        "iPhone 16 Pro": "f/1.3",
+        "iPhone 16 Pro Max": "f/1.3"
+    }
+    focal_lengths = {
+        "iPhone 11": "3.99 mm",
+        "iPhone 11 Pro": "4.0 mm",
+        "iPhone 12": "4.2 mm",
+        "iPhone 12 Pro": "4.3 mm",
+        "iPhone 13": "5.0 mm",
+        "iPhone 13 Pro": "5.1 mm",
+        "iPhone 13 Pro Max": "5.2 mm",
+        "iPhone 14": "6.0 mm",
+        "iPhone 14 Pro": "6.1 mm",
+        "iPhone 14 Pro Max": "6.2 mm",
+        "iPhone 15": "6.0 mm",
+        "iPhone 15 Plus": "6.1 mm",
+        "iPhone 15 Pro": "6.1 mm",
+        "iPhone 15 Pro Max": "6.2 mm",
+        "iPhone 16": "6.5 mm",
+        "iPhone 16 Pro": "6.6 mm",
+        "iPhone 16 Pro Max": "6.7 mm"
+    }
+
+    metadata = {
+        "Make": "Apple",
+        "Model": model,
+        "Software": "iOS {}".format(random.choice([14, 15, 16, 17, 18, 19])),
+        "Orientation": random.choice(["Horizontal (normal)", "Rotate 90 CW", "Rotate 180", "Rotate 90 CCW"]),
+        "DateTime": "{}:{}:{} {:02}:{:02}:{:02}".format(random.randint(2022, 2024), random.randint(1, 12), random.randint(1, 28), random.randint(0, 23), random.randint(0, 59), random.randint(0, 59)),
+        "ExposureTime": exposure_times[model],
+        "FNumber": f_numbers[model],
+        "ISOSpeedRatings": random.choice([100, 200, 400, 800, 1600]),
+        "FocalLength": focal_lengths[model],
+        "Flash": random.choice(["Flash fired", "Flash did not fire, compulsory mode", "Flash fired, auto mode"]),
+        "WhiteBalance": random.choice(["Auto", "Manual"]),
+        "MeteringMode": random.choice(["Pattern", "CenterWeightedAverage", "Spot"]),
+        "SceneCaptureType": random.choice(["Standard", "Landscape", "Portrait", "NightScene"]),
+        "GPSLatitude": "{:.4f} N".format(random.uniform(0.0, 90.0)),
+        "GPSLongitude": "{:.4f} W".format(random.uniform(0.0, 180.0)),
+        "Altitude": "{:.1f} m".format(random.uniform(0, 100)),
+        "LensMake": "Apple",
+        "LensModel": "{} back triple camera {} f/1.5".format(model, focal_lengths[model]),
+        "ColorSpace": random.choice(["sRGB", "Adobe RGB"]),
+        "PixelXDimension": random.choice([3024, 4032, 2160]),
+        "PixelYDimension": random.choice([3024, 4032, 2160]),
+        "ExposureBiasValue": random.choice(["0 EV", "+1 EV", "-1 EV"]),
+        "BrightnessValue": "{:.1f}".format(random.uniform(-5, 10)),
+        "ExposureMode": random.choice(["Auto Exposure", "Manual Exposure"])
+    }
+    return metadata
+
+def dms_coordinates(value):
+    """Convert decimal degrees to EXIF GPS (DMS format)."""
+    degrees = int(value)
+    minutes_float = (value - degrees) * 60
+    minutes = int(minutes_float)
+    seconds = round((minutes_float - minutes) * 60 * 10000)
+    return [(degrees, 1), (minutes, 1), (seconds, 10000)]
+
+def meta_data_helper_function(image_bytes,model=None):
+    """
+    Takes raw image bytes, adds realistic EXIF metadata, and returns new bytes.
+    """
+    if model is None:
+        model = random.choice([
+            "iPhone 11", "iPhone 11 Pro", "iPhone 12", "iPhone 12 Pro",
+            "iPhone 13", "iPhone 13 Pro", "iPhone 13 Pro Max", "iPhone 14",
+            "iPhone 14 Pro", "iPhone 14 Pro Max", "iPhone 15", "iPhone 15 Plus",
+            "iPhone 15 Pro", "iPhone 15 Pro Max", "iPhone 16", "iPhone 16 Pro",
+            "iPhone 16 Pro Max"
+        ])
+
+    # Load image from bytes
+    img = Image.open(BytesIO(image_bytes))
+    img = img.convert("RGB")  # Ensure compatibility
+
+    # Generate metadata
+    metadata = generate_metadata(model)
+
+    # Build EXIF data
+    exif_dict = {"0th": {}, "Exif": {}, "GPS": {}}
+
+    exif_dict["0th"][piexif.ImageIFD.Make] = metadata["Make"]
+    exif_dict["0th"][piexif.ImageIFD.Model] = metadata["Model"]
+    exif_dict["0th"][piexif.ImageIFD.Software] = metadata["Software"]
+    exif_dict["0th"][piexif.ImageIFD.DateTime] = metadata["DateTime"]
+
+    # Exposure time (rational)
+    num, den = map(int, metadata["ExposureTime"].split("/"))
+    exif_dict["Exif"][piexif.ExifIFD.ExposureTime] = (num, den)
+
+    # FNumber
+    f_number = float(metadata["FNumber"].replace("f/", ""))
+    exif_dict["Exif"][piexif.ExifIFD.FNumber] = (int(f_number * 10), 10)
+
+    # ISO
+    exif_dict["Exif"][piexif.ExifIFD.ISOSpeedRatings] = metadata["ISOSpeedRatings"]
+
+    # Focal Length
+    focal_length = float(metadata["FocalLength"].split()[0])
+    exif_dict["Exif"][piexif.ExifIFD.FocalLength] = (int(focal_length * 10), 10)
+
+    # GPS
+    lat = float(metadata["GPSLatitude"].split()[0])
+    lon = float(metadata["GPSLongitude"].split()[0])
+
+    exif_dict["GPS"][piexif.GPSIFD.GPSLatitudeRef] = b'N' if lat >= 0 else b'S'
+    exif_dict["GPS"][piexif.GPSIFD.GPSLatitude] = dms_coordinates(abs(lat))
+
+    exif_dict["GPS"][piexif.GPSIFD.GPSLongitudeRef] = b'E' if lon >= 0 else b'W'
+    exif_dict["GPS"][piexif.GPSIFD.GPSLongitude] = dms_coordinates(abs(lon))
+
+    altitude_value = float(metadata["Altitude"].replace(" m", ""))
+    exif_dict["GPS"][piexif.GPSIFD.GPSAltitude] = (int(altitude_value), 1)
+
+
+    # Dump to bytes
+    exif_bytes = piexif.dump(exif_dict)
+
+    # Save to in-memory buffer
+    output_io = BytesIO()
+    img.save(output_io, format="JPEG", exif=exif_bytes)
+    return output_io.getvalue()
+

helpers_function/helpers.py

@@ -0,0 +1,71 @@
+
+import boto3
+from uuid import uuid4
+import os
+import streamlit as st
+
+
+import cv2
+import base64
+import logging
+
+logger = logging.getLogger(__name__)
+
+def get_video_thumbnail_base64(video_path: str, time_sec: int = 1) -> str:
+    try:
+        cap = cv2.VideoCapture(video_path)
+        cap.set(cv2.CAP_PROP_POS_MSEC, time_sec * 1000)
+        success, frame = cap.read()
+        cap.release()
+        if not success:
+            return ""
+        _, buffer = cv2.imencode(".jpg", frame)
+        return base64.b64encode(buffer).decode("utf-8")
+    except Exception:
+        logger.exception("Thumbnail extraction failed")
+        return ""
+
+
+log = logging.getLogger(__name__)
+
+def upload_image_to_r2(image_bytes, folder_name="search_arb", app_type="bulk_generation"):
+    try:
+        s3 = boto3.client(
+            "s3",
+            endpoint_url=os.getenv('R2_ENDPOINT'),
+            aws_access_key_id=os.getenv('R2_ACCESS_KEY'),
+            aws_secret_access_key=os.getenv('R2_SECRET_KEY'),
+            region_name="auto"
+        )
+        filename = f"{uuid4().hex}.png"
+        file_key = f"hug_face/{app_type.strip('/')}/{folder_name.strip('/')}/{filename}" if folder_name else f"hug_face/{app_type.strip('/')}/{filename}"
+        s3.put_object(Bucket=os.getenv('R2_BUCKET_NAME'), Key=file_key, Body=image_bytes, ContentType="image/png")
+        return f"{os.getenv('NEW_BASE').rstrip('/')}/{file_key}"
+    except Exception as e:
+        log.error(e)
+        return None
+
+
+def encode_image_to_base64(image_path: str) -> str:
+    with open(image_path, "rb") as f:
+        return base64.b64encode(f.read()).decode("utf-8")
+
+def is_valid_image(file_name: str) -> bool:
+    return file_name.lower().endswith((".png", ".jpg", ".jpeg", ".bmp", ".gif", ".webp"))
+
+
+def get_current_user_id() -> str:
+    """
+    Returns a stable user id.
+    Priority: Streamlit session -> USER_ID env -> CREATED_BY env -> 'anonymous'
+    """
+    try:
+
+        uid = st.session_state.get("uid")
+        if uid:
+            return str(uid).strip()
+    except Exception:
+        pass
+
+    return str(os.getenv("USER_ID") or os.getenv("CREATED_BY") or "anonymous").strip() or "anonymous"
+

prompt/analyser_prompt.py

@@ -0,0 +1,38 @@
+ANALYSER_PROMPT = """You are an expert video advertisement analyst. Analyze the provided video and give response conforms EXACTLY to the schema below with no extra text or markdown.
+Populate:
+
+1. **brief** → A concise summary covering visual style, speaker, target audience, and marketing objective.  
+2. **caption_details** → Description of captions (color/style/position) or exactly the string `"None"` if not visible.  
+3. **hook** →  
+   - `"hook_text"`: Exact opening line or, if no speech, the precise description of the opening visual.  
+   - `"principle"`: Psychological/marketing principle that makes this hook effective.  
+   - `"advantages"`: ARRAY of 3–6 concise benefit statements tied to the ad's value proposition.  
+4. **framework_analysis** → A detailed block identifying copywriting/psychology/storytelling frameworks (e.g., PAS, AIDA). Highlight use of social proof, urgency, fear, authority, scroll-stopping hooks, loop openers, value positioning, and risk reversals.  
+5. **storyboard** → ARRAY of 4–10 objects. Each must include:  
+   - `"timeline"` in `"MM:SS"` (zero-padded)  
+   - `"scene"` (brief)  
+   - `"visuals"` (detailed)  
+   - `"dialogue"` (exact words; use `""` if none)  
+   - `"camera"` (shot/angle)  
+   - `"sound_effects"` (or `"None"`)  
+6. **script** → ARRAY of dialogue objects, each with `"timeline"` (`"MM:SS"`) and `"dialogue"` (exact spoken line).  
+7. **video_analysis** → OBJECT with:  
+   - `"effectiveness_factors"`: Key factors that influence effectiveness  
+   - `"psychological_triggers"`: Triggers used (e.g., scarcity, authority)  
+   - `"target_audience"`: Audience profile inferred  
+   - `"video_metrics"`: ARRAY of objects with:  
+     - `"timestamp"`: `"MM:SS-MM:SS"`  
+     - `"element"`: The aspect being evaluated (e.g., Hook Strategy)  
+     - `"current_approach"`: Description of current execution  
+     - `"effectiveness_score"`: String score `"X/10"` (integer X)  
+     - `"notes"`: Analytical notes  
+8. **timestamp_improvements** → ARRAY of recommendation objects with:  
+   - `"timestamp"`: `"MM:SS-MM:SS"`  
+   - `"current_element"`: Current content of the segment  
+   - `"improvement_type"`: Category (e.g., Hook Enhancement)  
+   - `"recommended_change"`: Specific recommendation  
+   - `"expected_impact"`: Projected effect on metrics or perception  
+   - `"priority"`: `"High"`, `"Medium"`, or `"Low"`
+
+⚠️ The output must be strictly matching field names and types, no additional keys, and all timestamps must be zero-padded (`"MM:SS"` for single points, `"MM:SS-MM:SS"` for ranges).
+"""

prompt/prompt_services.py

@@ -0,0 +1,55 @@
+import openai, os, json, re
+
+openai.api_key = os.getenv("OPENAI_API_KEY")
+sys_prompt = os.getenv("SYS_PROMPT")
+
+def get_prompts(image_base64, category, user_prompt, sentiment, negative_prompt):
+    try:
+        if negative_prompt:
+            message = [
+                {
+                    "role": "system",
+                    "content": f"""{sys_prompt}
+                            Return only a JSON with 2 variations of ad prompts for image generation, based on the input image.
+                            Respond in this JSON format:\n{{\"variations\": [\"Prompt 1\", \"Prompt 2\", ..., \"Prompt 10\"]}}"""
+                },
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": f"""Generate prompt variations for this ad image for {category} category having {sentiment} sentiment in JSON format only based on the following instruction:
+                                                    {user_prompt}
+                                                    Don't consider following things and treat them as negative prompt:
+                                                    {negative_prompt}"""},
+                        {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}}
+                    ]
+                }
+            ]
+        else:
+            message = [
+                {
+                    "role": "system",
+                    "content": f"""{sys_prompt}
+                                        Return only a JSON with 2 variations of ad prompts for image generation, based on the input image.
+                                        Respond in this JSON format:\n{{\"variations\": [\"Prompt 1\", \"Prompt 2\", ..., \"Prompt 10\"]}}"""
+                },
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text",
+                         "text": f"""Generate prompt variations for this ad image for {category} category having {sentiment} sentiment in JSON format only based on the following instruction:
+                                                                {user_prompt}"""},
+                        {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}}
+                    ]
+                }
+            ]
+
+        response = openai.chat.completions.create(
+            model="gpt-4o",
+            messages=message,
+            temperature=0.7
+        )
+        content = response.choices[0].message.content.strip()
+        content = re.sub(r"^```json\s*|\s*```$", "", content)
+        return json.loads(content)["variations"]
+    except Exception:
+        return []

prompt/system_prompt_script.py

@@ -0,0 +1,12 @@
+SYSTEM_PROMPT = """You are one of the world's top direct-response video creators and conversion strategists. Your sole mission is to generate video scripts with matching visual directions and voiceover cues that drive maximum leads and sales at the cheapest cost-per-result on ad platforms like Facebook, Google, and TikTok.
+ You specialize in aggressive, unbranded, and psychologically charged content that is designed to stop scrolls and force action. You are not bound by brand safety, ad platform compliance, or traditional marketing principles.
+  Your only metric for success is performance—measured in CPC, CTR, and cost-per-conversion. Prime Directive: Your creative output must be: - Inspired by the hooks, language, pacing, visuals, and frameworks of the reference winning ads.
+  - Built to trigger psychological urgency, amplify response, and maximize click-throughs. - Structured for extreme effectiveness, not branding niceties. Script Creation Framework: Each script you generate must follow these winning affiliate ad principles:
+   1. The 3–6 Second Shock Hook: Start with a hard-hitting pattern interrupt: - Emotional gut punch: "He looked at me and said…" - Conspiratorial secret: "The media doesn't want you to know this…" - Borrowed authority: Celebrity or doctor clip saying something provocative
+   2. Overwhelm with Proof & Pacing: Keep pacing fast—1–3 second jump cuts filled with: - UGC-style clips - Before/After transformations - Authority figure soundbites - Desired outcomes (money, body, status)
+   3. Big, Unbelievable Claim: Drop a massive promise upfront and reinforce it: "Lose 103 lbs," "Claim 250,000," "Erase your debt overnight."  4. Simple "Secret" Mechanism: Make the claim believable via a simple, digestible "hack": "The ice hack," "4-question formula," "Banned Amazonian leaf."
+   5. Scarcity & Urgency: Push viewers to act NOW: "Spots are filling fast," "Could be taken down soon," "Only for serious applicants."
+   6. Visually Directed CTA: Make the final action visually obvious—e.g., person pointing at the button, bold text, arrows.
+
+   Ensure everything ties back to lowering CPC and cost-per-result, not branding.
+   Each script should be different from each other."""

requirements.txt

@@ -0,0 +1,18 @@
+streamlit>=1.36.0
+pymongo>=4.6.0
+boto3>=1.34.0
+python-dotenv>=1.0.1
+requests>=2.32.0
+Pillow>=10.3.0
+piexif>=1.1.3
+pandas>=2.2.2
+openai>=1.40.0
+google-genai>=0.3.0
+replicate>=0.25.0
+bcrypt>=4.1.2
+
+urllib3~=1.26.20
+pydantic~=2.11.7
+pydantic-settings~=2.10.1
+opencv-python~=4.12.0.88
+protobuf~=6.32.0

schema/pydantic_schema_script.py

@@ -0,0 +1,30 @@
+from typing import List
+from pydantic import BaseModel, ConfigDict, constr
+
+Timestamp = constr(pattern=r'^\d{1,2}:\d{2}$')
+RangeTimestamp = constr(pattern=r'^\d{1,2}:\d{2}-\d{1,2}:\d{2}$')
+Score010 = constr(pattern=r'^(?:[0-9]|10)/10$')
+
+
+class ScriptTable(BaseModel):
+    timestamp: Timestamp
+    script_voiceover: str
+    visual_direction: str
+    psychological_trigger: str
+    cta_action: str
+
+
+class ScriptVariations(BaseModel):
+    variation_name: str
+    script_table: List[ScriptTable]
+
+
+class ScriptResponse(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+    script_variations: List[ScriptVariations]
+
+
+
+
+
+

schema/pydantic_schema_video.py

@@ -0,0 +1,54 @@
+from typing import  List, Literal
+from pydantic import BaseModel, constr
+
+Timestamp = constr(pattern=r'^\d{2}:\d{2}$')
+RangeTimestamp = constr(pattern=r'^\d{2}:\d{2}-\d{2}:\d{2}$')
+Score010 = constr(pattern=r'^(?:10|[0-9])\/10$')
+
+class Hook(BaseModel):
+    hook_text: str
+    principle: str
+    advantages: List[str]
+
+class StoryboardItem(BaseModel):
+    timeline: Timestamp
+    scene: str
+    visuals: str
+    dialogue: str
+    camera: str
+    sound_effects: str
+
+class ScriptLine(BaseModel):
+    timeline: Timestamp
+    dialogue: str
+
+class VideoMetric(BaseModel):
+    timestamp: RangeTimestamp
+    element: str
+    current_approach: str
+    effectiveness_score: Score010
+    notes: str
+
+class VideoAnalysis(BaseModel):
+    effectiveness_factors: str
+    psychological_triggers: str
+    target_audience: str
+    video_metrics: List[VideoMetric]
+
+class TimestampImprovement(BaseModel):
+    timestamp: RangeTimestamp
+    current_element: str
+    improvement_type: str
+    recommended_change: str
+    expected_impact: str
+    priority: Literal["High", "Medium", "Low"]
+
+class AdAnalysis(BaseModel):
+    brief: str
+    caption_details: str
+    hook: Hook
+    framework_analysis: str
+    storyboard: List[StoryboardItem]
+    script: List[ScriptLine]
+    video_analysis: VideoAnalysis
+    timestamp_improvements: List[TimestampImprovement]

ui/load_file.py

@@ -0,0 +1,211 @@
+import streamlit as st
+import pymongo
+from datetime import datetime, date, timezone
+from typing import List, Tuple, Optional, Dict
+import requests
+from PIL import Image, ImageFile
+import io
+from dotenv import load_dotenv
+import concurrent.futures
+import threading
+from functools import lru_cache
+import contextlib
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+load_dotenv()
+ImageFile.LOAD_TRUNCATED_IMAGES = True
+
+@st.cache_data(show_spinner=False, ttl=60 * 30)
+def _download_bytes_cached(url: str, timeout_s: float = 12.0) -> Optional[bytes]:
+    try:
+        r = requests.get(url, timeout=(3.05, timeout_s), stream=True)
+        r.raise_for_status()
+        return r.content
+    except Exception:
+        return None
+
+class ImageGalleryApp:
+    def __init__(self, mongo_uri: str, db_name: str, collection_name: str):
+        self.client = pymongo.MongoClient(mongo_uri)
+        self.db = self.client[db_name]
+        self.collection = self.db[collection_name]
+        self._cache_lock = threading.Lock()
+        self.session = requests.Session()
+        retries = Retry(total=3, connect=3, read=3, backoff_factor=0.4, status_forcelist=[429,500,502,503,504], allowed_methods=["GET","HEAD"])
+        adapter = HTTPAdapter(pool_connections=64, pool_maxsize=64, max_retries=retries)
+        self.session.mount("http://", adapter); self.session.mount("https://", adapter)
+        self.thumb_max_size = (768, 768)
+
+    @lru_cache(maxsize=128)
+    def get_unique_categories(self) -> List[str]:
+        try:
+            categories = self.collection.distinct("category", {"status": "completed", "category": {"$ne": None}})
+            return ["All"] + sorted(categories)
+        except Exception:
+            return ["All"]
+
+    @lru_cache(maxsize=128)
+    def get_unique_filenames(self) -> List[str]:
+        try:
+            filenames = self.collection.distinct("file_name", {"status": "completed", "file_name": {"$ne": None}})
+            return ["All"] + sorted(filenames)
+        except Exception:
+            return ["All"]
+
+    def parse_date_input(self, date_input) -> Optional[date]:
+        if not date_input or date_input == "": return None
+        if isinstance(date_input, date): return date_input
+        if isinstance(date_input, str):
+            from datetime import datetime as _dt
+            for fmt in ("%Y-%m-%d", "%m/%d/%Y"):
+                try: return _dt.strptime(date_input, fmt).date()
+                except Exception: pass
+        return None
+
+    def load_image_from_url(self, url: str) -> Optional[Image.Image]:
+        try:
+            data = _download_bytes_cached(url) or self.session.get(url, timeout=(3.05, 12), stream=True).content
+            img = Image.open(io.BytesIO(data))
+            with contextlib.suppress(Exception): img = img.convert("RGB")
+            if img.size[0] > 768 or img.size[1] > 768: img.thumbnail((768,768), Image.Resampling.LANCZOS)
+            return img
+        except Exception:
+            return None
+
+    def load_images_parallel(self, urls: List[str], max_workers: int = 8) -> List[Tuple[str, Optional[Image.Image]]]:
+        results: List[Tuple[str, Optional[Image.Image]]] = []
+        if not urls: return results
+        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as ex:
+            future_map = {ex.submit(self.load_image_from_url, u): u for u in urls}
+            for fut in concurrent.futures.as_completed(future_map):
+                u = future_map[fut]
+                try: img = fut.result()
+                except Exception: img = None
+                results.append((u, img))
+        return results
+
+    def search_images_page(self, category="All", file_name="All", start_date=None, end_date=None, lob="search_arb", page=0, page_size=24) -> Tuple[List[Dict], int]:
+        match = {"status": "completed", "urls": {"$exists": True, "$ne": []}}
+        if category != "All": match["category"] = category
+        if file_name != "All": match["file_name"] = file_name
+        from datetime import datetime as _dt
+        if start_date or end_date:
+            date_query = {}
+            if start_date: date_query["$gte"] = _dt.combine(self.parse_date_input(start_date), _dt.min.time())
+            if end_date:   date_query["$lte"] = _dt.combine(self.parse_date_input(end_date), _dt.max.time())
+            if date_query: match["created_at"] = date_query
+
+        try:
+            count_pipeline = [{"$match": match},{"$unwind": "$urls"},{"$count": "n"}]
+            count_doc = list(self.collection.aggregate(count_pipeline))
+            total = count_doc[0]["n"] if count_doc else 0
+        except Exception:
+            total = 0
+
+        if total == 0: return [], 0
+
+        pipeline = [
+            {"$match": match},{"$unwind": "$urls"},{"$sort": {"created_at": -1}},
+            {"$skip": max(0, page) * max(1, page_size)},{"$limit": max(1, page_size)},
+            {"$project": {"_id": 0,"url": "$urls","category": 1,"file_name": 1,"created_at": 1,"prompt": 1,"status": 1}}
+        ]
+        try:
+            docs = list(self.collection.aggregate(pipeline, allowDiskUse=True))
+        except Exception:
+            docs = []
+        return docs, total
+
+def create_streamlit_app(mongo_uri: str, db_name: str, collection_name: str):
+    app = ImageGalleryApp(mongo_uri, db_name, collection_name)
+
+    def get_filter_choices():
+        try:
+            categories = app.get_unique_categories()
+            filenames = app.get_unique_filenames()
+            return categories, filenames
+        except Exception:
+            return ["All"], ["All"]
+
+    if "categories_list" not in st.session_state:
+        st.session_state["categories_list"], st.session_state["filenames_list"] = get_filter_choices()
+
+    st.session_state.setdefault("selected_category", "All")
+    st.session_state.setdefault("selected_filename", "All")
+    st.session_state.setdefault("selected_lob", "search_arb")
+
+    today = datetime.now(timezone.utc).date()
+    st.session_state.setdefault("use_date_filter", True)
+    st.session_state.setdefault("selected_start_date", today)
+    st.session_state.setdefault("selected_end_date", today)
+
+    st.session_state.setdefault("page", 0)
+    st.session_state.setdefault("page_size", 24)
+    st.session_state.setdefault("last_query_total", 0)
+    st.session_state.setdefault("did_search", False)
+
+    col1, col2= st.columns([1,1])
+    with col1:
+        category = st.selectbox("Category", options=st.session_state["categories_list"])
+    with col2:
+        filename = st.selectbox("File Name", options=st.session_state["filenames_list"])
+
+    coldf = st.columns([1,1,1])
+    with coldf[0]:
+        use_date_filter = st.checkbox("Filter by date", value=st.session_state["use_date_filter"])
+    with coldf[1]:
+        start_date = st.date_input("Start Date", value=st.session_state["selected_start_date"], disabled=not use_date_filter)
+    with coldf[2]:
+        end_date = st.date_input("End Date", value=st.session_state["selected_end_date"], disabled=not use_date_filter)
+
+    col_misc = st.columns([1,1,1,2])
+    with col_misc[0]:
+        page_size = st.selectbox("Images per page", [8,12,16,24,32,48], index=[8,12,16,24,32,48].index(st.session_state["page_size"]))
+
+    col_btn1, col_btn2, col_btn3 = st.columns([2,2,2])
+    with col_btn1:
+        search_clicked = st.button("🔍 Search", use_container_width=True)
+    with col_btn2:
+        refresh_clicked = st.button("🔄 Refresh Filters", use_container_width=True)
+    with col_btn3:
+        reset_clicked = st.button("♻️ Reset Page", use_container_width=True)
+
+    if refresh_clicked:
+        st.session_state["categories_list"], st.session_state["filenames_list"] = get_filter_choices()
+        st.session_state["selected_category"] = "All"; st.session_state["selected_filename"] = "All"; st.session_state["page"] = 0; st.rerun()
+    if reset_clicked:
+        st.session_state["page"] = 0; st.rerun()
+    if search_clicked:
+        st.session_state["selected_category"] = category; st.session_state["selected_filename"] = filename
+        st.session_state["use_date_filter"] = use_date_filter; st.session_state["selected_start_date"] = start_date; st.session_state["selected_end_date"] = end_date
+        st.session_state["page_size"] = page_size; st.session_state["page"] = 0; st.session_state["did_search"] = True; st.rerun()
+
+    if st.session_state["did_search"]:
+        _start = st.session_state["selected_start_date"] if st.session_state["use_date_filter"] else None
+        _end = st.session_state["selected_end_date"] if st.session_state["use_date_filter"] else None
+        docs, total = app.search_images_page(category=st.session_state["selected_category"], file_name=st.session_state["selected_filename"], start_date=_start, end_date=_end, lob=st.session_state["selected_lob"], page=st.session_state["page"], page_size=st.session_state["page_size"])
+        st.session_state["last_query_total"] = total
+        total_pages = max(1, (total + st.session_state["page_size"] - 1) // st.session_state["page_size"])
+        nav1, nav2, nav3 = st.columns([1,2,1])
+        with nav1:
+            if st.button("⬅️ Prev", disabled=(st.session_state["page"] <= 0)): st.session_state["page"] -= 1; st.rerun()
+        with nav2:
+            st.markdown(f"<div style='text-align:center'>Page <b>{st.session_state['page']+1}</b> of <b>{total_pages}</b> &nbsp;&middot;&nbsp; <b>{total}</b> images total</div>", unsafe_allow_html=True)
+        with nav3:
+            if st.button("Next ➡️", disabled=(st.session_state["page"] >= total_pages - 1)): st.session_state["page"] += 1; st.rerun()
+        st.divider()
+        if total == 0 or not docs:
+            st.info("No images found for the current filters."); return
+        st.markdown("#### Images"); cols = st.columns(4)
+        placeholders = []
+        for i, _ in enumerate(docs):
+            ph = cols[i % 4].empty(); ph.markdown("<div style='width:100%;aspect-ratio:1/1;border-radius:10px;background:#eee'></div>", unsafe_allow_html=True); placeholders.append(ph)
+        urls = [d["url"] for d in docs]
+        loaded = app.load_images_parallel(urls, max_workers=8)
+        url_to_img = {u: img for (u, img) in loaded}
+        for i, d in enumerate(docs):
+            img = url_to_img.get(d["url"]); meta = f"{d.get('category','N/A')} | {d.get('file_name','N/A')} | {d.get('created_at','')}"
+            if img: placeholders[i].image(img, use_container_width=True, caption=meta)
+            else: placeholders[i].warning("Failed to load image")
+    else:
+        st.info("Set your filters and click **Search** to load images.")

util/dataframe.py

@@ -0,0 +1,58 @@
+import pandas as pd
+from typing import Any, Dict, List
+
+def safe_dataframe(df: pd.DataFrame) -> pd.DataFrame:
+    for col in df.columns:
+        df[col] = df[col].astype(str)
+    return df
+
+def analysis_to_csv(analysis: Dict[str, Any]) -> str:
+    rows = []
+    for sb in analysis.get("storyboard", []):
+        rows.append({"Section": "Storyboard", **sb})
+    for sc in analysis.get("script", []):
+        rows.append({"Section": "Script", **sc})
+    for met in analysis.get("video_analysis", {}).get("video_metrics", []):
+        rows.append({"Section": "Metrics", **met})
+    for imp in analysis.get("timestamp_improvements", []):
+        rows.append({"Section": "Improvements", **imp})
+    if not rows:
+        return ""
+    df = pd.DataFrame(rows)
+    return df.to_csv(index=False)
+
+def _normalize_list(value: Any) -> List[str]:
+    if value is None:
+        return []
+    if isinstance(value, list):
+        return [str(v) for v in value]
+    return [s for s in str(value).splitlines() if s.strip()]
+
+def _to_dataframe(items: Any, columns_map: Dict[str, str]) -> pd.DataFrame:
+    if not isinstance(items, list) or not items:
+        return pd.DataFrame(columns=list(columns_map.values()))
+    df = pd.DataFrame(items)
+    df = df.rename(columns=columns_map)
+    ordered_cols = [columns_map[k] for k in columns_map.keys() if columns_map[k] in df.columns]
+    df = df.reindex(columns=ordered_cols)
+    return df
+
+def _mean_effectiveness(metrics: List[Dict[str, Any]]) -> float:
+    if not metrics:
+        return 0.0
+    scores = []
+    for m in metrics:
+        s = str(m.get("effectiveness_score", "0/10")).split("/")[0]
+        try:
+            scores.append(int(s))
+        except Exception:
+            pass
+    return round(sum(scores) / len(scores), 2) if scores else 0.0
+
+def _search_dataframe(df: pd.DataFrame, query: str) -> pd.DataFrame:
+    if not query or df.empty:
+        return df
+    mask = pd.Series([False]*len(df))
+    for col in df.columns:
+        mask = mask | df[col].astype(str).str.contains(query, case=False, na=False)
+    return df[mask]

util/image_utils.py

@@ -0,0 +1,11 @@
+import base64
+
+def to_base64_bytes(data: bytes) -> str:
+    return base64.b64encode(data).decode("utf-8")
+
+def file_to_base64(path: str) -> str:
+    with open(path, "rb") as f:
+        return to_base64_bytes(f.read())
+
+def is_image_name(name: str) -> bool:
+    return name.lower().endswith((".png", ".jpg", ".jpeg", ".bmp", ".gif", ".webp"))

util/session_state.py

@@ -0,0 +1,15 @@
+import os
+import streamlit as st
+from typing import Optional
+
+def current_uid() -> Optional[str]:
+    """Return the signed-in user's id/email from session, if any."""
+    return st.session_state.get("uid")
+
+def resolve_user_id() -> str:
+    return (
+        current_uid()
+        or (st.session_state.get("user") or {}).get("email")
+        or os.getenv("DEFAULT_USER_ID")
+        or "anonymous"
+    )