app.py

import os
import logging
import time
import uuid
import streamlit as st
import streamlit.components.v1 as components
from google import genai
from langchain_core.prompts import PromptTemplate
from langchain_google_genai import ChatGoogleGenerativeAI
from dotenv import load_dotenv
from yt_dlp import YoutubeDL
import validators
from datetime import datetime
import json
from urllib.parse import urlparse
from collections import deque
from threading import Lock

logger = logging.getLogger("tubetotweet")
if not logger.handlers:
    logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))

load_dotenv()

REQUEST_TIMEOUT_SECONDS = int(os.getenv("REQUEST_TIMEOUT_SECONDS", "15"))
RATE_LIMIT_WINDOW_SECONDS = int(os.getenv("RATE_LIMIT_WINDOW_SECONDS", "300"))
RATE_LIMIT_MAX_REQUESTS = int(os.getenv("RATE_LIMIT_MAX_REQUESTS", "8"))

api_key = os.getenv("GOOGLE_API_KEY")
if not api_key:
    try:
        api_key = st.secrets.get("GOOGLE_API_KEY")
    except Exception:
        api_key = None

default_models = [
    "models/gemini-2.5-flash-lite",
    "models/gemini-2.5-flash",
]
model_env = os.getenv("GEMINI_MODELS", "").strip()
if not model_env:
    try:
        model_env = str(st.secrets.get("GEMINI_MODELS", "")).strip()
    except Exception:
        model_env = ""
raw_candidates = [m.strip() for m in model_env.split(",") if m.strip()] or default_models
model_candidates = []
for name in raw_candidates:
    model_candidates.append(name)
    if not name.startswith("models/"):
        model_candidates.append(f"models/{name}")
model_candidates = list(dict.fromkeys(model_candidates))

@st.cache_resource
def get_rate_limit_resources():
    return {"events": {}, "lock": Lock()}


def get_client_ip():
    try:
        headers = {str(k).lower(): str(v) for k, v in st.context.headers.items()}
    except Exception:
        return None

    forwarded_for = headers.get("x-forwarded-for", "").split(",")[0].strip()
    if forwarded_for:
        return forwarded_for

    return headers.get("x-real-ip") or headers.get("cf-connecting-ip")


def consume_rate_limit():
    now = time.time()
    resources = get_rate_limit_resources()

    session_id = st.session_state.get("client_session_id")
    identifiers = [f"session:{session_id}"]

    client_ip = get_client_ip()
    if client_ip:
        identifiers.append(f"ip:{client_ip}")

    retry_after_seconds = 0

    with resources["lock"]:
        events_map = resources["events"]

        for identifier in identifiers:
            events = events_map.setdefault(identifier, deque())
            while events and now - events[0] > RATE_LIMIT_WINDOW_SECONDS:
                events.popleft()

            if len(events) >= RATE_LIMIT_MAX_REQUESTS:
                wait_time = int(RATE_LIMIT_WINDOW_SECONDS - (now - events[0])) + 1
                retry_after_seconds = max(retry_after_seconds, wait_time)

        if retry_after_seconds > 0:
            return False, retry_after_seconds

        for identifier in identifiers:
            events_map.setdefault(identifier, deque()).append(now)

    return True, 0


def is_allowed_youtube_url(url):
    try:
        parsed = urlparse(url.strip())
    except Exception:
        return False

    if parsed.scheme not in {"http", "https"}:
        return False

    host = (parsed.hostname or "").lower()
    return host == "youtu.be" or host == "youtube.com" or host.endswith(".youtube.com")

def get_active_api_key():
    user_key = st.session_state.get("user_api_key", "").strip()
    return user_key or api_key

# Initialize Gemini model on-demand to allow fallback
def create_llm(model_name):
    active_api_key = get_active_api_key()
    if not active_api_key:
        raise ValueError("Gemini API key is missing. Enter your own API key to continue.")
    return ChatGoogleGenerativeAI(
        model=model_name,
        google_api_key=active_api_key,
        temperature=0.7  # Adjust for creativity: lower for factual, higher for viral flair
    )

def extract_response_text(response):
    content = getattr(response, "content", response)
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for item in content:
            if isinstance(item, dict) and "text" in item:
                parts.append(item["text"])
            elif hasattr(item, "text"):
                parts.append(item.text)
            else:
                parts.append(str(item))
        return "\n".join(parts).strip()
    return str(content)

