main.py

"""DripFilesBot — archivo → enlace DripFiles (key del bot o del usuario).

ZIP · botones · resubir · /apikey · /expire · /dev (wget|curl) · i18n · SQLite
"""

from __future__ import annotations

import asyncio
import logging
import os
import re
import shutil
import time
import uuid
import zipfile
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path

import aiohttp
from pyrogram import Client, enums, filters, idle
from pyrogram.errors import FloodWait, MessageNotModified
from pyrogram.types import (
    BotCommand,
    CallbackQuery,
    InlineKeyboardButton,
    InlineKeyboardMarkup,
    LinkPreviewOptions,
    Message,
    MessageEntity,
)

# kurigram depreca disable_web_page_preview
_NO_PREVIEW = LinkPreviewOptions(is_disabled=True)


def _utf16_len(s: str) -> int:
    """Longitud en unidades UTF-16 (offsets de entidades de Telegram)."""
    return len(s.encode("utf-16-le")) // 2

import dripfiles as dripfiles_mod
from config import load_config
from db import Database, MediaRef, PendingJob, UserSettings
from i18n import DEFAULT_LANG, LANG_LABELS, LANGS, normalize_lang, t

LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()

logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
    datefmt="%H:%M:%S",
)
if LOG_LEVEL != "DEBUG":
    logging.getLogger("pyrogram").setLevel(logging.WARNING)
log = logging.getLogger("bot")

cfg = load_config()

app = Client(
    "dripfiles_bot",
    api_id=cfg.api_id,
    api_hash=cfg.api_hash,
    bot_token=cfg.bot_token,
    # Replay messages sent while the bot was down (needs the .session file).
    skip_updates=False,
)

http: aiohttp.ClientSession
database: Database

background_tasks: set[asyncio.Task] = set()
zip_sessions: dict[int, "ZipSession"] = {}
pending_first_file: dict[int, Message] = {}
pending_first_link: dict[int, tuple[Message, str]] = {}
drip_selections: dict[str, "DripSelection"] = {}
_lang_prompted: set[int] = set()
_STALE_SECONDS = 180
_KNOWN_CMDS = frozenset(
    {
        "start",
        "help",
        "lang",
        "settings",
        "me",
        "apikey",
        "expire",
        "dev",
        "zip",
        "multi",
        "cancel",
        "done",
    }
)
_URL_RE = re.compile(r"(?i)^(https?://\S+|www\.\S+\.\S+)")
_DRIPFILES_URL_RE = re.compile(
    r"(?i)(?<![\w.-])(?:(?:https?://)?(?:www\.)?)"
    r"dripfiles\.com/[A-Za-z0-9_-]{4,64}(?:[/?#][^\s]*)?"
)
_SLOT_WAIT_SECONDS = 3600
_slot_cond = asyncio.Condition()
_user_inflight: dict[int, int] = {}
_global_inflight = 0

ZIP_SESSION_TIMEOUT = cfg.zip_timeout_minutes * 60
_SAFE_NAME_RE = re.compile(r"[^\w.\- ()\[\]]+", re.UNICODE)
_DRIP_SELECTION_PAGE_SIZE = 8


@dataclass
class StagedFile:
    path: str
    filename: str
    size: int
    file_id: str = ""


@dataclass
class ZipSession:
    user_id: int
    work_dir: str
    mode: str = "zip"  # zip | multi
    files: list[StagedFile] = field(default_factory=list)
    created_at: float = field(default_factory=time.monotonic)
    last_activity: float = field(default_factory=time.monotonic)
    status_msg: Message | None = None
    busy: bool = False
    closed: bool = False
    pending: int = 0
    lock: asyncio.Lock = field(default_factory=asyncio.Lock)

    @property
    def total_size(self) -> int:
        return sum(f.size for f in self.files)

    def touch(self) -> None:
        self.last_activity = time.monotonic()

    def expired(self) -> bool:
        return (time.monotonic() - self.last_activity) > ZIP_SESSION_TIMEOUT


@dataclass
class DripSelection:
    token: str
    user_id: int
    source_url: str
    files: list[dripfiles_mod.PublicFile]
    origin: Message
    prompt: Message | None = None
    selected: set[int] = field(default_factory=set)
    page: int = 0
    created_at: float = field(default_factory=time.monotonic)
    last_activity: float = field(default_factory=time.monotonic)
    busy: bool = False
    lock: asyncio.Lock = field(default_factory=asyncio.Lock)

    def expired(self) -> bool:
        return (time.monotonic() - self.last_activity) > ZIP_SESSION_TIMEOUT


def _auth(_, __, update) -> bool:
    user = update.from_user
    return bool(user) and (not cfg.allowed_users or user.id in cfg.allowed_users)


auth = filters.create(_auth)


def human_size(size: float) -> str:
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if size < 1024 or unit == "TB":
            return f"{size:.2f} {unit}" if unit != "B" else f"{int(size)} B"
        size /= 1024
    return f"{size:.2f} TB"


def progress_bar(pct: float) -> str:
    filled = min(20, int(pct / 5))
    return "█" * filled + "░" * (20 - filled)


async def acquire_transfer(user_id: int) -> bool:
    """Espera un hueco (cola al arrancar / ráfagas). False si pasa el timeout."""
    global _global_inflight
    deadline = time.monotonic() + _SLOT_WAIT_SECONDS
    async with _slot_cond:
        while True:
            under_global = _global_inflight < cfg.max_concurrent_global
            under_user = _user_inflight.get(user_id, 0) < cfg.max_concurrent_per_user
            if under_global and under_user:
                _user_inflight[user_id] = _user_inflight.get(user_id, 0) + 1
                _global_inflight += 1
                return True
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return False
            try:
                await asyncio.wait_for(_slot_cond.wait(), timeout=remaining)
            except asyncio.TimeoutError:
                return False


async def release_transfer(user_id: int) -> None:
    global _global_inflight
    async with _slot_cond:
        _global_inflight = max(0, _global_inflight - 1)
        n = _user_inflight.get(user_id, 1) - 1
        if n <= 0:
            _user_inflight.pop(user_id, None)
        else:
            _user_inflight[user_id] = n
        _slot_cond.notify_all()


def zip_session_active(session: ZipSession) -> bool:
    return (
        not session.closed
        and not session.busy
        and zip_sessions.get(session.user_id) is session
    )


def ensure_disk(lang: str, needed: int = 0) -> None:
    reserve = cfg.min_free_disk_bytes
    if reserve <= 0 and needed <= 0:
        return
    os.makedirs(cfg.download_dir, exist_ok=True)
    free = shutil.disk_usage(cfg.download_dir).free
    if free < reserve + max(needed, 0):
        raise dripfiles_mod.DripFilesError(t(lang, "err_disk"))


def failover_text(lang: str, user: UserSettings, kind: str, **kwargs) -> str:
    if user_own_api_key(user):
        key = f"failover_{kind}"
    elif cfg.allow_user_api_keys:
        key = f"failover_{kind}_bot"
    else:
        key = f"failover_{kind}_bot_off"
    return t(lang, key, **kwargs)


def spawn(coro) -> asyncio.Task:
    task = asyncio.create_task(coro)
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)
    return task


def safe_filename(name: str | None, fallback: str = "file") -> str:
    base = (name or fallback).strip() or fallback
    base = base.replace("/", "_").replace("\\", "_").replace("\x00", "")
    base = _SAFE_NAME_RE.sub("_", base).strip(" ._") or fallback
    return base[:200]


def extract_dripfiles_url(text: str) -> str | None:
    match = _DRIPFILES_URL_RE.search(text or "")
    return match.group(0).rstrip(".,;:!?)]}") if match else None


async def safe_edit(message: Message, text: str, **kwargs) -> None:
    try:
        await message.edit_text(text, **kwargs)
    except MessageNotModified:
        pass
    except Exception:
        log.debug("safe_edit falló", exc_info=True)


async def lang_of(user_id: int) -> str:
    user = await database.get_user(user_id)
    return normalize_lang(user.lang)


def mask_key(key: str | None, lang: str) -> str:
    if not key:
        return t(lang, "no_api_key")
    k = key.strip()
    if len(k) <= 8:
        return "`" + ("*" * len(k)) + "`"
    return f"`{k[:4]}…{k[-4:]}`"


def user_own_api_key(user: UserSettings) -> str | None:
    """Key propia del usuario solo si el host lo permite."""
    if not cfg.allow_user_api_keys:
        return None
    if not user.api_key:
        return None
    return user.api_key.strip() or None


def effective_api_key(user: UserSettings) -> str | None:
    """Key efectiva: propia (si se permite) → key del bot → free (None)."""
    own = user_own_api_key(user)
    if own:
        return own
    bot_key = cfg.dripfiles_api_key
    return bot_key.strip() if bot_key else None


def display_api_key(user: UserSettings, lang: str) -> str:
    """Texto de la key en /settings y /apikey (sin revelar la del bot en claro)."""
    if user_own_api_key(user):
        return mask_key(user.api_key, lang)
    if cfg.dripfiles_api_key:
        return t(lang, "using_bot_key")
    return t(lang, "no_api_key")


def help_text(lang: str) -> str:
    api_note = t(lang, "help_api_own_ok") if cfg.allow_user_api_keys else ""
    return t(
        lang,
        "help",
        api_note=api_note,
        auto_zip=cfg.dripfiles_auto_zip_files,
    )


def lang_keyboard() -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup(
        [
            [
                InlineKeyboardButton(LANG_LABELS[code], callback_data=f"lang:{code}")
                for code in LANGS
            ]
        ]
    )


def help_keyboard(lang: str) -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup(
        [
            [
                InlineKeyboardButton(
                    t(lang, "btn_zip_start"), callback_data="zip:start"
                ),
                InlineKeyboardButton(
                    t(lang, "btn_multi_start"), callback_data="multi:start"
                ),
            ],
            [
                InlineKeyboardButton(
                    t(lang, "btn_settings"), callback_data="ui:settings"
                ),
                InlineKeyboardButton(t(lang, "btn_lang"), callback_data="ui:lang"),
            ],
        ]
    )


def settings_keyboard(lang: str) -> InlineKeyboardMarkup:
    rows = [
        [
            InlineKeyboardButton(t(lang, "btn_lang"), callback_data="ui:lang"),
            InlineKeyboardButton(t(lang, "btn_dev"), callback_data="ui:dev"),
        ],
        [
            InlineKeyboardButton(t(lang, "btn_expire"), callback_data="ui:expire"),
            InlineKeyboardButton(t(lang, "btn_me"), callback_data="ui:me"),
        ],
    ]
    if cfg.allow_user_api_keys:
        rows.insert(
            1,
            [
                InlineKeyboardButton(
                    t(lang, "btn_apikey"), callback_data="ui:apikey"
                )
            ],
        )
    return InlineKeyboardMarkup(rows)


