signal-bot/bot.py

import asyncio
import base64
import binascii
import json
import logging
import os
import re
import secrets
import shlex
import shutil
import subprocess
import tempfile
import time
from datetime import datetime
from urllib.parse import quote

from signalbot import Command, Context, SignalBot
from signalbot.command import regex_triggered, triggered
from signalbot.message import MessageType

# Path segments before /status/ are [^/\s]+ (not .+): a greedy .+ matched across
# spaces, merging two space-separated tweet links into one malformed URL. The
# (?:/[^/\s]+)* allows multi-segment forms like x.com/i/web/status/<id> while
# still refusing whitespace (so two links can't merge).
TWITTER_URL_PATTERN = r"https?://(?:www\.)?(?:twitter\.com|x\.com|fxtwitter\.com|vxtwitter\.com|fixupx\.com)/[^/\s]+(?:/[^/\s]+)*/status/\d+"
INSTAGRAM_URL_PATTERN = r"https?://(?:www\.)?instagram\.com/(?:reel|p)/[\w-]+"
YOUTUBE_URL_PATTERN = r"https?://(?:www\.)?(?:youtube\.com/(?:watch\?v=|shorts/)|youtu\.be/)[\w-]+"
TIKTOK_URL_PATTERN = r"https?://(?:(?:www|m)\.tiktok\.com/(?:@[\w.-]+/video/\d+|t/\w+|v/\d+)|(?:vm|vt)\.tiktok\.com/\w+)"
VIDEO_URL_PATTERN = rf"(?:{TWITTER_URL_PATTERN}|{INSTAGRAM_URL_PATTERN}|{YOUTUBE_URL_PATTERN}|{TIKTOK_URL_PATTERN})"
MAX_FILE_SIZE = 100 * 1024 * 1024  # 100 MB
CLIP_DURATION = 60  # default seconds to grab around a shared ?t= timestamp
MAX_CLIP_DURATION = 600  # ceiling for a user-supplied /clip override
# Cap simultaneous heavy media jobs (yt-dlp/ffmpeg) so a busy group can't spawn
# unbounded subprocesses. Each job runs in a worker thread (see _run_blocking).
MAX_CONCURRENT_JOBS = 2
YTDLP = os.path.join(os.path.dirname(os.path.abspath(__file__)), "venv", "bin", "yt-dlp")
COOKIES = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cookies.txt")
STATE_DB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bot-state.db")
ADMIN_NUMBERS = {n.strip() for n in os.environ.get("BOT_ADMINS", "").split(",") if n.strip()}

# --- /force: host an oversized video as a temporary download link ---
FORCE_DURATION_THRESHOLD = 3600   # offer /force for videos longer than this (1 hr)
FORCE_OFFER_TTL = 600             # a /force offer stays valid for 10 min
FORCE_TIMEOUT = 1800             # max seconds for a forced full download (30 min)
FORCE_MAX_FILESIZE = "10G"        # yt-dlp aborts a single download larger than this
RSYNC_TIMEOUT = 1800             # max seconds to upload to the web box
FORCE_STAGE_ROOT = "/var/tmp/signal-bot-force"  # disk-backed staging (/tmp is tmpfs/RAM)
MEDIA_TTL = 24 * 3600             # delete hosted files after 24 h
MEDIA_MAX_BYTES = 10 * 1024 * 1024 * 1024  # evict oldest once the hosted folder exceeds 10 GB
TOKEN_RE = re.compile(r"^[0-9a-f]{24}$")   # cleanup only ever touches dirs we created

# /force hosting target — configured via env so the published code carries no
# infra details. The systemd unit supplies the real values; unset = feature off.
FORCE_REMOTE = os.environ.get("FORCE_REMOTE", "")             # ssh target, e.g. "user@host"
FORCE_REMOTE_DIR = os.environ.get("FORCE_REMOTE_DIR", "")     # web docroot on that host
FORCE_BASE_URL = os.environ.get("FORCE_BASE_URL", "").rstrip("/")  # public URL for that dir
FORCE_SSH_PORT = os.environ.get("FORCE_SSH_PORT", "22")
FORCE_SSH_KEY = os.environ.get("FORCE_SSH_KEY", "")          # ssh identity file (optional)
FORCE_ENABLED = bool(FORCE_REMOTE and FORCE_REMOTE_DIR and FORCE_BASE_URL)
_SSH_OPTS = ["-p", FORCE_SSH_PORT, "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=accept-new",
             *(["-i", FORCE_SSH_KEY] if FORCE_SSH_KEY else [])]
_RSYNC_RSH = "ssh " + " ".join(shlex.quote(o) for o in _SSH_OPTS)

# /force is known to be big and slow, so run its subprocesses (yt-dlp + the node
# n-sig solver and ffmpeg merge it spawns, plus rsync) at lowest CPU and idle IO
# priority. They yield to everything else under load but still run full-speed when
# the box is idle; child processes inherit the niceness. Normal short downloads
# are left at full priority for responsiveness.
_NICE = (["nice", "-n", "19"] if shutil.which("nice") else []) + \
        (["ionice", "-c", "3"] if shutil.which("ionice") else [])

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
log = logging.getLogger("signal-bot")

# (group_id, sender) -> {"b64": ..., "time": monotonic}. Keyed per sender so a
# stranger's later video doesn't silently become the target of your /speed or
# /rev; _get_video falls back to the group's latest video when you have none.
# Intentionally in-memory ONLY: each value is a base64 video up to ~133 MB, so
# persisting these to SQLite would bloat the DB. Only the dedup map below persists.
last_video = {}
VIDEO_TTL = 3600  # 1 hour

# (group_id, url) -> wall-clock epoch seconds the URL was successfully handled.
# When a user edits a message, signal-cli redelivers it as MessageType.EDIT_MESSAGE
# with the same text — without this guard the bot re-downloads and re-posts the video.
# Persisted to SQLite so a restart doesn't forget recent dedup; tiny and TTL-bounded.
# Wall-clock (not monotonic) so the stored timestamps stay meaningful across restarts.
recent_urls = {}
RECENT_URL_TTL = 600  # 10 min