def run_chain_with_fallback(input_text, prompt, candidates, max_retries=2, retry_delay=1.5):
    prompt_text = prompt.format(text=input_text)
    last_error = None

    for model_name in candidates:
        for attempt in range(1, max_retries + 1):
            try:
                llm = create_llm(model_name)
                response = llm.invoke(prompt_text)
                return extract_response_text(response), model_name
            except Exception as exc:
                last_error = exc
                logger.warning("Model call failed model=%s attempt=%s error=%s", model_name, attempt, exc)
                if attempt < max_retries:
                    time.sleep(retry_delay)

    raise RuntimeError("All configured Gemini models failed.") from last_error

def list_available_models(api_key_value):
    client = genai.Client(api_key=api_key_value or get_active_api_key())
    models = []
    for model in client.models.list():
        name = getattr(model, "name", None)
        if name:
            models.append(name)
    return models

# Prompt for full summary
summary_prompt_template = """
Provide a concise 300-500 word summary of the video described here:
{text}
Cover main topics, key points, and takeaways factually without hallucinations.
"""
summary_prompt = PromptTemplate(template=summary_prompt_template, input_variables=["text"])

# Prompt for viral X post (chained after summary)
x_post_prompt_template = """
You are a professional copywriter and expert social media manager specializing in viral X content.

From this video summary:
{text}

Write a viral X post under 280 characters that:
- Starts with a compelling hook to stop the scroll
- Includes one key insight or takeaway
- Uses 2-3 impactful emojis strategically placed
- Includes 2-3 relevant hashtags
- Ends with a clear, action-driving call-to-action
- Is engaging, impressive, memorable, and highly shareable

Focus on creating content that sparks curiosity, emotion, or urgency while maintaining authenticity.
"""
x_post_prompt = PromptTemplate(template=x_post_prompt_template, input_variables=["text"])

# Function to extract YouTube metadata (title, description, transcript hints if available)
def load_youtube_content(url):
    ydl_opts = {
        'format': 'bestaudio/best',
        'quiet': True,
        'socket_timeout': REQUEST_TIMEOUT_SECONDS,
        'noplaylist': True,
        'extractor_retries': 1,
    }
    with YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(url, download=False)
        title = info.get("title", "Unknown")
        description = info.get("description", "No description")
        thumbnail = info.get("thumbnail", None)
        duration = info.get("duration", 0)
        uploader = info.get("uploader", "Unknown")
        view_count = info.get("view_count", 0)
        return {
            "title": title,
            "description": description,
            "thumbnail": thumbnail,
            "duration": duration,
            "uploader": uploader,
            "view_count": view_count,
            "metadata_text": f"Video Title: {title}\nDescription: {description}"
        }