def zip_keyboard(lang: str, mode: str = "zip") -> InlineKeyboardMarkup:
    prefix = "multi" if mode == "multi" else "zip"
    done_key = "btn_multi_done" if mode == "multi" else "btn_zip_done"
    return InlineKeyboardMarkup(
        [
            [
                InlineKeyboardButton(
                    t(lang, done_key), callback_data=f"{prefix}:done"
                ),
                InlineKeyboardButton(
                    t(lang, "btn_zip_cancel"), callback_data=f"{prefix}:cancel"
                ),
            ]
        ]
    )


def success_keyboard(lang: str, url: str, token: str | None) -> InlineKeyboardMarkup:
    rows = [[InlineKeyboardButton(t(lang, "btn_open"), url=url)]]
    if token:
        rows.append(
            [InlineKeyboardButton(t(lang, "btn_reup"), callback_data=f"reup:{token}")]
        )
    return InlineKeyboardMarkup(rows)


def dev_keyboard(lang: str) -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup(
        [
            [
                InlineKeyboardButton(
                    t(lang, "btn_dev_wget"), callback_data="dev:wget"
                ),
                InlineKeyboardButton(
                    t(lang, "btn_dev_curl"), callback_data="dev:curl"
                ),
                InlineKeyboardButton(
                    t(lang, "btn_dev_off"), callback_data="dev:off"
                ),
            ]
        ]
    )


def drip_selection_text(session: DripSelection, lang: str) -> str:
    return t(
        lang,
        "drip_pick",
        count=len(session.files),
        selected=len(session.selected),
    )


def drip_selection_keyboard(
    session: DripSelection, lang: str
) -> InlineKeyboardMarkup:
    page_count = max(
        1,
        (len(session.files) + _DRIP_SELECTION_PAGE_SIZE - 1)
        // _DRIP_SELECTION_PAGE_SIZE,
    )
    session.page = min(max(session.page, 0), page_count - 1)
    start = session.page * _DRIP_SELECTION_PAGE_SIZE
    end = min(start + _DRIP_SELECTION_PAGE_SIZE, len(session.files))
    rows: list[list[InlineKeyboardButton]] = []
    for index in range(start, end):
        item = session.files[index]
        mark = "✅" if index in session.selected else "⬜"
        label = item.name
        if len(label) > 38:
            label = label[:35] + "…"
        if item.size:
            label += f" · {human_size(item.size)}"
        rows.append(
            [
                InlineKeyboardButton(
                    f"{mark} {label}",
                    callback_data=f"df:t:{session.token}:{index}",
                )
            ]
        )
    if page_count > 1:
        nav: list[InlineKeyboardButton] = []
        if session.page > 0:
            nav.append(
                InlineKeyboardButton(
                    "◀️", callback_data=f"df:p:{session.token}:{session.page - 1}"
                )
            )
        nav.append(
            InlineKeyboardButton(
                f"{session.page + 1}/{page_count}",
                callback_data=f"df:p:{session.token}:{session.page}",
            )
        )
        if session.page + 1 < page_count:
            nav.append(
                InlineKeyboardButton(
                    "▶️", callback_data=f"df:p:{session.token}:{session.page + 1}"
                )
            )
        rows.append(nav)
    rows.append(
        [
            InlineKeyboardButton(
                t(lang, "btn_select_all"), callback_data=f"df:a:{session.token}"
            ),
            InlineKeyboardButton(
                t(lang, "btn_select_none"), callback_data=f"df:n:{session.token}"
            ),
        ]
    )
    rows.append(
        [
            InlineKeyboardButton(
                t(lang, "btn_submit_files", n=len(session.selected)),
                callback_data=f"df:s:{session.token}",
            )
        ]
    )
    return InlineKeyboardMarkup(rows)


def media_filter(_, __, message: Message) -> bool:
    return bool(
        message.document
        or message.video
        or message.audio
        or message.voice
        or message.video_note
        or message.animation
        or message.photo
        or message.sticker
    )


has_media = filters.create(media_filter)


def media_file_id(message: Message) -> str | None:
    if message.document:
        return message.document.file_id
    if message.video:
        return message.video.file_id
    if message.audio:
        return message.audio.file_id
    if message.voice:
        return message.voice.file_id
    if message.video_note:
        return message.video_note.file_id
    if message.animation:
        return message.animation.file_id
    if message.sticker:
        return message.sticker.file_id
    if message.photo:
        return message.photo.file_id
    return None


def media_meta(message: Message) -> tuple[int | None, str]:
    if message.document:
        d = message.document
        return d.file_size, d.file_name or f"document_{d.file_unique_id}"
    if message.video:
        v = message.video
        return v.file_size, v.file_name or f"video_{v.file_unique_id}.mp4"
    if message.audio:
        a = message.audio
        name = a.file_name
        if not name:
            ext = ".mp3"
            if a.mime_type and "/" in a.mime_type:
                sub = a.mime_type.split("/", 1)[1].split(";")[0].strip()
                if sub and sub != "mpeg":
                    ext = f".{sub}"
            name = f"audio_{a.file_unique_id}{ext}"
        return a.file_size, name
    if message.voice:
        v = message.voice
        return v.file_size, f"voice_{v.file_unique_id}.ogg"
    if message.video_note:
        v = message.video_note
        return v.file_size, f"videonote_{v.file_unique_id}.mp4"
    if message.animation:
        a = message.animation
        return a.file_size, a.file_name or f"animation_{a.file_unique_id}.mp4"
    if message.sticker:
        s = message.sticker
        ext = ".tgs" if s.is_animated else (".webm" if s.is_video else ".webp")
        return s.file_size, f"sticker_{s.file_unique_id}{ext}"
    if message.photo:
        p = message.photo
        return p.file_size, f"photo_{p.file_unique_id}.jpg"
    return None, f"file_{uuid.uuid4().hex[:8]}"


def user_work_dir(user_id: int, kind: str = "single") -> str:
    path = os.path.join(
        cfg.download_dir, f"{kind}_{user_id}_{uuid.uuid4().hex[:10]}"
    )
    os.makedirs(path, exist_ok=True)
    return path


def rmtree_quiet(path: str | None) -> None:
    if path and os.path.isdir(path):
        shutil.rmtree(path, ignore_errors=True)


async def free_limits_capped() -> dripfiles_mod.AccountLimits:
    limits = await dripfiles_mod.get_free_limits(http)
    return dripfiles_mod.AccountLimits(
        tier="free",
        max_size_bytes=min(limits.max_size_bytes, dripfiles_mod.FREE_MAX_SIZE),
        max_files=limits.max_files,
        expire_seconds=limits.expire_seconds,
        chunk_size=limits.chunk_size,
        raw=limits.raw,
    )


async def user_limits(
    user: UserSettings,
) -> tuple[dripfiles_mod.AccountLimits, int | None, str | None]:
    """Devuelve (límites, expire_seconds, aviso). aviso: auth | me_error | None.

    Solo un 401/403 marca auth (subir por free). Un fallo de red en /me
    no descarta la key: se sigue el camino autenticado con tope Telegram.
    """
    api_key = effective_api_key(user)
    if not api_key:
        limits = await free_limits_capped()
        return limits, None, None

    preferred_expire = (
        int(user.expire_days * 86400)
        if user.expire_days and user.expire_days > 0
        else None
    )

    try:
        limits = await dripfiles_mod.get_account(http, api_key)
    except dripfiles_mod.DripFilesAuthError:
        free = await free_limits_capped()
        return free, None, "auth"
    except dripfiles_mod.DripFilesError:
        return (
            dripfiles_mod.AccountLimits(
                tier="authenticated",
                max_size_bytes=dripfiles_mod.TELEGRAM_MAX_SIZE,
                max_files=dripfiles_mod.FREE_MAX_FILES,
                expire_seconds=preferred_expire or 0,
                chunk_size=dripfiles_mod.DEFAULT_CHUNK,
                raw={},
            ),
            preferred_expire,
            "me_error",
        )

    effective_max = min(limits.max_size_bytes, dripfiles_mod.TELEGRAM_MAX_SIZE)
    expire = preferred_expire if preferred_expire else limits.expire_seconds
    return (
        dripfiles_mod.AccountLimits(
            tier=limits.tier,
            max_size_bytes=effective_max,
            max_files=limits.max_files,
            expire_seconds=limits.expire_seconds,
            chunk_size=limits.chunk_size,
            raw=limits.raw,
        ),
        expire,
        None,
    )


async def settings_text(user_id: int) -> str:
    user = await database.get_user(user_id)
    lang = normalize_lang(user.lang)
    try:
        limits, expire, failover = await user_limits(user)
        want = ""
        if effective_api_key(user) and expire and failover != "auth":
            want = t(lang, "want_expire", days=expire / 86400)
        lim_line = t(
            lang,
            "limits_block",
            tier=limits.tier,
            max_size=human_size(limits.max_size_bytes),
            expire_days=limits.expire_days,
            want_expire=want,
        )
        if failover == "auth":
            lim_line += failover_text(lang, user, "auth_note")
        elif failover == "me_error":
            lim_line += failover_text(lang, user, "me_note")
    except dripfiles_mod.DripFilesError as exc:
        lim_line = t(lang, "limits_error", err=exc)

    if user.expire_days:
        expire_s = t(lang, "expire_days", n=user.expire_days)
    else:
        expire_s = t(lang, "expire_default")

    if user.dev_mode:
        dev_s = t(lang, "dev_state_tool", tool=user.dev_tool)
    else:
        dev_s = t(lang, "dev_state_off")

    return t(
        lang,
        "settings",
        lang_label=LANG_LABELS.get(lang, lang),
        api_key=display_api_key(user, lang),
        expire=expire_s,
        dev=dev_s,
        limits=lim_line,
    )


# ── Download / upload ──────────────────────────────────────────────────────