# (group_id, url) currently being downloaded — guards against an edit redelivery
# arriving mid-download (before recent_urls is set) triggering a second download.
_inflight = set()

# Lazily created so it binds to the bot's running event loop, not import time.
_job_semaphore = None

# signalbot SQLite key/value store (set in main()); persists recent_urls only.
_storage = None
_RECENT_URLS_KEY = "recent_urls"
_RECENT_SEP = "\x1f"  # joins (group, url) into one JSON-safe storage key

# (group_id, sender) -> {"url","title","duration","time"}: a pending /force offer.
pending_force = {}

# Separate 1-at-a-time slot for forced downloads so a long one can't starve clips.
_force_semaphore = None


def _get_force_semaphore() -> asyncio.Semaphore:
    global _force_semaphore
    if _force_semaphore is None:
        _force_semaphore = asyncio.Semaphore(1)
    return _force_semaphore


def _get_job_semaphore() -> asyncio.Semaphore:
    global _job_semaphore
    if _job_semaphore is None:
        _job_semaphore = asyncio.Semaphore(MAX_CONCURRENT_JOBS)
    return _job_semaphore


def _sweep_videos() -> None:
    now = time.monotonic()
    for key in [k for k, e in last_video.items() if now - e["time"] > VIDEO_TTL]:
        del last_video[key]


def _set_video(group_id, sender, b64) -> None:
    _sweep_videos()
    last_video[(group_id, sender)] = {"b64": b64, "time": time.monotonic()}


def _get_video(group_id, sender):
    """The caller's own most recent video, else the group's latest non-expired one."""
    _sweep_videos()
    entry = last_video.get((group_id, sender))
    if entry:
        return entry["b64"]
    best = None
    for (g, _s), e in last_video.items():
        if g != group_id:
            continue
        if best is None or e["time"] > best["time"]:
            best = e
    return best["b64"] if best else None


def _sweep_recent_urls() -> None:
    now = time.time()
    for key in [k for k, t in recent_urls.items() if now - t > RECENT_URL_TTL]:
        del recent_urls[key]


def _load_recent_urls() -> None:
    """Restore the (pruned) dedup map from SQLite on startup."""
    if _storage is None:
        return
    try:
        if not _storage.exists(_RECENT_URLS_KEY):
            return
        data = _storage.read(_RECENT_URLS_KEY) or {}
    except Exception as e:  # noqa: BLE001
        log.warning("Could not load dedup state: %s", e)
        return
    now = time.time()
    for joined, ts in data.items():
        if not isinstance(ts, (int, float)) or now - ts > RECENT_URL_TTL:
            continue
        group_id, _, url = joined.partition(_RECENT_SEP)
        recent_urls[(group_id, url)] = ts
    log.info("Restored %d dedup entries from %s", len(recent_urls), STATE_DB)


def _persist_recent_urls() -> None:
    if _storage is None:
        return
    try:
        _storage.save(
            _RECENT_URLS_KEY,
            {f"{g}{_RECENT_SEP}{u}": ts for (g, u), ts in recent_urls.items()},
        )
    except Exception as e:  # noqa: BLE001
        log.warning("Could not persist dedup state: %s", e)


def _url_recently_handled(group_id, url) -> bool:
    _sweep_recent_urls()
    return (group_id, url) in recent_urls


def _url_busy(group_id, url) -> bool:
    return (group_id, url) in _inflight or _url_recently_handled(group_id, url)


def _mark_url_handled(group_id, url) -> None:
    _sweep_recent_urls()
    recent_urls[(group_id, url)] = time.time()
    _persist_recent_urls()


def _sweep_pending_force() -> None:
    now = time.monotonic()
    for k in [k for k, v in pending_force.items() if now - v["time"] > FORCE_OFFER_TTL]:
        del pending_force[k]


def _set_pending_force(group_id, sender, url, title, duration) -> None:
    _sweep_pending_force()
    pending_force[(group_id, sender)] = {
        "url": url, "title": title, "duration": duration, "time": time.monotonic(),
    }


def _pop_pending_force(group_id, sender):
    _sweep_pending_force()
    return pending_force.pop((group_id, sender), None)


async def _safe_reply(c: Context, text: str) -> None:
    try:
        await c.reply(text)
    except Exception as e:  # noqa: BLE001
        log.warning("Failed to send reply: %s", e)


async def _safe_send_video(c: Context, b64: str) -> bool:
    try:
        await c.send("", base64_attachments=[b64])
        return True
    except Exception as e:  # noqa: BLE001
        log.warning("Failed to send video: %s", e)
        await _safe_reply(c, "Couldn't send that video (Signal may have rejected it).")
        return False


# Errors that mean "the link simply has no downloadable video" rather than a
# genuine failure. We stay silent for these — the bot watches every message with
# a link, so most links legitimately have no video and shouldn't draw a complaint.
# NOTE: "unsupported url" is deliberately NOT here. Every URL that reaches yt-dlp
# already matched one of our platform patterns, so "Unsupported URL" means a real
# extractor/version break worth surfacing, not a quiet no-media link.
_NO_MEDIA_ERROR_PATTERNS = (
    "no video could be found",
    "there's no video",
    "no media found",
    "no video formats found",
)


def _is_no_media_error(err: str) -> bool:
    e = err.lower()
    return any(p in e for p in _NO_MEDIA_ERROR_PATTERNS)


def _summarize_ytdlp_error(stderr: str) -> str:
    """Pull a short, user-readable reason out of yt-dlp stderr."""
    if not stderr:
        return "unknown error"
    error_lines = [ln.strip() for ln in stderr.splitlines() if ln.startswith("ERROR:")]
    if error_lines:
        msg = error_lines[-1][len("ERROR:"):].strip()
        # Strip "[extractor] video_id:" prefix yt-dlp prepends.
        msg = re.sub(r"^\[[^\]]+\]\s+\S+?:\s*", "", msg)
        # Trim verbose "Use --cookies..." tails that aren't useful to a chat user.
        msg = re.split(r"\s+(?:Use --cookies|See https?://)", msg, maxsplit=1)[0]
        return msg[:240].rstrip(". ")
    lines = [ln.strip() for ln in stderr.splitlines() if ln.strip()]
    return lines[-1][:240] if lines else "unknown error"


