Spaces:
Sleeping
Sleeping
File size: 4,450 Bytes
3b1ecaa | 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 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 | """
predictor_audio.py — Replaces Component 2 for unpublished videos.
Uses audio energy + pacing to simulate a retention curve.
"""
import numpy as np
import pandas as pd
import json
def generate_simulated_retention(video_path, total_duration):
"""
Analyzes audio energy to simulate a retention curve.
Returns a DataFrame matching get_retention_curve() output format.
"""
try:
import librosa
audio, sr = librosa.load(video_path, sr=22050, mono=True)
# Calculate RMS energy per second
hop_length = sr # 1 second per frame
rms = librosa.feature.rms(y=audio, hop_length=hop_length)[0]
# Normalize energy 0 to 1
rms_norm = (rms - rms.min()) / (rms.max() - rms.min() + 1e-6)
seconds = np.arange(len(rms_norm))
except Exception as e:
print(f"[Predictor Audio] librosa failed ({e}), using duration-based fallback")
seconds = np.arange(int(total_duration))
rms_norm = np.ones(len(seconds))
# Build simulated retention curve
# Base: natural decay (viewers always drop over time)
n = len(seconds)
base_decay = np.linspace(1.0, 0.30, n)
# Modify decay based on audio energy
# Low energy = viewers more likely to leave
# High energy = viewers stay
energy_factor = 0.15 * (rms_norm - 0.5) # range: -0.075 to +0.075
# Add intro spike drop (first 30s many viewers leave)
intro_drop = np.ones(n)
intro_end = min(30, n)
intro_drop[:intro_end] = np.linspace(1.0, 0.85, intro_end)
# Combine all factors
retention = base_decay * intro_drop + energy_factor
retention = np.clip(retention, 0.05, 1.0)
# Smooth the curve
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)
df = pd.DataFrame({
"second": seconds,
"retention_pct": (retention * 100).round(2)
})
print(f"[Predictor Audio] Simulated retention curve — {len(df)} seconds")
return df
def run_component2_offline(video_path):
"""
Drop-in replacement for run_component2() in pipeline.py.
Uses audio analysis instead of YouTube Analytics API.
"""
with open("candidates.json") as f:
data = json.load(f)
candidates = data["candidates"]
total_duration = data["total_duration"]
# Generate simulated retention curve from audio
curve_df = generate_simulated_retention(video_path, total_duration)
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"[Predictor Audio] features.csv saved — {len(df)} candidates")
return df |