async def download_to_path(
    *,
    message: Message | None = None,
    file_id: str | None = None,
    dest: str,
    filename: str,
    size_hint: int | None,
    max_size: int,
    progress: Message | None,
    lang: str,
) -> StagedFile:
    if size_hint and size_hint > max_size:
        raise dripfiles_mod.DripFilesError(
            t(
                lang,
                "err_file_too_big",
                size=human_size(size_hint),
                limit=human_size(max_size),
            )
        )

    ensure_disk(lang, size_hint or 0)

    last_edit = [0.0]

    async def on_progress(current: int, total: int) -> None:
        if not progress:
            return
        now = time.monotonic()
        if now - last_edit[0] < 2.5 and current < total:
            return
        last_edit[0] = now
        total = total or size_hint or 0
        head = t(lang, "downloading", filename=filename)
        if total:
            pct = min(99.0, current * 100 / total)
            text = (
                f"{head}\n"
                f"`[{progress_bar(pct)}]` {pct:.1f}% "
                f"({human_size(current)}/{human_size(total)})"
            )
        else:
            text = f"{head}\n{human_size(current)}"
        await safe_edit(progress, text)

    async def _download_once() -> tuple[str, str]:
        if message is not None:
            got = await message.download(file_name=dest, progress=on_progress)
            return got, media_file_id(message) or ""
        if file_id:
            got = await app.download_media(
                file_id, file_name=dest, progress=on_progress
            )
            return got, file_id
        raise RuntimeError("message or file_id required")

    path = None
    fid = ""
    last_flood = 0
    for attempt in range(6):
        try:
            path, fid = await _download_once()
            break
        except FloodWait as exc:
            wait = int(getattr(exc, "value", None) or getattr(exc, "x", 1) or 1)
            wait = min(max(wait, 1), 180)
            last_flood = wait
            log.warning("FloodWait %ss (intento %s)", wait, attempt + 1)
            await asyncio.sleep(wait)
    else:
        raise RuntimeError(f"FloodWait {last_flood}s")

    if not path or not os.path.isfile(path):
        raise RuntimeError("Telegram did not return the file")

    real_size = os.path.getsize(path)
    if real_size <= 0:
        os.remove(path)
        raise dripfiles_mod.DripFilesError(t(lang, "err_empty"))
    if real_size > max_size:
        os.remove(path)
        raise dripfiles_mod.DripFilesError(
            t(
                lang,
                "err_file_too_big",
                size=human_size(real_size),
                limit=human_size(max_size),
            )
        )
    return StagedFile(path=path, filename=filename, size=real_size, file_id=fid)


async def download_telegram_media(
    message: Message,
    dest_dir: str,
    *,
    max_size: int,
    lang: str,
    progress: Message | None = None,
) -> StagedFile:
    size_hint, raw_name = media_meta(message)
    filename = safe_filename(raw_name)
    dest = os.path.join(dest_dir, filename)
    if os.path.exists(dest):
        stem = Path(filename).stem
        suffix = Path(filename).suffix
        filename = f"{stem}_{uuid.uuid4().hex[:6]}{suffix}"
        dest = os.path.join(dest_dir, filename)

    return await download_to_path(
        message=message,
        dest=dest,
        filename=filename,
        size_hint=size_hint,
        max_size=max_size,
        progress=progress,
        lang=lang,
    )


async def download_public_file(
    item: dripfiles_mod.PublicFile,
    dest_dir: str,
    *,
    max_size: int,
    progress: Message,
    lang: str,
    position: int,
    count: int,
) -> StagedFile:
    if item.size and item.size > max_size:
        raise dripfiles_mod.DripFilesError(
            t(
                lang,
                "err_file_too_big",
                size=human_size(item.size),
                limit=human_size(max_size),
            )
        )
    disk_check_step = 64 * 1024**2
    ensure_disk(lang, item.size or disk_check_step)
    filename = safe_filename(item.name, f"file_{position}")
    dest = os.path.join(dest_dir, filename)
    if os.path.exists(dest):
        stem, suffix = Path(filename).stem, Path(filename).suffix
        filename = f"{stem}_{uuid.uuid4().hex[:6]}{suffix}"
        dest = os.path.join(dest_dir, filename)

    last_edit = 0.0
    headers = {"User-Agent": "DripFilesBot/1.0 (+https://t.me/DripFilesBot)"}
    try:
        async with http.get(item.url, headers=headers) as resp:
            if resp.status in (404, 410):
                raise dripfiles_mod.DripFilesError(
                    t(lang, "drip_link_expired")
                )
            if resp.status >= 400:
                raise dripfiles_mod.DripFilesError(
                    f"HTTP {resp.status} descargando {filename}"
                )
            content_type = (resp.headers.get("Content-Type") or "").lower()
            disposition = resp.headers.get("Content-Disposition") or ""
            if "text/html" in content_type and "attachment" not in disposition.lower():
                raise dripfiles_mod.DripFilesError(
                    t(lang, "drip_link_expired")
                )
            header_size = dripfiles_mod.coerce_positive_int(
                resp.headers.get("Content-Length")
            )
            total = header_size or item.size or 0
            if total and total > max_size:
                raise dripfiles_mod.DripFilesError(
                    t(
                        lang,
                        "err_file_too_big",
                        size=human_size(total),
                        limit=human_size(max_size),
                    )
                )
            written = 0
            next_disk_check = disk_check_step
            with open(dest, "wb") as fh:
                async for chunk in resp.content.iter_chunked(1024 * 1024):
                    written += len(chunk)
                    if written > max_size:
                        raise dripfiles_mod.DripFilesError(
                            t(
                                lang,
                                "err_file_too_big",
                                size=human_size(written),
                                limit=human_size(max_size),
                            )
                        )
                    if not item.size and written >= next_disk_check:
                        ensure_disk(lang, disk_check_step)
                        next_disk_check = written + disk_check_step
                    fh.write(chunk)
                    now = time.monotonic()
                    if now - last_edit < 2.5 and (not total or written < total):
                        continue
                    last_edit = now
                    pct_text = ""
                    if total:
                        pct = min(99.0, written * 100 / total)
                        pct_text = (
                            f"\n`[{progress_bar(pct)}]` {pct:.1f}% "
                            f"({human_size(written)}/{human_size(total)})"
                        )
                    await safe_edit(
                        progress,
                        t(
                            lang,
                            "drip_downloading",
                            i=position,
                            n=count,
                            filename=filename,
                        )
                        + pct_text,
                    )
    except dripfiles_mod.DripFilesError:
        try:
            os.remove(dest)
        except OSError:
            pass
        raise
    except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as exc:
        try:
            os.remove(dest)
        except OSError:
            pass
        raise dripfiles_mod.DripFilesError(
            f"error descargando {filename}: {exc}"
        ) from exc

    if not os.path.isfile(dest) or os.path.getsize(dest) <= 0:
        try:
            os.remove(dest)
        except OSError:
            pass
        raise dripfiles_mod.DripFilesError(t(lang, "err_empty"))
    return StagedFile(
        path=dest,
        filename=filename,
        size=os.path.getsize(dest),
    )


async def send_public_document(
    origin: Message,
    staged: StagedFile,
    progress: Message,
    lang: str,
) -> None:
    last_edit = 0.0

    async def on_progress(current: int, total: int) -> None:
        nonlocal last_edit
        now = time.monotonic()
        if now - last_edit < 2.5 and current < total:
            return
        last_edit = now
        pct = min(99.0, current * 100 / total) if total else 0.0
        detail = (
            f"\n`[{progress_bar(pct)}]` {pct:.1f}% "
            f"({human_size(current)}/{human_size(total)})"
            if total
            else f"\n{human_size(current)}"
        )
        await safe_edit(
            progress,
            t(lang, "drip_sending", filename=staged.filename) + detail,
        )

    for attempt in range(6):
        try:
            await app.send_document(
                chat_id=origin.chat.id,
                document=staged.path,
                reply_to_message_id=origin.id,
                progress=on_progress,
            )
            return
        except FloodWait as exc:
            if attempt == 5:
                raise
            wait = int(getattr(exc, "value", None) or getattr(exc, "x", 1) or 1)
            await asyncio.sleep(min(max(wait, 1), 180))


async def deliver_public_files(
    origin: Message,
    files: list[dripfiles_mod.PublicFile],
    *,
    source_url: str,
    as_zip: bool,
    progress: Message | None = None,
) -> None:
    user_id = origin.from_user.id
    lang = await lang_of(user_id)
    if not await acquire_transfer(user_id):
        await origin.reply_text(t(lang, "err_busy"))
        return

    work: str | None = None
    status = progress
    try:
        work = user_work_dir(user_id, "drip_download")
        if status is None:
            status = await origin.reply_text(t(lang, "drip_preparing"))
        total_known = sum(item.size for item in files if item.size > 0)
        if as_zip and total_known > dripfiles_mod.TELEGRAM_MAX_SIZE:
            raise dripfiles_mod.DripFilesError(
                t(
                    lang,
                    "drip_total_too_big",
                    size=human_size(total_known),
                    limit=human_size(dripfiles_mod.TELEGRAM_MAX_SIZE),
                )
            )

        if not as_zip:
            for index, item in enumerate(files, start=1):
                staged = await download_public_file(
                    item,
                    work,
                    max_size=dripfiles_mod.TELEGRAM_MAX_SIZE,
                    progress=status,
                    lang=lang,
                    position=index,
                    count=len(files),
                )
                await safe_edit(
                    status, t(lang, "drip_sending", filename=staged.filename)
                )
                await send_public_document(origin, staged, status, lang)
                try:
                    os.remove(staged.path)
                except OSError:
                    pass
            await safe_edit(status, t(lang, "drip_sent", n=len(files)))
            return

        staged_files: list[StagedFile] = []
        for index, item in enumerate(files, start=1):
            remaining = dripfiles_mod.TELEGRAM_MAX_SIZE - sum(
                f.size for f in staged_files
            )
            staged = await download_public_file(
                item,
                work,
                max_size=remaining,
                progress=status,
                lang=lang,
                position=index,
                count=len(files),
            )
            staged_files.append(staged)

        ensure_disk(lang, sum(f.size for f in staged_files))
        slug = dripfiles_mod.normalize_public_url(source_url).rstrip("/").rsplit(
            "/", 1
        )[-1]
        zip_name = f"dripfiles_{slug}.zip"
        zip_path = os.path.join(work, zip_name)
        await safe_edit(status, t(lang, "drip_packing", n=len(staged_files)))
        with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_STORED) as zf:
            for staged in staged_files:
                zf.write(staged.path, arcname=staged.filename)
        zip_size = os.path.getsize(zip_path)
        if zip_size > dripfiles_mod.TELEGRAM_MAX_SIZE:
            raise dripfiles_mod.DripFilesError(
                t(
                    lang,
                    "drip_total_too_big",
                    size=human_size(zip_size),
                    limit=human_size(dripfiles_mod.TELEGRAM_MAX_SIZE),
                )
            )
        outgoing = StagedFile(path=zip_path, filename=zip_name, size=zip_size)
        await safe_edit(status, t(lang, "drip_sending", filename=outgoing.filename))
        await send_public_document(origin, outgoing, status, lang)
        await safe_edit(status, t(lang, "drip_sent", n=1))
    except dripfiles_mod.DripFilesError as exc:
        if status:
            await safe_edit(status, t(lang, "err_drip_download", err=exc))
        else:
            await origin.reply_text(t(lang, "err_drip_download", err=exc))
    except Exception as exc:
        log.exception("Error entregando enlace DripFiles")
        if status:
            await safe_edit(status, t(lang, "err_drip_download", err=exc))
        else:
            await origin.reply_text(t(lang, "err_drip_download", err=exc))
    finally:
        rmtree_quiet(work)
        await release_transfer(user_id)


