Become a Download Agent

#!/usr/bin/env python3
"""
FreeTheVideo Download Agent — earn sats by helping download videos.

Run this on your home machine. It polls the FreeTheVideo server for download
jobs, uses yt-dlp to download videos locally (your IP, your cookies), then
uploads them back to the server. You earn sats for each completed download.

Setup:
    pip install requests yt-dlp

Usage:
    python ftv-agent.py --server https://freethevideo.com --token ftv-abc123...
    python ftv-agent.py --once            # grab one job and exit
    python ftv-agent.py --stats           # show your stats

Config file (~/.ftv-agent.json):
    {"server": "https://freethevideo.com", "token": "ftv-abc123..."}
"""

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path

try:
    import requests
except ImportError:
    print("Missing dependency: pip install requests")
    sys.exit(1)

# --- Config ---

CONFIG_FILE = Path.home() / ".ftv-agent.json"
COOKIES_FILE = Path.home() / ".ftv-agent-cookies.txt"  # optional Netscape cookies
POLL_INTERVAL = 30  # seconds between polls
MAX_FILE_SIZE = 500 * 1024 * 1024  # 500MB
YTDLP_TIMEOUT = 600  # 10 min max per download

# Desktop UA pool — used for everything except Instagram (mobile UA) and TikTok (default).
_USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:133.0) Gecko/20100101 Firefox/133.0",
]
_INSTAGRAM_UA = (
    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 "
    "(KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1"
)

# Hosts this agent is willing to download from. A compromised or spoofed server
# must NOT be able to push an arbitrary URL — or a yt-dlp option disguised as one
# (e.g. "--exec=<cmd>") — to run against the volunteer's machine or LAN.
SUPPORTED_DOMAINS = {
    "instagram.com", "www.instagram.com",
    "tiktok.com", "www.tiktok.com", "vm.tiktok.com",
    "youtube.com", "www.youtube.com", "youtu.be", "m.youtube.com",
    "x.com", "twitter.com", "www.twitter.com",
    "facebook.com", "www.facebook.com", "m.facebook.com",
    "fb.watch", "www.fb.watch",
}


def _has_curl_cffi() -> bool:
    try:
        import curl_cffi  # noqa: F401
        return True
    except Exception:
        return False


def build_ytdlp_args(url: str, output_template: str) -> list[str]:
    """Mirror server-side _ytdlp_args with platform-specific tweaks.

    Per CLAUDE.md:
      - Instagram: iPhone UA, cookies if available, --ignore-errors for carousels.
      - Facebook: default desktop UA (NOT iPhone — breaks parsing).
      - TikTok: default UA + --impersonate (requires curl_cffi).
      - YouTube: EJS solver + concurrent fragments.
      - X/Twitter: default settings.
    """
    import random
    from urllib.parse import urlparse

    ytdlp = find_ytdlp()
    host = (urlparse(url).hostname or "").lower()

    # Reject anything that isn't a plain https URL on a supported host, before it
    # can reach the yt-dlp command line.
    if not url.lower().startswith("https://"):
        raise ValueError(f"Refusing non-https job URL: {url[:80]!r}")
    if not (host in SUPPORTED_DOMAINS or any(host.endswith("." + d) for d in SUPPORTED_DOMAINS)):
        raise ValueError(f"Refusing job URL for unsupported host: {host!r}")

    args = [
        ytdlp,
        # Prefer H.264 (avc1) — VP9-in-MP4 renders as a black screen on Safari/iOS.
        # Fall through to any codec only when H.264 isn't offered.
        # Avoid filesize-in-format predicate (manifests often lack filesize headers
        # and yt-dlp filters them out before merge).
        "-f", (
            "bv*[vcodec^=avc1][height<=720]+ba[ext=m4a]/"
            "bv*[vcodec^=avc1][height<=720]+ba/"
            "b[vcodec^=avc1][height<=720]/"
            "bv*[height<=720]+ba/b[height<=720]/best"
        ),
        "--merge-output-format", "mp4",
        "--no-playlist",
        "--no-warnings",
        "--no-check-certificates",
        "--socket-timeout", "30",
        "--retries", "3",
        "--max-filesize", "450M",
        "--write-info-json",
        "--write-thumbnail",
        "--convert-thumbnails", "jpg",
        "--ignore-errors",
        "-o", output_template,
    ]

    # Platform-specific user-agent
    if "instagram" in host:
        args.extend(["--user-agent", _INSTAGRAM_UA])
    elif "tiktok" in host:
        # Don't set UA — let yt-dlp + impersonation pick the right fingerprint.
        if _has_curl_cffi():
            args.extend(["--impersonate", "chrome"])
    elif "facebook" in host or "fb.watch" in host:
        # Desktop UA; mobile breaks Facebook parsing.
        args.extend(["--user-agent", random.choice(_USER_AGENTS)])
        args.extend(["--extractor-retries", "3"])
    else:
        args.extend(["--user-agent", random.choice(_USER_AGENTS)])

    # Optional cookies (age-restricted / login-required content)
    if COOKIES_FILE.exists():
        args.extend(["--cookies", str(COOKIES_FILE)])

    # YouTube: EJS solver + concurrent fragments
    if "youtube" in host or "youtu.be" in host:
        args.extend(["--remote-components", "ejs:github"])
        args.extend(["--concurrent-fragments", "4"])

    # End-of-options separator: everything after "--" is treated as a positional
    # URL, so a job "url" like "--exec=<cmd>" can't be parsed as a yt-dlp option.
    args.append("--")
    args.append(url)
    return args