def _parse_timestamp(value: str) -> int | None:
    """Parse a YouTube timestamp into seconds.

    Accepts plain seconds ("1509", "90s") and the h/m/s form ("25m9s",
    "1h2m3s"). Returns None for anything that isn't a recognizable time.
    """
    if not value:
        return None
    m = re.fullmatch(r"(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s?)?", value.strip(), re.IGNORECASE)
    if not m or not any(m.groups()):
        return None
    h, mi, s = (int(g) if g else 0 for g in m.groups())
    return h * 3600 + mi * 60 + s


def _extract_timestamp(url_token: str) -> int | None:
    """Pull the start offset (seconds) out of a YouTube URL's t=/start= param."""
    m = re.search(r"[?&#](?:t|start)=([0-9hms]+)", url_token, re.IGNORECASE)
    return _parse_timestamp(m.group(1)) if m else None


def _message_data(envelope: dict) -> dict:
    """Find the data_message dict (which carries attachments) inside a raw envelope.

    Mirrors signalbot's own extraction so we also see attachments delivered via
    edit envelopes and synced (linked-device) messages, not just plain dataMessage.
    """
    if "dataMessage" in envelope:
        return envelope["dataMessage"] or {}
    if "editMessage" in envelope:
        return envelope["editMessage"].get("dataMessage") or {}
    if "syncMessage" in envelope:
        sent = envelope["syncMessage"].get("sentMessage") or {}
        if "editMessage" in sent:
            return sent["editMessage"].get("dataMessage") or {}
        return sent
    return {}


class VideoTracker(Command):
    """Watches group messages for video attachments and stores the last one per sender."""
    async def handle(self, c: Context) -> None:
        if not c.message.is_group():
            return

        try:
            envelope = json.loads(c.message.raw_message)["envelope"]
        except Exception:  # noqa: BLE001
            return

        attachments = _message_data(envelope).get("attachments") or []
        att = next(
            (a for a in attachments if str(a.get("contentType", "")).startswith("video/")),
            None,
        )
        if not att or not att.get("id"):
            return

        # download_attachments is disabled globally, so fetch only this video on
        # demand instead of letting the library base64 every attachment of every
        # message on the producer loop.
        try:
            b64 = await c.bot._signal.get_attachment(att["id"])
        except Exception as e:  # noqa: BLE001
            log.warning("Failed to fetch video attachment %s: %s", att.get("id"), e)
            return

        _set_video(c.message.group, _sender_number(c.message), b64)
        log.info("Stored received video for group %s", c.message.group)


class VideoCommand(Command):
    @regex_triggered(VIDEO_URL_PATTERN)
    async def handle(self, c: Context) -> None:
        if not c.message.is_group():
            return

        text = c.message.text or ""
        matches = list(re.finditer(VIDEO_URL_PATTERN, text))
        if not matches:
            return

        # An optional "/clip <seconds>" anywhere in the message overrides the
        # default window length for any clip produced from this message.
        clip_len = CLIP_DURATION
        mclip = re.search(r"/clip\s+(\S+)", text, re.IGNORECASE)
        if mclip:
            secs = _parse_timestamp(mclip.group(1))
            if secs is None or secs < 1:
                await _safe_reply(c, "`/clip` needs a length in seconds, e.g. `/clip 30`.")
                return
            clip_len = min(secs, MAX_CLIP_DURATION)

        is_edit = c.message.type == MessageType.EDIT_MESSAGE
        group = c.message.group
        sender = _sender_number(c.message)

        # Process URLs sequentially in message order. Each download already runs
        # off the event loop (asyncio.to_thread), so this doesn't block the bot,
        # and sequential keeps last_video deterministic (the last link wins) and
        # posts in order. `seen` drops a URL repeated within one message.
        seen = set()
        for m in matches:
            url = m.group(0)

            # The URL pattern stops at the video id, so any ?t=/&t= timestamp
            # lives in the characters that follow. Grab the whole whitespace-
            # delimited token to recover it. Timestamps/clips apply to YouTube.
            clip = None
            if re.match(YOUTUBE_URL_PATTERN, url):
                tok = re.match(r"\S+", text[m.start():])
                start = _extract_timestamp(tok.group(0) if tok else url)
                if mclip:
                    # Explicit /clip clips even without a timestamp (from 0).
                    start = start or 0
                    clip = (start, start + clip_len)
                elif start is not None:
                    clip = (start, start + clip_len)

            # Normalize fxtwitter/vxtwitter wrappers to x.com
            url = re.sub(
                r"https?://(?:www\.)?(?:fxtwitter\.com|vxtwitter\.com|fixupx\.com)",
                "https://x.com",
                url,
            )

            if url in seen:
                continue
            seen.add(url)

            if is_edit and _url_busy(group, url):
                log.info("Skipping edited message; already handled %s", url)
                continue

            await self._process_one(c, group, sender, url, clip)

    async def _process_one(self, c: Context, group, sender, url, clip) -> None:
        key = (group, url)
        _inflight.add(key)
        try:
            # Clip path: a bounded window, just grab it (no oversize concern).
            if clip is not None:
                log.info("Clipping %s to window %d-%ds", url, clip[0], clip[1])
                async with _get_job_semaphore():
                    b64, err, silent = await asyncio.to_thread(_produce_video, url, clip)
                await self._deliver(c, group, sender, url, b64, err, silent,
                                    title=None, duration=None, allow_force=False)
                return

            # Full-video path. Only when /force hosting is enabled do we probe
            # duration first (to fast-path an over-long video to a /force offer);
            # otherwise behave like before and just download.
            async with _get_job_semaphore():
                if FORCE_ENABLED:
                    duration, title = await asyncio.to_thread(_probe_metadata, url)
                    too_long = duration is not None and duration > FORCE_DURATION_THRESHOLD
                else:
                    duration, title, too_long = None, None, False
                if too_long:
                    b64, err, silent = None, "", False
                else:
                    b64, err, silent = await asyncio.to_thread(_produce_video, url, clip)

            if too_long:
                _mark_url_handled(group, url)
                await _too_big_response(c, group, sender, url, title, duration)
                return
            await self._deliver(c, group, sender, url, b64, err, silent,
                                title=title, duration=duration, allow_force=True)
        except Exception as e:  # noqa: BLE001
            log.exception("Unexpected error handling %s: %s", url, e)
            await _safe_reply(c, "Something went wrong handling that video.")
        finally:
            _inflight.discard(key)

    async def _deliver(self, c: Context, group, sender, url, b64, err, silent,
                       *, title, duration, allow_force) -> None:
        if silent:
            # A genuine no-media determination: record it so an edit replay
            # doesn't pointlessly re-run yt-dlp.
            _mark_url_handled(group, url)
            return
        if err:
            if allow_force and _is_too_big_error(err):
                _mark_url_handled(group, url)
                await _too_big_response(c, group, sender, url, title, duration)
            else:
                # Hard error: do NOT mark handled, so a corrective edit can retry.
                await _safe_reply(c, f"Couldn't grab that video: {err}")
            return

        _mark_url_handled(group, url)
        _set_video(group, sender, b64)
        await _safe_send_video(c, b64)


