#!/usr/bin/env python3
import os
import time
import json
import math
# ── CONFIG ─────────────────────────────────────────────────────────────
LOG_FILE = 'ga_viewers.log'
OUTPUT_HTML = 'live_viewers.html'
POLL_INTERVAL = 10 # seconds between updates
# Cooldown strategy parameters (match your PHP settings)
COOLDOWN_DECAY_MULTIPLIER = 0.7 # when > COOLDOWN_MIN_VIEWERS
COOLDOWN_SOFT_DECAY = 0.85 # when <= COOLDOWN_MIN_VIEWERS
COOLDOWN_MIN_VIEWERS = 50
# ── HELPERS ────────────────────────────────────────────────────────────
def load_latest_snapshot():
"""Read the last JSON blob from the log."""
try:
with open(LOG_FILE, 'r') as f:
last = None
for last in f:
pass
return json.loads(last.strip().split('] ',1)[1])
except Exception as e:
print(f"⚠️ Error loading snapshot: {e}", flush=True)
return {}
def compute_projection(current, multiplier):
"""
Returns (seconds_until_<=1, projected_count) given
current viewer count and decay multiplier per step.
"""
if multiplier is None or multiplier >= 1 or current < 1:
return None, None
# steps needed: multiplier^n * current <= 1 → n ≥ log(1/current)/log(multiplier)
n = math.log(1.0/current, multiplier)
n = math.ceil(n)
secs = n * POLL_INTERVAL
projected = current * (multiplier ** n)
return secs, projected
def fmt_time(seconds):
if seconds is None:
return '-'
m, s = divmod(int(seconds), 60)
return f"{m}m{s:02d}s"
# ── MAIN LOOP ───────────────────────────────────────────────────────────
if __name__ == '__main__':
while True:
now_str = time.strftime('%Y-%m-%d %H:%M:%S')
snap = load_latest_snapshot()
print(f"[Snapshot @ {now_str}]", flush=True)
rows = []
for sid, info in snap.items():
cur = info.get('viewerCount', 0)
recent = info.get('recentTotalViewers', 0)
strat = info.get('strategy', '')
# decide which decay multiplier applies
if strat.startswith('cooldown'):
mult = COOLDOWN_SOFT_DECAY if cur <= COOLDOWN_MIN_VIEWERS else COOLDOWN_DECAY_MULTIPLIER
else:
mult = None
# full cooldown projection (down to ~1 viewer)
secs_full, proj_full = compute_projection(cur, mult)
# projection down to COOLDOWN_MIN_VIEWERS threshold
if mult is not None and cur > COOLDOWN_MIN_VIEWERS:
# solve for n: mult^n * cur <= COOLDOWN_MIN_VIEWERS
n_min = math.log(COOLDOWN_MIN_VIEWERS / cur, mult)
n_min = math.ceil(n_min)
secs_to_min = n_min * POLL_INTERVAL
else:
secs_to_min = None
rows.append({
'id': sid,
'viewerCount': cur,
'recentTotal': recent,
'strategy': strat,
'cooldownIn': fmt_time(secs_full), # time → ~1 viewer
'projected': int(proj_full) if proj_full is not None else '-',
'timeToMin': fmt_time(secs_to_min), # time → COOLDOWN_MIN_VIEWERS
'minThreshold': COOLDOWN_MIN_VIEWERS
})
if not rows:
print(" • No data to display", flush=True)
for r in rows:
proj_str = str(r['projected'])
print(
f" • {r['id']:20} "
f"current={int(r['viewerCount']):4d} "
f"recent={int(r['recentTotal']):4d} "
f"strat={r['strategy']:15} "
f"cd1_in={r['cooldownIn']:>8} "
f"proj1={proj_str:>4} "
f"cd50_in={r['timeToMin']:>8}",
flush=True
)
# build HTML
html = [
"Live Viewers",
f"", # ← auto-reload every POLL_INTERVAL seconds
"",
"",
f"Last update: {now_str}
",
""
"| Streamer | Current | Recent Total | "
"Strategy | "
"Cooldown↓1 In | Proj↓1 | "
"Cooldown↓50 In | MinThresh | "
"
"
]
for r in rows:
html.append(
""
f"| {r['id']} | "
f"{int(r['viewerCount'])} | "
f"{int(r['recentTotal'])} | "
f"{r['strategy']} | "
f"{r['cooldownIn']} | "
f"{r['projected']} | "
f"{r['timeToMin']} | "
f"{r['minThreshold']} | "
"
"
)
html.append("
")
try:
with open(OUTPUT_HTML, 'w') as f:
f.write("\n".join(html))
print(f" → Wrote {len(rows)} rows to {OUTPUT_HTML}", flush=True)
except Exception as e:
print(f"❌ Error writing HTML: {e}", flush=True)
time.sleep(POLL_INTERVAL)