# --- Colors ---

class C:
    RESET = "\033[0m"
    RED = "\033[91m"
    GREEN = "\033[92m"
    YELLOW = "\033[93m"
    BLUE = "\033[94m"
    CYAN = "\033[96m"
    BOLD = "\033[1m"
    DIM = "\033[2m"


def log(msg, color=C.RESET):
    ts = time.strftime("%H:%M:%S")
    print(f"{C.DIM}{ts}{C.RESET} {color}{msg}{C.RESET}")


def log_ok(msg): log(msg, C.GREEN)
def log_warn(msg): log(msg, C.YELLOW)
def log_err(msg): log(msg, C.RED)
def log_info(msg): log(msg, C.CYAN)


# --- Config loading ---

def load_config(args) -> dict:
    """Load config from args, env vars, or config file."""
    config = {}

    # Config file
    if CONFIG_FILE.exists():
        try:
            config = json.loads(CONFIG_FILE.read_text())
        except Exception:
            pass

    # Env vars override
    if os.environ.get("FTV_SERVER"):
        config["server"] = os.environ["FTV_SERVER"]
    if os.environ.get("FTV_TOKEN"):
        config["token"] = os.environ["FTV_TOKEN"]

    # CLI args override
    if args.server:
        config["server"] = args.server
    if args.token:
        config["token"] = args.token

    server = config.get("server", "").rstrip("/")
    token = config.get("token", "")

    if not server or not token:
        print(f"{C.RED}Missing config. Provide --server and --token, or create {CONFIG_FILE}{C.RESET}")
        print(f"\nExample config file ({CONFIG_FILE}):")
        print(json.dumps({"server": "https://freethevideo.com", "token": "ftv-your-token-here"}, indent=2))
        sys.exit(1)

    return {"server": server, "token": token}


def save_config(server: str, token: str):
    """Save config to file for future use."""
    CONFIG_FILE.write_text(json.dumps({"server": server, "token": token}, indent=2))
    log_ok(f"Config saved to {CONFIG_FILE}")


# --- API calls ---