async def upload_to_dripfiles(
    path: str,
    filename: str,
    size: int,
    progress: Message,
    *,
    user: UserSettings,
    limits: dripfiles_mod.AccountLimits,
    expire_seconds: int | None,
    lang: str,
    count: int = 1,
    prefer_free: bool = False,
) -> dict:
    """Sube a DripFiles.

    · Key del usuario o del bot → API autenticada.
    · Si la key falla (401/403) y el archivo ≤ free → reintenta free (failover).
    · `prefer_free=True` si /me ya falló por auth (no reintentar la key rota).
    """
    last_edit = [0.0]
    api_key = effective_api_key(user)

    def on_progress(uploaded: int, total: int) -> None:
        now = time.monotonic()
        if now - last_edit[0] < 3 and uploaded < total:
            return
        last_edit[0] = now
        head = t(lang, "uploading", filename=filename)
        if total:
            pct = min(99.0, uploaded * 100 / total)
            text = (
                f"{head}\n"
                f"`[{progress_bar(pct)}]` {pct:.1f}% "
                f"({human_size(uploaded)}/{human_size(total)})"
            )
        else:
            text = f"{head}\n{human_size(uploaded)}"
        spawn(safe_edit(progress, text))

    await safe_edit(progress, t(lang, "uploading_prep", filename=filename))
    msg = drip_message_for(filename, size, count=count)

    async def _do(
        *,
        api_key: str | None,
        expire: int | None,
        max_size: int,
        chunk: int,
    ) -> dict:
        result = await dripfiles_mod.upload_path(
            http,
            path,
            filename,
            api_key=api_key,
            message=msg,
            expire_seconds=expire,
            max_size=max_size,
            on_progress=on_progress,
            chunk_size=chunk,
        )
        if not result.get("url"):
            raise dripfiles_mod.DripFilesError("DripFiles returned no URL")
        return result

    async def _upload_free(*, reason: str | None) -> dict:
        free = await free_limits_capped()
        if size > free.max_size_bytes:
            raise dripfiles_mod.DripFilesError(
                failover_text(
                    lang, user, "too_big", limit=human_size(free.max_size_bytes)
                )
            )
        if reason:
            await safe_edit(
                progress,
                failover_text(lang, user, "retry_free", filename=filename),
            )
        result = await _do(
            api_key=None,
            expire=None,
            max_size=free.max_size_bytes,
            chunk=free.chunk_size,
        )
        result["_used_key"] = False
        result["_failover"] = bool(reason)
        result["_failover_reason"] = reason
        return result

    # Ya sabemos que la key no vale → free directo
    if prefer_free or not api_key:
        return await _upload_free(reason="auth" if prefer_free else None)

    # Intento autenticado (key del usuario o del bot)
    try:
        result = await _do(
            api_key=api_key,
            expire=expire_seconds,
            max_size=limits.max_size_bytes,
            chunk=limits.chunk_size,
        )
        result["_used_key"] = True
        result["_failover"] = False
        return result
    except dripfiles_mod.DripFilesAuthError:
        return await _upload_free(reason="auth")


async def upload_many_to_dripfiles(
    files: list[StagedFile],
    display_name: str,
    progress: Message,
    *,
    user: UserSettings,
    limits: dripfiles_mod.AccountLimits,
    expire_seconds: int | None,
    lang: str,
    prefer_free: bool = False,
) -> dict:
    """Sube varios archivos como una sola transferencia, sin comprimirlos."""
    total_size = sum(f.size for f in files)
    count = len(files)
    last_edit = [0.0]
    api_key = effective_api_key(user)

    def on_progress(uploaded: int, total: int) -> None:
        now = time.monotonic()
        if now - last_edit[0] < 3 and uploaded < total:
            return
        last_edit[0] = now
        head = t(lang, "multi_uploading", n=count)
        if total:
            pct = min(99.0, uploaded * 100 / total)
            text = (
                f"{head}\n"
                f"`[{progress_bar(pct)}]` {pct:.1f}% "
                f"({human_size(uploaded)}/{human_size(total)})"
            )
        else:
            text = f"{head}\n{human_size(uploaded)}"
        spawn(safe_edit(progress, text))

    await safe_edit(progress, t(lang, "multi_uploading_prep", n=count))
    msg = drip_message_for(display_name, total_size, count=count)
    paths = [(f.path, f.filename) for f in files]

    async def _do(
        *,
        api_key: str | None,
        expire: int | None,
        active_limits: dripfiles_mod.AccountLimits,
    ) -> dict:
        result = await dripfiles_mod.upload_paths(
            http,
            paths,
            api_key=api_key,
            message=msg,
            expire_seconds=expire,
            max_size=active_limits.max_size_bytes,
            max_files=active_limits.max_files,
            on_progress=on_progress,
            chunk_size=active_limits.chunk_size,
        )
        if not result.get("url"):
            raise dripfiles_mod.DripFilesError("DripFiles returned no URL")
        return result

    async def _upload_free(*, reason: str | None) -> dict:
        free = await free_limits_capped()
        if total_size > free.max_size_bytes:
            raise dripfiles_mod.DripFilesError(
                failover_text(
                    lang, user, "too_big", limit=human_size(free.max_size_bytes)
                )
            )
        if count > free.max_files:
            raise dripfiles_mod.DripFilesError(
                t(lang, "multi_too_many", limit=free.max_files)
            )
        if reason:
            await safe_edit(
                progress,
                failover_text(lang, user, "retry_free", filename=display_name),
            )
        result = await _do(api_key=None, expire=None, active_limits=free)
        result["_used_key"] = False
        result["_failover"] = bool(reason)
        result["_failover_reason"] = reason
        return result

    if prefer_free or not api_key:
        return await _upload_free(reason="auth" if prefer_free else None)
    try:
        result = await _do(
            api_key=api_key,
            expire=expire_seconds,
            active_limits=limits,
        )
        result["_used_key"] = True
        result["_failover"] = False
        return result
    except dripfiles_mod.DripFilesAuthError:
        return await _upload_free(reason="auth")


def drip_message_for(filename: str, size: int, count: int = 1) -> str:
    template = cfg.dripfiles_message or "{filename}"
    try:
        return template.format(
            filename=filename or "file",
            size=human_size(size),
            count=count,
        ).strip()
    except (KeyError, ValueError):
        return filename or "file"


async def show_success(
    progress: Message,
    *,
    url: str,
    filename: str,
    size: int,
    count: int,
    token: str | None,
    user: UserSettings,
    expire_seconds: int | None = None,
    expires_at: int | float | None = None,
    tier: str = "free",
    failover: bool = False,
    dev_download: bool = True,
) -> None:
    lang = normalize_lang(user.lang)
    url = (url or "").replace("`", "").replace("\n", "").replace("\r", "")
    eff_expire = dripfiles_mod.coerce_positive_int(expire_seconds)
    expires_at_n = dripfiles_mod.coerce_positive_int(expires_at)
    if not eff_expire and expires_at_n:
        remaining = int(expires_at_n - time.time())
        if remaining > 0:
            eff_expire = remaining

    count_line = (
        t(lang, "count_line", count=count) if count > 1 else ""
    )
    if eff_expire and eff_expire > 0:
        days = eff_expire / 86400
        if days >= 1:
            expire_note = f"⏱ {t(lang, 'expire_days', n=int(round(days)))}."
        else:
            expire_note = f"⏱ {t(lang, 'expire_hours', n=eff_expire // 3600)}."
    else:
        # free ~2 días
        expire_note = f"⏱ {t(lang, 'expire_days', n=2)}."

    tier_note = t(lang, "tier_note", tier=tier) if tier else ""
    failover_note = (
        f"\n{failover_text(lang, user, 'success_note')}" if failover else ""
    )
    text = t(
        lang,
        "success",
        filename=filename,
        count_line=count_line,
        url=url,
        expire_note=expire_note,
        tier_note=tier_note,
    ) + failover_note
    await safe_edit(
        progress,
        text,
        link_preview_options=_NO_PREVIEW,
        reply_markup=success_keyboard(lang, url, token),
    )
    if user.dev_mode and dev_download:
        tool = user.dev_tool if user.dev_tool in ("wget", "curl") else "wget"
        cmd = dripfiles_mod.download_command(tool, url, filename)
        # Entidad PRE (no markdown ```): evita línea vacía + etiqueta "shell"
        # que mete el cliente al parsear fences.
        header = t(lang, "dev_header", tool=tool)
        # texto plano: quitar markdown residual del header
        header_plain = (
            header.replace("**", "").replace("`", "").replace("__", "").strip()
        )
        body = f"{header_plain}\n{cmd}"
        pre_offset = _utf16_len(header_plain) + 1  # +1 por el \n
        pre_length = _utf16_len(cmd)
        await progress.reply_text(
            body,
            entities=[
                MessageEntity(
                    type=enums.MessageEntityType.PRE,
                    offset=pre_offset,
                    length=pre_length,
                    language="",  # vacío = sin badge "shell" ni intro fantasma
                )
            ],
            link_preview_options=_NO_PREVIEW,
        )


async def register_job(
    user_id: int,
    *,
    kind: str,
    output_name: str,
    files: list[MediaRef],
) -> str | None:
    usable = [f for f in files if f.file_id]
    if not usable:
        return None
    token = uuid.uuid4().hex[:12]
    await database.save_job(
        PendingJob(
            token=token,
            user_id=user_id,
            kind=kind,
            output_name=output_name,
            files=usable,
            created_at=time.time(),
        )
    )
    return token


def _pending_filename(message: Message) -> str:
    _, raw = media_meta(message)
    return safe_filename(raw)


def message_is_stale(message: Message, max_age: float = _STALE_SECONDS) -> bool:
    dt = getattr(message, "date", None)
    if dt is None:
        return False
    ts = dt.timestamp() if hasattr(dt, "timestamp") else float(dt)
    return (time.time() - ts) > max_age


def infer_lang(from_user) -> str | None:
    raw = (getattr(from_user, "language_code", None) or "").lower().replace("_", "-")
    if not raw:
        return None
    primary = raw.split("-", 1)[0]
    if primary in LANGS:
        return primary
    return None


async def prompt_lang(
    message: Message, *, pending_name: str | None = None
) -> None:
    key = "choose_lang_pending" if pending_name else "choose_lang"
    kwargs = {"filename": pending_name} if pending_name else {}
    await message.reply_text(
        t(DEFAULT_LANG, key, **kwargs),
        reply_markup=lang_keyboard(),
    )


def stash_pending_media(message: Message) -> None:
    if message.from_user and media_file_id(message):
        pending_first_link.pop(message.from_user.id, None)
        pending_first_file[message.from_user.id] = message