def _run_ytdlp(url: str, outpath: str, tmpdir: str,
               clip: tuple[int, int] | None = None) -> tuple[bool, str]:
    """Run yt-dlp with retries. Returns (success, short_reason).
    reason is empty on success; otherwise a one-line description suitable for user reply.
    When clip is (start, end), only that window is downloaded.
    Blocking — call via asyncio.to_thread, never directly on the event loop."""
    cmd = [
        YTDLP,
        "--no-playlist",
        # YouTube wraps URLs in a JS "n-sig" challenge; node solves it
        # via yt-dlp-ejs. Without this, only image/thumb formats resolve.
        "--js-runtimes", "node",
    ]
    if clip is not None:
        start, end = clip
        # Download only the requested window instead of the whole video —
        # essential for hour-long uploads shared with a ?t= timestamp.
        # force-keyframes-at-cuts makes the start boundary accurate.
        # The filesize filters below key off the *whole* video's size, which
        # is irrelevant to a 60s slice, so drop them and just take best mp4.
        cmd += [
            "--download-sections", f"*{start}-{end}",
            "--force-keyframes-at-cuts",
            "-f", "best[ext=mp4]/best",
        ]
    else:
        # Prefer the largest mp4 that already fits under 95 MB, so we avoid
        # re-encoding when a smaller variant exists (e.g. a 4K rendition
        # >100 MB alongside a 1080p ~50 MB).
        cmd += [
            "-f", "best[ext=mp4][filesize<95M]/best[ext=mp4][filesize_approx<95M]/best[ext=mp4]/best",
        ]
    cmd += [
        "--merge-output-format", "mp4",
        *(["--cookies", COOKIES] if os.path.exists(COOKIES) else []),
        "-o", outpath,
        "--", url,
    ]

    delays = [0, 3, 8]
    last_stderr = ""
    for attempt, delay in enumerate(delays, 1):
        if delay:
            time.sleep(delay)
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=120,
                cwd=tmpdir,
            )
        except subprocess.TimeoutExpired:
            # Don't retry timeouts — three 120s timeouts would block a worker for 6 min.
            log.warning("yt-dlp timed out for %s", url)
            return False, "yt-dlp timed out after 120s"

        if result.returncode == 0:
            return True, ""

        last_stderr = result.stderr
        log.warning(
            "yt-dlp failed for %s (attempt %d/%d): %s",
            url, attempt, len(delays), last_stderr.strip()[-300:],
        )

    return False, _summarize_ytdlp_error(last_stderr)


def _produce_video(url: str, clip: tuple[int, int] | None) -> tuple[str | None, str | None, bool]:
    """Blocking pipeline: download (+optional clip), re-encode if oversized, base64.

    Returns (b64, error, silent):
      success   -> (b64, None, False)
      no media  -> (None, None, True)   # stay quiet
      failure   -> (None, "reason", False)
    Run via asyncio.to_thread — does all subprocess/file/base64 work off the loop."""
    with tempfile.TemporaryDirectory() as tmpdir:
        outpath = os.path.join(tmpdir, "video.mp4")
        ok, err = _run_ytdlp(url, outpath, tmpdir, clip)
        if not ok:
            if _is_no_media_error(err):
                log.info("No video at %s (%s); staying silent", url, err)
                return None, None, True
            return None, err, False

        # Prefer the known output path; fall back to any mp4 yt-dlp produced.
        actual_file = outpath if os.path.exists(outpath) else None
        if actual_file is None:
            for f in os.listdir(tmpdir):
                if f.endswith(".mp4"):
                    actual_file = os.path.join(tmpdir, f)
                    break
        if actual_file is None:
            log.warning("No mp4 file found after yt-dlp for %s", url)
            return None, "yt-dlp finished but produced no mp4.", False

        file_size = os.path.getsize(actual_file)
        if file_size > MAX_FILE_SIZE:
            size_mb = file_size // (1024 * 1024)
            log.info("Video is %d MB, re-encoding to fit under %d MB", size_mb, MAX_FILE_SIZE // (1024 * 1024))
            new_path, reason = _reencode(actual_file, tmpdir)
            if new_path is None:
                return None, f"That video is too large ({size_mb} MB) and re-encoding failed: {reason}.", False
            actual_file = new_path

        with open(actual_file, "rb") as f:
            return base64.b64encode(f.read()).decode("utf-8"), None, False


def _probe_metadata(url: str) -> tuple[int | None, str | None]:
    """Fetch (duration_seconds, title) without downloading. Blocking — use to_thread."""
    cmd = [
        YTDLP, "--no-playlist", "--quiet", "--no-warnings",
        "--js-runtimes", "node",
        "--print", "%(duration)s\n%(title)s",
        *(["--cookies", COOKIES] if os.path.exists(COOKIES) else []),
        "--", url,
    ]
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=90)
    except subprocess.TimeoutExpired:
        return None, None
    if r.returncode != 0:
        return None, None
    lines = r.stdout.splitlines()
    duration = None
    if lines:
        try:
            duration = int(float(lines[0]))
        except ValueError:
            duration = None
    title = lines[1].strip() if len(lines) > 1 and lines[1].strip() not in ("", "NA") else None
    return duration, title


