Spaces:
Sleeping
Sleeping
File size: 3,948 Bytes
4dc7a21 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | """
component2_feature_extractor.py — Updated with real YouTube Analytics support
Simulated : python component2_feature_extractor.py
Real data : python component2_feature_extractor.py --video-id YOUR_VIDEO_ID
"""
import json
import sys
import pandas as pd
import numpy as np
def simulate_retention_curve(total_duration, seed=42):
np.random.seed(seed)
t = np.linspace(0, total_duration, int(total_duration))
base = 100 * np.exp(-0.003 * t)
noise = np.random.normal(0, 2, len(t))
spikes = np.zeros(len(t))
for _ in range(5):
spike_t = np.random.randint(0, len(t))
spikes[max(0, spike_t-10):spike_t+10] += np.random.uniform(3, 8)
return pd.DataFrame({"second": t, "retention_pct": np.clip(base + noise + spikes, 0, 100)})
def get_real_retention_curve(video_id):
try:
from youtube_analytics import get_retention_curve
df = get_retention_curve(video_id)
return df
except Exception as e:
print(f"[Component 2] Warning: {e}")
print("[Component 2] Falling back to simulated retention curve")
return None
def get_retention_at(curve_df, t, window=10):
mask = (curve_df["second"] >= t - window) & (curve_df["second"] <= t + window)
subset = curve_df[mask]
if subset.empty:
return 0.0, 0.0, 0.0
at_t_idx = (subset["second"] - t).abs().idxmin()
retention_at_t = curve_df.loc[at_t_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 = (before["retention_pct"].mean() - after["retention_pct"].mean()) if len(before) and len(after) else 0
recovery = after["retention_pct"].mean() - further["retention_pct"].mean() if len(after) and len(further) else 0
return round(float(retention_at_t), 3), round(float(drop_rate), 3), round(float(recovery), 3)
def extract_features(candidates_path="candidates.json", video_id=None):
with open(candidates_path) as f:
data = json.load(f)
candidates = data["candidates"]
total_duration = data["total_duration"]
if video_id:
print(f"[Component 2] Fetching REAL retention for video: {video_id}")
curve_df = get_real_retention_curve(video_id)
if curve_df is None:
curve_df = simulate_retention_curve(total_duration)
else:
print("[Component 2] Using SIMULATED retention curve")
curve_df = simulate_retention_curve(total_duration)
rows = []
for i, c in enumerate(candidates):
t = c["timestamp"]
ret_at_t, drop_rate, recovery = get_retention_at(curve_df, t)
time_since_last = t - candidates[i-1]["timestamp"] if i > 0 else t
rows.append({
"timestamp": t,
"type": c["type"],
"content_score": c["score"],
"retention_at_t": ret_at_t,
"retention_drop_rate": drop_rate,
"retention_recovery": recovery,
"relative_position": round(t / total_duration, 4),
"time_since_last_candidate": round(time_since_last, 2),
"label": None
})
df = pd.DataFrame(rows)
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("[Component 2] features.csv saved ✅")
print(df[["timestamp", "type", "retention_at_t", "retention_drop_rate", "label"]].to_string(index=False))
return df
if __name__ == "__main__":
video_id = None
if "--video-id" in sys.argv:
idx = sys.argv.index("--video-id")
video_id = sys.argv[idx + 1]
extract_features(video_id=video_id) |