async def ensure_lang(
    message: Message, *, force_prompt: bool = False
) -> UserSettings | None:
    """Idioma guardado, o el de Telegram. El picker solo una vez (nunca en backlog)."""
    if not message.from_user:
        return None
    uid = message.from_user.id
    user = await database.get_user(uid)
    if user.lang:
        return user

    inferred = infer_lang(message.from_user)
    if inferred:
        return await database.upsert_user(uid, lang=inferred)

    if media_file_id(message):
        stash_pending_media(message)
        pending_name = _pending_filename(message)
    else:
        pending = pending_first_file.get(uid)
        pending_name = _pending_filename(pending) if pending else None

    if message_is_stale(message) and not force_prompt:
        return None

    if force_prompt or uid not in _lang_prompted:
        _lang_prompted.add(uid)
        await prompt_lang(message, pending_name=pending_name)
    return None


# ── Handlers ───────────────────────────────────────────────────────────────


@app.on_message(filters.command(["start", "help"]) & filters.private & auth)
async def cmd_help(_: Client, message: Message):
    if message_is_stale(message):
        return
    user = await ensure_lang(message)
    if not user:
        return
    lang = normalize_lang(user.lang)
    await message.reply_text(
        help_text(lang),
        link_preview_options=_NO_PREVIEW,
        reply_markup=help_keyboard(lang),
    )


@app.on_message(filters.command("lang") & filters.private & auth)
async def cmd_lang(_: Client, message: Message):
    if message_is_stale(message):
        return
    await message.reply_text(
        t(await lang_of(message.from_user.id), "choose_lang"),
        reply_markup=lang_keyboard(),
    )


@app.on_message(filters.command("settings") & filters.private & auth)
async def cmd_settings(_: Client, message: Message):
    if message_is_stale(message):
        return
    if not await ensure_lang(message):
        return
    await message.reply_text(
        await settings_text(message.from_user.id),
        link_preview_options=_NO_PREVIEW,
        reply_markup=settings_keyboard(await lang_of(message.from_user.id)),
    )


async def send_me_card(origin: Message, user: UserSettings) -> None:
    lang = normalize_lang(user.lang)
    status = await origin.reply_text(t(lang, "me_loading"))
    try:
        api_key = effective_api_key(user)
        limits = await dripfiles_mod.resolve_limits(http, api_key)
        if user_own_api_key(user):
            mode = t(lang, "me_mode_key")
        elif cfg.dripfiles_api_key:
            mode = t(lang, "me_mode_bot")
        else:
            mode = t(lang, "me_mode_free")
        await safe_edit(
            status,
            t(
                lang,
                "me_ok",
                mode=mode,
                tier=limits.tier,
                max_size=human_size(limits.max_size_bytes),
                max_files=limits.max_files,
                expire_days=limits.expire_days,
                chunk=human_size(limits.chunk_size),
                key=display_api_key(user, lang),
            ),
        )
    except dripfiles_mod.DripFilesError as exc:
        await safe_edit(status, t(lang, "err_drip", err=exc))


@app.on_message(filters.command("me") & filters.private & auth)
async def cmd_me(_: Client, message: Message):
    if message_is_stale(message):
        return
    user = await ensure_lang(message)
    if not user:
        return
    await send_me_card(message, user)


@app.on_message(filters.command("apikey") & filters.private & auth)
async def cmd_apikey(_: Client, message: Message):
    if message_is_stale(message):
        return
    user = await ensure_lang(message)
    if not user:
        return
    lang = normalize_lang(user.lang)
    user_id = message.from_user.id
    parts = (message.text or "").split(maxsplit=1)
    arg = parts[1].strip() if len(parts) > 1 else ""

    if not cfg.allow_user_api_keys:
        await message.reply_text(
            t(lang, "apikey_disabled", key=display_api_key(user, lang)),
            link_preview_options=_NO_PREVIEW,
        )
        return

    async def scrub():
        try:
            await message.delete()
        except Exception:
            pass

    if not arg:
        await message.reply_text(
            t(lang, "apikey_help", key=display_api_key(user, lang)),
            link_preview_options=_NO_PREVIEW,
        )
        return

    if arg.lower() in ("clear", "del", "delete", "remove", "none", "free"):
        await database.upsert_user(user_id, api_key=None)
        await scrub()
        if cfg.dripfiles_api_key:
            await message.reply_text(t(lang, "apikey_cleared_bot"))
        else:
            await message.reply_text(t(lang, "apikey_cleared"))
        return

    key = arg.strip().strip("\"'")
    status = await message.reply_text(t(lang, "apikey_validating"))
    try:
        limits = await dripfiles_mod.get_account(http, key)
        await database.upsert_user(user_id, api_key=key)
        await scrub()
        await safe_edit(
            status,
            t(
                lang,
                "apikey_ok",
                tier=limits.tier,
                max_size=human_size(limits.max_size_bytes),
                expire_days=limits.expire_days,
                key=mask_key(key, lang),
            ),
        )
    except dripfiles_mod.DripFilesError as exc:
        await safe_edit(status, t(lang, "apikey_bad", err=exc))


@app.on_message(filters.command("expire") & filters.private & auth)
async def cmd_expire(_: Client, message: Message):
    if message_is_stale(message):
        return
    user = await ensure_lang(message)
    if not user:
        return
    lang = normalize_lang(user.lang)
    user_id = message.from_user.id
    parts = (message.text or "").split(maxsplit=1)
    arg = parts[1].strip() if len(parts) > 1 else ""

    if not arg:
        cur = (
            t(lang, "expire_days", n=user.expire_days)
            if user.expire_days
            else t(lang, "expire_default")
        )
        await message.reply_text(t(lang, "expire_help", cur=cur))
        return

    if arg.lower() in ("clear", "default", "reset", "0"):
        await database.upsert_user(user_id, expire_days=None)
        await message.reply_text(t(lang, "expire_cleared"))
        return

    try:
        days = int(arg)
    except ValueError:
        await message.reply_text(t(lang, "expire_bad_num"))
        return
    if days < 1 or days > 3650:
        await message.reply_text(t(lang, "expire_range"))
        return

    await database.upsert_user(user_id, expire_days=days)
    if effective_api_key(user):
        note = ""
    elif cfg.allow_user_api_keys:
        note = t(lang, "expire_no_key_note")
    else:
        note = t(lang, "expire_no_key_note_off")
    await message.reply_text(t(lang, "expire_set", days=days, note=note))


@app.on_message(filters.command("dev") & filters.private & auth)
async def cmd_dev(_: Client, message: Message):
    if message_is_stale(message):
        return
    user = await ensure_lang(message)
    if not user:
        return
    lang = normalize_lang(user.lang)
    parts = (message.text or "").split(maxsplit=1)
    arg = parts[1].strip().lower() if len(parts) > 1 else ""

    if arg in ("wget", "curl"):
        await database.upsert_user(
            message.from_user.id, dev_mode=True, dev_tool=arg
        )
        await message.reply_text(t(lang, f"dev_on_{arg}"))
        return
    if arg in ("off", "0", "false", "no"):
        await database.upsert_user(message.from_user.id, dev_mode=False)
        await message.reply_text(t(lang, "dev_off"))
        return

    state = (
        t(lang, "dev_state_tool", tool=user.dev_tool)
        if user.dev_mode
        else t(lang, "dev_state_off")
    )
    await message.reply_text(
        t(lang, "dev_pick") + "\n" + t(lang, "dev_current", state=state),
        reply_markup=dev_keyboard(lang),
    )


@app.on_message(filters.command("zip") & filters.private & auth)
async def cmd_zip(_: Client, message: Message):
    if message_is_stale(message):
        return
    if not await ensure_lang(message):
        return
    await start_zip_session(message)


@app.on_message(filters.command("multi") & filters.private & auth)
async def cmd_multi(_: Client, message: Message):
    if message_is_stale(message):
        return
    if not await ensure_lang(message):
        return
    await start_zip_session(message, mode="multi")


@app.on_message(filters.command("cancel") & filters.private & auth)
async def cmd_cancel(_: Client, message: Message):
    if message_is_stale(message):
        return
    user = await ensure_lang(message)
    if not user:
        return
    lang = normalize_lang(user.lang)
    user_id = message.from_user.id
    if user_id not in zip_sessions:
        await message.reply_text(t(lang, "zip_none"))
        return
    mode = zip_sessions[user_id].mode
    await clear_zip_session(user_id)
    await message.reply_text(
        t(lang, "multi_cancelled" if mode == "multi" else "zip_cancelled")
    )


@app.on_message(filters.command("done") & filters.private & auth)
async def cmd_done(_: Client, message: Message):
    if message_is_stale(message):
        return
    user = await ensure_lang(message)
    if not user:
        return
    lang = normalize_lang(user.lang)
    user_id = message.from_user.id
    session = zip_sessions.get(user_id)
    if not session or session.expired():
        if session:
            await clear_zip_session(user_id)
        await message.reply_text(t(lang, "zip_none_done"))
        return

    custom_name = None
    parts = (message.text or "").split(maxsplit=1)
    if len(parts) > 1:
        custom_name = parts[1].strip()

    spawn(finish_zip(message, session, custom_name=custom_name))


@app.on_callback_query(filters.regex(r"^lang:(es|en|pt)$") & auth)
async def lang_callback(_: Client, query: CallbackQuery):
    code = query.data.split(":", 1)[1]
    user_id = query.from_user.id
    await database.upsert_user(user_id, lang=code)
    _lang_prompted.discard(user_id)
    await query.answer()
    try:
        await query.message.edit_text(t(code, "lang_set"))
    except Exception:
        pass
    await query.message.reply_text(
        help_text(code),
        link_preview_options=_NO_PREVIEW,
        reply_markup=help_keyboard(code),
    )
    pending = pending_first_file.pop(user_id, None)
    if pending:
        spawn(handle_single_file(pending))
        return
    pending_link = pending_first_link.pop(user_id, None)
    if pending_link:
        pending_message, url = pending_link
        spawn(handle_dripfiles_link(pending_message, url))