def _fmt_duration(seconds: int | None) -> str:
    if not seconds:
        return ""
    h, rem = divmod(int(seconds), 3600)
    m, s = divmod(rem, 60)
    return f"{h}h{m:02d}m" if h else f"{m}m{s:02d}s"


def _too_big_offer(title: str | None, duration: int | None) -> str:
    name = f' "{title}"' if title else ""
    dur = f" ({_fmt_duration(duration)})" if duration else ""
    return (
        f"That video{name}{dur} is too big to post here. "
        f"Reply `/force` within 10 min and I'll download it and give you a temporary "
        f"link (auto-deleted in 24h)."
    )


# Bot-generated oversize/slow sentinels only — NOT yt-dlp's transient network
# "read operation timed out", which should stay a retryable hard error.
_TOO_BIG_PATTERNS = ("too large", "after re-encode", "yt-dlp timed out after", "ffmpeg timed out")


def _is_too_big_error(err: str) -> bool:
    e = (err or "").lower()
    return any(p in e for p in _TOO_BIG_PATTERNS)


async def _too_big_response(c: Context, group, sender, url, title, duration) -> None:
    """Reply to an oversized video: offer /force if hosting is set up, else just note it."""
    if FORCE_ENABLED:
        _set_pending_force(group, sender, url, title, duration)
        await _safe_reply(c, _too_big_offer(title, duration))
    else:
        name = f' "{title}"' if title else ""
        dur = f" ({_fmt_duration(duration)})" if duration else ""
        await _safe_reply(c, f"That video{name}{dur} is too big to post here.")


def _force_download_and_publish(url: str, token: str) -> tuple[str | None, int | None, str | None]:
    """Blocking: download the full video, upload it to the web box, return
    (public_url, size_mb, error). Stages on disk (/tmp is RAM) and cleans up."""
    os.makedirs(FORCE_STAGE_ROOT, exist_ok=True)
    # Self-heal: drop staging dirs orphaned by a hard crash (only one force at a time).
    for old in os.listdir(FORCE_STAGE_ROOT):
        if old.startswith("force-"):
            shutil.rmtree(os.path.join(FORCE_STAGE_ROOT, old), ignore_errors=True)
    stage = tempfile.mkdtemp(dir=FORCE_STAGE_ROOT, prefix="force-")
    try:
        tokdir = os.path.join(stage, token)
        os.makedirs(tokdir)
        cmd = [
            *_NICE,
            YTDLP, "--no-playlist", "--no-warnings", "--js-runtimes", "node",
            "-f", "best[ext=mp4]/best", "--merge-output-format", "mp4",
            "--max-filesize", FORCE_MAX_FILESIZE,
            *(["--cookies", COOKIES] if os.path.exists(COOKIES) else []),
            "-o", os.path.join(tokdir, "video.%(ext)s"),
            "--", url,
        ]
        try:
            r = subprocess.run(cmd, capture_output=True, text=True, timeout=FORCE_TIMEOUT)
        except subprocess.TimeoutExpired:
            return None, None, f"timed out after {FORCE_TIMEOUT // 60} min"

        files = [f for f in os.listdir(tokdir) if os.path.isfile(os.path.join(tokdir, f))]
        if r.returncode != 0 or not files:
            # yt-dlp prints the max-filesize abort to stdout (and exits 0), so check both.
            blob = (r.stdout + r.stderr).lower()
            if "larger than" in blob or "max-filesize" in blob:
                return None, None, f"video is larger than the {FORCE_MAX_FILESIZE}B hosting limit"
            return None, None, _summarize_ytdlp_error(r.stderr)

        # Pick the actual video (largest file), not an arbitrary listdir entry.
        fname = max(files, key=lambda f: os.path.getsize(os.path.join(tokdir, f)))
        size_mb = os.path.getsize(os.path.join(tokdir, fname)) // (1024 * 1024)

        up = subprocess.run(
            [*_NICE, "rsync", "-a", "--chmod=F644,D755", "-e", _RSYNC_RSH,
             tokdir, f"{FORCE_REMOTE}:{FORCE_REMOTE_DIR}/"],
            capture_output=True, text=True, timeout=RSYNC_TIMEOUT,
        )
        if up.returncode != 0:
            return None, None, f"upload failed ({up.stderr.strip()[-160:]})"

        return f"{FORCE_BASE_URL}/{token}/{quote(fname)}", size_mb, None
    except Exception as e:  # noqa: BLE001
        log.exception("forced download failed for %s", url)
        return None, None, f"unexpected error ({e})"
    finally:
        shutil.rmtree(stage, ignore_errors=True)


