Upload 15 files

Dockerfile

@@ -23,4 +23,4 @@ RUN pip install --no-cache-dir -r requirements.txt
 EXPOSE 7860
 
 # Command to run on start
-CMD ["python", "main.py"]
+CMD ["python", "app.py"]

app.py

@@ -0,0 +1,492 @@
+import gradio as gr
+import sqlite3
+import requests
+import os
+
+def fetch_data(db_path, limit_links=None):
+    os.makedirs(os.path.dirname(db_path), exist_ok=True)
+    if not os.path.exists(db_path):
+        # We can either create it or return empty. I will let sqlite create it if it doesn't exist.
+        pass
+    conn = sqlite3.connect(db_path)
+    cursor = conn.cursor()
+
+    cursor.execute('''CREATE TABLE IF NOT EXISTS links (
+                        id INTEGER PRIMARY KEY,
+                        url TEXT,
+                        status TEXT DEFAULT 'pending'
+                    )''')
+    try:
+        cursor.execute("SELECT api_key FROM accounts")
+        api_keys = [row[0] for row in cursor.fetchall() if row[0]]
+    except Exception as e:
+        api_keys = []
+        
+    try:
+        query = "SELECT id, url FROM links WHERE status='pending'"
+        if limit_links:
+            query += f" LIMIT {limit_links}"
+            
+        cursor.execute(query)
+        link_records = cursor.fetchall()
+        links = [row[1] for row in link_records if row[1]]
+        
+        if link_records:
+            ids = [str(row[0]) for row in link_records]
+            cursor.execute(f"UPDATE links SET status='processing' WHERE id IN ({','.join(ids)})")
+            conn.commit()
+    except Exception as e:
+        links = []
+        
+    conn.close()
+    return api_keys, links
+
+def trigger_workers(db_path, worker_urls_str, links_per_worker):
+    worker_urls = [u.strip() for u in worker_urls_str.split(',') if u.strip()]
+    if not worker_urls:
+        return "Error: No worker URLs provided."
+        
+    api_keys, links = fetch_data(db_path, limit_links=int(links_per_worker) * len(worker_urls))
+    if not api_keys:
+        return "Error: No API keys found in DB."
+    if not links:
+        return "Error: No pending links found in DB."
+        
+    link_chunks = [links[i:i + int(links_per_worker)] for i in range(0, len(links), int(links_per_worker))]
+    
+    logs = []
+    logs.append(f"Loaded {len(api_keys)} API keys and {len(links)} links.")
+    logs.append(f"Divided into {len(link_chunks)} worker chunks.")
+    
+    for idx, chunk in enumerate(link_chunks):
+        if idx >= len(worker_urls):
+            logs.append("More chunks than workers. Remaining chunks won't be sent.")
+            break
+            
+        worker_url = worker_urls[idx]
+        api_key = api_keys[idx % len(api_keys)]
+        
+        payload = {
+            "links": chunk,
+            "api_key": api_key
+        }
+        
+        endpoint = f"{worker_url.rstrip('/')}/process"
+        try:
+            resp = requests.post(endpoint, json=payload, timeout=10)
+            logs.append(f"Sent {len(chunk)} links to {worker_url}: {resp.status_code} - {resp.text}")
+        except Exception as e:
+            logs.append(f"Failed to send to {worker_url}: {str(e)}")
+            
+    return "\n".join(logs)
+
+import json
+import uuid
+import shutil
+import subprocess
+from pathlib import Path
+
+def get_kaggle_username():
+    kaggle_json = Path.home() / ".kaggle" / "kaggle.json"
+    if not kaggle_json.exists():
+        return "your-username"
+    with open(kaggle_json, "r") as f:
+        creds = json.load(f)
+        return creds.get("username", "your-username")
+
+def trigger_kaggle_test():
+    username = get_kaggle_username()
+    worker_title = f"yt-test-worker-{uuid.uuid4().hex[:6]}"
+    worker_dir = Path("kaggle_test_worker")
+    
+    if worker_dir.exists():
+        shutil.rmtree(worker_dir)
+    worker_dir.mkdir(parents=True)
+    
+    notebook_content = {
+        "cells": [
+            {
+                "cell_type": "code",
+                "execution_count": None,
+                "metadata": {},
+                "outputs": [],
+                "source": [
+                    "print('Hello from Kaggle Worker!')\n",
+                    "print('Test successful.')\n"
+                ]
+            }
+        ],
+        "metadata": {
+            "kernelspec": {"display_name": "Python 3", "language": "python", "name": "python3"},
+            "language_info": {"name": "python"}
+        },
+        "nbformat": 4,
+        "nbformat_minor": 4
+    }
+    
+    with open(worker_dir / "notebook.ipynb", "w") as f:
+        json.dump(notebook_content, f, indent=2)
+        
+    metadata = {
+        "id": f"{username}/{worker_title}",
+        "title": worker_title,
+        "code_file": "notebook.ipynb",
+        "language": "python",
+        "kernel_type": "notebook",
+        "is_private": True,
+        "enable_gpu": False,
+        "enable_internet": True,
+        "dataset_sources": [],
+        "competition_sources": [],
+        "kernel_sources": []
+    }
+    
+    with open(worker_dir / "kernel-metadata.json", "w") as f:
+        json.dump(metadata, f, indent=2)
+        
+    logs = [f"Prepared Kaggle Test Worker: {worker_title}"]
+    
+    # Use kaggle from PATH (assuming it is installed via requirements.txt)
+    command = ["kaggle", "kernels", "push", "-p", str(worker_dir)]
+    logs.append(f"Pushing {worker_title} to Kaggle...")
+    
+    try:
+        res = subprocess.run(command, capture_output=True, text=True)
+        if res.returncode == 0:
+            logs.append("Push successful.")
+            logs.append(res.stdout)
+        else:
+            logs.append("Push failed.")
+            logs.append(res.stderr)
+    except Exception as e:
+        logs.append(f"Error running kaggle command: {e}")
+        logs.append("Make sure Kaggle API credentials are provided in ~/.kaggle/kaggle.json")
+        
+    return "\n".join(logs)
+
+def create_worker_notebook(worker_dir, config_name):
+    notebook_content = {
+        "cells": [
+            {
+                "cell_type": "code",
+                "execution_count": None,
+                "metadata": {},
+                "outputs": [],
+                "source": [
+                    "!pip install openai huggingface_hub yt-dlp opencv-python-headless\n"
+                ]
+            },
+            {
+                "cell_type": "code",
+                "execution_count": None,
+                "metadata": {},
+                "outputs": [],
+                "source": [
+                    "import json\n",
+                    "import os\n",
+                    "import subprocess\n",
+                    "import sys\n",
+                    "\n",
+                    f"with open('{config_name}') as f:\n",
+                    "    config = json.load(f)\n",
+                    "\n",
+                    "links = config['links']\n",
+                    "api_key = config['api_key']\n",
+                    "os.environ['FEATHERLESS_API_KEY'] = api_key\n",
+                    "\n",
+                    "print(f'Starting worker with {len(links)} links using API Key: {api_key[:5]}...')\n",
+                    "\n",
+                    "for link in links:\n",
+                    "    print(f'\\n[+] Processing: {link}')\n",
+                    "    # 1. Download Video\n",
+                    "    dl_cmd = [sys.executable, 'worker/youtube_worker/downloader.py', link, '--resolution', 'HD', '-o', 'test_output']\n",
+                    "    res = subprocess.run(dl_cmd, capture_output=True, text=True)\n",
+                    "    if res.returncode != 0:\n",
+                    "        print(f'Download failed:\\n{res.stderr}')\n",
+                    "        continue\n",
+                    "    \n",
+                    "    # Find the downloaded video path from stdout (last line)\n",
+                    "    lines = res.stdout.strip().split('\\n')\n",
+                    "    video_path = lines[-1]\n",
+                    "    if not os.path.exists(video_path):\n",
+                    "        print(f'Could not find downloaded video at: {video_path}')\n",
+                    "        continue\n",
+                    "        \n",
+                    "    print(f'[+] Video downloaded to: {video_path}')\n",
+                    "    \n",
+                    "    # 2. Curate Model (Video Worker)\n",
+                    "    prompt = 'Analyze this video comprehensively. Focus on visuals, objects, and text.'\n",
+                    "    vw_cmd = [sys.executable, 'worker/models_worker/video_worker.py', '--video-url', video_path, '--prompt', prompt, '--api-key', api_key]\n",
+                    "    print('[+] Running model curation...')\n",
+                    "    subprocess.run(vw_cmd)\n",
+                    "    \n",
+                    "    # 3. Optional: Upload to HF\n",
+                    "    # hf_cmd = [sys.executable, 'worker/hf_uploader.py', 'test_output']\n",
+                    "    # subprocess.run(hf_cmd)\n",
+                    "\n",
+                    "print('\\nAll tasks completed!')\n"
+                ]
+            }
+        ],
+        "metadata": {
+            "kernelspec": {
+                "display_name": "Python 3",
+                "language": "python",
+                "name": "python3"
+            },
+            "language_info": {
+                "name": "python"
+            }
+        },
+        "nbformat": 4,
+        "nbformat_minor": 4
+    }
+    
+    with open(worker_dir / "notebook.ipynb", "w") as f:
+        json.dump(notebook_content, f, indent=2)
+
+def trigger_kaggle_workers(db_path, links_per_worker, max_workers):
+    api_keys, links = fetch_data(db_path, limit_links=int(links_per_worker) * int(max_workers))
+    if not api_keys:
+        return "Error: No API keys found in DB."
+    if not links:
+        return "Error: No pending links found in DB."
+        
+    username = get_kaggle_username()
+    workers_dir = Path("kaggle_workers_out")
+    if workers_dir.exists():
+        shutil.rmtree(workers_dir)
+    workers_dir.mkdir(parents=True)
+    
+    logs = []
+    logs.append(f"Loaded {len(api_keys)} API keys and {len(links)} links.")
+    
+    link_chunks = [links[i:i + int(links_per_worker)] for i in range(0, len(links), int(links_per_worker))]
+    logs.append(f"Divided into {len(link_chunks)} worker chunks.")
+    
+    for idx, chunk in enumerate(link_chunks):
+        worker_id = f"worker-{uuid.uuid4().hex[:8]}"
+        worker_title = f"yt-worker-{idx+1}-{uuid.uuid4().hex[:4]}"
+        
+        api_key = api_keys[idx % len(api_keys)]
+        worker_path = workers_dir / worker_id
+        worker_path.mkdir(parents=True)
+        
+        # Copy worker if exists locally
+        if Path("worker").exists():
+            shutil.copytree("worker", worker_path / "worker")
+        else:
+            # We copy from parent directory if this is running locally, otherwise it's an error on HF unless packed
+            parent_worker = Path("../hf_space_worker/worker")
+            if parent_worker.exists():
+                shutil.copytree(parent_worker, worker_path / "worker")
+            else:
+                logs.append("Warning: 'worker' directory not found. Kaggle notebook may fail.")
+                
+        if Path("cookies.txt").exists():
+            shutil.copy("cookies.txt", worker_path / "cookies.txt")
+        elif Path("../cookies.txt").exists():
+            shutil.copy("../cookies.txt", worker_path / "cookies.txt")
+            
+        if Path("cookies.json").exists():
+            shutil.copy("cookies.json", worker_path / "cookies.json")
+        elif Path("../cookies.json").exists():
+            shutil.copy("../cookies.json", worker_path / "cookies.json")
+            
+        config_name = "worker_config.json"
+        config_data = {
+            "links": chunk,
+            "api_key": api_key,
+        }
+        with open(worker_path / config_name, "w") as f:
+            json.dump(config_data, f, indent=2)
+            
+        create_worker_notebook(worker_path, config_name)
+        
+        metadata = {
+            "id": f"{username}/{worker_title}",
+            "title": worker_title,
+            "code_file": "notebook.ipynb",
+            "language": "python",
+            "kernel_type": "notebook",
+            "is_private": True,
+            "enable_gpu": False,
+            "enable_internet": True,
+            "dataset_sources": [],
+            "competition_sources": [],
+            "kernel_sources": []
+        }
+        with open(worker_path / "kernel-metadata.json", "w") as f:
+            json.dump(metadata, f, indent=2)
+            
+        logs.append(f"Prepared Worker {idx+1}: {worker_title} with {len(chunk)} links.")
+        
+        command = ["kaggle", "kernels", "push", "-p", str(worker_path)]
+        logs.append(f"Pushing {worker_title} to Kaggle...")
+        try:
+            res = subprocess.run(command, capture_output=True, text=True)
+            if res.returncode == 0:
+                logs.append("Push successful.")
+            else:
+                logs.append("Push failed.")
+                logs.append(res.stderr)
+        except Exception as e:
+            logs.append(f"Error: {e}")
+            
+    return "\n".join(logs)
+
+def sync_db(db_path, api_keys_text, links_text):
+    if not db_path:
+        return "Error: DB Path is empty."
+    
+    os.makedirs(os.path.dirname(db_path), exist_ok=True)
+    conn = sqlite3.connect(db_path)
+    cursor = conn.cursor()
+    
+    # Ensure tables exist
+    cursor.execute('''CREATE TABLE IF NOT EXISTS links (
+                        id INTEGER PRIMARY KEY,
+                        url TEXT,
+                        status TEXT DEFAULT 'pending'
+                    )''')
+    
+    cursor.execute('''CREATE TABLE IF NOT EXISTS accounts (
+                        id INTEGER PRIMARY KEY,
+                        email TEXT,
+                        api_key TEXT,
+                        created_at TEXT DEFAULT CURRENT_TIMESTAMP
+                    )''')
+    
+    logs = []
+    
+    # Add API keys
+    api_keys = [k.strip() for k in api_keys_text.split('\n') if k.strip()]
+    if api_keys:
+        added_keys = 0
+        for key in api_keys:
+            try:
+                # Basic check to avoid exact duplicates
+                cursor.execute("SELECT id FROM accounts WHERE api_key=?", (key,))
+                if not cursor.fetchone():
+                    cursor.execute("INSERT INTO accounts (email, api_key) VALUES (?, ?)", (f"user_{uuid.uuid4().hex[:4]}@example.com", key))
+                    added_keys += 1
+            except Exception as e:
+                logs.append(f"Error adding key {key}: {e}")
+        conn.commit()
+        logs.append(f"Successfully added {added_keys} new API keys.")
+        
+    # Add Links
+    links = [l.strip() for l in links_text.split('\n') if l.strip()]
+    if links:
+        added_links = 0
+        for link in links:
+            try:
+                # Basic check to avoid exact duplicates
+                cursor.execute("SELECT id FROM links WHERE url=?", (link,))
+                if not cursor.fetchone():
+                    cursor.execute("INSERT INTO links (url, status) VALUES (?, 'pending')", (link,))
+                    added_links += 1
+            except Exception as e:
+                logs.append(f"Error adding link {link}: {e}")
+        conn.commit()
+        logs.append(f"Successfully added {added_links} new pending links.")
+        
+    conn.close()
+    
+    if not api_keys and not links:
+        return "No API keys or links provided to sync."
+        
+    return "\n".join(logs)
+
+def run_v3_launch(max_workers, videos_per_worker):
+    cmd = ["python", "pipeline/orchestrator.py", "launch", 
+           "--max-workers", str(int(max_workers)), 
+           "--videos-per-worker", str(int(videos_per_worker)), 
+           "--push"]
+    try:
+        # Run in a subprocess and capture output
+        res = subprocess.run(cmd, capture_output=True, text=True, cwd=os.path.dirname(os.path.abspath(__file__)))
+        return f"--- LAUNCH OUTPUT ---\n{res.stdout}\n{res.stderr}"
+    except Exception as e:
+        return f"Error: {e}"
+
+def run_v3_sync():
+    cmd = ["python", "pipeline/orchestrator.py", "sync"]
+    try:
+        res = subprocess.run(cmd, capture_output=True, text=True, cwd=os.path.dirname(os.path.abspath(__file__)))
+        return f"--- SYNC OUTPUT ---\n{res.stdout}\n{res.stderr}"
+    except Exception as e:
+        return f"Error: {e}"
+
+def run_v3_status():
+    cmd = ["python", "pipeline/orchestrator.py", "status"]
+    try:
+        res = subprocess.run(cmd, capture_output=True, text=True, cwd=os.path.dirname(os.path.abspath(__file__)))
+        return f"--- STATUS OUTPUT ---\n{res.stdout}\n{res.stderr}"
+    except Exception as e:
+        return f"Error: {e}"
+
+with gr.Blocks() as app:
+    gr.Markdown("# YouTube Video Orchestrator Host")
+    gr.Markdown("Deploy this to Hugging Face Spaces or run locally. It fetches pending links from your database and delegates them to FastAPI workers or Kaggle.")
+    
+    with gr.Tabs():
+        with gr.Tab("V3 Pipeline Orchestrator"):
+            gr.Markdown("Directly run the advanced `pipeline/orchestrator.py` commands from this dashboard!")
+            
+            with gr.Row():
+                v3_max_workers = gr.Number(label="Max Workers", value=5)
+                v3_videos = gr.Number(label="Videos per Worker", value=250)
+                
+            with gr.Row():
+                v3_launch_btn = gr.Button("Launch Workers (--push)", variant="primary")
+                v3_sync_btn = gr.Button("Sync DB & Status")
+                v3_status_btn = gr.Button("Check Status")
+                
+            v3_output_log = gr.Textbox(label="Terminal Output", lines=15)
+            
+            v3_launch_btn.click(fn=run_v3_launch, inputs=[v3_max_workers, v3_videos], outputs=v3_output_log)
+            v3_sync_btn.click(fn=run_v3_sync, inputs=[], outputs=v3_output_log)
+            v3_status_btn.click(fn=run_v3_status, inputs=[], outputs=v3_output_log)
+
+        with gr.Tab("Database Sync"):
+            gr.Markdown("Sync (Add) new API keys and YouTube Links into your database.")
+            with gr.Row():
+                sync_db_input = gr.Textbox(label="DB Path", value="/data/db/accounts.db")
+            with gr.Row():
+                api_keys_input = gr.Textbox(label="API Keys (One per line)", lines=5)
+                links_input = gr.Textbox(label="YouTube Links (One per line)", lines=5)
+                
+            sync_btn = gr.Button("Sync to Database", variant="primary")
+            sync_log = gr.Textbox(label="Sync Logs", lines=5)
+            
+            sync_btn.click(fn=sync_db, inputs=[sync_db_input, api_keys_input, links_input], outputs=sync_log)
+
+        with gr.Tab("FastAPI Workers"):
+            with gr.Row():
+                db_input = gr.Textbox(label="DB Path", value="/data/db/accounts.db")
+                workers_input = gr.Textbox(label="Worker URLs (comma separated)", placeholder="http://localhost:7860, https://your-worker.hf.space")
+                links_per_worker = gr.Number(label="Links per worker", value=2)
+                
+            run_btn = gr.Button("Run Orchestration", variant="primary")
+            output_log = gr.Textbox(label="Logs", lines=10)
+            
+            run_btn.click(fn=trigger_workers, inputs=[db_input, workers_input, links_per_worker], outputs=output_log)
+
+        with gr.Tab("Simple Kaggle Workers"):
+            gr.Markdown("Trigger simple Kaggle notebook workers. Note: You must mount `kaggle.json` to `~/.kaggle/kaggle.json` in the container for this to work.")
+            
+            with gr.Row():
+                k_db_input = gr.Textbox(label="DB Path", value="/data/db/accounts.db")
+                k_links_per_worker = gr.Number(label="Links per worker", value=2)
+                k_max_workers = gr.Number(label="Max Workers", value=5)
+                
+            k_run_btn = gr.Button("Run Simple Kaggle Orchestration", variant="primary")
+            k_test_run_btn = gr.Button("Run Simple Kaggle Test Worker", variant="secondary")
+            kaggle_log = gr.Textbox(label="Kaggle Logs", lines=15)
+            
+            k_run_btn.click(fn=trigger_kaggle_workers, inputs=[k_db_input, k_links_per_worker, k_max_workers], outputs=kaggle_log)
+            k_test_run_btn.click(fn=trigger_kaggle_test, inputs=[], outputs=kaggle_log)
+
+if __name__ == "__main__":
+    app.launch(server_name="0.0.0.0", server_port=7860)