@app.on_callback_query(
    filters.regex(r"^ui:(settings|dev|lang|apikey|expire|me)$") & auth
)
async def ui_callback(_: Client, query: CallbackQuery):
    action = query.data.split(":", 1)[1]
    user_id = query.from_user.id
    user = await database.get_user(user_id)
    if not user.lang:
        await query.answer()
        await query.message.reply_text(
            t(DEFAULT_LANG, "choose_lang"),
            reply_markup=lang_keyboard(),
        )
        return
    lang = normalize_lang(user.lang)
    await query.answer()

    if action == "settings":
        await query.message.reply_text(
            await settings_text(user_id),
            link_preview_options=_NO_PREVIEW,
            reply_markup=settings_keyboard(lang),
        )
    elif action == "dev":
        state = (
            t(lang, "dev_state_tool", tool=user.dev_tool)
            if user.dev_mode
            else t(lang, "dev_state_off")
        )
        await query.message.reply_text(
            t(lang, "dev_pick") + "\n" + t(lang, "dev_current", state=state),
            reply_markup=dev_keyboard(lang),
        )
    elif action == "lang":
        await query.message.reply_text(
            t(lang, "choose_lang"), reply_markup=lang_keyboard()
        )
    elif action == "apikey":
        if not cfg.allow_user_api_keys:
            await query.message.reply_text(
                t(lang, "apikey_disabled", key=display_api_key(user, lang)),
                link_preview_options=_NO_PREVIEW,
            )
            return
        await query.message.reply_text(
            t(lang, "apikey_help", key=display_api_key(user, lang)),
            link_preview_options=_NO_PREVIEW,
        )
    elif action == "expire":
        cur = (
            t(lang, "expire_days", n=user.expire_days)
            if user.expire_days
            else t(lang, "expire_default")
        )
        await query.message.reply_text(t(lang, "expire_help", cur=cur))
    elif action == "me":
        await send_me_card(query.message, user)


@app.on_callback_query(filters.regex(r"^dev:(wget|curl|off)$") & auth)
async def dev_callback(_: Client, query: CallbackQuery):
    action = query.data.split(":", 1)[1]
    user_id = query.from_user.id
    lang = await lang_of(user_id)
    if action == "off":
        await database.upsert_user(user_id, dev_mode=False)
        await query.answer("OFF")
        try:
            await query.message.edit_text(t(lang, "dev_off"))
        except Exception:
            await query.message.reply_text(t(lang, "dev_off"))
        return
    await database.upsert_user(user_id, dev_mode=True, dev_tool=action)
    await query.answer(action)
    text = t(lang, f"dev_on_{action}")
    try:
        await query.message.edit_text(text)
    except Exception:
        await query.message.reply_text(text)


@app.on_callback_query(filters.regex(r"^(zip|multi):(done|cancel|start)$") & auth)
async def zip_callback(_: Client, query: CallbackQuery):
    user_id = query.from_user.id
    user = await database.get_user(user_id)
    lang = normalize_lang(user.lang)
    mode, action = query.data.split(":", 1)
    session = zip_sessions.get(user_id)

    if action == "start":
        await query.answer()
        if not user.lang:
            await query.message.reply_text(
                t(DEFAULT_LANG, "choose_lang"),
                reply_markup=lang_keyboard(),
            )
            return
        await start_zip_session(
            query.message,
            reply_to_user_id=user_id,
            mode=mode,
        )
        return

    if session and session.mode != mode:
        await query.answer(t(lang, "session_button_old"), show_alert=True)
        return

    if action == "cancel":
        if user_id in zip_sessions:
            await clear_zip_session(user_id)
            await query.answer("OK")
            try:
                await query.message.edit_text(
                    t(
                        lang,
                        "multi_cancelled" if mode == "multi" else "zip_cancelled",
                    )
                )
            except Exception:
                pass
        else:
            await query.answer(t(lang, "zip_none"), show_alert=True)
        return

    if not session or session.expired():
        if session:
            await clear_zip_session(user_id)
        await query.answer(t(lang, "zip_none_done"), show_alert=True)
        return
    if session.pending:
        await query.answer(
            t(lang, "session_pending", n=session.pending), show_alert=True
        )
        return
    await query.answer(
        t(lang, "multi_submit_answer" if mode == "multi" else "packing_answer")
    )
    spawn(
        finish_zip(query.message, session, custom_name=None, from_callback=True)
    )


@app.on_callback_query(filters.regex(r"^reup:[a-f0-9]{12}$") & auth)
async def reup_callback(_: Client, query: CallbackQuery):
    token = query.data.split(":", 1)[1]
    lang = await lang_of(query.from_user.id)
    job = await database.get_job(token)
    if not job:
        await query.answer(t(lang, "reup_gone"), show_alert=True)
        return
    if query.from_user.id != job.user_id:
        await query.answer(t(lang, "reup_not_yours"), show_alert=True)
        return
    await query.answer(t(lang, "reup_answer"))
    spawn(do_reupload(query.message, job))


@app.on_callback_query(
    filters.regex(r"^df:(?:[tp]:[a-f0-9]{10}:\d+|[ans]:[a-f0-9]{10})$")
    & auth
)
async def drip_selection_callback(_: Client, query: CallbackQuery):
    parts = query.data.split(":")
    action, token = parts[1], parts[2]
    session = drip_selections.get(token)
    lang = await lang_of(query.from_user.id)
    if not session or session.expired():
        if session:
            drip_selections.pop(token, None)
        await query.answer(t(lang, "drip_selection_expired"), show_alert=True)
        return
    if session.user_id != query.from_user.id:
        await query.answer(t(lang, "reup_not_yours"), show_alert=True)
        return

    if action == "s":
        async with session.lock:
            if session.busy:
                await query.answer(t(lang, "drip_already_sending"), show_alert=True)
                return
            if not session.selected:
                await query.answer(t(lang, "drip_select_one"), show_alert=True)
                return
            session.busy = True
            chosen = [session.files[i] for i in sorted(session.selected)]
            drip_selections.pop(token, None)
        await query.answer(t(lang, "drip_submit_answer"))
        await safe_edit(
            query.message,
            t(lang, "drip_preparing_selected", n=len(chosen)),
            reply_markup=None,
        )
        spawn(
            deliver_public_files(
                session.origin,
                chosen,
                source_url=session.source_url,
                as_zip=False,
                progress=query.message,
            )
        )
        return

    async with session.lock:
        session.last_activity = time.monotonic()
        if action == "t":
            index = int(parts[3])
            if index < 0 or index >= len(session.files):
                await query.answer(t(lang, "drip_selection_expired"), show_alert=True)
                return
            if index in session.selected:
                session.selected.remove(index)
            else:
                session.selected.add(index)
        elif action == "p":
            session.page = max(0, int(parts[3]))
        elif action == "a":
            session.selected = set(range(len(session.files)))
        elif action == "n":
            session.selected.clear()
    await query.answer()
    await safe_edit(
        query.message,
        drip_selection_text(session, lang),
        reply_markup=drip_selection_keyboard(session, lang),
    )


@app.on_message(filters.private & auth & has_media)
async def handle_media(_: Client, message: Message):
    user = await ensure_lang(message)
    if not user:
        return
    user_id = message.from_user.id
    session = zip_sessions.get(user_id)
    lang = normalize_lang(user.lang)

    if session:
        if session.expired():
            await clear_zip_session(user_id)
            if not message_is_stale(message):
                await message.reply_text(t(lang, "zip_expired"))
            return
        async with session.lock:
            if not zip_session_active(session):
                return
            session.pending += 1
            session.touch()
        spawn(stage_for_zip(message, session, reserved=True))
        return

    spawn(handle_single_file(message))


@app.on_message(filters.private & auth & filters.text, group=8)
async def fallback_text(_: Client, message: Message):
    if message_is_stale(message):
        return
    text = (message.text or "").strip()
    if not text:
        return
    if text.startswith("/"):
        cmd = text[1:].split()[0].split("@", 1)[0].lower()
        if cmd in _KNOWN_CMDS:
            return
    drip_url = extract_dripfiles_url(text)
    user = await ensure_lang(message)
    if not user:
        if drip_url and message.from_user:
            pending_first_file.pop(message.from_user.id, None)
            pending_first_link[message.from_user.id] = (message, drip_url)
        return
    lang = normalize_lang(user.lang)
    if text.startswith("/"):
        await message.reply_text(
            t(lang, "unknown_cmd"),
            reply_markup=help_keyboard(lang),
        )
        return
    if drip_url:
        spawn(handle_dripfiles_link(message, drip_url))
        return
    if _URL_RE.match(text):
        await message.reply_text(t(lang, "hint_url"))
        return
    await message.reply_text(
        t(lang, "hint_send_file"),
        reply_markup=help_keyboard(lang),
        link_preview_options=_NO_PREVIEW,
    )


async def handle_dripfiles_link(message: Message, url: str) -> None:
    user_id = message.from_user.id
    lang = await lang_of(user_id)
    progress = await message.reply_text(t(lang, "drip_inspecting"))
    try:
        source_url, files = await dripfiles_mod.inspect_public_link(http, url)
        if len(files) == 1:
            await safe_edit(
                progress,
                t(lang, "drip_found_one", filename=safe_filename(files[0].name)),
            )
            spawn(
                deliver_public_files(
                    message,
                    files,
                    source_url=source_url,
                    as_zip=False,
                    progress=progress,
                )
            )
            return

        if len(files) > cfg.dripfiles_auto_zip_files:
            await safe_edit(
                progress,
                t(
                    lang,
                    "drip_auto_zip",
                    count=len(files),
                    threshold=cfg.dripfiles_auto_zip_files,
                ),
            )
            spawn(
                deliver_public_files(
                    message,
                    files,
                    source_url=source_url,
                    as_zip=True,
                    progress=progress,
                )
            )
            return

        # Una sola selección activa por usuario: el enlace nuevo sustituye al viejo.
        for old_token, old in list(drip_selections.items()):
            if old.user_id == user_id:
                drip_selections.pop(old_token, None)
        token = uuid.uuid4().hex[:10]
        session = DripSelection(
            token=token,
            user_id=user_id,
            source_url=source_url,
            files=files,
            origin=message,
            prompt=progress,
        )
        drip_selections[token] = session
        await safe_edit(
            progress,
            drip_selection_text(session, lang),
            reply_markup=drip_selection_keyboard(session, lang),
        )
    except dripfiles_mod.DripFilesError as exc:
        await safe_edit(progress, t(lang, "err_drip_download", err=exc))
    except Exception as exc:
        log.exception("Error inspeccionando enlace DripFiles")
        await safe_edit(progress, t(lang, "err_drip_download", err=exc))


async def start_zip_session(
    message: Message,
    *,
    reply_to_user_id: int | None = None,
    mode: str = "zip",
) -> None:
    mode = "multi" if mode == "multi" else "zip"
    user_id = reply_to_user_id or message.from_user.id
    user = await database.get_user(user_id)
    lang = normalize_lang(user.lang)
    existing = zip_sessions.get(user_id)
    if existing and not existing.expired():
        await message.reply_text(
            t(
                lang,
                "multi_already" if existing.mode == "multi" else "zip_already",
                n=len(existing.files),
                size=human_size(existing.total_size),
            ),
            reply_markup=zip_keyboard(lang, existing.mode),
        )
        return
    if existing:
        await clear_zip_session(user_id)

    work = user_work_dir(user_id, mode)
    session = ZipSession(user_id=user_id, work_dir=work, mode=mode)
    try:
        limits, _, _ = await user_limits(user)
        lim = human_size(limits.max_size_bytes)
    except Exception:
        lim = "2 GB"

    status = await message.reply_text(
        t(
            lang,
            "multi_active" if mode == "multi" else "zip_active",
            limit=lim,
            timeout=cfg.zip_timeout_minutes,
            n=0,
            size="0 B",
        ),
        reply_markup=zip_keyboard(lang, mode),
    )
    session.status_msg = status
    zip_sessions[user_id] = session


