""" pipeline.py — Runs full pipeline as functions (no subprocess needed on Render) """ import json import os import tempfile import datetime import numpy as np import pandas as pd def save_to_history(entry): """Save an analysis result to channel history.""" history_file = "channel_history.json" history = [] if os.path.exists(history_file): with open(history_file, "r") as f: try: history = json.load(f) except: pass history.append(entry) with open(history_file, "w") as f: json.dump(history, f, indent=2) # ── Component 1: Candidate Generator ────────────────────────── def run_component1(video_path): """Extract scene changes, silences, transcript boundaries from video.""" import cv2 from pydub import AudioSegment from pydub.silence import detect_silence candidates = [] total_duration = 0 cap = cv2.VideoCapture(video_path) fps = cap.get(cv2.CAP_PROP_FPS) frame_count = cap.get(cv2.CAP_PROP_FRAME_COUNT) total_duration = frame_count / fps if fps > 0 else 0 # Scene change detection — sample every 5 frames to save memory prev_frame = None frame_idx = 0 scene_threshold = 30.0 sample_every = 5 while True: cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx) ret, frame = cap.read() if not ret: break small = cv2.resize(frame, (160, 90)) gray = cv2.cvtColor(small, cv2.COLOR_BGR2GRAY) if prev_frame is not None: diff = cv2.absdiff(gray, prev_frame) score = diff.mean() if score > scene_threshold: timestamp = frame_idx / fps if timestamp > 30: candidates.append({ "timestamp": round(timestamp, 2), "type": "scene_change", "score": round(float(score), 3) }) prev_frame = gray frame_idx += sample_every cap.release() # Silence detection try: audio = AudioSegment.from_file(video_path) silences = detect_silence(audio, min_silence_len=800, silence_thresh=-40) for start_ms, end_ms in silences: mid = (start_ms + end_ms) / 2 / 1000 if mid > 30 and mid < total_duration - 30: candidates.append({ "timestamp": round(mid, 2), "type": "silence", "score": 0.6 }) except Exception as e: print(f"[Component 1] Audio extraction warning: {e}") # Transcript boundary detection try: import whisper print("[Component 1] Running Whisper transcription (base model)...") model = whisper.load_model("base") result = model.transcribe(video_path) segments = result.get("segments", []) for segment in segments: text = segment.get("text", "").strip() # If segment ends with terminal punctuation, treat as a break if text and text[-1] in ".!?": end_time = segment["end"] if end_time > 30 and end_time < total_duration - 30: candidates.append({ "timestamp": round(end_time, 2), "type": "transcript_boundary", "score": 0.7 }) except Exception as e: print(f"[Component 1] Whisper transcription warning: {e}") # Deduplicate — remove candidates within 5s of each other candidates = sorted(candidates, key=lambda x: x["timestamp"]) deduped = [] for c in candidates: if not deduped or abs(c["timestamp"] - deduped[-1]["timestamp"]) > 5: deduped.append(c) data = {"candidates": deduped, "total_duration": round(total_duration, 2)} with open("candidates.json", "w") as f: json.dump(data, f, indent=2) print(f"[Component 1] {len(deduped)} candidates found, duration: {total_duration:.1f}s") return data # ── Component 2: Feature Extractor (Published Videos) ───────── def run_component2(video_id, creds): """Fetch retention data from YouTube Analytics and extract features.""" from youtube_analytics import get_retention_curve with open("candidates.json") as f: data = json.load(f) candidates = data["candidates"] total_duration = data["total_duration"] curve_df = get_retention_curve(video_id, creds=creds) rows = [] for i, c in enumerate(candidates): t = c["timestamp"] mask = (curve_df["second"] >= t - 10) & (curve_df["second"] <= t + 10) subset = curve_df[mask] if subset.empty: ret_at_t, drop_rate, recovery = 0.0, 0.0, 0.0 else: idx = (subset["second"] - t).abs().idxmin() ret_at_t = float(curve_df.loc[idx, "retention_pct"]) before = curve_df[curve_df["second"] < t].tail(30) after = curve_df[curve_df["second"] > t].head(30) further = curve_df[curve_df["second"] > t + 30].head(30) drop_rate = float(before["retention_pct"].mean() - after["retention_pct"].mean()) if len(before) and len(after) else 0 recovery = float(after["retention_pct"].mean() - further["retention_pct"].mean()) if len(after) and len(further) else 0 time_since_last = t - candidates[i-1]["timestamp"] if i > 0 else t rows.append({ "timestamp": round(t, 2), "type": c["type"], "content_score": c["score"], "retention_at_t": round(ret_at_t, 3), "retention_drop_rate": round(drop_rate, 3), "retention_recovery": round(recovery, 3), "relative_position": round(t / total_duration, 4), "time_since_last_candidate": round(time_since_last, 2), "label": 0 }) df = pd.DataFrame(rows) if len(df) > 1: df["label"] = ( (df["retention_at_t"] > df["retention_at_t"].median()) & (df["retention_drop_rate"] < df["retention_drop_rate"].median()) ).astype(int) df.to_csv("features.csv", index=False) print(f"[Component 2] features.csv saved — {len(df)} candidates") return df # ── Component 2 Offline: Audio-Based Retention Simulator ────── def run_component2_offline(video_path): """ Replaces Component 2 for unpublished videos. Simulates retention curve using audio energy analysis. No YouTube API needed. """ with open("candidates.json") as f: data = json.load(f) candidates = data["candidates"] total_duration = data["total_duration"] # ── Generate simulated retention curve ── try: import librosa audio, sr = librosa.load(video_path, sr=22050, mono=True) hop_length = sr # 1 second per frame rms = librosa.feature.rms(y=audio, hop_length=hop_length)[0] rms_norm = (rms - rms.min()) / (rms.max() - rms.min() + 1e-6) seconds = np.arange(len(rms_norm)) print(f"[Component 2 Offline] Audio loaded — {len(seconds)} seconds analyzed") except Exception as e: print(f"[Component 2 Offline] librosa fallback ({e})") seconds = np.arange(int(total_duration)) rms_norm = np.ones(len(seconds)) n = len(seconds) # Base: natural viewer decay over time base_decay = np.linspace(1.0, 0.30, n) # Audio energy modifier — high energy = viewers stay energy_factor = 0.15 * (rms_norm[:n] - 0.5) # Intro drop — first 30s many viewers leave quickly intro_drop = np.ones(n) intro_end = min(30, n) intro_drop[:intro_end] = np.linspace(1.0, 0.85, intro_end) # Combine & smooth retention = base_decay * intro_drop + energy_factor retention = np.clip(retention, 0.05, 1.0) window = min(30, n // 4) if window > 1: retention = np.convolve(retention, np.ones(window)/window, mode='same') retention = np.clip(retention, 0.05, 1.0) curve_df = pd.DataFrame({ "second": seconds, "retention_pct": (retention * 100).round(2) }) # ── Extract features same as Component 2 ── rows = [] for i, c in enumerate(candidates): t = c["timestamp"] mask = (curve_df["second"] >= t - 10) & (curve_df["second"] <= t + 10) subset = curve_df[mask] if subset.empty: ret_at_t, drop_rate, recovery = 50.0, 0.0, 0.0 else: idx = (subset["second"] - t).abs().idxmin() ret_at_t = float(curve_df.loc[idx, "retention_pct"]) before = curve_df[curve_df["second"] < t].tail(30) after = curve_df[curve_df["second"] > t].head(30) further = curve_df[curve_df["second"] > t + 30].head(30) drop_rate = float(before["retention_pct"].mean() - after["retention_pct"].mean()) if len(before) and len(after) else 0 recovery = float(after["retention_pct"].mean() - further["retention_pct"].mean()) if len(after) and len(further) else 0 time_since_last = t - candidates[i-1]["timestamp"] if i > 0 else t rows.append({ "timestamp": round(t, 2), "type": c["type"], "content_score": c["score"], "retention_at_t": round(ret_at_t, 3), "retention_drop_rate": round(drop_rate, 3), "retention_recovery": round(recovery, 3), "relative_position": round(t / total_duration, 4), "time_since_last_candidate": round(time_since_last, 2), "label": 0 }) df = pd.DataFrame(rows) if len(df) > 1: df["label"] = ( (df["retention_at_t"] > df["retention_at_t"].median()) & (df["retention_drop_rate"] < df["retention_drop_rate"].median()) ).astype(int) df.to_csv("features.csv", index=False) print(f"[Component 2 Offline] features.csv saved — {len(df)} candidates") return df # ── Component 3: ML Ranker ───────────────────────────────────── def run_component3(): """Train LightGBM ranker and score all candidates.""" import lightgbm as lgb from sklearn.preprocessing import LabelEncoder df = pd.read_csv("features.csv") feature_cols = [ "retention_at_t", "retention_drop_rate", "retention_recovery", "relative_position", "time_since_last_candidate", "content_score" ] le = LabelEncoder() df["type_enc"] = le.fit_transform(df["type"]) feature_cols.append("type_enc") X = df[feature_cols].fillna(0) y = df["label"].fillna(0).astype(int) n_pos = y.sum() print(f"[Component 3] Training LightGBM — {len(df)} samples, {n_pos} positive") model = lgb.LGBMClassifier( n_estimators=100, learning_rate=0.05, num_leaves=15, random_state=42, verbose=-1 ) model.fit(X, y) scores = model.predict_proba(X)[:, 1] df["placement_score"] = scores def fmt(t): return f"{int(t//60)}m {int(t%60):02d}s" placements = [] for _, row in df.iterrows(): placements.append({ "timestamp": row["timestamp"], "timestamp_formatted": fmt(row["timestamp"]), "type": row["type"], "placement_score": float(row["placement_score"]), "retention_at_t": float(row["retention_at_t"]), "label": int(row["label"]) }) placements = sorted(placements, key=lambda x: x["placement_score"], reverse=True) for i, p in enumerate(placements): p["rank"] = i + 1 result = {"ranked_placements": placements} with open("ranked_candidates.json", "w") as f: json.dump(result, f, indent=2) print(f"[Component 3] Ranked {len(placements)} candidates") return placements # ── Component 4: Recommender ─────────────────────────────────── def run_component4(): """Apply business rules and generate final recommendations.""" CONFIG = { "min_placement_score": 0.0, "min_gap_seconds": 120, "skip_intro_pct": 0.20, "skip_outro_pct": 0.10, "max_placements": 3, "short_video_threshold": 480, } with open("ranked_candidates.json") as f: data = json.load(f) placements = data["ranked_placements"] total_duration = max(p["timestamp"] for p in placements) / 0.80 filtered = [p for p in placements if p["placement_score"] >= CONFIG["min_placement_score"]] intro_cut = total_duration * CONFIG["skip_intro_pct"] filtered = [p for p in filtered if p["timestamp"] >= intro_cut] outro_cut = total_duration * (1 - CONFIG["skip_outro_pct"]) filtered = [p for p in filtered if p["timestamp"] <= outro_cut] max_allow = 1 if total_duration <= CONFIG["short_video_threshold"] else CONFIG["max_placements"] selected = [] for p in sorted(filtered, key=lambda x: x["placement_score"], reverse=True): if len(selected) >= max_allow: break too_close = any(abs(p["timestamp"] - s["timestamp"]) < CONFIG["min_gap_seconds"] for s in selected) if not too_close: selected.append(p) def fmt(t): return f"{int(t//60)}m {int(t%60):02d}s" output = { "video_duration_seconds": round(total_duration, 1), "video_duration_formatted": fmt(total_duration), "total_placements_recommended": len(selected), "config_used": CONFIG, "recommendations": [] } for i, p in enumerate(sorted(selected, key=lambda x: x["timestamp"])): ret = p["retention_at_t"] output["recommendations"].append({ "placement_number": i + 1, "timestamp_seconds": p["timestamp"], "timestamp_formatted": p["timestamp_formatted"], "type": p["type"], "placement_score": p["placement_score"], "retention_at_t": ret, "confidence": ( "HIGH" if p["placement_score"] >= 0.75 else "MEDIUM" if p["placement_score"] >= 0.50 else "LOW" ), "creator_note": ( f"Place sponsored segment at {p['timestamp_formatted']} — " f"natural {p['type'].replace('_', ' ')} detected, " f"{ret:.1f}% viewers still watching." ) }) with open("final_recommendations.json", "w") as f: json.dump(output, f, indent=2) print(f"[Component 4] {len(selected)} final placements recommended") return output # ── Full Pipeline (Published Videos) ────────────────────────── def run_full_pipeline(video_path, video_id, creds, progress_callback=None): """Run all 4 components for a published YouTube video.""" def update(msg): print(msg) if progress_callback: progress_callback(msg) update("🔍 Step 1/4 — Detecting natural break points in your video...") run_component1(video_path) update("📊 Step 2/4 — Fetching viewer retention data from YouTube...") run_component2(video_id, creds) update("🤖 Step 3/4 — Running ML ranking engine...") run_component3() update("🎯 Step 4/4 — Generating final recommendations...") result = run_component4() # Save to history entry = { "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), "video_id": video_id, "video_path": os.path.basename(video_path), "duration": result.get("video_duration_formatted", "0m 0s"), "placements": result.get("total_placements_recommended", 0), "is_prediction": False } save_to_history(entry) update("✅ Analysis complete!") return result # ── Offline Pipeline (Unpublished Videos — No YouTube API) ──── def run_full_pipeline_offline(video_path, progress_callback=None): """ Run pipeline WITHOUT YouTube Analytics API. Uses audio energy analysis to simulate retention curve. Perfect for pre-publish ad spot prediction. """ def update(msg): print(msg) if progress_callback: progress_callback(msg) update("🔍 Step 1/4 — Detecting natural break points in your video...") run_component1(video_path) update("🎵 Step 2/4 — Analyzing audio energy to predict viewer retention...") run_component2_offline(video_path) update("🤖 Step 3/4 — Running ML ranking engine...") run_component3() update("🎯 Step 4/4 — Generating final predictions...") result = run_component4() # Save to history entry = { "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), "video_id": None, "video_path": os.path.basename(video_path), "duration": result.get("video_duration_formatted", "0m 0s"), "placements": result.get("total_placements_recommended", 0), "is_prediction": True } save_to_history(entry) update("✅ Prediction complete!") return result