pipeline/__init__.py

@@ -0,0 +1 @@
+# YT Video Analysis Pipeline

pipeline/orchestrator.py

@@ -0,0 +1,599 @@
+#!/usr/bin/env python3
+"""
+YT Pipeline Orchestrator v3 — Host System (Part 2)
+===================================================
+Kaggle limits: max 10 concurrent sessions, 12hr max runtime.
+Auto-rotates workers, syncs results, retries failed tasks.
+
+Usage:
+  python orchestrator.py run --videos-per-worker 3    # auto-pilot loop
+  python orchestrator.py launch --max-workers 5       # one-shot launch
+  python orchestrator.py status
+  python orchestrator.py sync
+  python orchestrator.py retry
+  python orchestrator.py import
+"""
+
+import argparse, json, logging, os, shutil, sqlite3, subprocess, sys
+import time, uuid
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    datefmt="%H:%M:%S",
+)
+log = logging.getLogger("orchestrator")
+
+# ── paths & constants ────────────────────────────────────────────────
+BASE_DIR       = Path(__file__).resolve().parent.parent
+DB_ACCOUNTS    = BASE_DIR / "db" / "accounts.db"
+DB_CHANNELS    = BASE_DIR / "db" / "channel_links.db"
+DB_TRACKER     = BASE_DIR / "db" / "tracker.db"
+WORKERS_DIR    = BASE_DIR / "kaggle_workers"
+WORKER_SCRIPT  = Path(__file__).resolve().parent / "worker.py"
+HF_REPO        = "AdhyanshVerma/YT"
+
+# kaggle limits
+MAX_KAGGLE_SESSIONS  = 5
+KAGGLE_TIMEOUT_HOURS = 12
+POLL_INTERVAL_SEC    = 120   # check every 2 min in run loop
+SAFETY_MARGIN_MIN    = 30    # requeue if within 30min of 12hr limit
+
+# video states
+S_PENDING    = "pending"
+S_ASSIGNED   = "assigned"
+S_DONE       = "done"
+S_FAILED     = "failed"
+S_TIMEOUT    = "timeout"
+
+# worker states
+WS_CREATED   = "created"
+WS_PUSHED    = "pushed"
+WS_RUNNING   = "running"
+WS_COMPLETE  = "complete"
+WS_ERROR     = "error"
+WS_TIMEOUT   = "timeout"
+WS_CANCELLED = "cancelled"
+
+# ── tracker database ─────────────────────────────────────────────────
+class Tracker:
+    def __init__(self, db_path=DB_TRACKER):
+        self.db_path = str(db_path)
+        self._init_db()
+
+    def _conn(self):
+        c = sqlite3.connect(self.db_path)
+        c.row_factory = sqlite3.Row
+        c.execute("PRAGMA journal_mode=WAL")
+        return c
+
+    def _init_db(self):
+        with self._conn() as c:
+            c.execute("""CREATE TABLE IF NOT EXISTS videos (
+                video_id TEXT PRIMARY KEY, url TEXT NOT NULL,
+                title TEXT DEFAULT '', channel TEXT DEFAULT '',
+                duration REAL DEFAULT 0, status TEXT DEFAULT 'pending',
+                worker_id TEXT DEFAULT '', assigned_at TEXT DEFAULT '',
+                completed_at TEXT DEFAULT '', error_msg TEXT DEFAULT '',
+                retry_count INTEGER DEFAULT 0
+            )""")
+            c.execute("""CREATE TABLE IF NOT EXISTS workers (
+                worker_id TEXT PRIMARY KEY, kaggle_title TEXT DEFAULT '',
+                video_count INTEGER DEFAULT 0, status TEXT DEFAULT 'created',
+                created_at TEXT DEFAULT '', pushed_at TEXT DEFAULT '',
+                finished_at TEXT DEFAULT '', kaggle_status TEXT DEFAULT ''
+            )""")
+            c.execute("CREATE INDEX IF NOT EXISTS idx_vid_status ON videos(status)")
+            c.execute("CREATE INDEX IF NOT EXISTS idx_w_status ON workers(status)")
+            # migrate old tables missing kaggle_status
+            try:
+                c.execute("ALTER TABLE workers ADD COLUMN kaggle_status TEXT DEFAULT ''")
+            except Exception:
+                pass  # column already exists
+
+    def import_from_channels_db(self):
+        if not DB_CHANNELS.exists():
+            return 0
+        src = sqlite3.connect(str(DB_CHANNELS))
+        rows = src.execute("SELECT video_id,url,title,channel_url,duration FROM videos").fetchall()
+        src.close()
+        count = 0
+        with self._conn() as c:
+            for r in rows:
+                try:
+                    c.execute("INSERT OR IGNORE INTO videos (video_id,url,title,channel,duration,status) VALUES (?,?,?,?,?,?)",
+                              (r[0], r[1], r[2] or "", r[3] or "", r[4] or 0, S_PENDING))
+                    count += 1
+                except: pass
+        log.info(f"Imported {count} videos")
+        return count
+
+    def get_pending(self, limit):
+        with self._conn() as c:
+            return c.execute("SELECT video_id,url,title FROM videos WHERE status=? ORDER BY duration ASC LIMIT ?",
+                             (S_PENDING, limit)).fetchall()
+
+    def assign_batch(self, video_ids, worker_id):
+        now = datetime.now(timezone.utc).isoformat()
+        with self._conn() as c:
+            for vid in video_ids:
+                c.execute("UPDATE videos SET status=?,worker_id=?,assigned_at=? WHERE video_id=?",
+                          (S_ASSIGNED, worker_id, now, vid))
+
+    def requeue_worker_videos(self, worker_id, new_status=S_PENDING):
+        """Put assigned videos from a dead/timed-out worker back to pending."""
+        with self._conn() as c:
+            r = c.execute("UPDATE videos SET status=?,worker_id='' WHERE worker_id=? AND status=?",
+                          (new_status, worker_id, S_ASSIGNED))
+            return r.rowcount
+
+    def mark_worker(self, worker_id, status, kaggle_status=""):
+        now = datetime.now(timezone.utc).isoformat()
+        with self._conn() as c:
+            if status in (WS_COMPLETE, WS_ERROR, WS_TIMEOUT, WS_CANCELLED):
+                c.execute("UPDATE workers SET status=?,finished_at=?,kaggle_status=? WHERE worker_id=?",
+                          (status, now, kaggle_status, worker_id))
+            else:
+                c.execute("UPDATE workers SET status=?,kaggle_status=? WHERE worker_id=?",
+                          (status, kaggle_status, worker_id))
+
+    def update_video_statuses(self, status_dict):
+        with self._conn() as c:
+            for vid, st in status_dict.items():
+                if st in (S_DONE, S_FAILED):
+                    c.execute("UPDATE videos SET status=? WHERE video_id=? AND status!=?",
+                              (st, vid, st))
+
+
+    def register_worker(self, worker_id, kaggle_title, video_count):
+        with self._conn() as c:
+            c.execute("INSERT OR REPLACE INTO workers VALUES (?,?,?,?,?,?,?,?)",
+                      (worker_id, kaggle_title, video_count, WS_CREATED,
+                       datetime.now(timezone.utc).isoformat(), "", "", ""))
+
+    def mark_worker_pushed(self, worker_id):
+        with self._conn() as c:
+            c.execute("UPDATE workers SET status=?,pushed_at=? WHERE worker_id=?",
+                      (WS_PUSHED, datetime.now(timezone.utc).isoformat(), worker_id))
+
+    def active_workers(self):
+        """Workers that are pushed/running (consuming Kaggle slots)."""
+        with self._conn() as c:
+            return c.execute("SELECT * FROM workers WHERE status IN (?,?)",
+                             (WS_PUSHED, WS_RUNNING)).fetchall()
+
+    def pushed_workers(self):
+        with self._conn() as c:
+            return c.execute("SELECT * FROM workers WHERE status IN (?,?)",
+                             (WS_PUSHED, WS_RUNNING)).fetchall()
+
+    def timed_out_workers(self):
+        """Workers pushed more than 12hrs ago still active."""
+        cutoff = (datetime.now(timezone.utc) - timedelta(hours=KAGGLE_TIMEOUT_HOURS,
+                  minutes=-SAFETY_MARGIN_MIN)).isoformat()
+        with self._conn() as c:
+            return c.execute("SELECT * FROM workers WHERE status IN (?,?) AND pushed_at<? AND pushed_at!=''",
+                             (WS_PUSHED, WS_RUNNING, cutoff)).fetchall()
+
+    def retry_failed(self, max_retries=3):
+        with self._conn() as c:
+            r = c.execute("UPDATE videos SET status=? WHERE status IN (?,?) AND retry_count<?",
+                          (S_PENDING, S_FAILED, S_TIMEOUT, max_retries))
+            return r.rowcount
+
+    def stats(self):
+        with self._conn() as c:
+            rows = c.execute("SELECT status,COUNT(*) FROM videos GROUP BY status").fetchall()
+            return {r[0]: r[1] for r in rows}
+
+    def worker_stats(self):
+        with self._conn() as c:
+            return c.execute("SELECT * FROM workers ORDER BY created_at DESC LIMIT 20").fetchall()
+
+    def available_slots(self):
+        active = len(self.active_workers())
+        return max(0, MAX_KAGGLE_SESSIONS - active)
+
+# ── helpers ──────────────────────────────────────────────────────────
+def load_api_keys():
+    conn = sqlite3.connect(str(DB_ACCOUNTS))
+    rows = conn.execute("SELECT api_key FROM accounts WHERE api_key IS NOT NULL AND api_key!=''").fetchall()
+    conn.close()
+    return [r[0] for r in rows]
+
+def load_hf_token():
+    token = os.environ.get("HF_TOKEN", "")
+    if not token:
+        p = Path.home() / ".cache" / "huggingface" / "token"
+        if p.exists(): token = p.read_text().strip()
+    return token
+
+def get_kaggle_username():
+    kf = Path.home() / ".kaggle" / "kaggle.json"
+    if kf.exists(): return json.loads(kf.read_text()).get("username", "adhyanshverma")
+    return "adhyanshverma"
+
+def check_kaggle_kernel_status(kaggle_title, username):
+    """Query Kaggle API for kernel status. Returns: queued/running/complete/error/cancelled or None."""
+    try:
+        r = subprocess.run(["kaggle","kernels","status",f"{username}/{kaggle_title}"],
+                           capture_output=True, text=True, timeout=30)
+        out = r.stdout.strip().lower()
+        for s in ["complete", "error", "cancelled", "running", "queued"]:
+            if s in out: return s
+        return out if out else None
+    except Exception:
+        return None
+
+# ── notebook generation ──────────────────────────────────────────────
+def generate_notebook(worker_dir, config, worker_code):
+    config_json = json.dumps(config, indent=2)
+    
+    cookies_txt = Path("cookies.txt").read_text() if Path("cookies.txt").exists() else ""
+    cookies_json_str = Path("cookies.json").read_text() if Path("cookies.json").exists() else ""
+    
+    cells = [
+        {"cell_type":"code","execution_count":None,"metadata":{},"outputs":[],
+         "source":["!pip install -q yt-dlp huggingface_hub openai faster-whisper pyarrow\n",
+                   "!apt-get install -y -qq ffmpeg > /dev/null 2>&1\n",
+                   "print('Dependencies installed ✅')\n"]},
+        {"cell_type":"code","execution_count":None,"metadata":{},"outputs":[],
+         "source":["from pathlib import Path\n",
+                   f"Path('cookies.txt').write_text({repr(cookies_txt)})\n",
+                   f"Path('cookies.json').write_text({repr(cookies_json_str)})\n",
+                   "print('Cookies written ✅')\n"]},
+        {"cell_type":"code","execution_count":None,"metadata":{},"outputs":[],
+         "source": worker_code.split("\n")},
+        {"cell_type":"code","execution_count":None,"metadata":{},"outputs":[],
+         "source":["import json, os\n",
+                   f"config = json.loads('''{config_json}''')\n",
+                   "os.environ['HF_TOKEN'] = config['hf_token']\n",
+                   "status = run_worker(config)\n",
+                   "print(f'Worker finished: {json.dumps(status, indent=2)}')\n"]},
+    ]
+    for cell in cells:
+        cell["source"] = [l if l.endswith("\n") else l+"\n" for l in cell["source"]]
+    nb = {"cells":cells,"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},
+          "language_info":{"name":"python"}},"nbformat":4,"nbformat_minor":4}
+    (worker_dir/"notebook.ipynb").write_text(json.dumps(nb, indent=2))
+
+def generate_kernel_metadata(worker_dir, kaggle_title, username):
+    meta = {"id":f"{username}/{kaggle_title}","title":kaggle_title,"code_file":"notebook.ipynb",
+            "language":"python","kernel_type":"notebook","is_private":True,
+            "enable_gpu":False,"enable_internet":True,
+            "dataset_sources":[],"competition_sources":[],"kernel_sources":[]}
+    (worker_dir/"kernel-metadata.json").write_text(json.dumps(meta, indent=2))
+
+def load_vision_models():
+    VISION_MODELS_FILE = BASE_DIR / "models" / "vision_models.txt"
+    if VISION_MODELS_FILE.exists():
+        with open(VISION_MODELS_FILE) as f:
+            models = [l.strip() for l in f if l.strip() and not l.strip().startswith("#")]
+            if models:
+                return models
+    return [
+        "MiniMaxAI/MiniMax-M3",
+        "XiaomiMiMo/MiMo-V2.5",
+        "Qwen/Qwen3-VL-32B-Instruct",
+        "Qwen/Qwen3.6-27B",
+        "Qwen/Qwen3-VL-235B-A22B-Thinking"
+    ]
+
+# ── launch N workers (respecting slot limit) ─────────────────────────
+def launch_workers(tracker, num_workers, videos_per_worker, push=False):
+    slots = tracker.available_slots()
+    if slots <= 0:
+        log.info(f"No Kaggle slots available (all {MAX_KAGGLE_SESSIONS} in use)")
+        return 0
+    num_workers = min(num_workers, slots)
+    total_fetch = num_workers * videos_per_worker
+    pending = tracker.get_pending(total_fetch)
+    if not pending:
+        log.info("No pending videos")
+        return 0
+
+    api_keys = load_api_keys()
+    hf_token = load_hf_token()
+    if not api_keys or not hf_token:
+        log.error("Missing API keys or HF token")
+        return 0
+
+    username = get_kaggle_username()
+    worker_code = WORKER_SCRIPT.read_text()
+
+    chunks = [pending[i:i+videos_per_worker] for i in range(0, len(pending), videos_per_worker)]
+    WORKERS_DIR.mkdir(parents=True, exist_ok=True)
+    keys_per_w = max(1, len(api_keys) // max(1, len(chunks)))
+    launched = 0
+    vmodels = load_vision_models()
+
+    for widx, chunk in enumerate(chunks):
+        worker_id = f"w-{uuid.uuid4().hex[:8]}"
+        kaggle_title = f"yt-w-{uuid.uuid4().hex[:6]}"
+
+        start_k = (widx * keys_per_w) % len(api_keys)
+        wkeys = [api_keys[(start_k+k) % len(api_keys)] for k in range(keys_per_w)]
+
+        video_list = [{"video_id":v["video_id"],"url":v["url"]} for v in chunk]
+        video_ids = [v["video_id"] for v in chunk]
+
+        config = {"worker_id":worker_id,"videos":video_list,"api_keys":wkeys,
+                  "hf_token":hf_token,"vision_models":vmodels,
+                  "text_model":"Qwen/Qwen3-32B"}
+
+        wdir = WORKERS_DIR / worker_id
+        wdir.mkdir(parents=True, exist_ok=True)
+        generate_notebook(wdir, config, worker_code)
+        generate_kernel_metadata(wdir, kaggle_title, username)
+        tracker.assign_batch(video_ids, worker_id)
+        tracker.register_worker(worker_id, kaggle_title, len(chunk))
+
+        log.info(f"  Worker: {kaggle_title} — {len(chunk)} videos, {len(wkeys)} keys")
+
+        if push:
+            r = subprocess.run(["kaggle","kernels","push","-p",str(wdir)],
+                               capture_output=True, text=True)
+            if r.returncode == 0:
+                tracker.mark_worker_pushed(worker_id)
+                launched += 1
+                log.info(f"  ✅ Pushed")
+            else:
+                log.error(f"  ❌ Push failed: {r.stderr[:200]}")
+                tracker.requeue_worker_videos(worker_id)
+                tracker.mark_worker(worker_id, WS_ERROR, "push_failed")
+        else:
+            launched += 1
+
+    return launched
+
+# ── sync: poll kaggle + HF status ───────────────────────────────────
+def sync_workers(tracker):
+    """Check Kaggle status for all active workers, handle timeouts."""
+    username = get_kaggle_username()
+    active = tracker.pushed_workers()
+    if not active:
+        return
+
+    log.info(f"Checking {len(active)} active workers...")
+
+    # 1. Check for 12hr timeouts
+    timed_out = tracker.timed_out_workers()
+    for w in timed_out:
+        wid = w["worker_id"]
+        log.warning(f"  ⏰ Worker {wid} ({w['kaggle_title']}) exceeded 12hr limit")
+        requeued = tracker.requeue_worker_videos(wid)
+        tracker.mark_worker(wid, WS_TIMEOUT, "12hr_timeout")
+        log.info(f"     Requeued {requeued} videos back to pending")
+
+    # 2. Poll Kaggle API for each active worker
+    for w in active:
+        wid = w["worker_id"]
+        title = w["kaggle_title"]
+        if wid in [t["worker_id"] for t in timed_out]:
+            continue  # already handled
+
+        kstatus = check_kaggle_kernel_status(title, username)
+        if not kstatus:
+            continue
+
+        if kstatus == "complete":
+            log.info(f"  ✅ {title} completed")
+            tracker.mark_worker(wid, WS_COMPLETE, kstatus)
+        elif kstatus == "error":
+            log.warning(f"  ❌ {title} errored")
+            requeued = tracker.requeue_worker_videos(wid)
+            tracker.mark_worker(wid, WS_ERROR, kstatus)
+            log.info(f"     Requeued {requeued} videos")
+        elif kstatus == "cancelled":
+            log.warning(f"  🚫 {title} cancelled")
+            requeued = tracker.requeue_worker_videos(wid)
+            tracker.mark_worker(wid, WS_CANCELLED, kstatus)
+            log.info(f"     Requeued {requeued} videos")
+        elif kstatus in ("running", "queued"):
+            tracker.mark_worker(wid, WS_RUNNING, kstatus)
+        time.sleep(1)  # rate limit kaggle API
+
+# ── sync HF status files to update video-level tracking ──────────────
+def sync_hf_results(tracker):
+    hf_token = load_hf_token()
+    if not hf_token:
+        return
+    try:
+        from huggingface_hub import HfApi, hf_hub_download
+        api = HfApi(token=hf_token)
+        files = api.list_repo_files(repo_id=HF_REPO, repo_type="dataset")
+
+        status_files = [f for f in files if f.startswith("status/") and f.endswith(".json")]
+        data_files = [f for f in files if f.startswith("data/") and f.endswith(".parquet")]
+
+        for sf in status_files:
+            try:
+                local = hf_hub_download(repo_id=HF_REPO, filename=sf, repo_type="dataset",
+                                        token=hf_token, force_download=True)
+                with open(local) as f:
+                    st = json.load(f)
+                wid = st.get("worker_id","")
+                state = st.get("state","")
+                v_status = st.get("video_status", {})
+                if v_status:
+                    tracker.update_video_statuses(v_status)
+                log.info(f"  HF status: {wid} → done={st.get('done',0)} "
+                         f"failed={st.get('failed',0)} state={state}")
+                if state == "completed":
+                    tracker.mark_worker(wid, WS_COMPLETE, "hf_confirmed")
+            except Exception as e:
+                log.warning(f"  Could not read {sf}: {e}")
+
+        log.info(f"HF: {len(data_files)} parquet files, {len(status_files)} status files")
+    except Exception as e:
+        log.error(f"HF sync error: {e}")
+
+# ── commands ─────────────────────────────────────────────────────────
+def cmd_run(args):
+    """Autopilot: continuously launch workers, sync, retry — respecting Kaggle limits."""
+    tracker = Tracker()
+    if not tracker.stats():
+        tracker.import_from_channels_db()
+
+    log.info(f"🚀 Autopilot started — max {MAX_KAGGLE_SESSIONS} sessions, "
+             f"{args.videos_per_worker} videos/worker, polling every {POLL_INTERVAL_SEC}s")
+    log.info(f"   Press Ctrl+C to stop\n")
+
+    cycle = 0
+    while True:
+        cycle += 1
+        stats = tracker.stats()
+        pending = stats.get(S_PENDING, 0)
+        done = stats.get(S_DONE, 0)
+        total = sum(stats.values())
+
+        log.info(f"── Cycle {cycle} ───────────────��─────────────────")
+        log.info(f"   Videos: {done}/{total} done, {pending} pending, "
+                 f"{stats.get(S_ASSIGNED,0)} assigned, {stats.get(S_FAILED,0)} failed")
+
+        # 1. Sync — check which workers finished/failed/timed out
+        sync_workers(tracker)
+
+        # 2. Sync HF results
+        if cycle % 5 == 0:  # every 5 cycles (~10min)
+            sync_hf_results(tracker)
+
+        # 3. Auto-retry failed (every 10 cycles)
+        if cycle % 10 == 0:
+            n = tracker.retry_failed(max_retries=3)
+            if n: log.info(f"   Auto-retried {n} failed videos")
+
+        # 4. Launch new workers if slots available
+        slots = tracker.available_slots()
+        if slots > 0 and pending > 0:
+            log.info(f"   {slots} Kaggle slots free — launching workers...")
+            n = launch_workers(tracker, slots, args.videos_per_worker, push=True)
+            log.info(f"   Launched {n} workers")
+        elif slots == 0:
+            log.info(f"   All {MAX_KAGGLE_SESSIONS} Kaggle slots in use — waiting...")
+        elif pending == 0:
+            assigned = stats.get(S_ASSIGNED, 0)
+            if assigned == 0:
+                log.info("✅ All videos processed! Exiting autopilot.")
+                break
+            log.info(f"   Waiting for {assigned} assigned videos to complete...")
+
+        # 5. Sleep
+        log.info(f"   Next check in {POLL_INTERVAL_SEC}s...\n")
+        try:
+            time.sleep(POLL_INTERVAL_SEC)
+        except KeyboardInterrupt:
+            log.info("\n⛔ Autopilot stopped by user")
+            break
+
+def cmd_launch(args):
+    tracker = Tracker()
+    if not tracker.stats():
+        tracker.import_from_channels_db()
+    stats = tracker.stats()
+    log.info(f"Video stats: {dict(stats)}")
+    slots = tracker.available_slots()
+    log.info(f"Kaggle slots available: {slots}/{MAX_KAGGLE_SESSIONS}")
+    n = min(args.max_workers, slots)
+    if n <= 0:
+        log.error(f"No slots! {len(tracker.active_workers())} workers active.")
+        return
+    launched = launch_workers(tracker, n, args.videos_per_worker, push=args.push)
+    log.info(f"Launched {launched} workers" + (" (dry run)" if not args.push else ""))
+
+def cmd_status(args):
+    tracker = Tracker()
+    stats = tracker.stats()
+    total = sum(stats.values()) or 1
+    active = tracker.active_workers()
+    slots = tracker.available_slots()
+
+    print(f"\n{'='*60}")
+    print(f"  YT Pipeline Status")
+    print(f"{'='*60}")
+    print(f"  Total videos:    {total}")
+    print(f"  Kaggle sessions: {len(active)}/{MAX_KAGGLE_SESSIONS} (slots free: {slots})")
+    print(f"{'─'*60}")
+    for s in [S_PENDING, S_ASSIGNED, S_DONE, S_FAILED, S_TIMEOUT]:
+        c = stats.get(s, 0)
+        pct = c/total*100
+        bar = "█"*int(pct/2) + "░"*(50-int(pct/2))
+        print(f"  {s:12s} {c:7d} {pct:5.1f}%  {bar}")
+    print(f"{'─'*60}")
+
+    workers = tracker.worker_stats()
+    if workers:
+        print(f"\n  Recent Workers:")
+        for w in workers:
+            age = ""
+            if w["pushed_at"]:
+                try:
+                    pushed = datetime.fromisoformat(w["pushed_at"])
+                    hrs = (datetime.now(timezone.utc)-pushed).total_seconds()/3600
+                    age = f" ({hrs:.1f}h ago)"
+                except: pass
+            ks = '?'
+            try: ks = w['kaggle_status'] or '?'
+            except: pass
+            print(f"    {w['worker_id']}  {w['kaggle_title']:25s}  "
+                  f"v={w['video_count']}  {w['status']:10s}  "
+                  f"kaggle={ks}{age}")
+    print()
+
+def cmd_retry(args):
+    tracker = Tracker()
+    n = tracker.retry_failed(max_retries=args.max_retries)
+    print(f"Reset {n} failed/timed-out videos to pending")
+
+def cmd_sync(args):
+    tracker = Tracker()
+    sync_workers(tracker)
+    sync_hf_results(tracker)
+
+def cmd_import(args):
+    tracker = Tracker()
+    n = tracker.import_from_channels_db()
+    print(f"Imported {n} videos")
+
+# ── main ─────────────────────────────────────────────────────────────
+def main():
+    p = argparse.ArgumentParser(description="YT Pipeline Orchestrator v3",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+Commands:
+  run          Autopilot — continuously launch, sync, retry (respects 10-session limit)
+  launch       One-shot launch (respects slot limit)
+  status       Show pipeline + Kaggle session status
+  sync         Poll Kaggle API + HF for updates
+  retry        Reset failed/timed-out videos for retry
+  import       Import videos from channel_links.db
+""")
+    sub = p.add_subparsers(dest="command")
+
+    pr = sub.add_parser("run", help="Autopilot loop")
+    pr.add_argument("--videos-per-worker", type=int, default=3)
+
+    pl = sub.add_parser("launch", help="One-shot launch")
+    pl.add_argument("--max-workers", type=int, default=5)
+    pl.add_argument("--videos-per-worker", type=int, default=3)
+    pl.add_argument("--push", action="store_true")
+
+    sub.add_parser("status", help="Show status")
+
+    pt = sub.add_parser("retry", help="Retry failed")
+    pt.add_argument("--max-retries", type=int, default=3)
+
+    sub.add_parser("sync", help="Sync from Kaggle+HF")
+    sub.add_parser("import", help="Import from channel_links.db")
+
+    args = p.parse_args()
+    if not args.command:
+        p.print_help(); return
+
+    {"run":cmd_run,"launch":cmd_launch,"status":cmd_status,
+     "retry":cmd_retry,"sync":cmd_sync,"import":cmd_import}[args.command](args)
+
+if __name__ == "__main__":
+    main()

pipeline/worker.py

@@ -0,0 +1,741 @@
+#!/usr/bin/env python3
+"""
+YT Video Analysis Worker v2 — Runs on Kaggle None runtime
+Constraints: 30GB RAM · 20GB disk · 4 vCPUs · CPU-only
+Pipeline: Download → Split Video 60s → Compress → Upload chunks to HF bucket
+          → Vision model per chunk → Extract audio → Split if >3min
+          → Transcribe → LLM analysis → Collect → Parquet → Upload dataset
+"""
+
+import base64, glob, hashlib, json, logging, os, re, shutil, subprocess
+import sys, tempfile, time, traceback, uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    datefmt="%H:%M:%S",
+)
+log = logging.getLogger("worker")
+
+# ── constants ────────────────────────────────────────────────────────
+HF_REPO          = "AdhyanshVerma/YT"
+STAGING_PREFIX   = "staging"
+DATA_PREFIX      = "data"
+VIDEO_CHUNK_SEC  = 60
+AUDIO_SPLIT_SEC  = 180       # 3 minutes
+MIN_TAIL_SEC     = 15
+FRAME_INTERVAL   = 5         # extract 1 frame per N seconds
+MAX_RETRIES      = 3
+RETRY_DELAY      = 5
+DISK_HEADROOM_GB = 4
+
+# ── disk helpers ─────────────────────────────────────────────────────
+def disk_free_gb(path="/"):
+    st = os.statvfs(path)
+    return (st.f_bavail * st.f_frsize) / (1024**3)
+
+def safe_cleanup(*paths):
+    for p in paths:
+        try:
+            p = str(p)
+            if os.path.isdir(p):
+                shutil.rmtree(p, ignore_errors=True)
+            elif os.path.isfile(p):
+                os.remove(p)
+        except Exception:
+            pass
+
+# ── ffprobe duration ─────────────────────────────────────────────────
+def media_duration(path):
+    try:
+        r = subprocess.run(
+            ["ffprobe","-v","error","-show_entries","format=duration",
+             "-of","default=noprint_wrappers=1:nokey=1", path],
+            capture_output=True, text=True, check=True)
+        return float(r.stdout.strip())
+    except Exception:
+        return None
+
+# ── video download ───────────────────────────────────────────────────
+def download_video(url, out_dir):
+    """Download video at HD, return filepath or None."""
+    os.makedirs(out_dir, exist_ok=True)
+    
+    # Path to cookies file (useful for bypassing bot blocks)
+    # Checks either current working dir or the parent of the script
+    cookies_path = "cookies.txt"
+    if not os.path.exists(cookies_path):
+        # Safely get directory, falling back if __file__ is not defined (e.g. in notebooks)
+        try:
+            base_d = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+            alt_cookies = os.path.join(base_d, "cookies.txt")
+            if os.path.exists(alt_cookies):
+                cookies_path = alt_cookies
+        except NameError:
+            pass
+
+    opts = {
+        "format": "bestvideo[height<=1080]+bestaudio/best[height<=1080]/bestvideo+bestaudio/best",
+        "outtmpl": os.path.join(out_dir, "%(id)s.%(ext)s"),
+        "merge_output_format": "mp4",
+        "quiet": True, "no_warnings": True,
+        "nocheckcertificate": True, "ignoreerrors": False,
+        "geo_bypass": True,
+        "socket_timeout": 60,
+        "retries": 5,
+        "sleep_requests": 2,
+        "sleep_interval": 5,
+        "max_sleep_interval": 15,
+        "extractor_args": {"youtube": {"player_client": ["android", "ios", "tv", "web"]}},
+        "writeinfojson": True,
+        "writecomments": True,
+        "writesubtitles": True,
+        "subtitleslangs": ["en", "auto"],
+        "http_headers": {
+            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                          "AppleWebKit/537.36 Chrome/125.0 Safari/537.36"
+        },
+    }
+    
+    # if os.path.exists(cookies_path):
+    #     opts["cookiefile"] = cookies_path
+        
+    try:
+        import yt_dlp
+        with yt_dlp.YoutubeDL(opts) as ydl:
+            info = ydl.extract_info(url, download=True)
+            if not info:
+                return None, {}
+            fp = None
+            if "requested_downloads" in info:
+                fp = info["requested_downloads"][0]["filepath"]
+            else:
+                fp = ydl.prepare_filename(info)
+                fp = os.path.splitext(fp)[0] + ".mp4"
+            meta = {
+                "title": info.get("title",""),
+                "duration": info.get("duration",0),
+                "duration_string": info.get("duration_string",""),
+                "view_count": info.get("view_count",0),
+                "like_count": info.get("like_count",0),
+                "comment_count": info.get("comment_count",0),
+                "channel": info.get("channel",""),
+                "channel_id": info.get("channel_id",""),
+                "channel_url": info.get("channel_url",""),
+                "channel_follower_count": info.get("channel_follower_count",0),
+                "upload_date": info.get("upload_date",""),
+                "timestamp": info.get("timestamp",0),
+                "description": info.get("description",""),
+                "tags": info.get("tags",[]),
+                "categories": info.get("categories",[]),
+                "language": info.get("language",""),
+                "license": info.get("license",""),
+                "availability": info.get("availability",""),
+                "age_limit": info.get("age_limit",0),
+                "width": info.get("width",0),
+                "height": info.get("height",0),
+                "resolution": info.get("resolution",""),
+                "fps": info.get("fps",0),
+                "aspect_ratio": info.get("aspect_ratio",0),
+                "vcodec": info.get("vcodec",""),
+                "acodec": info.get("acodec",""),
+                "vbr": info.get("vbr",0),
+                "abr": info.get("abr",0),
+                "asr": info.get("asr",0),
+                "audio_channels": info.get("audio_channels",0),
+                "dynamic_range": info.get("dynamic_range",""),
+                "filesize_approx": info.get("filesize_approx",0),
+                "tbr": info.get("tbr",0),
+                "ext": info.get("ext",""),
+                "protocol": info.get("protocol",""),
+                "format_id": info.get("format_id",""),
+                "format_note": info.get("format_note",""),
+                "playlist_title": info.get("playlist_title",""),
+                "playlist_index": info.get("playlist_index",0),
+                "live_status": info.get("live_status",""),
+                "video_id": info.get("id",""),
+                "webpage_url": info.get("webpage_url", url),
+            }
+            return fp, meta
+    except Exception as e:
+        log.error(f"Download failed for {url}: {e}")
+        return None, {}
+
+# ── video splitting ──────────────────────────────────────────────────
+def split_video(input_path, chunk_sec=VIDEO_CHUNK_SEC, out_dir=None):
+    name = Path(input_path).stem
+    if out_dir is None:
+        out_dir = os.path.join(os.path.dirname(input_path), f"{name}_vchunks")
+    os.makedirs(out_dir, exist_ok=True)
+    dur = media_duration(input_path)
+    if not dur:
+        return []
+    # calculate split points
+    pts, cur = [], chunk_sec
+    while cur < dur:
+        if (dur - cur) < MIN_TAIL_SEC:
+            break
+        pts.append(cur); cur += chunk_sec
+    pattern = os.path.join(out_dir, f"{name}_chunk_%03d.mp4")
+    cmd = ["ffmpeg","-y","-i",input_path,"-map","0","-c","copy",
+           "-f","segment","-reset_timestamps","1"]
+    if pts:
+        cmd += ["-segment_times", ",".join(str(round(p,3)) for p in pts)]
+    else:
+        cmd += ["-segment_time", str(dur+10)]
+    cmd.append(pattern)
+    try:
+        subprocess.run(cmd, check=True, capture_output=True)
+    except subprocess.CalledProcessError as e:
+        log.error(f"Video split error: {e.stderr.decode(errors='replace')[:300]}")
+        return []
+    return sorted(glob.glob(os.path.join(out_dir, f"{name}_chunk_*.mp4")))
+
+# ── video compression (same resolution, smaller size) ────────────────
+def compress_chunk(inp, out):
+    """Re-encode with h264 CRF 28 — same res, ~40-60% smaller."""
+    cmd = ["ffmpeg","-y","-i",inp,
+           "-c:v","libx264","-crf","28","-preset","fast",
+           "-c:a","aac","-b:a","128k", out]
+    try:
+        subprocess.run(cmd, check=True, capture_output=True)
+        return True
+    except Exception:
+        # if compression fails, just copy
+        shutil.copy2(inp, out)
+        return False
+
+# ── audio extraction ─────────────────────────────────────────────────
+def extract_audio(video_path, out_path):
+    cmd = ["ffmpeg","-y","-i",video_path,
+           "-vn","-acodec","libmp3lame","-q:a","4", out_path]
+    try:
+        subprocess.run(cmd, check=True, capture_output=True, text=True)
+        return True
+    except subprocess.CalledProcessError as e:
+        log.error(f"Audio extraction failed for {video_path}: {e.stderr}")
+        return False
+    except Exception as e:
+        log.error(f"Unexpected error in audio extraction: {e}")
+        return False
+
+# ── audio silence detection + splitting ──────────────────────────────
+def detect_silences(audio_path):
+    cmd = ["ffmpeg","-i",audio_path,
+           "-af","silencedetect=noise=-20dB:d=0.3","-f","null","-"]
+    r = subprocess.run(cmd, capture_output=True, text=True)
+    starts, ends = [], []
+    for line in r.stderr.split("\n"):
+        m = re.search(r"silence_start:\s+([\d.]+)", line)
+        if m: starts.append(float(m.group(1)))
+        m = re.search(r"silence_end:\s+([\d.]+)", line)
+        if m: ends.append(float(m.group(1)))
+    return [(s+e)/2 for s,e in zip(starts, ends)]
+
+def split_audio(audio_path, chunk_sec=AUDIO_SPLIT_SEC, out_dir=None):
+    name = Path(audio_path).stem
+    if out_dir is None:
+        out_dir = os.path.join(os.path.dirname(audio_path), f"{name}_achunks")
+    os.makedirs(out_dir, exist_ok=True)
+    dur = media_duration(audio_path)
+    if not dur or dur <= chunk_sec:
+        return [audio_path]  # no split needed
+    silences = detect_silences(audio_path)
+    pts, target = [], chunk_sec
+    while target < dur:
+        if (dur - target) < MIN_TAIL_SEC:
+            break
+        valid = [p for p in silences if abs(p - target) <= 15]
+        if valid:
+            best = min(valid, key=lambda p: abs(p - target))
+            pts.append(best); target = best + chunk_sec
+        else:
+            pts.append(target); target += chunk_sec
+    ext = Path(audio_path).suffix
+    pattern = os.path.join(out_dir, f"{name}_achunk_%03d{ext}")
+    cmd = ["ffmpeg","-y","-i",audio_path,"-c","copy","-map","0",
+           "-f","segment","-reset_timestamps","1"]
+    if pts:
+        cmd += ["-segment_times", ",".join(str(round(p,3)) for p in pts)]
+    else:
+        cmd += ["-segment_time", str(dur+10)]
+    cmd.append(pattern)
+    try:
+        subprocess.run(cmd, check=True, capture_output=True)
+        return sorted(glob.glob(os.path.join(out_dir, f"{name}_achunk_*{ext}")))
+    except Exception:
+        return [audio_path]
+
+# ── frame extraction for vision API ──────────────────────────────────
+def extract_frames(video_path, out_dir, interval=FRAME_INTERVAL):
+    os.makedirs(out_dir, exist_ok=True)
+    pattern = os.path.join(out_dir, "frame_%04d.jpg")
+    cmd = ["ffmpeg","-y","-i",video_path,"-vf",f"fps=1/{interval}",
+           "-q:v","5", pattern]
+    try:
+        subprocess.run(cmd, check=True, capture_output=True)
+        return sorted(glob.glob(os.path.join(out_dir, "frame_*.jpg")))
+    except Exception:
+        return []
+
+def encode_image_b64(path):
+    with open(path, "rb") as f:
+        return "data:image/jpeg;base64," + base64.b64encode(f.read()).decode()
+
+# ── HF storage bucket ───────────────────────────────────────────────
+def hf_upload_file(local_path, remote_path, token):
+    from huggingface_hub import HfApi
+    api = HfApi(token=token)
+    for attempt in range(MAX_RETRIES):
+        try:
+            api.upload_file(
+                path_or_fileobj=local_path,
+                path_in_repo=remote_path,
+                repo_id=HF_REPO, repo_type="dataset",
+            )
+            return True
+        except Exception as e:
+            log.warning(f"HF upload attempt {attempt+1} failed: {e}")
+            time.sleep(RETRY_DELAY * (attempt+1))
+    return False
+
+def hf_download_url(remote_path):
+    return f"https://huggingface.co/datasets/{HF_REPO}/resolve/main/{remote_path}"
+
+def hf_delete_folder(prefix, token):
+    from huggingface_hub import HfApi
+    api = HfApi(token=token)
+    try:
+        files = api.list_repo_files(repo_id=HF_REPO, repo_type="dataset")
+        to_del = [f for f in files if f.startswith(prefix)]
+        if to_del:
+            ops = []
+            from huggingface_hub import CommitOperationDelete
+            for f in to_del:
+                ops.append(CommitOperationDelete(path_in_repo=f))
+            api.create_commit(
+                repo_id=HF_REPO, repo_type="dataset",
+                operations=ops, commit_message=f"cleanup {prefix}")
+    except Exception as e:
+        log.warning(f"HF cleanup failed: {e}")
+
+# ── model API calls ──────────────────────────────────────────────────
+def _api_call(messages, api_key, model, base_url="https://api.featherless.ai/v1",
+              max_tokens=4096, temperature=0.4):
+    """OpenAI-compatible API call with retries."""
+    from openai import OpenAI
+    client = OpenAI(base_url=base_url, api_key=api_key)
+    for attempt in range(MAX_RETRIES):
+        try:
+            resp = client.chat.completions.create(
+                model=model, messages=messages,
+                max_tokens=max_tokens, temperature=temperature)
+            return resp.choices[0].message.content
+        except Exception as e:
+            log.warning(f"API attempt {attempt+1} failed ({model}): {e}")
+            time.sleep(RETRY_DELAY * (attempt+1))
+    return None
+
+DEFAULT_VISION_MODELS = [
+    "MiniMaxAI/MiniMax-M3",
+    "XiaomiMiMo/MiMo-V2.5",
+    "Qwen/Qwen3-VL-32B-Instruct",
+    "Qwen/Qwen3.6-27B",
+    "Qwen/Qwen3-VL-235B-A22B-Thinking",
+]
+
+def analyze_video_chunk(frames_dir, chunk_idx, total_chunks, api_key,
+                        vision_models=None):
+    """Send extracted frames to vision model for analysis with priority fallback."""
+    frames = sorted(glob.glob(os.path.join(frames_dir, "frame_*.jpg")))
+    if not frames:
+        return None
+    prompt = (
+        f"You are analyzing video chunk {chunk_idx+1}/{total_chunks}. "
+        f"These are frames extracted every {FRAME_INTERVAL}s from a 60-second segment.\n\n"
+        "Provide an EXTREMELY detailed analysis:\n"
+        "1. Scene description — what is happening visually\n"
+        "2. Objects, people, text/UI visible on screen\n"
+        "3. Motion and transitions between frames\n"
+        "4. Any on-screen text, code, diagrams, or slides\n"
+        "5. Visual style, colors, camera angles\n"
+        "6. Key information conveyed visually\n\n"
+        "Be thorough — every detail matters for the dataset."
+    )
+    content = [{"type": "text", "text": prompt}]
+    # send up to 12 frames to avoid token limits
+    step = max(1, len(frames) // 12)
+    selected = frames[::step][:12]
+    for fp in selected:
+        content.append({
+            "type": "image_url",
+            "image_url": {"url": encode_image_b64(fp)}
+        })
+    messages = [{"role": "user", "content": content}]
+
+    models = vision_models if vision_models else DEFAULT_VISION_MODELS
+    for model in models:
+        log.info(f"Attempting vision analysis with model: {model}")
+        res = _api_call(messages, api_key, model, max_tokens=4096)
+        if res:
+            log.info(f"Vision analysis succeeded with model: {model}")
+            return res
+        log.warning(f"Vision model {model} failed or busy. Trying next in priority...")
+    return None
+
+
+def analyze_audio_transcript(transcript, chunk_idx, total_chunks, api_key,
+                             model="Qwen/Qwen3-32B"):
+    """Send audio transcript to LLM for detailed analysis."""
+    prompt = (
+        f"Audio transcript chunk {chunk_idx+1}/{total_chunks}:\n\n"
+        f"---\n{transcript}\n---\n\n"
+        "Provide a detailed analysis of this audio:\n"
+        "1. Summary of what is being discussed\n"
+        "2. Key points and arguments made\n"
+        "3. Technical terms or concepts mentioned\n"
+        "4. Speaker tone and intent\n"
+        "5. Notable quotes or statements\n"
+        "6. Questions raised or answered\n\n"
+        "Be comprehensive and precise."
+    )
+    messages = [{"role": "user", "content": [{"type": "text", "text": prompt}]}]
+    return _api_call(messages, api_key, model, max_tokens=4096)
+
+def transcribe_audio_chunk(audio_path):
+    """Transcribe audio using faster-whisper (CPU)."""
+    try:
+        from faster_whisper import WhisperModel
+        model = WhisperModel("base", device="cpu", compute_type="int8")
+        segments, info = model.transcribe(audio_path, beam_size=3)
+        text_parts = []
+        for seg in segments:
+            text_parts.append(f"[{seg.start:.1f}s-{seg.end:.1f}s] {seg.text.strip()}")
+        return "\n".join(text_parts)
+    except Exception as e:
+        log.error(f"Transcription failed: {e}")
+        return None
+
+# ── per-video pipeline ───────────────────────────────────────────────
+def process_video(video_url, video_id, api_key, hf_token, worker_id, work_dir,
+                  vision_models=None):
+    """Full pipeline for one video. Returns result dict or None."""
+    log.info(f"{'='*60}")
+    log.info(f"Processing: {video_id} — {video_url}")
+    log.info(f"Disk free: {disk_free_gb():.1f} GB")
+
+    vid_dir = os.path.join(work_dir, video_id)
+    os.makedirs(vid_dir, exist_ok=True)
+    staging_prefix = f"{STAGING_PREFIX}/{worker_id}/{video_id}"
+    
+    result = {
+        "video_id": video_id,
+        "url": video_url,
+        "worker_id": worker_id,
+        "processed_at": datetime.now(timezone.utc).isoformat(),
+        "status": "failed",
+        "video_analysis": [],
+        "audio_analysis": [],
+        "audio_transcript": "",
+        "metadata": {},
+        "hf_video_urls": [],
+        "errors": [],
+    }
+
+    try:
+        # ── 1. Download ──────────────────────────────────────────
+        log.info("Step 1: Downloading video...")
+        dl_dir = os.path.join(vid_dir, "download")
+        video_path, meta = download_video(video_url, dl_dir)
+        if not video_path or not os.path.exists(video_path):
+            log.error(f"Download failed for {video_id}. Skipping commit to avoid empty records.")
+            return None
+        result["metadata"] = meta
+        fsize = os.path.getsize(video_path) / (1024**2)
+        log.info(f"Downloaded: {fsize:.0f} MB — {meta.get('title','?')}")
+
+        # ── 2. Split video into 60s chunks (skip for short videos) ──
+        vid_duration = media_duration(video_path) or meta.get("duration", 0)
+        if vid_duration and vid_duration <= VIDEO_CHUNK_SEC + MIN_TAIL_SEC:
+            log.info(f"Step 2: Short video ({vid_duration:.0f}s ≤ {VIDEO_CHUNK_SEC}s) — skipping split")
+            video_chunks = [video_path]
+        else:
+            log.info(f"Step 2: Splitting {vid_duration:.0f}s video into {VIDEO_CHUNK_SEC}s chunks...")
+            chunks_dir = os.path.join(vid_dir, "vchunks")
+            video_chunks = split_video(video_path, VIDEO_CHUNK_SEC, chunks_dir)
+            if not video_chunks:
+                video_chunks = [video_path]
+        log.info(f"Video chunks: {len(video_chunks)}")
+
+        # ── 3. Compress + upload chunks + vision analysis ────────
+        log.info("Step 3: Compress → Upload → Vision analysis...")
+        for cidx, chunk_path in enumerate(video_chunks):
+            log.info(f"  Chunk {cidx+1}/{len(video_chunks)}: {Path(chunk_path).name}")
+            # compress
+            comp_path = chunk_path.replace(".mp4", "_comp.mp4")
+            compress_chunk(chunk_path, comp_path)
+            if os.path.exists(comp_path):
+                if chunk_path != video_path:
+                    safe_cleanup(chunk_path)
+                chunk_path = comp_path
+
+            # upload to HF bucket
+            remote = f"{staging_prefix}/v_chunk_{cidx:03d}.mp4"
+            uploaded = hf_upload_file(chunk_path, remote, hf_token)
+            if not uploaded:
+                result["errors"].append(f"Upload failed chunk {cidx}")
+                if chunk_path != video_path:
+                    safe_cleanup(chunk_path)
+                continue
+
+            # extract frames for vision API
+            result["hf_video_urls"].append(hf_download_url(remote))
+            frames_dir = os.path.join(vid_dir, f"frames_{cidx}")
+            frames = extract_frames(chunk_path, frames_dir)
+            if chunk_path != video_path:
+                safe_cleanup(chunk_path)  # free disk
+
+            if frames:
+                analysis = analyze_video_chunk(
+                    frames_dir, cidx, len(video_chunks), api_key, vision_models)
+                result["video_analysis"].append({
+                    "chunk_index": cidx,
+                    "chunk_url": hf_download_url(remote),
+                    "analysis": analysis or "ANALYSIS_FAILED",
+
+                    "frame_count": len(frames),
+                })
+            safe_cleanup(frames_dir)
+
+        # ── 4. Extract audio ─────────────────────────────────────
+        log.info("Step 4: Extracting audio...")
+        audio_path = os.path.join(vid_dir, f"{video_id}.mp3")
+        audio_ok = extract_audio(video_path, audio_path)
+        # We no longer delete the original video here so it can be stored in the scalable directory system
+        # safe_cleanup(video_path, dl_dir)
+
+        if audio_ok and os.path.exists(audio_path):
+            audio_dur = media_duration(audio_path)
+            log.info(f"Audio duration: {audio_dur:.0f}s")
+
+            # ── 5. Split audio if > 3 min ────────────────────────
+            if audio_dur and audio_dur > AUDIO_SPLIT_SEC:
+                log.info("Step 5: Splitting audio (>3min)...")
+                audio_chunks = split_audio(audio_path, AUDIO_SPLIT_SEC)
+                if audio_chunks and audio_chunks[0] != audio_path:
+                    safe_cleanup(audio_path)
+            else:
+                audio_chunks = [audio_path]
+            log.info(f"Audio chunks: {len(audio_chunks)}")
+
+            # ── 6. Transcribe + analyze each audio chunk ─────────
+            log.info("Step 6: Transcribing + analyzing audio...")
+            all_transcript = []
+            for aidx, achunk in enumerate(audio_chunks):
+                log.info(f"  Audio chunk {aidx+1}/{len(audio_chunks)}")
+                transcript = transcribe_audio_chunk(achunk)
+                if transcript:
+                    all_transcript.append(transcript)
+                    analysis = analyze_audio_transcript(
+                        transcript, aidx, len(audio_chunks), api_key)
+                    result["audio_analysis"].append({
+                        "chunk_index": aidx,
+                        "transcript": transcript,
+                        "analysis": analysis or "ANALYSIS_FAILED",
+                    })
+                safe_cleanup(achunk)
+            result["audio_transcript"] = "\n\n".join(all_transcript)
+        else:
+            result["errors"].append("Audio extraction failed")
+
+        # ── 7. Mark success and Store in Scalable Directory ──────────
+        if result["video_analysis"] or result["audio_analysis"]:
+            result["status"] = "done"
+        log.info(f"Video {video_id} → {result['status']}")
+
+        # Move to scalable directory system
+        channel_id = result.get("metadata", {}).get("channel_id", "unknown_channel")
+        store_dir = os.path.join(os.path.dirname(work_dir), "video_store", channel_id, video_id)
+        os.makedirs(store_dir, exist_ok=True)
+        
+        # Copy original video and yt-dlp metadata to store_dir
+        if os.path.exists(dl_dir):
+            for f in glob.glob(os.path.join(dl_dir, "*")):
+                shutil.copy2(f, store_dir)
+        
+        # Save AI generated descriptions
+        ai_desc_path = os.path.join(store_dir, "ai_description.json")
+        with open(ai_desc_path, "w", encoding="utf-8") as f:
+            json.dump({
+                "video_analysis": result.get("video_analysis", []),
+                "audio_analysis": result.get("audio_analysis", []),
+                "audio_transcript": result.get("audio_transcript", "")
+            }, f, indent=2, ensure_ascii=False)
+
+    except Exception as e:
+        log.error(f"Pipeline failed for {video_id}: {e}")
+        result = None
+    finally:
+        safe_cleanup(vid_dir)
+        # clean HF staging
+        try:
+            # We skip HF deletion if we actually want to keep it on HF as requested
+            # hf_delete_folder(staging_prefix, hf_token)
+            pass
+        except Exception:
+            pass
+
+    return result
+
+# ── results → parquet + upload ───────────────────────────────────────
+def save_and_upload(results, worker_id, hf_token, work_dir):
+    """Save results as parquet and upload to HF dataset."""
+    import pyarrow as pa
+    import pyarrow.parquet as pq
+
+    if not results:
+        log.warning("No results to save")
+        return False
+
+    rows = []
+    for r in results:
+        rows.append({
+            "video_id": r["video_id"],
+            "url": r["url"],
+            "title": r.get("metadata",{}).get("title",""),
+            "channel": r.get("metadata",{}).get("channel",""),
+            "duration": r.get("metadata",{}).get("duration",0),
+            "view_count": r.get("metadata",{}).get("view_count",0),
+            "upload_date": r.get("metadata",{}).get("upload_date",""),
+            "description": r.get("metadata",{}).get("description",""),
+            "video_analysis": json.dumps(r.get("video_analysis",[])),
+            "audio_transcript": r.get("audio_transcript",""),
+            "audio_analysis": json.dumps(r.get("audio_analysis",[])),
+            "hf_video_urls": json.dumps(r.get("hf_video_urls",[])),
+            "status": r["status"],
+            "errors": json.dumps(r.get("errors",[])),
+            "worker_id": r["worker_id"],
+            "processed_at": r["processed_at"],
+        })
+
+    table = pa.table({k: [row[k] for row in rows] for k in rows[0]})
+    ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
+    fname = f"{worker_id}_{ts}.parquet"
+    local_pq = os.path.join(work_dir, fname)
+    pq.write_table(table, local_pq, compression="zstd")
+    log.info(f"Saved {len(rows)} rows → {fname} ({os.path.getsize(local_pq)/1024:.0f} KB)")
+
+    remote = f"{DATA_PREFIX}/{fname}"
+    ok = hf_upload_file(local_pq, remote, hf_token)
+    safe_cleanup(local_pq)
+    if ok:
+        log.info(f"✅ Uploaded to HF: {remote}")
+    else:
+        log.error("❌ Failed to upload parquet")
+    return ok
+
+# ── status reporting ─────────────────────────────────────────────────
+def report_status(worker_id, status_data, hf_token, work_dir):
+    fp = os.path.join(work_dir, "status.json")
+    with open(fp, "w") as f:
+        json.dump(status_data, f, indent=2)
+    hf_upload_file(fp, f"status/{worker_id}.json", hf_token)
+    safe_cleanup(fp)
+
+# ── main entry ───────────────────────────────────────────────────────
+def run_worker(config):
+    """
+    config = {
+        "worker_id": "w-abc123",
+        "videos": [{"video_id": "xxx", "url": "https://..."}],
+        "api_keys": ["key1", "key2", ...],
+        "hf_token": "hf_xxx",
+        "vision_model": "Qwen/Qwen3-VL-32B-Instruct",
+        "text_model": "Qwen/Qwen3-32B",
+    }
+    """
+    worker_id = config["worker_id"]
+    videos    = config["videos"]
+    api_keys  = config["api_keys"]
+    hf_token  = config["hf_token"]
+    vision_models = config.get("vision_models")
+
+    work_dir  = os.path.join(
+        os.environ.get("KAGGLE_WORKING_DIR", "/kaggle/working"), "yt_pipeline")
+    os.makedirs(work_dir, exist_ok=True)
+
+    log.info(f"Worker {worker_id} starting — {len(videos)} videos, "
+             f"{len(api_keys)} API keys, disk free: {disk_free_gb():.1f}GB")
+
+    results = []
+    status  = {
+        "worker_id": worker_id,
+        "total": len(videos),
+        "done": 0, "failed": 0,
+        "started_at": datetime.now(timezone.utc).isoformat(),
+        "state": "running",
+        "video_status": {},
+    }
+
+    for vidx, vinfo in enumerate(videos):
+        vid_id  = vinfo["video_id"]
+        vid_url = vinfo["url"]
+        api_key = api_keys[vidx % len(api_keys)]  # round-robin
+
+        log.info(f"\n[{vidx+1}/{len(videos)}] {vid_id}")
+
+        # check disk
+        if disk_free_gb() < DISK_HEADROOM_GB:
+            log.error("Disk too full, stopping early")
+            status["state"] = "disk_full"
+            break
+
+        result = process_video(vid_url, vid_id, api_key, hf_token,
+                               worker_id, work_dir, vision_models=vision_models)
+
+        if result is None:
+            status["failed"] += 1
+            status["video_status"][vid_id] = "failed"
+        else:
+            results.append(result)
+            status["video_status"][vid_id] = result["status"]
+            if result["status"] == "done":
+                status["done"] += 1
+            else:
+                status["failed"] += 1
+
+        # periodic save + status report
+        if (vidx + 1) % 10 == 0 or vidx == len(videos) - 1:
+            save_and_upload(results, worker_id, hf_token, work_dir)
+            results = []  # reset batch
+            status["last_update"] = datetime.now(timezone.utc).isoformat()
+            report_status(worker_id, status, hf_token, work_dir)
+
+    # final save
+    if results:
+        save_and_upload(results, worker_id, hf_token, work_dir)
+
+    status["state"] = "completed"
+    status["finished_at"] = datetime.now(timezone.utc).isoformat()
+    report_status(worker_id, status, hf_token, work_dir)
+    log.info(f"Worker {worker_id} finished: {status['done']} done, {status['failed']} failed")
+    return status
+
+
+def _is_notebook():
+    """Detect if running inside Jupyter/Kaggle notebook."""
+    try:
+        return "ipykernel" in sys.modules or any("-f" in a for a in sys.argv)
+    except Exception:
+        return False
+
+if __name__ == "__main__" and not _is_notebook():
+    # For local testing only — skipped on Kaggle
+    if len(sys.argv) > 1:
+        with open(sys.argv[1]) as f:
+            cfg = json.load(f)
+        run_worker(cfg)
+    else:
+        print("Usage: python worker.py config.json")

worker/db_worker/get_apis.py

@@ -0,0 +1,23 @@
+import sqlite3
+def print_db_data(db_path):
+    # Connect to the SQLite database
+    conn = sqlite3.connect(db_path)
+    cursor = conn.cursor()
+    
+    try:
+        # Execute a query to read standard data columns
+        cursor.execute("SELECT id, email, api_key, created_at FROM accounts")
+        rows = cursor.fetchall()
+        
+        print(f"Found {len(rows)} records:")
+        for row in rows:
+            print(f"ID: {row[0]}, Email: {row[1]}, API Key: {row[2]}, Created At: {row[3]}")
+
+    except sqlite3.Error as e:
+        print(f"An error occurred: {e}")
+    finally:
+        # Always ensure the connection is closed
+        conn.close()
+  
+if __name__ == "__main__":
+    d = print_db_data('db/accounts.db')

worker/db_worker/get_links.py

@@ -0,0 +1,8 @@
+import sqlite3
+
+conn = sqlite3.connect('db/channel_links.db')
+
+cursor = conn.execute('SELECT * FROM videos')
+for row in cursor:
+    print(row)
+conn.close()

worker/hf_uploader.py

@@ -0,0 +1,74 @@
+import os
+import sys
+import argparse
+from huggingface_hub import HfApi, login
+
+def upload_to_hf(local_path, repo_id, repo_type="dataset", path_in_repo=None, token=None):
+    if token:
+        login(token=token)
+    elif not os.environ.get("HF_TOKEN"):
+        print("Warning: No HF_TOKEN found. Upload might fail if the repo is private or requires authentication.")
+        
+    api = HfApi()
+    
+    if not os.path.exists(local_path):
+        print(f"Error: Path does not exist: {local_path}", file=sys.stderr)
+        sys.exit(1)
+
+    # Ensure repo exists
+    try:
+        api.create_repo(repo_id=repo_id, repo_type=repo_type, exist_ok=True)
+        print(f"Verified repository: {repo_type}s/{repo_id}")
+    except Exception as e:
+        print(f"Note: Could not verify or create repo (it might already exist or there's a permission issue). Continuing... ({e})")
+
+    if os.path.isfile(local_path):
+        if not path_in_repo:
+            path_in_repo = os.path.basename(local_path)
+        print(f"Uploading file '{local_path}' to 'hf://{repo_type}s/{repo_id}/{path_in_repo}'...")
+        try:
+            api.upload_file(
+                path_or_fileobj=local_path,
+                path_in_repo=path_in_repo,
+                repo_id=repo_id,
+                repo_type=repo_type
+            )
+            print("✅ File upload complete!")
+        except Exception as e:
+            print(f"❌ Failed to upload file: {e}", file=sys.stderr)
+            
+    elif os.path.isdir(local_path):
+        if path_in_repo is None:
+            path_in_repo = "" # Root of repo by default
+        print(f"Syncing folder '{local_path}' to 'hf://{repo_type}s/{repo_id}/{path_in_repo}'...")
+        try:
+            api.upload_folder(
+                folder_path=local_path,
+                path_in_repo=path_in_repo,
+                repo_id=repo_id,
+                repo_type=repo_type
+            )
+            print("✅ Folder sync complete!")
+        except Exception as e:
+            print(f"❌ Failed to sync folder: {e}", file=sys.stderr)
+
+def main():
+    parser = argparse.ArgumentParser(description="Upload any video, file, or folder to Hugging Face")
+    parser.add_argument("input_path", type=str, help="Path to the local video file or directory (e.g. ./data or video.mp4)")
+    parser.add_argument("--repo", type=str, default="AdhyanshVerma/YT", help="Hugging Face repo ID (default: AdhyanshVerma/YT)")
+    parser.add_argument("--repo-type", type=str, default="dataset", choices=["dataset", "model", "space"], help="Type of HF repo (default: dataset)")
+    parser.add_argument("--path-in-repo", type=str, default=None, help="Destination path in the repo (defaults to filename or root of repo)")
+    parser.add_argument("--token", type=str, default=os.environ.get("HF_TOKEN"), help="HF Access Token (or set HF_TOKEN env var)")
+    
+    args = parser.parse_args()
+    
+    upload_to_hf(
+        local_path=args.input_path,
+        repo_id=args.repo,
+        repo_type=args.repo_type,
+        path_in_repo=args.path_in_repo,
+        token=args.token
+    )
+
+if __name__ == "__main__":
+    main()

worker/models_worker/call_model.py

@@ -0,0 +1,117 @@
+import os
+import sys
+import base64
+import argparse
+from pathlib import Path
+from openai import OpenAI
+
+def encode_image(image_path: str) -> str:
+    """Read a local image file and return its base64 data URI."""
+    path = Path(image_path)
+    if not path.is_file():
+        raise FileNotFoundError(f"Image file not found: {image_path}")
+    
+    # Determine MIME type from file extension
+    suffix = path.suffix.lower()
+    mime_type = {
+        ".jpg": "image/jpeg",
+        ".jpeg": "image/jpeg",
+        ".png": "image/png",
+        ".gif": "image/gif",
+        ".webp": "image/webp",
+    }.get(suffix, "application/octet-stream")
+    
+    with open(path, "rb") as f:
+        image_data = base64.b64encode(f.read()).decode("utf-8")
+    return f"data:{mime_type};base64,{image_data}"
+
+def describe_image(
+    image_source: str,
+    model: str = "MiniMaxAI/MiniMax-M3",
+    api_key: str = None,
+    base_url: str = "https://api.featherless.ai/v1",
+) -> str:
+    """
+    Send an image (URL or local path) to the Featherless AI vision API
+    and return the generated description.
+    """
+    # Use provided API key or fall back to environment variable
+    api_key = api_key or os.getenv("FEATHERLESS_API_KEY")
+    if not api_key:
+        raise ValueError("API key must be provided or set in FEATHERLESS_API_KEY env var")
+
+    client = OpenAI(base_url=base_url, api_key=api_key)
+
+    # Determine if image_source is a URL or a local file
+    if image_source.startswith(("http://", "https://")):
+        image_url = image_source
+    else:
+        # Assume it's a local file; encode to data URI
+        image_url = encode_image(image_source)
+
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "Describe the image in one paragraph."},
+                {"type": "image_url", "image_url": {"url": image_url}},
+            ],
+        }
+    ]
+
+    try:
+        response = client.chat.completions.create(
+            model=model,
+            messages=messages,
+            # You can add more parameters, e.g., temperature, max_tokens
+        )
+        return response.choices[0].message.content
+    except Exception as e:
+        # Catch and re-raise with a user-friendly message
+        raise RuntimeError(f"API call failed: {e}") from e
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Describe an image using Featherless AI vision models."
+    )
+    parser.add_argument(
+        "image",
+        help="Image URL or local file path.",
+    )
+    parser.add_argument(
+        "--model",
+        default="MiniMaxAI/MiniMax-M3",
+        help="Model name (default: MiniMaxAI/MiniMax-M3).",
+    )
+    parser.add_argument(
+        "--api-key",
+        help="Featherless API key (overrides FEATHERLESS_API_KEY env var).",
+    )
+    parser.add_argument(
+        "--base-url",
+        default="https://api.featherless.ai/v1",
+        help="Base URL for the API (default: https://api.featherless.ai/v1).",
+    )
+    args = parser.parse_args()
+
+    try:
+        description = describe_image(
+            image_source=args.image,
+            model=args.model,
+            api_key=args.api_key,
+            base_url=args.base_url,
+        )
+        print(description)
+    except Exception as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+
+if __name__ == "__main__":
+    main()
+
+
+
+
+"""
+Bro, sif you are here then please add ahere video support and make this as a function as a whole!
+"""

worker/models_worker/video_worker.py

@@ -0,0 +1,87 @@
+import cv2
+import base64
+import os
+import argparse
+import sys
+from openai import OpenAI
+
+def extract_frames(video_path, num_frames=5):
+    """Extract a few evenly spaced frames from a video and encode to base64."""
+    cap = cv2.VideoCapture(video_path)
+    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+    frames = []
+    
+    if total_frames == 0:
+        return frames
+        
+    step = max(1, total_frames // num_frames)
+    
+    for i in range(num_frames):
+        cap.set(cv2.CAP_PROP_POS_FRAMES, min(i * step, total_frames - 1))
+        ret, frame = cap.read()
+        if ret:
+            # encode to jpeg
+            _, buffer = cv2.imencode('.jpg', frame)
+            b64_img = base64.b64encode(buffer).decode('utf-8')
+            frames.append(f"data:image/jpeg;base64,{b64_img}")
+            
+    cap.release()
+    return frames
+
+def analyze_video(video_source, prompt, api_key=None, model="MiniMaxAI/MiniMax-M3", base_url="https://api.featherless.ai/v1"):
+    api_key = api_key or os.getenv("FEATHERLESS_API_KEY")
+    if not api_key:
+        raise ValueError("API key must be provided or set in FEATHERLESS_API_KEY env var")
+        
+    client = OpenAI(base_url=base_url, api_key=api_key)
+    
+    print(f"[+] Extracting frames from {video_source}...")
+    frames = extract_frames(video_source, num_frames=5)
+    
+    if not frames:
+        raise RuntimeError("Failed to extract any frames from the video.")
+    
+    content = [{"type": "text", "text": prompt}]
+    for frame in frames:
+        content.append({"type": "image_url", "image_url": {"url": frame}})
+        
+    messages = [
+        {
+            "role": "user",
+            "content": content
+        }
+    ]
+    
+    print(f"[+] Sending {len(frames)} frames to vision model...")
+    response = client.chat.completions.create(
+        model=model,
+        messages=messages
+    )
+    return response.choices[0].message.content
+
+def main():
+    parser = argparse.ArgumentParser(description="Analyze video using Featherless AI vision models.")
+    parser.add_argument("--video-url", required=True, help="Video file path or URL.")
+    parser.add_argument("--prompt", required=True, help="Prompt for analysis.")
+    parser.add_argument("--api-key", help="Featherless API key (overrides FEATHERLESS_API_KEY env var).")
+    parser.add_argument("--model", default="MiniMaxAI/MiniMax-M3", help="Model name.")
+    parser.add_argument("--base-url", default="https://api.featherless.ai/v1", help="Base API URL.")
+    args = parser.parse_args()
+
+    try:
+        description = analyze_video(
+            video_source=args.video_url,
+            prompt=args.prompt,
+            api_key=args.api_key,
+            model=args.model,
+            base_url=args.base_url,
+        )
+        print("\n--- Video Analysis ---\n")
+        print(description)
+        print("\n----------------------\n")
+    except Exception as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+
+if __name__ == "__main__":
+    main()

worker/youtube_worker/audio_downloader.py

@@ -0,0 +1,67 @@
+import os
+import yt_dlp
+import logging
+
+logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
+
+def download_audio(url, output_path='./', cookies_from_browser=None, audio_format='mp3'):
+    """
+    Downloads the best audio from a given URL and converts it to the specified format.
+    
+    Args:
+        url (str): The URL of the video to extract audio from.
+        output_path (str): The directory to save the audio.
+        cookies_from_browser (str): Browser name to extract cookies from (e.g. 'chrome').
+        audio_format (str): Desired audio format (mp3, wav, m4a). Default is 'mp3'.
+    """
+    ydl_opts = {
+        'format': 'bestaudio/best',
+        'outtmpl': os.path.join(output_path, '%(title)s.%(ext)s'),
+        'postprocessors': [{
+            'key': 'FFmpegExtractAudio',
+            'preferredcodec': audio_format,
+            'preferredquality': '192',
+        }],
+        'quiet': False,
+        'no_warnings': False,
+        'ignoreerrors': True,
+        'geo_bypass': True,
+        'sleep_interval_requests': 1,
+    }
+    
+    if cookies_from_browser:
+        ydl_opts['cookiesfrombrowser'] = (cookies_from_browser, )
+    elif os.path.exists('cookies.txt'):
+        ydl_opts['cookiefile'] = 'cookies.txt'
+        
+    try:
+        logging.info(f"Starting audio download for {url}")
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            error_code = ydl.download([url])
+            if error_code == 0:
+                logging.info(f"Successfully downloaded audio to {output_path}")
+                return True
+            else:
+                logging.error(f"yt-dlp encountered an error (code {error_code}).")
+                return False
+    except yt_dlp.utils.DownloadError as e:
+        logging.error(f"Download error: {e}")
+        return False
+    except Exception as e:
+        logging.error(f"Unexpected error: {e}")
+        return False
+
+if __name__ == "__main__":
+    import argparse
+    parser = argparse.ArgumentParser(description="Download audio using yt-dlp.")
+    parser.add_argument("url", help="URL of the video")
+    parser.add_argument("--output", "-o", default="./", help="Output directory")
+    parser.add_argument("--format", "-f", choices=["mp3", "wav", "m4a"], default="mp3", help="Audio format")
+    parser.add_argument("--cookies", "-c", help="Browser to extract cookies from (e.g. chrome, firefox)")
+    
+    args = parser.parse_args()
+    
+    if not os.path.exists(args.output):
+        os.makedirs(args.output, exist_ok=True)
+        
+    download_audio(args.url, args.output, args.cookies, args.format)

worker/youtube_worker/audio_worker/audio_splitter.py

@@ -0,0 +1,153 @@
+import os
+import subprocess
+import argparse
+import logging
+import re
+
+logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
+
+def get_audio_duration(input_path):
+    try:
+        result = subprocess.run(
+            ["ffprobe", "-v", "error", "-show_entries",
+             "format=duration", "-of",
+             "default=noprint_wrappers=1:nokey=1", input_path],
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=True
+        )
+        return float(result.stdout.strip())
+    except:
+        return None
+
+def detect_silences(input_path, noise_level="-20dB", duration=0.3):
+    """
+    Runs ffmpeg silencedetect to find natural pauses in the audio.
+    Returns a list of timestamps (middle of the silence).
+    """
+    logging.info("Analyzing audio to find silences (avoiding mid-sentence cuts)...")
+    command = [
+        "ffmpeg", "-i", input_path,
+        "-af", f"silencedetect=noise={noise_level}:d={duration}",
+        "-f", "null", "-"
+    ]
+    
+    result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+    if result.returncode != 0:
+        logging.error(f"Silence detection failed: {result.stderr}")
+        return []
+    output = result.stderr
+    
+    silence_starts = []
+    silence_ends = []
+    for line in output.split('\n'):
+        if "silence_start" in line:
+            match = re.search(r"silence_start:\s+([\d\.]+)", line)
+            if match:
+                silence_starts.append(float(match.group(1)))
+        elif "silence_end" in line:
+            match = re.search(r"silence_end:\s+([\d\.]+)", line)
+            if match:
+                silence_ends.append(float(match.group(1)))
+                
+    silence_points = []
+    for s, e in zip(silence_starts, silence_ends):
+        # Calculate the midpoint of the silence to use as the cut point
+        silence_points.append((s + e) / 2.0)
+        
+    return silence_points
+
+def calculate_smart_audio_splits(total_duration, silence_points, target_chunk=60, min_tail=15):
+    """
+    Finds split points close to the target chunk size, prioritizing silences
+    and avoiding tiny fragments at the end.
+    """
+    split_points = []
+    current_target = target_chunk
+    
+    while current_target < total_duration:
+        if (total_duration - current_target) < min_tail:
+            logging.info(f"Remaining {total_duration - current_target:.1f}s is too short, merging with the last clip.")
+            break
+            
+        # Search for silences within +/- 15 seconds of our target time
+        valid_silences = [p for p in silence_points if abs(p - current_target) <= 15]
+        
+        if not valid_silences and silence_points:
+            last_split = split_points[-1] if split_points else 0
+            valid_silences = [p for p in silence_points if p > last_split + 5]
+            
+        if valid_silences:
+            # Pick the silence closest to the target
+            best_point = min(valid_silences, key=lambda p: abs(p - current_target))
+            split_points.append(best_point)
+            logging.info(f"Found natural silence at {best_point:.1f}s")
+            current_target = best_point + target_chunk
+        else:
+            # Hard cut if no silence is found near the target
+            logging.warning(f"No silence found near {current_target}s, falling back to exact cut.")
+            split_points.append(current_target)
+            current_target += target_chunk
+            
+    return split_points
+
+def split_audio(input_path, chunk_duration=60, output_dir=None):
+    if not os.path.exists(input_path):
+        logging.error(f"Input audio '{input_path}' not found.")
+        return False
+        
+    input_path = os.path.abspath(input_path)
+    filename = os.path.basename(input_path)
+    name, ext = os.path.splitext(filename)
+    name = name.rstrip('.')
+    
+    if output_dir is None:
+        output_dir = os.path.join(os.path.dirname(input_path), f"{name}_chunks")
+        
+    os.makedirs(output_dir, exist_ok=True)
+    output_pattern = os.path.join(output_dir, f"{name}_chunk_%03d{ext}")
+    
+    total_duration = get_audio_duration(input_path)
+    if not total_duration:
+        return False
+        
+    logging.info(f"Total audio duration: {total_duration:.1f}s")
+    
+    silences = detect_silences(input_path)
+    
+    min_tail = max(15, chunk_duration // 4)
+    split_points = calculate_smart_audio_splits(total_duration, silences, target_chunk=chunk_duration, min_tail=min_tail)
+    
+    command = [
+        "ffmpeg",
+        "-i", input_path,
+        "-c", "copy",
+        "-map", "0",
+        "-f", "segment",
+        "-reset_timestamps", "1"
+    ]
+    
+    if split_points:
+        times_str = ",".join(str(round(p, 3)) for p in split_points)
+        command.extend(["-segment_times", times_str])
+    else:
+        command.extend(["-segment_time", str(total_duration + 10)])
+        
+    command.append(output_pattern)
+    
+    try:
+        logging.info("Executing split...")
+        subprocess.run(command, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
+        logging.info("Successfully split the audio!")
+        return True
+    except subprocess.CalledProcessError as e:
+        logging.error("Error during audio splitting:")
+        logging.error(e.stderr.decode('utf-8', errors='replace'))
+        return False
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Smart split audio into chunks.")
+    parser.add_argument("audio_path", help="Path to the audio file")
+    parser.add_argument("--duration", "-d", type=int, default=60, help="Target duration (default: 60)")
+    parser.add_argument("--output", "-o", help="Output directory")
+    args = parser.parse_args()
+    
+    split_audio(args.audio_path, chunk_duration=args.duration, output_dir=args.output)

worker/youtube_worker/downloader.py

@@ -0,0 +1,98 @@
+import os
+import yt_dlp
+import logging
+
+logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
+
+def download_video(url, resolution='HD', output_path='./', cookies_from_browser=None):
+    """
+    Download a video from a given URL using yt-dlp at a specified resolution.
+
+    Args:
+        url (str): The URL of the video to download.
+        resolution (str): The desired resolution. Options: '4K', '2K', 'HD'. Default is 'HD'.
+        output_path (str): The directory to save the video. Default is current directory.
+        cookies_from_browser (str): Browser to extract cookies from (e.g., 'chrome', 'firefox', 'edge').
+                                    Use this if you encounter blocking or age-restricted content.
+
+    Returns:
+        bool: True if successful, False otherwise.
+    """
+    # Map resolutions to their approximate vertical pixel height
+    resolution_map = {
+        '4K': '2160',
+        '2K': '1440',
+        'HD': '1080'
+    }
+    
+    height = resolution_map.get(resolution.upper(), '1080')
+    
+    # We ask for the best video with height <= the specified height,
+    # plus the best audio, and merge them. If that's not available, 
+    # it falls back to the best single file with height <= specified height.
+    format_string = f'bestvideo[height<={height}]+bestaudio/best[height<={height}]'
+    
+    ydl_opts = {
+        'format': format_string,
+        'outtmpl': os.path.join(output_path, '%(title)s.%(ext)s'),
+        'merge_output_format': 'mp4',  # standardizing on mp4 where possible
+        'quiet': False,
+        'no_warnings': False,
+        'nocheckcertificate': True,
+        'ignoreerrors': True, # skip unavailable videos in playlists
+        'geo_bypass': True,   # try to bypass geographic restrictions
+        # Adds sleep to reduce likelihood of getting rate-limited or banned
+        'sleep_interval_requests': 1,
+        'sleep_interval': 1,
+        'max_sleep_interval': 3,
+        # Fake user agent can help avoid bot detection
+        'http_headers': {
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36',
+        }
+    }
+    
+    if cookies_from_browser:
+        ydl_opts['cookiesfrombrowser'] = (cookies_from_browser, )
+    elif os.path.exists('cookies.txt'):
+        ydl_opts['cookiefile'] = 'cookies.txt'
+        
+    try:
+        logging.info(f"Starting download for {url} at max resolution {resolution} ({height}p)")
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=True)
+            if info:
+                filename = None
+                if 'requested_downloads' in info:
+                    filename = info['requested_downloads'][0]['filepath']
+                else:
+                    filename = ydl.prepare_filename(info)
+                    if ydl_opts.get('merge_output_format'):
+                        filename = os.path.splitext(filename)[0] + '.' + ydl_opts['merge_output_format']
+                
+                logging.info(f"Successfully downloaded to {filename}")
+                print(os.path.abspath(filename))
+                return True
+            else:
+                logging.error(f"yt-dlp encountered an error.")
+                return False
+    except yt_dlp.utils.DownloadError as e:
+        logging.error(f"Download error: {e}")
+        return False
+    except Exception as e:
+        logging.error(f"Unexpected error: {e}")
+        return False
+
+if __name__ == "__main__":
+    import argparse
+    parser = argparse.ArgumentParser(description="Download videos using yt-dlp.")
+    parser.add_argument("url", help="URL of the video to download")
+    parser.add_argument("--resolution", "-r", choices=["4K", "2K", "HD"], default="HD", help="Target resolution (4K, 2K, HD). Default: HD")
+    parser.add_argument("--output", "-o", default="./", help="Output directory path. Default: current directory")
+    parser.add_argument("--cookies", "-c", help="Browser to extract cookies from (e.g. chrome, firefox, edge) to prevent being blocked.")
+    
+    args = parser.parse_args()
+    
+    if not os.path.exists(args.output):
+        os.makedirs(args.output, exist_ok=True)
+        
+    download_video(args.url, args.resolution, args.output, args.cookies)

worker/youtube_worker/video_worker/splitter.py

@@ -0,0 +1,108 @@
+import os
+import subprocess
+import argparse
+import logging
+
+logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
+
+def get_video_duration(input_path):
+    try:
+        result = subprocess.run(
+            ["ffprobe", "-v", "error", "-show_entries",
+             "format=duration", "-of",
+             "default=noprint_wrappers=1:nokey=1", input_path],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            check=True
+        )
+        return float(result.stdout.strip())
+    except Exception as e:
+        logging.error(f"Failed to get duration: {e}")
+        return None
+
+def calculate_split_points(total_duration, chunk_duration=60, min_tail_duration=15):
+    """
+    Calculator system that prevents tiny clips at the end by merging them.
+    If the remaining clip is less than min_tail_duration, it won't split,
+    merging it into the previous clip (making it slightly longer).
+    """
+    split_points = []
+    current_time = chunk_duration
+    while current_time < total_duration:
+        # Check if the remaining part is too small
+        if (total_duration - current_time) < min_tail_duration:
+            logging.info(f"Remaining {total_duration - current_time:.1f}s is too short, merging with the last clip.")
+            break
+        split_points.append(current_time)
+        current_time += chunk_duration
+    return split_points
+
+def split_video(input_path, chunk_duration=60, output_dir=None):
+    """
+    Splits a video into smaller chunks of a specified duration using ffmpeg,
+    with smart distribution logic.
+    """
+    if not os.path.exists(input_path):
+        logging.error(f"Input video '{input_path}' not found.")
+        return False
+        
+    input_path = os.path.abspath(input_path)
+    filename = os.path.basename(input_path)
+    name, ext = os.path.splitext(filename)
+    name = name.rstrip('.')
+    
+    if output_dir is None:
+        output_dir = os.path.join(os.path.dirname(input_path), f"{name}_chunks")
+        
+    os.makedirs(output_dir, exist_ok=True)
+    output_pattern = os.path.join(output_dir, f"{name}_chunk_%03d{ext}")
+    
+    total_duration = get_video_duration(input_path)
+    if total_duration is None:
+        return False
+        
+    logging.info(f"Total video duration: {total_duration:.1f}s")
+    
+    # Calculate the smart split points
+    # Min tail is a quarter of chunk size or at least 15s
+    min_tail = max(15, chunk_duration // 4) 
+    split_points = calculate_split_points(total_duration, chunk_duration, min_tail_duration=min_tail)
+    
+    logging.info(f"Calculated split points: {[round(p, 1) for p in split_points]}")
+    
+    command = [
+        "ffmpeg",
+        "-i", input_path,
+        "-map", "0",
+        "-f", "segment",
+        "-reset_timestamps", "1"
+    ]
+    
+    if split_points:
+        times_str = ",".join(str(round(p, 3)) for p in split_points)
+        command.extend(["-segment_times", times_str])
+    else:
+        # If no split points (e.g. video is too short), just use segment_time 
+        # effectively keeping it whole or just standard copy
+        command.extend(["-segment_time", str(total_duration + 10)])
+        
+    command.append(output_pattern)
+    
+    try:
+        subprocess.run(command, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
+        logging.info("Successfully split the video!")
+        return True
+    except subprocess.CalledProcessError as e:
+        logging.error("Error during video splitting:")
+        logging.error(e.stderr.decode('utf-8', errors='replace'))
+        return False
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Smart split video into chunks.")
+    parser.add_argument("video_path", help="Path to the video file")
+    parser.add_argument("--duration", "-d", type=int, default=60, help="Target duration (default: 60)")
+    parser.add_argument("--output", "-o", help="Output directory")
+    args = parser.parse_args()
+    
+    split_video(args.video_path, chunk_duration=args.duration, output_dir=args.output)