async def stage_for_zip(
    message: Message, session: ZipSession, *, reserved: bool = False
) -> None:
    user = await database.get_user(session.user_id)
    lang = normalize_lang(user.lang)
    acquired = False
    progress: Message | None = None
    try:
        if not reserved:
            async with session.lock:
                if not zip_session_active(session):
                    return
                session.pending += 1
                session.touch()
                reserved = True
        elif not zip_session_active(session):
            return
        await refresh_zip_status(session, lang)
        if not await acquire_transfer(session.user_id):
            await message.reply_text(t(lang, "err_busy"))
            return
        acquired = True
        session.touch()
        try:
            limits, _, _ = await user_limits(user)
        except dripfiles_mod.DripFilesError as exc:
            await message.reply_text(t(lang, "err_drip", err=exc))
            return

        if not zip_session_active(session):
            return

        size_hint, raw_name = media_meta(message)
        name = safe_filename(raw_name)

        async with session.lock:
            if not zip_session_active(session):
                return
            over_hint = bool(
                size_hint
                and session.total_size + size_hint > limits.max_size_bytes
            )
            current = session.total_size
        if over_hint:
            await message.reply_text(
                t(
                    lang,
                    "zip_too_big",
                    limit=human_size(limits.max_size_bytes),
                    current=human_size(current),
                    extra=human_size(size_hint or 0),
                )
            )
            return

        progress = await message.reply_text(
            t(
                lang,
                "multi_add" if session.mode == "multi" else "zip_add",
                name=name,
            )
        )
        staged = await download_telegram_media(
            message,
            session.work_dir,
            max_size=limits.max_size_bytes,
            progress=progress,
            lang=lang,
        )

        n = 0
        total = 0
        over_after = False
        too_many_after = False
        async with session.lock:
            if not zip_session_active(session):
                try:
                    os.remove(staged.path)
                except OSError:
                    pass
                return
            if session.total_size + staged.size > limits.max_size_bytes:
                os.remove(staged.path)
                over_after = True
            elif session.mode == "multi" and len(session.files) >= limits.max_files:
                os.remove(staged.path)
                too_many_after = True
            else:
                session.files.append(staged)
                session.touch()
                n = len(session.files)
                total = session.total_size

        if over_after:
            await safe_edit(
                progress,
                t(
                    lang,
                    "zip_too_big_short",
                    limit=human_size(limits.max_size_bytes),
                ),
            )
            return
        if too_many_after:
            await safe_edit(
                progress,
                t(lang, "multi_too_many", limit=limits.max_files),
            )
            return

        await safe_edit(
            progress,
            t(
                lang,
                "multi_added" if session.mode == "multi" else "zip_added",
                n=n,
                name=staged.filename,
                size=human_size(staged.size),
                total=human_size(total),
            ),
        )
        await refresh_zip_status(session, lang)
    except dripfiles_mod.DripFilesError as exc:
        if progress:
            await safe_edit(progress, t(lang, "err_drip", err=exc))
        else:
            await message.reply_text(t(lang, "err_drip", err=exc))
    except Exception as exc:
        log.exception("Error añadiendo archivo a sesión %s", session.mode)
        if progress:
            await safe_edit(progress, t(lang, "err_download", err=exc))
        else:
            await message.reply_text(t(lang, "err_download", err=exc))
    finally:
        if reserved:
            async with session.lock:
                session.pending = max(0, session.pending - 1)
                session.touch()
            await refresh_zip_status(session, lang)
        if acquired:
            await release_transfer(session.user_id)


async def refresh_zip_status(session: ZipSession, lang: str) -> None:
    if not session.status_msg or not zip_session_active(session):
        return
    n = len(session.files)
    title = "🔗 **MULTI**" if session.mode == "multi" else "📦 **ZIP**"
    pending = (
        f" · ⏳ {session.pending}" if session.pending else ""
    )
    body = [
        f"{title} · **{n}** · {human_size(session.total_size)}{pending}",
        "",
    ]
    for i, f in enumerate(session.files[-12:], start=max(1, n - 11)):
        body.append(f"{i}. `{f.filename}` ({human_size(f.size)})")
    if n > 12:
        body.append(f"… {n - 12}")
    body.append(
        t(
            lang,
            "multi_status_footer" if session.mode == "multi" else "zip_status_footer",
        ).strip()
    )
    await safe_edit(
        session.status_msg,
        "\n".join(body),
        reply_markup=zip_keyboard(lang, session.mode),
    )


async def finish_zip(
    message: Message,
    session: ZipSession,
    *,
    custom_name: str | None,
    from_callback: bool = False,
) -> None:
    user_id = session.user_id
    user = await database.get_user(user_id)
    lang = normalize_lang(user.lang)

    async with session.lock:
        if session.busy or session.closed:
            if not from_callback:
                await message.reply_text(t(lang, "zip_busy"))
            return
        if session.pending:
            await message.reply_text(
                t(lang, "session_pending", n=session.pending)
            )
            return
        if not session.files:
            await message.reply_text(
                t(lang, "multi_empty" if session.mode == "multi" else "zip_empty")
            )
            return

    if not await acquire_transfer(user_id):
        await message.reply_text(t(lang, "err_busy"))
        return

    async with session.lock:
        if (
            session.busy
            or session.closed
            or session.pending
            or zip_sessions.get(user_id) is not session
        ):
            await release_transfer(user_id)
            return
        if not session.files:
            await release_transfer(user_id)
            await message.reply_text(
                t(lang, "multi_empty" if session.mode == "multi" else "zip_empty")
            )
            return
        session.busy = True
        session.closed = True
        session.touch()
        files = list(session.files)
        zip_sessions.pop(user_id, None)

    refs = [
        MediaRef(file_id=f.file_id, filename=f.filename, size=f.size)
        for f in files
        if f.file_id
    ]

    progress: Message | None = None
    try:
        progress = await message.reply_text(
            t(
                lang,
                "multi_uploading_prep" if session.mode == "multi" else "zip_packing",
                n=len(files),
            )
        )
        limits, expire, failover = await user_limits(user)
        prefer_free = failover == "auth"

        if session.mode == "multi":
            if len(files) > limits.max_files:
                raise dripfiles_mod.DripFilesError(
                    t(lang, "multi_too_many", limit=limits.max_files)
                )
            total_size = sum(f.size for f in files)
            if total_size > limits.max_size_bytes:
                raise dripfiles_mod.DripFilesError(
                    t(
                        lang,
                        "file_too_big_result",
                        size=human_size(total_size),
                        limit=human_size(limits.max_size_bytes),
                    )
                )
            stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            display_name = (
                safe_filename(custom_name, f"drip_multi_{stamp}")
                if custom_name
                else f"drip_multi_{stamp}"
            )
            result = await upload_many_to_dripfiles(
                files,
                display_name,
                progress,
                user=user,
                limits=limits,
                expire_seconds=expire,
                lang=lang,
                prefer_free=prefer_free,
            )
            token = await register_job(
                user_id,
                kind="multi",
                output_name=display_name,
                files=refs,
            )
            used_tier = (
                "free" if result.get("_failover") or prefer_free else limits.tier
            )
            await show_success(
                progress,
                url=result["url"],
                filename=display_name,
                size=total_size,
                count=len(files),
                token=token,
                user=user,
                expire_seconds=result.get("expire")
                or (None if result.get("_failover") or prefer_free else expire),
                expires_at=result.get("expires_at"),
                tier=used_tier,
                failover=bool(result.get("_failover") or prefer_free),
                dev_download=False,
            )
            if session.status_msg:
                await safe_edit(
                    session.status_msg,
                    t(lang, "multi_closed", n=len(files)),
                    reply_markup=None,
                )
            return

        if custom_name:
            zip_name = safe_filename(custom_name)
            if not zip_name.lower().endswith(".zip"):
                zip_name += ".zip"
        else:
            stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            zip_name = f"drip_{stamp}.zip"

        zip_path = os.path.join(session.work_dir, zip_name)
        with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_STORED) as zf:
            for staged in files:
                zf.write(staged.path, arcname=staged.filename)

        zip_size = os.path.getsize(zip_path)
        if zip_size > limits.max_size_bytes:
            await safe_edit(
                progress,
                t(
                    lang,
                    "zip_too_big_result",
                    size=human_size(zip_size),
                    limit=human_size(limits.max_size_bytes),
                ),
            )
            return

        await safe_edit(
            progress,
            t(
                lang,
                "zip_ready",
                name=zip_name,
                size=human_size(zip_size),
            ),
        )
        result = await upload_to_dripfiles(
            zip_path,
            zip_name,
            zip_size,
            progress,
            user=user,
            limits=limits,
            expire_seconds=expire,
            lang=lang,
            count=len(files),
            prefer_free=prefer_free,
        )
        token = await register_job(
            user_id, kind="zip", output_name=zip_name, files=refs
        )
        used_tier = "free" if result.get("_failover") or prefer_free else limits.tier
        await show_success(
            progress,
            url=result["url"],
            filename=zip_name,
            size=zip_size,
            count=len(files),
            token=token,
            user=user,
            expire_seconds=result.get("expire") or (
                None if result.get("_failover") or prefer_free else expire
            ),
            expires_at=result.get("expires_at"),
            tier=used_tier,
            failover=bool(result.get("_failover") or prefer_free),
        )
        if session.status_msg:
            try:
                await session.status_msg.edit_text(
                    t(lang, "zip_closed", name=zip_name)
                )
            except Exception:
                pass
    except dripfiles_mod.DripFilesError as exc:
        if progress:
            await safe_edit(progress, t(lang, "err_drip", err=exc))
        else:
            await message.reply_text(t(lang, "err_drip", err=exc))
    except Exception as exc:
        log.exception("Error finalizando sesión %s", session.mode)
        if progress:
            await safe_edit(progress, t(lang, "err_generic", err=exc))
        else:
            await message.reply_text(t(lang, "err_generic", err=exc))
    finally:
        rmtree_quiet(session.work_dir)
        await release_transfer(user_id)


async def clear_zip_session(
    user_id: int, *, expect: ZipSession | None = None
) -> ZipSession | None:
    session = zip_sessions.get(user_id)
    if not session:
        return None
    if expect is not None and session is not expect:
        return None
    async with session.lock:
        if zip_sessions.get(user_id) is not session:
            return None
        if expect is not None and session is not expect:
            return None
        session.closed = True
        session.busy = True
        zip_sessions.pop(user_id, None)
    rmtree_quiet(session.work_dir)
    return session