def api_headers(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


def get_jobs(server: str, token: str) -> list:
    try:
        r = requests.get(f"{server}/api/agent/jobs", headers=api_headers(token), timeout=15)
        if r.status_code == 401:
            log_err("Invalid token. Check your agent token.")
            log_err("Exiting — fix your token and restart.")
            sys.exit(1)
        r.raise_for_status()
        return r.json().get("jobs", [])
    except requests.ConnectionError:
        log_err(f"Cannot connect to {server}")
        return []
    except Exception as e:
        log_err(f"Failed to get jobs: {e}")
        return []


def claim_job(server: str, token: str, job_id: str) -> dict | None:
    try:
        r = requests.post(f"{server}/api/agent/claim/{job_id}", headers=api_headers(token), timeout=15)
        if r.status_code == 409:
            log_warn(f"Job {job_id[:8]} already claimed by another agent")
            return None
        r.raise_for_status()
        return r.json()
    except Exception as e:
        log_err(f"Failed to claim job: {e}")
        return None


# Cloudflare caps proxied request bodies at ~100MB, so anything bigger is uploaded
# in ordered <100MB chunks that the server reassembles.
CHUNK_THRESHOLD = 90 * 1024 * 1024
CHUNK_SIZE = 90 * 1024 * 1024


def upload_video(server: str, token: str, job_id: str, filepath: Path, metadata: dict, thumb_path: Path | None = None) -> dict | None:
    try:
        size = filepath.stat().st_size
    except Exception:
        size = 0
    if size > CHUNK_THRESHOLD:
        return _upload_chunked(server, token, job_id, filepath, metadata, thumb_path, size)
    open_handles = []
    try:
        vf = open(filepath, "rb")
        open_handles.append(vf)
        files = {"video": (filepath.name, vf, "video/mp4")}
        if thumb_path and thumb_path.exists():
            tf = open(thumb_path, "rb")
            open_handles.append(tf)
            files["thumbnail"] = (thumb_path.name, tf, "image/jpeg")
        r = requests.post(
            f"{server}/api/agent/upload/{job_id}",
            headers=api_headers(token),
            files=files,
            data={"metadata": json.dumps(metadata)},
            timeout=300,
        )
        r.raise_for_status()
        return r.json()
    except Exception as e:
        log_err(f"Failed to upload: {e}")
        return None
    finally:
        for h in open_handles:
            h.close()


def _upload_chunked(server: str, token: str, job_id: str, filepath: Path, metadata: dict, thumb_path: Path | None, size: int) -> dict | None:
    """Upload a large video in ordered chunks (server reassembles on the final chunk)."""
    upload_id = os.urandom(12).hex()
    total = (size + CHUNK_SIZE - 1) // CHUNK_SIZE
    log_info(f"Uploading {size / 1024 / 1024:.0f}MB in {total} chunks (large file)...")
    try:
        with open(filepath, "rb") as vf:
            for idx in range(total):
                buf = vf.read(CHUNK_SIZE)
                data = {
                    "upload_id": upload_id,
                    "chunk_index": str(idx),
                    "total_chunks": str(total),
                    "filename": filepath.name,
                }
                files = {"chunk": (filepath.name, buf, "application/octet-stream")}
                thandle = None
                if idx == total - 1:
                    data["metadata"] = json.dumps(metadata)
                    if thumb_path and thumb_path.exists():
                        thandle = open(thumb_path, "rb")
                        files["thumbnail"] = (thumb_path.name, thandle, "image/jpeg")
                try:
                    r = requests.post(
                        f"{server}/api/agent/upload-chunk/{job_id}",
                        headers=api_headers(token),
                        files=files,
                        data=data,
                        timeout=300,
                    )
                    r.raise_for_status()
                finally:
                    if thandle:
                        thandle.close()
                if idx == total - 1:
                    return r.json()
                log_info(f"  chunk {idx + 1}/{total} uploaded")
        return None
    except Exception as e:
        log_err(f"Chunked upload failed: {e}")
        return None


def report_failure(server: str, token: str, job_id: str, error: str) -> bool:
    """Report a download failure to the server so the job is released immediately."""
    try:
        r = requests.post(
            f"{server}/api/agent/fail/{job_id}",
            headers={**api_headers(token), "Content-Type": "application/json"},
            json={"error": error[:300]},
            timeout=15,
        )
        if r.status_code == 200:
            data = r.json()
            log_warn(f"Reported failure (attempt {data.get('attempts', '?')}/{data.get('max_attempts', '?')})")
            return True
        log_warn(f"Failed to report failure: HTTP {r.status_code}")
    except Exception as e:
        log_warn(f"Failed to report failure: {e}")
    return False


def get_stats(server: str, token: str) -> dict | None:
    try:
        r = requests.get(f"{server}/api/agent/stats", headers=api_headers(token), timeout=15)
        r.raise_for_status()
        return r.json()
    except Exception as e:
        log_err(f"Failed to get stats: {e}")
        return None


def request_payout(server: str, token: str) -> dict | None:
    try:
        r = requests.post(f"{server}/api/agent/payout", headers=api_headers(token), timeout=30)
        return r.json()
    except Exception as e:
        log_err(f"Payout request failed: {e}")
        return None


# --- Download ---

def find_ytdlp() -> str:
    """Find yt-dlp binary."""
    for cmd in ["yt-dlp", "yt_dlp"]:
        if shutil.which(cmd):
            return cmd
    print(f"{C.RED}yt-dlp not found. Install it: pip install yt-dlp{C.RESET}")
    sys.exit(1)


def download_video(url: str, url_hash: str, tmpdir: Path) -> tuple[Path | None, dict, Path | None]:
    """Download a video using yt-dlp. Returns (filepath, metadata, thumb)."""
    # Never trust server-supplied values blindly: url_hash is used in a filesystem
    # path and url goes onto the yt-dlp command line.
    if not re.fullmatch(r"[a-f0-9]{16}", url_hash or ""):
        log_err(f"Refusing job with invalid url_hash: {url_hash!r}")
        return None, {}, None
    output_template = str(tmpdir / f"{url_hash}.%(ext)s")
    try:
        args = build_ytdlp_args(url, output_template)
    except ValueError as e:
        log_err(str(e))
        return None, {}, None

    log_info(f"Downloading: {url}")
    start = time.time()

    try:
        result = subprocess.run(
            args,
            capture_output=True,
            text=True,
            timeout=YTDLP_TIMEOUT,
        )
    except subprocess.TimeoutExpired:
        log_err("Download timed out")
        return None, {}, None

    # Find the video file — tolerate non-zero exit (carousel posts: image items
    # fail with --ignore-errors but the video item may still succeed).
    video_file = None
    for f in tmpdir.iterdir():
        if f.suffix in (".mp4", ".webm", ".mkv") and ".info.json" not in f.name:
            video_file = f
            break

    if not video_file:
        stderr = (result.stderr or "")[-500:] if result.stderr else "Unknown error"
        log_err(f"Download failed: {stderr}")
        return None, {}, None

    elapsed = time.time() - start
    size_mb = video_file.stat().st_size / 1024 / 1024
    log_ok(f"Downloaded: {size_mb:.1f} MB in {elapsed:.0f}s")

    # Parse metadata
    metadata = {}
    info_file = tmpdir / f"{url_hash}.info.json"
    if info_file.exists():
        try:
            raw = json.loads(info_file.read_text())
            metadata = {
                "title": raw.get("title", ""),
                "uploader": raw.get("uploader", raw.get("channel", "")),
                "duration": raw.get("duration", 0),
                "description": (raw.get("description", "") or "")[:300],
            }
        except Exception:
            pass

    # Find thumbnail (yt-dlp may name it with video ID, not our hash)
    thumb_file = None
    for f in tmpdir.iterdir():
        if f.suffix == ".jpg" and ".info" not in f.name:
            thumb_file = f
            break

    return video_file, metadata, thumb_file


# --- Main loop ---

def print_banner():
    print(f"""
{C.BOLD}{C.CYAN}  ╔═══════════════════════════════════════╗
  ║  FreeTheVideo Download Agent          ║
  ║  Earn sats by helping download videos ║
  ╚═══════════════════════════════════════╝{C.RESET}
""")


def print_stats(stats: dict):
    s = stats.get("stats", {})
    balance = s.get("earned_sats", 0) - s.get("paid_sats", 0)
    print(f"""
{C.BOLD}Agent:{C.RESET} {stats.get('name', '?')} ({stats.get('agent_id', '?')})
{C.BOLD}Downloads completed:{C.RESET} {s.get('completed', 0)}
{C.BOLD}Total earned:{C.RESET} {C.GREEN}{s.get('earned_sats', 0)} sats{C.RESET}
{C.BOLD}Paid out:{C.RESET} {s.get('paid_sats', 0)} sats
{C.BOLD}Balance:{C.RESET} {C.YELLOW}{balance} sats{C.RESET}
{C.BOLD}Lightning address:{C.RESET} {stats.get('lightning_address', 'not set')}
""")


def run_once(server: str, token: str) -> bool:
    """Try to grab and complete one job. Returns True if successful."""
    jobs = get_jobs(server, token)
    if not jobs:
        log_info("No jobs available")
        return False

    log_info(f"Found {len(jobs)} available job(s)")
    job = jobs[0]
    log(f"  {C.BOLD}{job['source']}{C.RESET}: {job['url'][:80]}")
    log(f"  Reward: {C.GREEN}{job['sats_reward']} sats{C.RESET}")
    if job.get("error_hint"):
        log(f"  Hint: {C.DIM}{job['error_hint'][:100]}{C.RESET}")

    # Claim it
    claimed = claim_job(server, token, job["job_id"])
    if not claimed:
        return False
    log_ok(f"Claimed job {job['job_id'][:8]}")

    # Download
    with tempfile.TemporaryDirectory(prefix="ftv-agent-") as tmpdir:
        tmppath = Path(tmpdir)
        video_file, metadata, thumb_file = download_video(claimed["url"], claimed["url_hash"], tmppath)

        if not video_file:
            log_err("Download failed, reporting to server")
            report_failure(server, token, job["job_id"], "yt-dlp download failed")
            return False

        # Check file size
        if video_file.stat().st_size > MAX_FILE_SIZE:
            err = f"File too large ({video_file.stat().st_size / 1024 / 1024:.0f} MB > {MAX_FILE_SIZE / 1024 / 1024:.0f} MB)"
            log_err(err)
            report_failure(server, token, job["job_id"], err)
            return False

        # Upload
        log_info("Uploading to server...")
        start = time.time()
        result = upload_video(server, token, job["job_id"], video_file, metadata, thumb_file)
        elapsed = time.time() - start

        if result and result.get("status") == "ok":
            sats = result.get("sats_earned", 0)
            total = result.get("total_earned", 0)
            log_ok(f"Upload complete ({elapsed:.0f}s) — earned {C.BOLD}{sats} sats{C.GREEN} (total: {total})")
            return True
        else:
            err = f"Upload failed: {result.get('error', 'unknown') if result else 'no response'}"
            log_err(err)
            report_failure(server, token, job["job_id"], err)
            return False


def main():
    parser = argparse.ArgumentParser(
        description="FreeTheVideo Download Agent — earn sats by downloading videos",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="Config file: ~/.ftv-agent.json\nEnv vars: FTV_SERVER, FTV_TOKEN",
    )
    parser.add_argument("--server", help="Server URL (e.g., https://freethevideo.com)")
    parser.add_argument("--token", help="Agent auth token")
    parser.add_argument("--interval", type=int, default=POLL_INTERVAL, help=f"Poll interval in seconds (default: {POLL_INTERVAL})")
    parser.add_argument("--once", action="store_true", help="Process one job and exit")
    parser.add_argument("--stats", action="store_true", help="Show agent stats and exit")
    parser.add_argument("--payout", action="store_true", help="Request sats payout")
    parser.add_argument("--save-config", action="store_true", help="Save --server and --token to config file")
    args = parser.parse_args()

    print_banner()
    config = load_config(args)
    server = config["server"]
    token = config["token"]

    if args.save_config:
        save_config(server, token)
        return

    if args.stats:
        stats = get_stats(server, token)
        if stats:
            print_stats(stats)
        return

    if args.payout:
        result = request_payout(server, token)
        if result:
            if result.get("status") == "ok":
                log_ok(f"Payout successful: {result.get('amount_sats', 0)} sats sent!")
            else:
                log_err(f"Payout failed: {result.get('error', 'unknown')}")
        return

    if args.once:
        success = run_once(server, token)
        sys.exit(0 if success else 1)

    # Continuous mode
    log_info(f"Connected to {server}")
    stats = get_stats(server, token)
    if stats:
        s = stats.get("stats", {})
        balance = s.get("earned_sats", 0) - s.get("paid_sats", 0)
        log(f"Agent: {C.BOLD}{stats.get('name', '?')}{C.RESET} | Completed: {s.get('completed', 0)} | Balance: {C.GREEN}{balance} sats{C.RESET}")
    log_info(f"Polling every {args.interval}s for download jobs...")
    print()

    consecutive_empty = 0
    consecutive_errors = 0
    while True:
        try:
            jobs = get_jobs(server, token)
            if jobs:
                consecutive_empty = 0
                consecutive_errors = 0
                run_once(server, token)
            else:
                consecutive_empty += 1
                if consecutive_empty % 10 == 1:  # log every 10th empty poll
                    log(f"{C.DIM}No jobs available... waiting{C.RESET}")
            time.sleep(args.interval)
        except KeyboardInterrupt:
            print(f"\n{C.YELLOW}Shutting down agent...{C.RESET}")
            stats = get_stats(server, token)
            if stats:
                print_stats(stats)
            break
        except Exception as e:
            consecutive_errors += 1
            log_err(f"Unexpected error: {e}")
            if consecutive_errors >= 5:
                log_err("Too many consecutive errors, exiting")
                sys.exit(1)
            backoff = min(60, args.interval * consecutive_errors)
            log_warn(f"Retrying in {backoff}s (error {consecutive_errors}/5)")
            time.sleep(backoff)


if __name__ == "__main__":
    main()