def _cleanup_media() -> None:
    """Delete hosted videos older than MEDIA_TTL and evict the oldest past
    MEDIA_MAX_BYTES. Only ever touches token dirs we created. Blocking."""
    ssh_base = ["ssh", *_SSH_OPTS, FORCE_REMOTE]
    # Emit "age_seconds<TAB>bytes<TAB>name" using the REMOTE clock, so the TTL
    # decision doesn't depend on this host's clock matching the web box's.
    list_script = (
        f"cd {shlex.quote(FORCE_REMOTE_DIR)} 2>/dev/null || exit 0; now=$(date +%s); "
        f'for d in */; do [ -d "$d" ] || continue; n="${{d%/}}"; '
        f'printf "%s\\t%s\\t%s\\n" "$((now - $(stat -c %Y "$d")))" '
        f'"$(du -sb "$d" 2>/dev/null | cut -f1)" "$n"; done'
    )
    try:
        r = subprocess.run([*ssh_base, list_script], capture_output=True, text=True, timeout=120)
    except subprocess.TimeoutExpired:
        log.warning("media cleanup: listing timed out")
        return
    if r.returncode != 0:
        log.warning("media cleanup: listing failed (%s)", r.stderr.strip()[-160:])
        return

    entries = []  # (age_seconds, bytes, name)
    for line in r.stdout.splitlines():
        parts = line.split("\t")
        if len(parts) != 3 or not TOKEN_RE.match(parts[2]):
            continue
        try:
            age = int(parts[0])
        except ValueError:
            continue  # no usable age -> can't make a TTL decision, skip this round
        # A du failure leaves the size field empty; treat as 0 so TTL can still fire.
        nbytes = int(parts[1]) if parts[1].isdigit() else 0
        entries.append((age, nbytes, parts[2]))

    doomed = {name for age, _b, name in entries if age > MEDIA_TTL}
    fresh = sorted(e for e in entries if e[2] not in doomed)  # ascending age = newest first
    total = 0
    for _age, nbytes, name in fresh:
        total += nbytes
        if total > MEDIA_MAX_BYTES:
            doomed.add(name)

    doomed = [n for n in doomed if TOKEN_RE.match(n)]
    if not doomed:
        return
    paths = " ".join(shlex.quote(f"{FORCE_REMOTE_DIR}/{n}") for n in doomed)
    try:
        rm = subprocess.run([*ssh_base, f"rm -rf -- {paths}"],
                            capture_output=True, text=True, timeout=120)
    except subprocess.TimeoutExpired:
        log.warning("media cleanup: rm timed out")
        return
    if rm.returncode == 0:
        log.info("media cleanup: removed %d hosted video(s)", len(doomed))
    else:
        log.warning("media cleanup: rm failed (%s)", rm.stderr.strip()[-160:])


async def _cleanup_media_job() -> None:
    try:
        await asyncio.to_thread(_cleanup_media)
    except Exception as e:  # noqa: BLE001
        log.warning("media cleanup job error: %s", e)


def _reencode(input_file: str, tmpdir: str) -> tuple[str | None, str]:
    """Re-encode video with ffmpeg to fit under MAX_FILE_SIZE.

    Returns (path, reason). On success, path is set and reason is "".
    On failure, path is None and reason is a short human-readable cause.
    Blocking — only called from within _produce_video/_speed_video/_reverse_video."""
    outpath = os.path.join(tmpdir, "reencoded.mp4")
    try:
        probe = subprocess.run(
            ["ffprobe", "-v", "error", "-show_entries", "format=duration",
             "-of", "default=noprint_wrappers=1:nokey=1", input_file],
            capture_output=True, text=True, timeout=30,
        )
        duration = float(probe.stdout.strip())
    except Exception:  # noqa: BLE001
        log.warning("Could not probe video duration")
        return None, "could not read video duration"

    # Guard against 0/NaN duration before the bitrate math (else ZeroDivisionError
    # or "cannot convert float NaN to integer" would escape and silently kill the
    # handler with no reply to the user). `not (duration > 0)` catches 0, negative, NaN.
    if not (duration > 0):
        log.warning("Invalid video duration from ffprobe: %r", duration)
        return None, "could not read video duration"

    # Target 95 MB to leave headroom
    target_bytes = 95 * 1024 * 1024
    # Total bitrate in kbps; reserve 128k for audio
    audio_bitrate = 128
    total_bitrate = int((target_bytes * 8) / duration / 1000)
    video_bitrate = max(total_bitrate - audio_bitrate, 200)

    try:
        result = subprocess.run(
            [
                "ffmpeg", "-y", "-i", input_file,
                "-c:v", "libx264", "-b:v", f"{video_bitrate}k",
                # Cap peaks so VBR overshoot can't blow past the target size.
                "-maxrate", f"{video_bitrate}k", "-bufsize", f"{2 * video_bitrate}k",
                "-c:a", "aac", "-b:a", f"{audio_bitrate}k",
                "-preset", "fast",
                "-movflags", "+faststart",
                outpath,
            ],
            capture_output=True, text=True, timeout=300,
        )
    except subprocess.TimeoutExpired:
        log.warning("ffmpeg re-encode timed out")
        return None, "ffmpeg timed out after 300s"

    if result.returncode != 0:
        log.warning("ffmpeg re-encode failed: %s", result.stderr[-500:])
        stderr_lines = [ln for ln in result.stderr.strip().splitlines() if ln.strip()]
        last_line = stderr_lines[-1] if stderr_lines else "no stderr"
        return None, f"ffmpeg exited {result.returncode} ({last_line[:160]})"

    final_size = os.path.getsize(outpath)
    if final_size > MAX_FILE_SIZE:
        final_mb = final_size // (1024 * 1024)
        log.warning("Re-encoded video still too large: %d MB", final_mb)
        return None, f"output still {final_mb} MB after re-encode (duration {int(duration)}s)"

    log.info("Re-encoded video from %d MB to %d MB",
             os.path.getsize(input_file) // (1024 * 1024), final_size // (1024 * 1024))
    return outpath, ""


def _has_audio_stream(path: str) -> bool:
    try:
        result = subprocess.run(
            ["ffprobe", "-v", "error", "-select_streams", "a",
             "-show_entries", "stream=index", "-of", "csv=p=0", path],
            capture_output=True, text=True, timeout=15,
        )
    except Exception:  # noqa: BLE001
        return True  # assume yes; ffmpeg will fail loudly if it's wrong
    return bool(result.stdout.strip())


def _audio_sample_rate(path: str) -> int | None:
    try:
        r = subprocess.run(
            ["ffprobe", "-v", "error", "-select_streams", "a:0",
             "-show_entries", "stream=sample_rate", "-of", "csv=p=0", path],
            capture_output=True, text=True, timeout=15,
        )
    except Exception:  # noqa: BLE001
        return None
    out = r.stdout.strip()
    try:
        return int(out) if out else None
    except ValueError:
        return None


def _speed_video(b64: str, speed: float) -> tuple[str | None, str | None]:
    """Blocking: change playback speed. Returns (b64, error). Run via to_thread."""
    with tempfile.TemporaryDirectory() as tmpdir:
        inpath = os.path.join(tmpdir, "input.mp4")
        outpath = os.path.join(tmpdir, "sped.mp4")

        try:
            data = base64.b64decode(b64)
        except (binascii.Error, ValueError):
            return None, "That video looks corrupted — share it again."
        with open(inpath, "wb") as f:
            f.write(data)

        cmd = [
            "ffmpeg", "-y", "-i", inpath,
            "-filter:v", f"setpts={1.0/speed:g}*PTS",
        ]
        if _has_audio_stream(inpath):
            # asetrate scales the sample rate (which shifts pitch AND
            # tempo, the tape-speed effect); aresample brings the data
            # rate back to a standard playback rate without undoing it.
            rate = _audio_sample_rate(inpath) or 48000
            cmd += [
                "-filter:a",
                f"asetrate={int(rate * speed)},aresample={rate}",
            ]
        else:
            cmd += ["-an"]
        cmd += ["-preset", "fast", "-movflags", "+faststart", outpath]

        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=180)
        except subprocess.TimeoutExpired:
            log.warning("ffmpeg timed out speeding up video")
            return None, "Timed out speeding up that video."

        if result.returncode != 0:
            log.warning("ffmpeg /speed failed: %s", result.stderr[-500:])
            return None, "Failed to speed up that video."

        final_file = outpath
        size = os.path.getsize(final_file)
        if size > MAX_FILE_SIZE:
            size_mb = size // (1024 * 1024)
            log.info("Sped-up video is %d MB, re-encoding to fit", size_mb)
            new_path, reason = _reencode(final_file, tmpdir)
            if new_path is None:
                return None, f"Sped-up video is too large ({size_mb} MB) and re-encoding failed: {reason}."
            final_file = new_path

        with open(final_file, "rb") as f:
            return base64.b64encode(f.read()).decode("utf-8"), None