def get_metrics(text):
    """Calculate metrics for the text"""
    word_count = len(text.split())
    char_count = len(text)
    reading_time = max(1, word_count // 200)  # Assume 200 words per minute
    return {
        "words": word_count,
        "chars": char_count,
        "reading_time": reading_time,
        "engagement_score": min(100, (len(text.split('#')) - 1) * 10 + (text.count('🎯') + text.count('💡') + text.count('🚀')) * 20)
    }

def format_duration(seconds):
    """Format duration in seconds to HH:MM:SS"""
    if not seconds:
        return "Unknown"
    hours = seconds // 3600
    minutes = (seconds % 3600) // 60
    secs = seconds % 60
    if hours > 0:
        return f"{hours}h {minutes}m"
    return f"{minutes}m {secs}s"

def render_copy_button(label, text, key):
        button_id = f"copy-btn-{key}"
        escaped_text = json.dumps(text)
        html = f"""
        <div style='display:flex;align-items:center;gap:8px;'>
            <button id='{button_id}' style='background:#1f2937;color:#e5e7eb;border:1px solid #374151;padding:6px 10px;border-radius:8px;cursor:pointer;'>
                {label}
            </button>
        </div>
        <script>
            const btn = document.getElementById('{button_id}');
            if (btn && !btn.dataset.bound) {{
                btn.dataset.bound = 'true';
                const originalText = btn.textContent;

                const fallbackCopy = (text) => {{
                    const textarea = document.createElement('textarea');
                    textarea.value = text;
                    textarea.setAttribute('readonly', '');
                    textarea.style.position = 'absolute';
                    textarea.style.left = '-9999px';
                    document.body.appendChild(textarea);
                    textarea.select();
                    document.execCommand('copy');
                    document.body.removeChild(textarea);
                }};

                btn.onclick = async () => {{
                    try {{
                        if (navigator.clipboard && window.isSecureContext) {{
                            await navigator.clipboard.writeText({escaped_text});
                        }} else {{
                            fallbackCopy({escaped_text});
                        }}
                        btn.textContent = 'Copied!';
                    }} catch (e) {{
                        btn.textContent = 'Copy failed';
                    }}
                    setTimeout(() => {{ btn.textContent = originalText; }}, 1500);
                }};
            }}
        </script>
        """
        components.html(html, height=42)

# Initialize session state for history and current output
if "post_history" not in st.session_state:
    st.session_state.post_history = []
if "current_summary" not in st.session_state:
    st.session_state.current_summary = None
if "current_xpost" not in st.session_state:
    st.session_state.current_xpost = None
if "current_video_data" not in st.session_state:
    st.session_state.current_video_data = None
if "user_api_key" not in st.session_state:
    st.session_state.user_api_key = ""
if "client_session_id" not in st.session_state:
    st.session_state.client_session_id = str(uuid.uuid4())

# Streamlit UI
st.set_page_config(page_title="TubeToTweet", layout="wide")
st.title("🎬 TubeToTweet : YouTube to Viral X Post Generator")
st.write("Paste a public YouTube URL to get a summary and a ready-to-post viral X tweet.")

st.text_input(
    "Enter your Gemini API key",
    type="password",
    key="user_api_key",
    placeholder="Paste your Gemini API key here",
    help="Your key is used only for this session. If left blank, the app will use the server-side key from Streamlit Secrets (if configured)."
)

entered_api_key = st.session_state.user_api_key.strip()
if entered_api_key:
    if entered_api_key.startswith("AIza") and len(entered_api_key) >= 35:
        st.success("Personal API key loaded for this session.")
    else:
        st.error("Invalid API key. Please enter a valid Gemini API key.")

url = st.text_input("Enter YouTube URL:")

if st.button("Generate", use_container_width=True):
    if not get_active_api_key():
        st.error("Please enter your Gemini API key before generating content.")
    elif not url.strip() or not validators.url(url):
        st.error("Please enter a valid YouTube URL.")
    elif not is_allowed_youtube_url(url):
        st.error("Only YouTube links are allowed (youtube.com or youtu.be).")
    else:
        allowed, retry_after = consume_rate_limit()
        if not allowed:
            st.error(f"Too many requests. Please wait {retry_after}s before trying again.")
        else:
            with st.spinner("Processing video..."):
                try:
                    # Step 1: Load content and metadata
                    video_data = load_youtube_content(url)
                    metadata_text = video_data["metadata_text"]

                    # Step 2: Generate summary
                    summary_input = metadata_text + f"\nURL: {url}"
                    summary, model_used = run_chain_with_fallback(summary_input, summary_prompt, model_candidates)

                    # Step 3: Generate viral X post from summary
                    x_post_candidates = [model_used] + [m for m in model_candidates if m != model_used]
                    x_post, _ = run_chain_with_fallback(summary, x_post_prompt, x_post_candidates)

                    summary_editable = summary
                    x_post_editable = x_post

                    # Add to history and session state
                    post_entry = {
                        "url": url,
                        "title": video_data["title"],
                        "summary": summary_editable,
                        "x_post": x_post_editable,
                        "timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                    }
                    st.session_state.post_history.insert(0, post_entry)

                    # Store in session state for persistence
                    st.session_state.current_summary = summary_editable
                    st.session_state.current_xpost = x_post_editable
                    st.session_state.current_video_data = video_data

                except Exception:
                    logger.exception("Generation failed for URL: %s", url)
                    st.error("Something went wrong while processing this request. Please try again.")

# Display persistent output if data exists in session state
if st.session_state.current_video_data and st.session_state.current_summary:
    video_data = st.session_state.current_video_data
    summary_editable = st.session_state.current_summary
    x_post_editable = st.session_state.current_xpost
    
    st.subheader("📺 Video Preview")
    col1, col2 = st.columns([1, 2])
    
    with col1:
        if video_data["thumbnail"]:
            st.image(video_data["thumbnail"], use_container_width=True)
    
    with col2:
        st.markdown(f"**{video_data['title']}**")
        st.caption(f"Channel: {video_data['uploader']}")
        col_duration, col_views = st.columns(2)
        with col_duration:
            st.metric("Duration", format_duration(video_data['duration']))
        with col_views:
            st.metric("Views", f"{video_data['view_count']:,}")

    # Create tabs for organized output
    tab1, tab2, tab3 = st.tabs(["📝 Summary", "𝕏 Post", "📊 History"])
    
    with tab1:
        st.subheader("Video Summary")

        col_metrics1, col_metrics2, col_metrics3 = st.columns(3)
        metrics = get_metrics(summary_editable)
        with col_metrics1:
            st.metric("Word Count", metrics["words"])
        with col_metrics2:
            st.metric("Reading Time", f"{metrics['reading_time']} min")
        with col_metrics3:
            st.metric("Characters", metrics["chars"])
        
        st.code(summary_editable, language=None)
        render_copy_button("Copy Summary", summary_editable, "summary")
        
        st.download_button(
            label="⬇️ Export Summary (.txt)",
            data=summary_editable,
            file_name=f"summary_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt",
            mime="text/plain",
            use_container_width=True
        )

    with tab2:
        st.subheader("Viral X Post (280 characters)")
        x_post_editable = st.text_area("Edit X post:", value=x_post_editable, height=120, key="xpost_edit_persist")
        
        # Character count progress
        char_count = len(x_post_editable)
        progress = min(char_count / 280, 1.0)
        if char_count <= 280:
            st.progress(progress, text=f"{char_count}/280 characters")
        else:
            st.error(f"⚠️ {char_count}/280 characters (Over limit by {char_count - 280})")
        
        # Metrics for X post
        xpost_metrics = get_metrics(x_post_editable)
        col_engagement1, col_engagement2 = st.columns(2)
        with col_engagement1:
            st.metric("Hashtags", len(x_post_editable.split('#')) - 1)
        with col_engagement2:
            st.metric("Engagement Score", f"{xpost_metrics['engagement_score']}/100")
        
        col_copy2, col_export2 = st.columns(2)
        with col_copy2:
            st.code(x_post_editable, language=None)
            render_copy_button("Copy X Post", x_post_editable, "xpost")
        with col_export2:
            st.download_button(
                label="⬇️ Export X Post (.txt)",
                data=x_post_editable,
                file_name=f"xpost_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt",
                mime="text/plain",
                use_container_width=True
            )

    with tab3:
        st.subheader("Recent Posts History")
        if st.session_state.post_history:
            for idx, entry in enumerate(st.session_state.post_history[:10]):
                with st.expander(f"📌 {entry['title'][:50]}... - {entry['timestamp']}"):
                    st.write(f"**URL:** {entry['url']}")
                    st.write(f"**Summary preview:** {entry['summary'][:200]}...")
                    st.subheader("X Post:")
                    st.code(entry['x_post'], language=None)
                    st.caption("📋 Click the copy icon to copy this post")
                    st.download_button(
                        label=f"⬇️ Export Post #{idx}",
                        data=entry['x_post'],
                        file_name=f"{entry['title'][:30]}_{entry['timestamp'].replace(':', '-')}.txt",
                        mime="text/plain",
                        use_container_width=True,
                        key=f"export_{idx}"
                    )
        else:
            st.info("No posts generated yet. Generate your first post above!")

# Footer
st.sidebar.title("Tips & Features")
st.sidebar.markdown("""
**Features:**

1. **Customizable Content** – Tweak the generated summary and X post according to your preferences for enhanced quality and alignment with your requirements.

2. **Seamless Clipboard Integration** – One-click copy-to-clipboard functionality for both summary and viral X posts, enabling quick sharing and publishing.

3. **Universal Video Support** – Compatible with all public YouTube videos, allowing you to generate content from any accessible video source.

4. **Engagement Analytics** – Real-time engagement score calculation based on hashtags, emojis, and content structure to optimize viral potential.

5. **Flexible Model Configuration** – Customize Gemini models directly in the code or via the `.env` file to match your computational requirements and API quota.

6. **Content History** – Session-based storage of all generated posts, enabling you to review, compare, and reuse previous content with full export capabilities.

---

**Developed by:** Abhay Aditya
""")