async def handle_single_file(message: Message) -> None:
    size_hint, raw_name = media_meta(message)
    name = safe_filename(raw_name)
    user_id = message.from_user.id
    user = await database.get_user(user_id)
    lang = normalize_lang(user.lang)
    if not await acquire_transfer(user_id):
        await message.reply_text(t(lang, "err_busy"))
        return
    work: str | None = None
    progress: Message | None = None
    try:
        work = user_work_dir(user_id, "single")
        progress = await message.reply_text(
            t(lang, "downloading", filename=name) + "…"
        )
        limits, expire, failover = await user_limits(user)
        prefer_free = failover == "auth"
        staged = await download_telegram_media(
            message,
            work,
            max_size=limits.max_size_bytes,
            progress=progress,
            lang=lang,
        )
        result = await upload_to_dripfiles(
            staged.path,
            staged.filename,
            staged.size,
            progress,
            user=user,
            limits=limits,
            expire_seconds=expire,
            lang=lang,
            prefer_free=prefer_free,
        )
        token = await register_job(
            user_id,
            kind="single",
            output_name=staged.filename,
            files=[
                MediaRef(
                    file_id=staged.file_id,
                    filename=staged.filename,
                    size=staged.size,
                )
            ],
        )
        used_tier = "free" if result.get("_failover") or prefer_free else limits.tier
        await show_success(
            progress,
            url=result["url"],
            filename=staged.filename,
            size=staged.size,
            count=1,
            token=token,
            user=user,
            expire_seconds=result.get("expire") or (
                None if result.get("_failover") or prefer_free else expire
            ),
            expires_at=result.get("expires_at"),
            tier=used_tier,
            failover=bool(result.get("_failover") or prefer_free),
        )
    except dripfiles_mod.DripFilesError as exc:
        if progress:
            await safe_edit(progress, t(lang, "err_drip", err=exc))
        else:
            await message.reply_text(t(lang, "err_drip", err=exc))
    except Exception as exc:
        log.exception("Error con archivo suelto")
        if progress:
            await safe_edit(progress, t(lang, "err_generic", err=exc))
        else:
            await message.reply_text(t(lang, "err_generic", err=exc))
    finally:
        rmtree_quiet(work)
        await release_transfer(user_id)


async def do_reupload(origin: Message, job: PendingJob) -> None:
    user = await database.get_user(job.user_id)
    lang = normalize_lang(user.lang)
    if not await acquire_transfer(job.user_id):
        await origin.reply_text(t(lang, "err_busy"))
        return
    work: str | None = None
    progress: Message | None = None
    try:
        work = user_work_dir(job.user_id, "reup")
        progress = await origin.reply_text(
            t(lang, "reup_start", name=job.output_name)
        )
        limits, expire, failover = await user_limits(user)
        prefer_free = failover == "auth"
        staged_files: list[StagedFile] = []
        n = len(job.files)
        for i, ref in enumerate(job.files, start=1):
            dest_name = safe_filename(ref.filename)
            dest = os.path.join(work, dest_name)
            if os.path.exists(dest):
                stem = Path(dest_name).stem
                suffix = Path(dest_name).suffix
                dest_name = f"{stem}_{uuid.uuid4().hex[:6]}{suffix}"
                dest = os.path.join(work, dest_name)

            if n > 1:
                await safe_edit(
                    progress,
                    t(lang, "reup_progress", i=i, n=n, name=dest_name),
                )
            staged = await download_to_path(
                file_id=ref.file_id,
                dest=dest,
                filename=dest_name,
                size_hint=ref.size or None,
                max_size=limits.max_size_bytes,
                progress=progress if n == 1 else None,
                lang=lang,
            )
            staged.filename = safe_filename(ref.filename) or staged.filename
            staged_files.append(staged)

        if job.kind == "multi":
            total_size = sum(f.size for f in staged_files)
            if len(staged_files) > limits.max_files:
                raise dripfiles_mod.DripFilesError(
                    t(lang, "multi_too_many", limit=limits.max_files)
                )
            result = await upload_many_to_dripfiles(
                staged_files,
                job.output_name,
                progress,
                user=user,
                limits=limits,
                expire_seconds=expire,
                lang=lang,
                prefer_free=prefer_free,
            )
            used_tier = (
                "free" if result.get("_failover") or prefer_free else limits.tier
            )
            await show_success(
                progress,
                url=result["url"],
                filename=job.output_name,
                size=total_size,
                count=len(staged_files),
                token=job.token,
                user=user,
                expire_seconds=result.get("expire")
                or (None if result.get("_failover") or prefer_free else expire),
                expires_at=result.get("expires_at"),
                tier=used_tier,
                failover=bool(result.get("_failover") or prefer_free),
                dev_download=False,
            )
            return

        if job.kind == "zip" or len(staged_files) > 1:
            zip_name = safe_filename(job.output_name)
            if not zip_name.lower().endswith(".zip"):
                zip_name += ".zip"
            zip_path = os.path.join(work, zip_name)
            await safe_edit(
                progress, t(lang, "zip_packing", n=len(staged_files))
            )
            with zipfile.ZipFile(
                zip_path, "w", compression=zipfile.ZIP_STORED
            ) as zf:
                for staged in staged_files:
                    zf.write(staged.path, arcname=staged.filename)
            out_path = zip_path
            out_name = zip_name
            out_size = os.path.getsize(zip_path)
            count = len(staged_files)
        else:
            staged = staged_files[0]
            out_path = staged.path
            out_name = job.output_name or staged.filename
            out_size = staged.size
            count = 1

        if out_size > limits.max_size_bytes:
            await safe_edit(
                progress,
                t(
                    lang,
                    "file_too_big_result",
                    size=human_size(out_size),
                    limit=human_size(limits.max_size_bytes),
                ),
            )
            return

        result = await upload_to_dripfiles(
            out_path,
            out_name,
            out_size,
            progress,
            user=user,
            limits=limits,
            expire_seconds=expire,
            lang=lang,
            count=count,
            prefer_free=prefer_free,
        )
        used_tier = "free" if result.get("_failover") or prefer_free else limits.tier
        await show_success(
            progress,
            url=result["url"],
            filename=out_name,
            size=out_size,
            count=count,
            token=job.token,
            user=user,
            expire_seconds=result.get("expire") or (
                None if result.get("_failover") or prefer_free else expire
            ),
            expires_at=result.get("expires_at"),
            tier=used_tier,
            failover=bool(result.get("_failover") or prefer_free),
        )
        await database.touch_job(job.token)
    except dripfiles_mod.DripFilesError as exc:
        if progress:
            await safe_edit(progress, t(lang, "err_drip", err=exc))
        else:
            await origin.reply_text(t(lang, "err_drip", err=exc))
    except Exception as exc:
        log.exception("Error en resubida %s", job.token)
        if progress:
            await safe_edit(progress, t(lang, "err_reup", err=exc))
        else:
            await origin.reply_text(t(lang, "err_reup", err=exc))
    finally:
        rmtree_quiet(work)
        await release_transfer(job.user_id)


async def zip_janitor() -> None:
    while True:
        await asyncio.sleep(60)
        expired = [
            (uid, s)
            for uid, s in list(zip_sessions.items())
            if s.expired() and not s.busy and not s.closed and not s.pending
        ]
        for uid, session in expired:
            lang = await lang_of(uid)
            current = zip_sessions.get(uid)
            if (
                current is not session
                or session.busy
                or session.closed
                or session.pending
                or not session.expired()
            ):
                continue
            cleared = await clear_zip_session(uid, expect=session)
            if not cleared:
                continue
            if session.status_msg:
                try:
                    await session.status_msg.edit_text(
                        t(lang, "zip_expired_idle", m=cfg.zip_timeout_minutes)
                    )
                except Exception:
                    pass
            log.info("Sesión %s expirada para user %s", session.mode, uid)
        for token, selection in list(drip_selections.items()):
            if selection.busy or not selection.expired():
                continue
            if drip_selections.get(token) is not selection:
                continue
            drip_selections.pop(token, None)
            if selection.prompt:
                lang = await lang_of(selection.user_id)
                await safe_edit(
                    selection.prompt,
                    t(lang, "drip_selection_expired"),
                    reply_markup=None,
                )
        try:
            await database.prune_jobs()
        except Exception:
            log.debug("prune_jobs falló", exc_info=True)


async def main() -> None:
    global http, database
    os.makedirs(cfg.download_dir, exist_ok=True)

    database = Database(
        cfg.database_path,
        max_jobs_per_user=cfg.pending_jobs_per_user,
    )
    await database.connect()

    timeout = aiohttp.ClientTimeout(total=None, sock_connect=60, sock_read=600)
    http = aiohttp.ClientSession(timeout=timeout)

    if cfg.allowed_users:
        log.info("Whitelist: %s user(s)", len(cfg.allowed_users))
    else:
        log.info("Public bot (ALLOWED_USER_IDS empty)")
    log.info(
        "Límites: %s/user, %s global, disco mín. %s, jobs/user %s",
        cfg.max_concurrent_per_user,
        cfg.max_concurrent_global,
        human_size(cfg.min_free_disk_bytes) if cfg.min_free_disk_bytes else "off",
        cfg.pending_jobs_per_user,
    )
    if cfg.dripfiles_api_key:
        log.info(
            "DripFiles bot key: set (uploads auth by default); "
            "user keys: %s",
            "allowed" if cfg.allow_user_api_keys else "disabled",
        )
    else:
        log.info(
            "DripFiles bot key: none (free unless user /apikey); "
            "user keys: %s",
            "allowed" if cfg.allow_user_api_keys else "disabled",
        )

    await app.start()
    commands = [
        BotCommand("start", "Help / Ayuda"),
        BotCommand("help", "Help / Ayuda"),
        BotCommand("lang", "Language / Idioma"),
        BotCommand("multi", "Several files in one link"),
        BotCommand("zip", "ZIP mode"),
        BotCommand("done", "Finish multi-file / ZIP"),
        BotCommand("cancel", "Cancel multi-file / ZIP"),
    ]
    if cfg.allow_user_api_keys:
        commands.append(BotCommand("apikey", "DripFiles API key"))
    commands.extend(
        [
            BotCommand("expire", "Preferred expiry days"),
            BotCommand("dev", "Dev: wget / curl"),
            BotCommand("settings", "Settings"),
            BotCommand("me", "DripFiles account limits"),
        ]
    )
    await app.set_bot_commands(commands)
    me = await app.get_me()
    log.info("DripFilesBot @%s (db=%s)", me.username, cfg.database_path)
    janitor = asyncio.create_task(zip_janitor())
    try:
        await idle()
    finally:
        janitor.cancel()
        for uid in list(zip_sessions.keys()):
            await clear_zip_session(uid)
        await app.stop()
        await http.close()
        await database.close()


if __name__ == "__main__":
    app.loop.run_until_complete(main())