#!/usr/bin/env python3
import subprocess
import time
import signal
import sys
import os
from datetime import datetime
import sentry_sdk
# Configuration
CHANNEL = os.getenv("CHANNEL_NAME")
TIMEOUT = os.getenv("TIMEOUT", "60")
COOKIES_FILE = "/cookies/cookies.txt"
DOWNLOAD_DIR = "/downloads"
POT_PROVIDER = os.getenv("POT_PROVIDER_URL", "http://pot-provider:4416")
SENTRY_DSN = os.getenv("SENTRY_DSN")
WAIT_INTERVAL = 10
# State
running = True
proc = None
alert_history = {}
ALERT_COOLDOWN = 60
# Initialize Sentry
if SENTRY_DSN:
print(f"[Wrapper] Sentry monitoring enabled.")
sentry_sdk.init(dsn=SENTRY_DSN, traces_sample_rate=1.0)
def graceful_exit(signum, frame):
global running, proc
print(f"\n[Wrapper] Signal {signum} received. Stopping...")
running = False
if proc:
proc.terminate()
def check_log_line(line):
"""Analyze a log line to see if we should alert Sentry."""
global alert_history
if "WARNING" not in line and "ERROR" not in line:
return
ignore_phrases = [
"channel is not currently live",
"Press Ctrl+C to try now",
]
for phrase in ignore_phrases:
if phrase in line:
return
msg_key = line.strip()
now = time.time()
last_time = alert_history.get(msg_key, 0)
if now - last_time > ALERT_COOLDOWN:
if SENTRY_DSN:
with sentry_sdk.push_scope() as scope:
scope.set_tag("channel", CHANNEL)
sentry_sdk.capture_message(f"yt-dlp Warning: {msg_key}", level="warning")
alert_history[msg_key] = now
def main():
global proc, running
signal.signal(signal.SIGINT, graceful_exit)
signal.signal(signal.SIGTERM, graceful_exit)
channel_dir = os.path.join(DOWNLOAD_DIR, CHANNEL)
os.makedirs(channel_dir, exist_ok=True)
while running:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_template = f"{channel_dir}/{CHANNEL}-{timestamp}-[%(title)s][%(id)s].%(ext)s"
cmd = [
"yt-dlp",
# Standard monitoring URL
f"https://www.youtube.com/@{CHANNEL}/live",
"--live-from-start",
"--wait-for-video", TIMEOUT,
"--cookies", COOKIES_FILE,
"--remote-components", "ejs:github",
"--js-runtimes", "node",
"--extractor-args", f"youtubepot-bgutilhttp:base_url={POT_PROVIDER}",
"-o", output_template,
]
try:
print(f"[Wrapper] Monitoring @{CHANNEL}...")
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1
)
for line in proc.stdout:
print(line, end='')
sys.stdout.flush()
check_log_line(line)
if not running:
break
proc.wait()
except Exception as e:
print(f"[Wrapper] Critical Python Error: {e}")
if SENTRY_DSN:
sentry_sdk.capture_exception(e)
if running:
time.sleep(WAIT_INTERVAL)
if __name__ == "__main__":
main()