def _reverse_video(b64: str) -> tuple[str | None, str | None]:
    """Blocking: reverse a video. Returns (b64, error). Run via to_thread."""
    with tempfile.TemporaryDirectory() as tmpdir:
        inpath = os.path.join(tmpdir, "input.mp4")
        outpath = os.path.join(tmpdir, "reversed.mp4")

        try:
            data = base64.b64decode(b64)
        except (binascii.Error, ValueError):
            return None, "That video looks corrupted — share it again."
        with open(inpath, "wb") as f:
            f.write(data)

        try:
            result = subprocess.run(
                [
                    "ffmpeg", "-i", inpath,
                    "-vf", "reverse",
                    "-af", "areverse",
                    "-preset", "fast",
                    "-movflags", "+faststart",
                    outpath,
                ],
                capture_output=True, text=True, timeout=120,
            )
        except subprocess.TimeoutExpired:
            log.warning("ffmpeg timed out reversing video")
            return None, "Timed out reversing that video."

        if result.returncode != 0:
            log.warning("ffmpeg reverse failed: %s", result.stderr[-500:])
            return None, "Failed to reverse that video."

        final_file = outpath
        size = os.path.getsize(final_file)
        if size > MAX_FILE_SIZE:
            size_mb = size // (1024 * 1024)
            log.info("Reversed video is %d MB, re-encoding to fit", size_mb)
            new_path, reason = _reencode(final_file, tmpdir)
            if new_path is None:
                return None, f"Reversed video is too large ({size_mb} MB) and re-encoding failed: {reason}."
            final_file = new_path

        with open(final_file, "rb") as f:
            return base64.b64encode(f.read()).decode("utf-8"), None


class SpeedCommand(Command):
    SPEED_RE = re.compile(r"^/speed(?:\s+(\S+))?$", re.IGNORECASE)

    async def handle(self, c: Context) -> None:
        if not c.message.is_group():
            return
        text = (c.message.text or "").strip()
        m = self.SPEED_RE.match(text)
        if not m:
            return

        speed_arg = m.group(1)
        if speed_arg is None:
            speed = 2.0
        else:
            try:
                speed = float(speed_arg.rstrip("xX"))
            except ValueError:
                await _safe_reply(c, f"`{speed_arg}` isn't a number. Try `/speed 2`.")
                return

        if not (0.1 <= speed <= 100.0):
            await _safe_reply(c, f"Speed must be between 0.1 and 100 (got {speed:g}).")
            return
        if speed == 1.0:
            await _safe_reply(c, "1x wouldn't change anything.")
            return

        sender = _sender_number(c.message)
        b64 = _get_video(c.message.group, sender)
        if not b64:
            await _safe_reply(c, "No video to speed up.")
            return

        async with _get_job_semaphore():
            out, err = await asyncio.to_thread(_speed_video, b64, speed)
        if err:
            await _safe_reply(c, err)
            return

        _set_video(c.message.group, sender, out)
        await _safe_send_video(c, out)


class ReverseCommand(Command):
    @triggered("/rev")
    async def handle(self, c: Context) -> None:
        if not c.message.is_group():
            return

        sender = _sender_number(c.message)
        b64 = _get_video(c.message.group, sender)
        if not b64:
            await _safe_reply(c, "No video to reverse.")
            return

        async with _get_job_semaphore():
            out, err = await asyncio.to_thread(_reverse_video, b64)
        if err:
            await _safe_reply(c, err)
            return

        _set_video(c.message.group, sender, out)
        await _safe_send_video(c, out)


class ForceCommand(Command):
    """`/force` downloads a video the bot refused as too big and hosts it as a
    temporary link (auto-deleted in 24h)."""
    async def handle(self, c: Context) -> None:
        if not c.message.is_group():
            return
        if (c.message.text or "").strip().lower() != "/force":
            return
        if not FORCE_ENABLED:
            await _safe_reply(c, "Big-video hosting isn't set up, sorry.")
            return

        group = c.message.group
        sender = _sender_number(c.message)
        pend = _pop_pending_force(group, sender)
        if not pend:
            await _safe_reply(c, "Nothing to force (offers expire after 10 min). Share the video link again and I'll re-offer `/force` if it's too big.")
            return

        url = pend["url"]
        title = pend.get("title")
        duration = pend.get("duration")
        nm = f' "{title}"' if title else ""
        dur = f" ({_fmt_duration(duration)})" if duration else ""
        await _safe_reply(c, f"Downloading{nm}{dur}… this may take a while.")

        token = secrets.token_hex(12)
        async with _get_force_semaphore():
            public_url, size_mb, err = await asyncio.to_thread(_force_download_and_publish, url, token)
        if err:
            # Re-arm the offer so a transient failure (network/rsync) can be retried.
            _set_pending_force(group, sender, url, title, duration)
            await _safe_reply(c, f"Couldn't download that video: {err}. Reply `/force` to try again.")
            return
        await _safe_reply(c, f"{public_url}\n({size_mb} MB — will be deleted in 24 hours)")


