""", unsafe_allow_html=True)
if st.button("🚪 Logout"):
logout()
st.rerun()
st.markdown('', unsafe_allow_html=True)
st.markdown("### ℹ️ How It Works")
steps = [
("1", "ᯓ➤ Paste YouTube URL (optional for unpublished)"),
("2", "🎬 Upload your video as .mp4"),
("3", "🚀 Click Analyze"),
("4", "📍 See exactly where to place ads"),
]
for num, text in steps:
st.markdown(f"""
', unsafe_allow_html=True)
col1, col2 = st.columns([1, 1])
with col1:
yt_input = st.text_input(
"[ ▶︎ ] YouTube Video URL (optional for unpublished videos)",
placeholder="https://youtu.be/4Rq-LY16WxM",
help="Leave blank if video is not yet published on YouTube"
)
with col2:
uploaded_file = st.file_uploader(
"🎬 Upload Your Video File (.mp4)",
type=["mp4"],
help="Upload your video file — required for analysis"
)
analyze_btn = st.button("🚀 Analyze My Video & Find Best Ad Spots")
if analyze_btn:
# ── Validate: video file is always required ──
if not uploaded_file:
st.warning("⚠️ Please upload your video file (.mp4) to continue.")
else:
# ── Determine pipeline mode ──
use_offline = not yt_input.strip()
video_id = extract_video_id(yt_input.strip()) if yt_input.strip() else None
if yt_input.strip() and not video_id:
st.error("❌ Could not extract video ID. Please check your YouTube URL.")
else:
# Save uploaded file to temp path
os.makedirs("test_video", exist_ok=True)
save_path = os.path.join("test_video", uploaded_file.name)
with open(save_path, "wb") as f:
f.write(uploaded_file.getbuffer())
# Loading UI
loading_container = st.empty()
with loading_container.container():
if use_offline:
st.markdown("""
[ ▶︎ ]
Predicting Ad Spots...
Analyzing audio energy to simulate viewer retention
Processing frames and fetching YouTube retention data
""", unsafe_allow_html=True)
progress_bar = st.progress(0)
steps_done = [0]
def on_progress(msg):
steps_done[0] += 1
progress_bar.progress(min(steps_done[0] / 4, 1.0))
try:
if use_offline:
# ── Unpublished video — audio-based prediction ──
from pipeline import run_full_pipeline_offline
run_full_pipeline_offline(save_path, progress_callback=on_progress)
st.session_state["prediction_mode"] = True
else:
# ── Published video — full YouTube Analytics pipeline ──
from pipeline import run_full_pipeline
run_full_pipeline(save_path, video_id, creds, progress_callback=on_progress)
st.session_state["prediction_mode"] = False
progress_bar.progress(1.0)
loading_container.empty()
st.session_state["analysis_done"] = True
st.rerun()
except Exception as e:
loading_container.empty()
st.error(f"❌ Pipeline error: {str(e)}")
if not use_offline:
st.info("💡 Make sure your video is public or unlisted and belongs to your connected YouTube channel.")
st.markdown('', unsafe_allow_html=True)
# ═══════════════════════════════════════════════════════════════
# LOAD RESULTS
# ═══════════════════════════════════════════════════════════════
if not st.session_state.get("analysis_done", False):
data = None
else:
data = load_json("final_recommendations.json")
candidates = load_json("ranked_candidates.json") if st.session_state.get("analysis_done") else None
retention = load_json("retention_curve.json") if st.session_state.get("analysis_done") else None
if not data:
st.markdown("""
📹
Ready to analyze your video
Upload your video above — paste a YouTube URL for published videos,
or leave it blank to predict spots for an unpublished video.
""", unsafe_allow_html=True)
st.stop()
recs = data.get("recommendations", [])
duration = data.get("video_duration_formatted", "N/A")
total_placed = data.get("total_placements_recommended", 0)
cands_list = candidates.get("ranked_placements", []) if candidates else []
top_ret = max((c["retention_at_t"] for c in cands_list), default=0)
is_predict = st.session_state.get("prediction_mode", False)
# ═══════════════════════════════════════════════════════════════
# PREDICTION MODE BANNER
# ═══════════════════════════════════════════════════════════════
if is_predict:
st.markdown("""
[ ▶︎ ] Pre-Publish Prediction Mode
No YouTube data available yet — ad spots are predicted using audio energy analysis.
Accuracy improves once you publish and re-analyze with real viewer retention data.
', unsafe_allow_html=True)
if is_predict:
st.caption("Simulated retention curve based on audio energy analysis. Markers show predicted ad spots.")
else:
st.caption("How many viewers are still watching at each moment. Markers show recommended ad spots.")
if retention:
times = [p["time_seconds"] for p in retention]
values = [p["retention_percent"] for p in retention]
fig = go.Figure()
fig.add_trace(go.Scatter(
x=times, y=values, mode="lines",
name="Viewer Retention",
line=dict(color="#6366f1", width=3, shape="spline", smoothing=1.2),
fill="tozeroy",
fillgradient=dict(
type="vertical",
colorscale=[[0.0, "rgba(99,102,241,0.0)"], [1.0, "rgba(99,102,241,0.2)"]]
),
hovertemplate="%{x:.0f}s Retention: %{y:.1f}%"
))
spot_colors = ["#10b981", "#f59e0b", "#ef4444"]
for i, r in enumerate(recs):
color = spot_colors[i % len(spot_colors)]
ts = r["timestamp_seconds"]
ret_val = r.get("retention_at_t", 50)
fig.add_trace(go.Scatter(
x=[ts], y=[ret_val],
mode="markers+text",
marker=dict(symbol="diamond", size=14, color=color, line=dict(width=2, color="white")),
text=[f"Ad {r['placement_number']}"],
textposition="top center",
textfont=dict(color=color, size=12, family="Inter"),
hovertemplate=f"Ad Spot {r['placement_number']} {r['timestamp_formatted']} Retention: {ret_val:.1f}%",
showlegend=False
))
fig.add_vline(x=ts, line_dash="dot", line_color=color, line_width=1.5, opacity=0.5)
fig.update_layout(
plot_bgcolor="rgba(0,0,0,0)", paper_bgcolor="rgba(0,0,0,0)",
font=dict(color="#94a3b8", family="Inter"), height=400,
xaxis=dict(title="Time (seconds)", gridcolor="rgba(255,255,255,0.04)", zeroline=False, tickfont=dict(size=11)),
yaxis=dict(title="Viewers Still Watching (%)", gridcolor="rgba(255,255,255,0.04)", range=[0, 105], zeroline=False, tickfont=dict(size=11)),
margin=dict(l=10, r=10, t=20, b=40),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1, bgcolor="rgba(0,0,0,0)"),
hoverlabel=dict(bgcolor="#161c2d", bordercolor="#6366f1", font_size=13, font_family="Inter")
)
st.plotly_chart(fig, width='stretch')
st.markdown('', unsafe_allow_html=True)
# ═══════════════════════════════════════════════════════════════
# RECOMMENDATIONS
# ═══════════════════════════════════════════════════════════════
st.markdown(
f'
{"[ ▶︎ ] Predicted" if is_predict else "🎯"} Where to Place Your Ads
',
unsafe_allow_html=True
)
if not recs:
st.markdown("""
😕
No strong ad spots found for this video
This usually means viewer retention dropped too quickly.
Try a video where viewers watch past the halfway point.
""", unsafe_allow_html=True)
else:
for idx, r in enumerate(recs):
emoji, quality, card_class, badge_class = get_spot_quality(r["retention_at_t"])
type_label = get_type_label(r["type"])
type_tip = get_type_tip(r["type"])
ret = r["retention_at_t"]
conf = r["confidence"]
conf_color = "#10b981" if conf == "HIGH" else "#f59e0b" if conf == "MEDIUM" else "#94a3b8"
ring_color = "#10b981" if ret >= 50 else "#f59e0b" if ret >= 30 else "#ef4444"
ring_svg = build_retention_ring(ret, ring_color)
is_predict = st.session_state.get("prediction_mode", False)
predict_note = "(predicted)" if is_predict else ""
html = (
f'
'
f'
'
f'
'
f'📍 Ad Spot {r["placement_number"]}'
f' '
f'{r["timestamp_formatted"]}'
f' {predict_note}'
f'
'
f'{emoji} {quality}'
f'
'
f'
'
f'
'
f'
'
f'
VIEWERS WATCHING
'
f'
{ret:.0f}%
'
f'
'
f'
'
f'
BREAK TYPE
'
f'
{type_label}
'
f'
'
f'
'
f'
CONFIDENCE
'
f'
{conf}
'
f'
'
f'
'
f'
{ring_svg}
'
f'
'
f'
'
f'Why this spot? {type_tip}
'
f'What to do: Place your sponsorship or enable mid-roll at {r["timestamp_formatted"]}. '
f'At this moment, {ret:.0f}% of your audience is still watching.'
f'
'
f'
'
)
st.markdown(html, unsafe_allow_html=True)
st.markdown('', unsafe_allow_html=True)
# ═══════════════════════════════════════════════════════════════
# INSIGHTS FOR NEXT VIDEO
# ═══════════════════════════════════════════════════════════════
st.markdown('
📊 Insights for Your Next Video
', unsafe_allow_html=True)
if cands_list:
df = pd.DataFrame(cands_list)
best_ret = df["retention_at_t"].max()
worst_ret = df["retention_at_t"].min()
avg_ret = df["retention_at_t"].mean()
drop_time = df.loc[df["retention_at_t"].idxmin(), "timestamp_formatted"]
insight_metrics = [
("🏆", "Peak Retention", f"{best_ret:.0f}%", "#10b981", "rgba(16,185,129,0.15)", "viewers at best moment"),
("📉", "Biggest Drop-Off", drop_time, "#ef4444", "rgba(239,68,68,0.15)", "most viewers left here"),
("📊", "Avg Retention", f"{avg_ret:.0f}%", "#f59e0b", "rgba(245,158,11,0.15)", "across all break points"),
]
i1, i2, i3 = st.columns(3)
for i, (col, (icon, label, value, color, icon_bg, sub)) in enumerate(zip([i1, i2, i3], insight_metrics)):
with col:
st.markdown(f"""
', unsafe_allow_html=True)
tips = []
if best_ret >= 60:
tips.append(("✅", "Strong early retention — your intro hooks viewers well. Keep doing what you did in the first 2 minutes."))
if worst_ret < 25:
tips.append(("⚠️", f"Viewers drop off near {drop_time} — tighten that section or add a re-engagement hook."))
if avg_ret < 35:
tips.append(("📉", "Overall retention is low — try shorter videos or stronger pacing."))
if avg_ret >= 50:
tips.append(("🎉", "Great overall retention — you can confidently place ads and expect good visibility."))
if total_placed == 0:
tips.append(("🔴", "No strong ad spots this time — aim to keep 40%+ viewers watching past the 2-minute mark."))
if is_predict:
tips.append(("[ ▶︎ ]", "This was a predicted analysis. Once published, re-analyze with your YouTube URL for real retention data and more accurate spots."))
for icon, text in tips:
st.markdown(f"""
{icon}
{text}
""", unsafe_allow_html=True)
# ═══════════════════════════════════════════════════════════════
# ADVANCED TABLE
# ═══════════════════════════════════════════════════════════════
st.markdown('', unsafe_allow_html=True)
with st.expander("🔬 View All Analyzed Moments (Advanced)"):
if cands_list:
df_show = pd.DataFrame(cands_list)[["timestamp_formatted", "type", "retention_at_t", "placement_score"]]
df_show.columns = ["Timestamp", "Break Type", "Viewers Watching (%)", "ML Score"]
df_show["Break Type"] = df_show["Break Type"].map(get_type_label)
df_show["Viewers Watching (%)"] = df_show["Viewers Watching (%)"].map(lambda x: f"{x:.1f}%")
df_show["ML Score"] = df_show["ML Score"].map(lambda x: f"{x:.4f}")
st.dataframe(df_show, width='stretch', hide_index=True)
# ═══════════════════════════════════════════════════════════════
# PAST ANALYSES (CHANNEL HISTORY)
# ═══════════════════════════════════════════════════════════════
history_data = load_json("channel_history.json")
if history_data:
st.markdown('', unsafe_allow_html=True)
with st.expander("📚 Past Analyses (Channel History)"):
st.markdown("""
### Why It's Important
Right now, your app analyzes each video in isolation — it has no memory. Every analysis starts fresh with zero knowledge of your channel's patterns. History storage gives the system memory and intelligence over time.
**Without History (Now)**
* Video 1 analyzed → recommendations → forgotten ❌
* Video 2 analyzed → recommendations → forgotten ❌
* *Each video treated like the first video ever*
**With History (After)**
* Video 1 analyzed → saved to history ✅
* Video 2 analyzed → compared to Video 1 → better prediction ✅
* Video 10 analyzed → compared to 9 past videos → highly accurate ✅
* *System gets SMARTER with every video you analyze*
""", unsafe_allow_html=True)
# Reverse the list so newest is on top
history_data = history_data[::-1]
df_history = pd.DataFrame(history_data)
df_history = df_history[["timestamp", "video_path", "duration", "placements", "is_prediction"]]
df_history.columns = ["Date", "Video", "Duration", "Spots Found", "Predicted?"]
df_history["Predicted?"] = df_history["Predicted?"].apply(lambda x: "[ ▶︎ ] Yes" if x else "✅ No")
st.dataframe(df_history, width='stretch', hide_index=True)
# ═══════════════════════════════════════════════════════════════
# FOOTER
# ═══════════════════════════════════════════════════════════════
st.markdown("""
""", unsafe_allow_html=True)
' if favicon_b64 else "[ ▶︎ ]"
st.markdown(f"""