Create video_export.py

src/video_export.py

@@ -0,0 +1,167 @@
+# src/video_export.py
+import os
+from datetime import datetime
+import numpy as np
+import pandas as pd
+import streamlit as st
+from PIL import Image, ImageDraw, ImageFont
+from moviepy.editor import ImageClip, concatenate_videoclips
+from moviepy.audio.io.AudioFileClip import AudioFileClip
+from gtts import gTTS
+
+from src.utils import APP_DIR
+
+def ensure_video_deps():
+    # ensure font exists; DejaVu Sans is usually present via fonts-dejavu-core
+    pass
+
+def _get_font(size: int):
+    for p in [
+        "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
+        "/usr/share/fonts/truetype/dejavu/DejaVuSansCondensed.ttf",
+    ]:
+        if os.path.exists(p):
+            return ImageFont.truetype(p, size=size)
+    return ImageFont.load_default()
+
+def _draw_multiline(draw, text, xy, font, fill, max_w, leading=6):
+    text = (text or "").strip()
+    if not text:
+        text = "."
+    words = text.split()
+    lines, cur = [], []
+    for w in words:
+        test = " ".join(cur + [w]).strip()
+        w_px = draw.textbbox((0,0), test, font=font)[2]
+        if w_px <= max_w: cur.append(w)
+        else:
+            lines.append(" ".join(cur).strip()); cur = [w]
+    if cur: lines.append(" ".join(cur).strip())
+    x, y = xy
+    for line in lines:
+        draw.text((x,y), line, font=font, fill=fill)
+        _,_,_,h = draw.textbbox((x,y), line, font=font)
+        y += h + leading
+    return y
+
+def build_slide(title, body, w=1280, h=720, seconds=3.5, bg=(245,247,250)):
+    img = Image.new("RGB", (w,h), color=bg)
+    d = ImageDraw.Draw(img)
+    title_font = _get_font(48)
+    body_font  = _get_font(36)
+    left, right = 60, w-60
+    max_w = right - left
+    y0 = 80
+    y1 = _draw_multiline(d, title, (left,y0), title_font, "black", max_w, 8)
+    y_body = max(y0 + 80, y1 + 20)
+    _draw_multiline(d, body, (left,y_body), body_font, "black", max_w, 6)
+    frame = np.array(img)
+    return ImageClip(frame).set_duration(seconds)
+
+def tts(text: str, lang: str, out_mp3: str):
+    safe = (text or ".").strip()
+    gTTS(text=safe, lang=lang, slow=False).save(out_mp3)
+    return out_mp3
+
+def narration_text(title: str, body: str, max_chars: int = 600) -> str:
+    txt = f"{title}. {body}".strip()
+    return (txt[:max_chars-1].rsplit(" ", 1)[0] + "…") if len(txt) > max_chars else txt
+
+def make_video(rows, title="Olist Chatbot — Conversation", narrate=True, tts_lang="en",
+               seconds_per_slide=3.5, out_path="chat_video.mp4"):
+    tmp_dir = APP_DIR / ".cache" / "tts_tmp"
+    if narrate:
+        os.makedirs(tmp_dir, exist_ok=True)
+
+    clips = []
+    def _add(title, body, bg=None):
+        duration = seconds_per_slide
+        audio = None
+        if narrate:
+            try:
+                mp3 = str(tmp_dir / f"tts_{len(clips):04d}.mp3")
+                tts(narration_text(title, body), tts_lang, mp3)
+                audio = AudioFileClip(mp3)
+                duration = max(seconds_per_slide, float(audio.duration) + 0.4)
+            except Exception:
+                audio = None
+        clip = build_slide(title, body, seconds=duration, bg=(245,247,250) if bg is None else bg)
+        if narrate and audio is not None:
+            clip = clip.set_audio(audio)
+        clips.append(clip)
+
+    _add(title, "Auto-generated summary video")
+    for _, r in rows.iterrows():
+        ts = str(r.get("timestamp",""))
+        msg = str(r.get("user_message",""))
+        rep = str(r.get("bot_reply",""))
+        _add(f"User • {ts}", msg, bg=(255,255,255))
+        _add("Agent", rep, bg=(242,255,242))
+    _add("Thanks for watching", "Exported from the Olist app")
+
+    video = concatenate_videoclips(clips, method="compose")
+    video.write_videofile(
+        out_path, fps=24, codec="libx264",
+        audio=narrate, audio_codec="aac" if narrate else None,
+        threads=2, preset="medium", ffmpeg_params=["-movflags","+faststart"]
+    )
+    # cleanup tts mp3s best effort
+    try:
+        if narrate and os.path.isdir(tmp_dir):
+            for f in os.listdir(tmp_dir):
+                if f.endswith(".mp3"):
+                    os.remove(tmp_dir / f)
+    except Exception:
+        pass
+    return out_path
+
+def render_admin_export_video_tab(conn):
+    st.title("📥 Export + 🎬 Video (Admin)")
+    st.write("Download CSV/TXT for all users, or render a narrated MP4.")
+
+    df = pd.read_sql_query(
+        "SELECT user_id, timestamp, user_message, bot_reply FROM chat_history ORDER BY timestamp DESC",
+        conn
+    )
+    if df.empty:
+        st.info("No chats yet.")
+        return
+
+    # CSV/TXT export for admins (all users)
+    csv_bytes = df.to_csv(index=False).encode("utf-8")
+    st.download_button("Download all chats (CSV)", csv_bytes, file_name="all_chats.csv", mime="text/csv")
+
+    txt_lines = []
+    for _, r in df.iterrows():
+        txt_lines.append(f"[{r['timestamp']}] user_id={r['user_id']} | {r['user_message']}")
+        txt_lines.append(f"Bot: {r['bot_reply']}")
+        txt_lines.append("---")
+    st.download_button("Download all chats (TXT)", "\n".join(txt_lines).encode("utf-8"),
+                       file_name="all_chats.txt", mime="text/plain")
+
+    st.markdown("---")
+    st.subheader("🎬 Build video")
+    c1, c2 = st.columns(2)
+    with c1:
+        narrate = st.checkbox("Add voiceover (gTTS)", value=True)
+        seconds_per_slide = st.slider("Seconds per slide", 2.0, 8.0, 3.5, 0.5)
+    with c2:
+        tts_lang = st.selectbox("TTS language", ["en","hi","pt","es","fr"], index=0)
+        max_rows = st.number_input("Max messages (0 = all)", min_value=0, max_value=200, value=50, step=1)
+
+    rows = df if max_rows == 0 else df.head(int(max_rows))
+
+    if st.button("Make Video"):
+        try:
+            out_path = make_video(rows, narrate=narrate, tts_lang=tts_lang,
+                                  seconds_per_slide=seconds_per_slide, out_path="chat_video.mp4")
+            with open(out_path, "rb") as f:
+                data = f.read()
+            st.success("Video ready!")
+            st.video(data)
+            st.download_button(
+                "Download MP4", data, file_name=f"chat_video_{datetime.now().strftime('%Y%m%d_%H%M%S')}.mp4",
+                mime="video/mp4"
+            )
+        except Exception as e:
+            st.error(f"Video generation failed: {e}")