HELP_TEXT = f"""🎬 Video bot — what I can do

Share a video link (X/Twitter, Instagram, YouTube, TikTok) and I'll post the video back to the group.
  e.g.  https://x.com/user/status/123456789

A YouTube link with a timestamp → I post a {CLIP_DURATION}s clip starting at that moment.
  e.g.  https://youtu.be/dQw4w9WgXcQ?t=90

/clip <seconds> — set the clip length for a link in the same message (max {MAX_CLIP_DURATION}s). With a ?t= it sets the window; without one it clips from the start.
  e.g.  /clip 30 https://youtu.be/dQw4w9WgXcQ?t=90
  e.g.  /clip 15 https://youtu.be/dQw4w9WgXcQ

/speed [factor] — speed up the last video (default 2x).
  e.g.  /speed     /speed 4     /speed 0.5

/rev — reverse the last video.

/force — if a video's too big to post (e.g. over an hour), I'll offer this; reply `/force` and I'll download it and post a temporary link (auto-deleted in 24h).

/help — show this message.

(In a DM, admins can run /cookies to refresh Instagram login cookies.)"""


class HelpCommand(Command):
    async def handle(self, c: Context) -> None:
        if not c.message.is_group():
            return
        if (c.message.text or "").strip().lower() not in ("/help", "/commands"):
            return
        await _safe_reply(c, HELP_TEXT)


def _sender_number(msg) -> str | None:
    for attr in ("source", "source_number", "sourceNumber"):
        v = getattr(msg, attr, None)
        if v:
            return v
    try:
        env = json.loads(msg.raw_message)["envelope"]
        return env.get("source") or env.get("sourceNumber")
    except Exception:  # noqa: BLE001
        return None


class CookiesCommand(Command):
    async def handle(self, c: Context) -> None:
        text = c.message.text or ""
        if not text.startswith("/cookies"):
            return
        if c.message.is_group():
            return

        sender = _sender_number(c.message)
        if not ADMIN_NUMBERS or sender not in ADMIN_NUMBERS:
            log.warning("Refused /cookies from %r (admins=%s)", sender, ADMIN_NUMBERS or "<unset>")
            return

        body = text.split("\n", 1)[1] if "\n" in text else ""
        cookie_re = re.compile(r"^\.?instagram\.com\b")
        ig_lines = []
        for line in body.splitlines():
            if not cookie_re.match(line):
                continue
            # Browser pastes can replace tabs with runs of spaces; restore tabs.
            normalized = re.sub(r"[ \t]{2,}", "\t", line)
            ig_lines.append(normalized)

        if not ig_lines:
            await _safe_reply(c, "No `.instagram.com` cookie lines found.")
            return

        if not any("\tsessionid\t" in ln for ln in ig_lines):
            await _safe_reply(c, "Missing `sessionid` cookie — that's the one that proves you're logged in. Re-export and try again.")
            return

        try:
            with open(COOKIES, "r") as f:
                existing = f.read().splitlines()
        except FileNotFoundError:
            existing = ["# Netscape HTTP Cookie File", ""]

        kept = [ln for ln in existing if not cookie_re.match(ln)]
        new_content = "\n".join(kept + ig_lines) + "\n"

        # Create the temp file 0600 from the start: open()+chmod left a brief
        # world-readable window exposing the Instagram session cookie.
        tmppath = COOKIES + ".tmp"
        fd = os.open(tmppath, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w") as f:
            f.write(new_content)
        os.replace(tmppath, COOKIES)

        log.info("Installed %d Instagram cookies from %s", len(ig_lines), sender)
        await _safe_reply(c, f"Installed {len(ig_lines)} Instagram cookies.")


def main():
    global _storage

    phone_number = os.environ.get("SIGNAL_PHONE_NUMBER")
    signal_service = os.environ.get("SIGNAL_SERVICE", "127.0.0.1:8080")

    if not phone_number:
        print("Error: SIGNAL_PHONE_NUMBER environment variable is required.")
        print("Example: export SIGNAL_PHONE_NUMBER='+15551234567'")
        raise SystemExit(1)

    bot = SignalBot({
        "signal_service": signal_service,
        "phone_number": phone_number,
        # Don't let the library download+base64 every attachment of every message
        # on the producer loop. VideoTracker fetches only video attachments lazily.
        "download_attachments": False,
        # Tiny SQLite KV (a few KB): persists only the URL dedup map across
        # restarts. The big last-video blobs stay in memory by design.
        "storage": {"type": "sqlite", "sqlite_db": STATE_DB},
    })
    _storage = bot.storage
    _load_recent_urls()
    bot.register(VideoTracker(), contacts=False, groups=True)
    bot.register(VideoCommand(), contacts=False, groups=True)
    bot.register(ReverseCommand(), contacts=False, groups=True)
    bot.register(SpeedCommand(), contacts=False, groups=True)
    bot.register(ForceCommand(), contacts=False, groups=True)
    bot.register(HelpCommand(), contacts=False, groups=True)
    bot.register(CookiesCommand(), contacts=True, groups=False)

    # Sweep hosted /force videos hourly (delete >24h old, evict oldest past 10 GB),
    # starting shortly after launch. Only when /force hosting is configured.
    if FORCE_ENABLED:
        bot.scheduler.add_job(_cleanup_media_job, "interval", hours=1, next_run_time=datetime.now())
        log.info("/force hosting enabled -> %s", FORCE_BASE_URL)
    else:
        log.info("/force hosting disabled (FORCE_REMOTE/FORCE_REMOTE_DIR/FORCE_BASE_URL unset)")

    log.info("Starting Signal video bot...")
    bot.start()


if __name__ == "__main__":
    main()