Automated sync from GitHub

.gitattributes

@@ -1,8 +1,8 @@
+*.ttf filter=lfs diff=lfs merge=lfs -text
+*.otf filter=lfs diff=lfs merge=lfs -text
+*.webp filter=lfs diff=lfs merge=lfs -text
 *.mp4 filter=lfs diff=lfs merge=lfs -text
 *.mp3 filter=lfs diff=lfs merge=lfs -text
 *.jpg filter=lfs diff=lfs merge=lfs -text
 *.jpeg filter=lfs diff=lfs merge=lfs -text
 *.png filter=lfs diff=lfs merge=lfs -text
-*.ttf filter=lfs diff=lfs merge=lfs -text
-*.otf filter=lfs diff=lfs merge=lfs -text
-*.webp filter=lfs diff=lfs merge=lfs -text

bot/app.py

@@ -27,7 +27,7 @@ from .db import MAX_FORMATS, Database
 from .errors import FormatError
 from .formats import FIELD_KINDS, NAME_MAX, FormatSettings, validate_name
 from .parser import compute_new_name, missing_variables, parse_caption, parse_info, split_ext
-from .pipeline import PROBE_EXTS, Renamer, original_name_of, source_ext
+from .pipeline import PROBE_EXTS, Renamer, embedded_thumb, original_name_of, source_ext
 from .queue import Job, JobQueue
 from .sequence import SeqItem, SequenceState, analyse, format_report
 from .settings_ops import META_COMMANDS, apply_setting, parse_channel, preview_template
@@ -1117,15 +1117,7 @@ class App:
         await self._reply(event, text + (f"\n\n{mode.dropped}" if mode.dropped else ""), rows)
 
     async def _embedded_thumb(self, msg: Any, allow_stripped: bool = False) -> Optional[bytes]:
-        """The preview Telegram keeps with a file.  The tiny blurred one (a "stripped" size) only counts when ``allow_stripped``."""
-        thumbs = getattr(getattr(msg, "document", None), "thumbs", None) or []
-        if not any(not isinstance(t, types.PhotoStrippedSize) for t in thumbs) and not (allow_stripped and thumbs):
-            return None
-        try:
-            return await self.client.download_media(msg, file=bytes, thumb=-1) or None
-        except Exception as exc:  # noqa: BLE001
-            log.debug("could not download the embedded thumbnail: %s", exc)
-            return None
+        return await embedded_thumb(self.client, msg, allow_stripped)
 
     async def _extract_thumbnail(self, event: Any, msg: Any) -> Optional[bytes]:
         data = await self._embedded_thumb(msg)
@@ -1455,7 +1447,7 @@ class App:
         )
 
     def _speed_text(self) -> str:
-        return sysinfo.format_speed(self.cfg.fast_transfer, self.cfg.connections, transfer.stats)
+        return sysinfo.format_speed(self.cfg.fast_transfer, self.cfg.connections, transfer.stats, self.cfg.stream_transfer)
 
     async def cmd_stats(self, event: Any) -> None:
         await self._reply(event, await self._stats_text())
@@ -1733,6 +1725,10 @@ class App:
             await self.db.set_format_thumb(uid, fid, None)
             await self._show_format(event, uid, fid)
             await event.answer("Thumbnail deleted.")
+        elif action == "cr":
+            await apply_setting(FormatSettings(self.db, fid), uid, "caption", "reset")
+            await self._show_format(event, uid, fid)
+            await event.answer("Caption reset.")
         elif action == "sh":  # Share format, shown while inline mode wasn't known to be on: just try the real button
             text, rows = panels.format_view(fmt, share_inline=True)
             try:
@@ -1780,7 +1776,7 @@ class App:
                 ok, value = validate_name(text)
                 if ok and field == "new":
                     fmt = await self.db.create_format(uid, value, template=(await self.db.get_settings(uid))["template"])
-                    msg = f"✅ Format <b>{esc(value)}</b> created. Now set its template, thumbnail and metadata."
+                    msg = f"✅ Format <b>{esc(value)}</b> created. Now set its template, caption, thumbnail and metadata."
                     p.fid = fmt["id"]
                 elif ok:
                     if await self.db.update_format(uid, p.fid, name=value) is None:

bot/config.py

@@ -57,6 +57,7 @@ class Config:
     max_file_mb: int
     fast_transfer: bool
     progress_interval: float
+    stream_transfer: bool = True  # big files go straight from the download into the upload instead of through the disk
 
     @property
     def max_file_bytes(self) -> int:
@@ -97,4 +98,5 @@ def load_config(env: Optional[Mapping[str, str]] = None) -> Config:
         max_file_mb=max(1, _get_int(env, "MAX_FILE_MB", 2000)),
         fast_transfer=_get_bool(env, "FAST_TRANSFER", True),
         progress_interval=max(2.0, float(env.get("PROGRESS_INTERVAL", "4") or 4)),
+        stream_transfer=_get_bool(env, "STREAM_RENAME", True),
     )

bot/db.py

@@ -41,9 +41,11 @@ OLD_DEFAULT_CAPTION = "{filename}"
 BOLD_CAPTION_MIGRATION = "migrated:bold_caption"
 
 MAX_FORMATS = 12
-# what a saved /rename format holds; everything else (mode, caption, file type, ...) comes from the user's settings
+# what a saved /rename format holds; everything else (mode, file type, ...) comes from the user's settings.  A format never
+# borrows the user's own caption: without a caption of its own it sends the new file name in bold.
 DEFAULT_FORMAT: Dict[str, Any] = {
     "template": DEFAULT_TEMPLATE,
+    "caption": DEFAULT_SETTINGS["caption"],
     "meta": {"title": "", "author": "", "artist": "", "audio": "", "subtitle": "", "video": ""},
     "keep_audio": [],
     "keep_subtitle": [],
@@ -142,8 +144,9 @@ def _merge_settings(stored: Dict[str, Any]) -> Dict[str, Any]:
 
 def _merge_format(stored: Dict[str, Any]) -> Dict[str, Any]:
     merged = copy.deepcopy(DEFAULT_FORMAT)
-    if isinstance(stored.get("template"), str) and stored["template"]:
-        merged["template"] = stored["template"]
+    for key in ("template", "caption"):
+        if isinstance(stored.get(key), str) and stored[key].strip():
+            merged[key] = stored[key]
     if isinstance(stored.get("meta"), dict):
         merged["meta"].update({k: str(v) for k, v in stored["meta"].items() if k in merged["meta"]})
     for key in ("keep_audio", "keep_subtitle"):
@@ -327,7 +330,7 @@ class Database:
         return _format_dict(fid, name, json.dumps(data, ensure_ascii=False), False)
 
     async def update_format(self, uid: int, fid: int, **changes: Any) -> Optional[Dict[str, Any]]:
-        """Change ``name``, ``template``, ``keep_audio``, ``keep_subtitle`` or merge a ``meta`` dict; ``None`` if it is gone."""
+        """Change ``name``, ``template``, ``caption``, ``keep_audio``, ``keep_subtitle`` or merge a ``meta`` dict; ``None`` if it is gone."""
         async with self._sessions() as session:
             row = await session.get(Format, fid)
             if row is None or row.user_id != uid:

bot/fasttransfer.py

@@ -17,7 +17,7 @@ import os
 import threading
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Awaitable, Callable, List, Optional, Protocol, Sequence
+from typing import Any, Awaitable, Callable, List, Optional, Protocol, Sequence, Tuple
 
 from .errors import Cancelled
 
@@ -26,6 +26,7 @@ log = logging.getLogger(__name__)
 DOWNLOAD_PART = 1024 * 1024  # upload.getFile allows up to 1 MiB per request
 UPLOAD_PART = 512 * 1024  # upload.saveBigFilePart parts are at most 512 KiB
 BIG_FILE_THRESHOLD = 10 * 1024 * 1024  # Telegram wants saveBigFilePart above 10 MB
+RELAY_BUFFER_MIN = 4  # downloaded parts a streamed rename may hold back at least (2 per upload connection otherwise)
 PART_TIMEOUT = 120.0
 MAX_RETRIES = 5
 RETRY_BASE_DELAY = 1.0
@@ -52,11 +53,18 @@ class TransferStats:
     standard_up: int = 0  # uploads by Telethon itself: parallel transfer is off, or it failed
     fallbacks: int = 0  # how often the parallel path failed and the standard one took over
     last_fallback: str = ""  # what failed last, e.g. "download: RuntimeError: ..."
+    streamed: int = 0  # renames that went straight from the download into the upload, without touching the disk
+    stream_fallbacks: int = 0  # streamed renames that failed, so the file was saved to disk first
+    last_stream_fallback: str = ""  # what failed last, e.g. "RuntimeError: ..."
 
     def fell_back(self, what: str, exc: BaseException) -> None:
         self.fallbacks += 1
         self.last_fallback = f"{what}: {type(exc).__name__}: {exc}"
 
+    def stream_failed(self, exc: BaseException) -> None:
+        self.stream_fallbacks += 1
+        self.last_stream_fallback = f"{type(exc).__name__}: {exc}"
+
 
 stats = TransferStats()
 
@@ -243,6 +251,83 @@ async def upload_parts(
         src.close()
 
 
+async def relay_parts(
+    down: Sequence[Transport],
+    up: Sequence[Transport],
+    location: Any,
+    size: int,
+    file_id: int,
+    *,
+    part_size: int = DOWNLOAD_PART,
+    send_size: int = UPLOAD_PART,
+    progress: Optional[ProgressCb] = None,
+    cancel: Optional[asyncio.Event] = None,
+) -> int:
+    """Copy ``size`` bytes of ``location`` into an upload without saving them anywhere; returns the number of upload parts.
+
+    The ``down`` connections fetch ``part_size`` pieces and hand them to the ``up`` connections through a small
+    buffer, which cut each piece into ``send_size`` parts for ``saveBigFilePart``.  Parts can be uploaded in any
+    order, so the two sides only wait for each other when the buffer is full or empty: the file takes about as long as
+    the slower direction, not as long as both together.  Progress counts what has been uploaded.
+    """
+    if size <= 0:
+        raise ValueError("there is nothing to copy")
+    if part_size % send_size:
+        raise ValueError("a download part must hold a whole number of upload parts")
+    if not down or not up:
+        raise ValueError("a streamed copy needs connections both ways")
+    n_fetch = math.ceil(size / part_size)
+    n_parts = math.ceil(size / send_size)
+    per_piece = part_size // send_size
+    buffer: asyncio.Queue = asyncio.Queue(maxsize=max(RELAY_BUFFER_MIN, 2 * len(up)))
+    fetch_counter, take_counter = itertools.count(), itertools.count()
+    sent = 0
+
+    async def fetch_worker(transport: Transport) -> None:
+        while True:
+            if cancel is not None and cancel.is_set():
+                raise Cancelled()
+            idx = next(fetch_counter)
+            if idx >= n_fetch:
+                return
+            offset = idx * part_size
+            want = min(part_size, size - offset)
+
+            async def fetch() -> bytes:
+                data = await transport.get_file(location, offset, part_size)
+                if len(data) < want:
+                    raise IOError(f"short read: got {len(data)} of {want} bytes at offset {offset}")
+                return data
+
+            data = await _retry(fetch, cancel=cancel, what=f"relay download part {idx + 1}/{n_fetch}")
+            await buffer.put((idx, data[:want]))
+
+    async def send_worker(transport: Transport) -> None:
+        nonlocal sent
+        while True:
+            if cancel is not None and cancel.is_set():
+                raise Cancelled()
+            if next(take_counter) >= n_fetch:  # every downloaded part already has a sender
+                return
+            idx, data = await buffer.get()
+            for start in range(0, len(data), send_size):
+                part = idx * per_piece + start // send_size
+                piece = data[start:start + send_size]
+                await _retry(
+                    lambda p=piece, i=part: transport.save_part(file_id, i, n_parts, p),
+                    cancel=cancel,
+                    what=f"relay upload part {part + 1}/{n_parts}",
+                )
+                sent += len(piece)
+                if progress is not None:
+                    progress(sent, size)
+
+    await _run_workers([lambda t=t: fetch_worker(t) for t in down] + [lambda t=t: send_worker(t) for t in up])
+    if sent != size:
+        raise IOError(f"relay incomplete: {sent} of {size} bytes uploaded")
+    return n_parts
+
+
 async def close_all(transports: Sequence[Transport]) -> None:
     await asyncio.gather(*(t.close() for t in transports), return_exceptions=True)
 
@@ -374,6 +459,56 @@ async def fast_upload(
     return InputFileBig(id=file_id, parts=parts, name=file_name)
 
 
+async def open_relay_transports(
+    client: Any, dc_id: int, connections: int
+) -> Tuple[List[TelethonTransport], List[TelethonTransport]]:
+    """Connections to fetch from ``dc_id`` and connections to upload to our own DC, opened together."""
+    results = await asyncio.gather(
+        open_transports(client, dc_id, connections),
+        open_transports(client, client.session.dc_id, connections),
+        return_exceptions=True,
+    )
+    failed = next((r for r in results if isinstance(r, BaseException)), None)
+    if failed is not None:
+        for r in results:
+            if not isinstance(r, BaseException):
+                await close_all(r)
+        raise failed
+    return results[0], results[1]  # type: ignore[return-value]
+
+
+async def relay(
+    client: Any,
+    document: Any,
+    file_name: str,
+    *,
+    connections: int = 8,
+    progress: Optional[ProgressCb] = None,
+    cancel: Optional[asyncio.Event] = None,
+) -> Any:
+    """Upload a copy of ``document`` named ``file_name`` while it downloads, without saving it; returns an ``InputFileBig``.
+
+    Telethon has no equivalent to fall back to here: when this fails the caller downloads and uploads the usual way.
+    """
+    from telethon import utils
+    from telethon.tl.types import InputFileBig
+
+    dc_id, location = utils.get_input_location(document)
+    size = int(document.size)
+    file_id = int.from_bytes(os.urandom(8), "big", signed=True)
+    down, up = await open_relay_transports(client, dc_id, connections)
+    try:
+        parts = await relay_parts(down, up, location, size, file_id, progress=progress, cancel=cancel)
+    finally:
+        await close_all(down + up)
+    return InputFileBig(id=file_id, parts=parts, name=file_name)
+
+
+def is_file_part_error(exc: BaseException) -> bool:
+    """Is this Telegram refusing an upload because of its parts (one missing, or the wrong size or number of them)?"""
+    return type(exc).__name__.startswith("FilePart")
+
+
 # ---------------------------------------------------------------------------
 # public wrappers with fallback
 # ---------------------------------------------------------------------------

bot/formats.py

@@ -1,8 +1,9 @@
 """Saved /rename formats: the pieces shared by the menus, the handlers and the pipeline.
 
-A format is a named bundle of a naming template, a thumbnail, the six metadata titles and the Keep Audio /
-Keep Subtitle lists.  Every other setting (mode, caption template, file type, remove words, target channel)
-still comes from the user's normal auto-rename settings.
+A format is a named bundle of a naming template, a caption, a thumbnail, the six metadata titles and the Keep
+Audio / Keep Subtitle lists.  Every other setting (mode, file type, remove words, target channel) still comes
+from the user's normal auto-rename settings.  The caption never does: a format that was given none sends the new
+file name in bold, whatever caption the user set for their own renames.
 """
 from __future__ import annotations
 
@@ -14,12 +15,12 @@ from .db import DEFAULT_SETTINGS, Database
 NAME_MAX = 30
 # format field -> the kind settings_ops.apply_setting() understands
 FIELD_KINDS: Dict[str, str] = {
-    "template": "template",
+    "template": "template", "caption": "caption",
     "title": "mt_title", "author": "mt_author", "artist": "mt_artist",
     "audio": "mt_audio", "subtitle": "mt_subtitle", "video": "mt_video",
     "keep_audio": "keep_audio", "keep_subtitle": "keep_subtitle",
 }
-_EDITABLE = ("template", "meta", "keep_audio", "keep_subtitle")
+_EDITABLE = ("template", "caption", "meta", "keep_audio", "keep_subtitle")
 
 
 def validate_name(text: Optional[str]) -> Tuple[bool, str]:
@@ -37,13 +38,14 @@ def has_metadata(fmt: Dict[str, Any]) -> bool:
 
 
 def apply_format(settings: Dict[str, Any], fmt: Dict[str, Any]) -> Dict[str, Any]:
-    """The user's settings with the format's template, metadata and keep lists laid over them.
+    """The user's settings with the format's template, caption, metadata and keep lists laid over them.
 
     Metadata is on whenever the format sets anything, whatever the global Metadata switch says: the format
     carries its own metadata, and a format with none leaves the file alone (no ffmpeg pass).
     """
     out = copy.deepcopy(settings)
     out["template"] = fmt["template"]
+    out["caption"] = fmt.get("caption") or DEFAULT_SETTINGS["caption"]  # never the user's own caption
     out["meta"] = dict(fmt["meta"])
     out["keep_audio"] = list(fmt["keep_audio"])
     out["keep_subtitle"] = list(fmt["keep_subtitle"])
@@ -67,7 +69,10 @@ class FormatSettings:
         if fmt is None:
             raise LookupError("that format no longer exists")
         out = copy.deepcopy(DEFAULT_SETTINGS)
-        out.update(template=fmt["template"], keep_audio=list(fmt["keep_audio"]), keep_subtitle=list(fmt["keep_subtitle"]), metadata_on=True)
+        out.update(
+            template=fmt["template"], caption=fmt["caption"], keep_audio=list(fmt["keep_audio"]),
+            keep_subtitle=list(fmt["keep_subtitle"]), metadata_on=True,
+        )
         out["meta"] = dict(fmt["meta"])
         return out
 

bot/main.py

@@ -71,9 +71,9 @@ async def amain() -> int:
     channel_log = errorlog.install(app._log_send, errorlog.secrets_from(cfg)) if cfg.log_channel else None  # errors -> LOG_CHANNEL
     log.info("@%s is online (DC %s, %d connection(s) per transfer, fast transfer %s)",
              me.username, client.session.dc_id, cfg.connections, "on" if cfg.fast_transfer else "off")
-    speed = sysinfo.speed_summary(cfg.fast_transfer, cfg.connections)
+    speed = sysinfo.speed_summary(cfg.fast_transfer, cfg.connections, cfg.stream_transfer)
     log.info("speed: %s", speed)
-    for _short, detail, fine in sysinfo.speed_checks(cfg.fast_transfer, cfg.connections):
+    for _short, detail, fine in sysinfo.speed_checks(cfg.fast_transfer, cfg.connections, cfg.stream_transfer):
         if not fine:
             log.warning("speed: %s", detail)
     if cfg.log_channel:

bot/panels.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 
 from typing import Any, Dict, Iterable, List, Sequence, Tuple
 
+from .db import DEFAULT_SETTINGS
 from .media import LANGUAGES, canonical_lang
 from .parser import CAPTION_VARS, TEMPLATE_VARS
 from .utils import clip, code_each, esc
@@ -207,7 +208,7 @@ def help_panel() -> Panel:
     text = (
         "<b>How to use</b>\n\n"
         "1. Send a file — I rename it with your template and send it back.\n"
-        "2. /formats — save up to 12 named formats (template, thumbnail, metadata, tracks) and share them; "
+        "2. /formats — save up to 12 named formats (template, caption, thumbnail, metadata, tracks) and share them; "
         "/rename — pick one, then send a file (send /rename again for the next one).\n"
         "3. /template and /caption — set the naming and caption templates.\n"
         "4. /mode — read details from the file name or the caption.\n"
@@ -267,7 +268,7 @@ def _format_buttons(formats: Sequence[Dict[str, Any]], prefix: str) -> List[Row]
 def formats_menu(formats: Sequence[Dict[str, Any]], limit: int) -> Panel:
     text = (
         f"<b>🗂 FORMATS</b> ({len(formats)}/{limit})\n\n"
-        "A format saves its own template, thumbnail, metadata titles and Keep Audio / Keep Subtitle lists "
+        "A format saves its own template, caption, thumbnail, metadata titles and Keep Audio / Keep Subtitle lists "
         "under a name, like <i>One Piece</i> or <i>Dragon Ball</i>.\n"
         "Tap one to view or edit it, then use /rename to apply it to your files."
     )
@@ -281,11 +282,17 @@ def formats_menu(formats: Sequence[Dict[str, Any]], limit: int) -> Panel:
     return text, rows + nav("home")
 
 
+def _custom_caption(fmt: Dict[str, Any]) -> bool:
+    """Has the format a caption of its own?  Otherwise it sends the new file name in bold."""
+    return fmt["caption"] != DEFAULT_SETTINGS["caption"]
+
+
 def _format_lines(fmt: Dict[str, Any], *, thumb_yes: str, thumb_no: str) -> List[str]:
     """What a format holds, one line per field."""
     m = fmt["meta"]
     lines = [
         f"{DOT} <b>Template</b> ▸ {_code(fmt['template'], 60)}",
+        f"{DOT} <b>Caption</b> ▸ {_code(fmt['caption'], 60)}" + ("" if _custom_caption(fmt) else " · default"),
         f"{DOT} <b>Thumbnail</b> ▸ " + (thumb_yes if fmt.get("has_thumb") else thumb_no),
     ]
     for key, label in FORMAT_FIELD_LABELS:
@@ -300,14 +307,18 @@ def format_view(fmt: Dict[str, Any], *, share_inline: bool = False) -> Panel:
     while inline mode is on for the bot (Telegram refuses the message otherwise), so without it the button explains how."""
     lines = [f"<b>🗂 {esc(clip(fmt['name'], 30))}</b>", ""]
     lines += _format_lines(fmt, thumb_yes="✅ saved", thumb_no="— (your default thumbnail is used)")
-    lines += ["", "Use it with /rename. Mode, caption, file type and the rest come from your normal settings."]
+    lines += ["", "Use it with /rename. Mode, file type and the rest come from your normal settings."]
     fid = fmt["id"]
     thumb_row: Row = [("Thumbnail", f"fm:th:{fid}")]
     if fmt.get("has_thumb"):
         thumb_row.append(("🗑 Delete Thumb", f"fm:thd:{fid}"))
     rows: List[Row] = [
-        [("Name", f"fm:in:{fid}:name"), ("Template", f"fm:in:{fid}:template")],
+        [("Name", f"fm:in:{fid}:name"), ("Template", f"fm:in:{fid}:template"), ("Caption", f"fm:in:{fid}:caption")],
         thumb_row,
+    ]
+    if _custom_caption(fmt):
+        rows.append([("↩ Reset Caption", f"fm:cr:{fid}")])
+    rows += [
         [("Title", f"fm:in:{fid}:title"), ("Author", f"fm:in:{fid}:author"), ("Artist", f"fm:in:{fid}:artist")],
         [("Audio", f"fm:in:{fid}:audio"), ("Subtitle", f"fm:in:{fid}:subtitle"), ("Video", f"fm:in:{fid}:video")],
         [("Keep Audio", f"fm:in:{fid}:keep_audio"), ("Keep Subtitle", f"fm:in:{fid}:keep_subtitle")],
@@ -350,7 +361,7 @@ def shared_added(fmt: Dict[str, Any]) -> Panel:
 
 
 def format_delete_confirm(fmt: Dict[str, Any]) -> Panel:
-    text = f"<b>Delete the format “{esc(clip(fmt['name'], 30))}”?</b>\n\nIts template, thumbnail and metadata are removed. This can't be undone."
+    text = f"<b>Delete the format “{esc(clip(fmt['name'], 30))}”?</b>\n\nIts template, caption, thumbnail and metadata are removed. This can't be undone."
     return text, [[("🗑 Yes, delete", f"fm:ok:{fmt['id']}"), ("◀ Keep it", f"fm:v:{fmt['id']}")]]
 
 
@@ -358,7 +369,7 @@ def rename_picker(formats: Sequence[Dict[str, Any]]) -> Panel:
     if not formats:
         text = (
             "<b>🏷 RENAME</b>\n\nYou have no saved formats yet.\n"
-            "A format keeps a template, thumbnail and metadata under a name, like <i>One Piece</i>. Create one, then come back here."
+            "A format keeps a template, caption, thumbnail and metadata under a name, like <i>One Piece</i>. Create one, then come back here."
         )
         return text, [[("➕ Create a format", "fm:new")], [("✖ Close", "x:close")]]
     text = "<b>🏷 RENAME</b>\n\nChoose the format to rename with. I'll ask for your files next."
@@ -536,6 +547,12 @@ FORMAT_PROMPTS: Dict[str, str] = {
     "name": "<b>Send the new name</b> for this format (30 characters max).",
     "thumb": "<b>Send a photo</b> to use as this format's thumbnail.",
     "template": PROMPTS["template"],
+    "caption": (
+        "<b>Send the caption for this format.</b>\n\n"
+        f"Variables: {CAPTION_VARIABLES}\n"
+        "Telegram HTML works (<code>&lt;b&gt;</code>, <code>&lt;i&gt;</code>, <code>&lt;code&gt;</code>).\n"
+        "Send <code>reset</code> to go back to the default: the new file name in bold."
+    ),
     "title": PROMPTS["mt_title"], "author": PROMPTS["mt_author"], "artist": PROMPTS["mt_artist"],
     "audio": PROMPTS["mt_audio"], "subtitle": PROMPTS["mt_subtitle"], "video": PROMPTS["mt_video"],
     "keep_audio": PROMPTS["keep_audio"], "keep_subtitle": PROMPTS["keep_subtitle"],

bot/pipeline.py

@@ -1,12 +1,13 @@
-"""The rename pipeline: download -> (ffmpeg) -> upload."""
+"""The rename pipeline: download -> (ffmpeg) -> upload, or for a plain rename of a big file a copy straight from the download into the upload."""
 from __future__ import annotations
 
 import logging
 import mimetypes
 import shutil
 import time
+from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 from telethon import errors, types
 
@@ -65,6 +66,47 @@ def original_name_of(message: Any) -> str:
     return f"{kind}_{message.id}{(f.ext if f is not None else '') or ''}"
 
 
+def make_caption(template: str, *, new_name: str, orig_name: str, ext: str, size: int, duration: float) -> Optional[str]:
+    """The caption of a renamed file: the user's (or the format's) template with its variables filled in."""
+    parsed = parse_info(split_ext(orig_name)[0])
+    values = build_values(parsed, split_ext(orig_name)[0], ext)
+    values.update(
+        filename=split_ext(new_name)[0], original=orig_name, ext=ext.lstrip("."),
+        size=human_size(size), duration=human_duration(duration) if duration else "",
+    )
+    return render_caption(template, values)[:1000] or None
+
+
+def source_attributes(message: Any) -> Tuple[Optional[Any], Optional[Any]]:
+    """The video and the audio details Telegram keeps with a file: ``(DocumentAttributeVideo, DocumentAttributeAudio)``, ``None`` for a missing one."""
+    attributes = getattr(getattr(message, "document", None), "attributes", None) or []
+    video = next((a for a in attributes if isinstance(a, types.DocumentAttributeVideo) and not a.round_message), None)
+    audio = next((a for a in attributes if isinstance(a, types.DocumentAttributeAudio) and not a.voice), None)
+    return video, audio
+
+
+async def embedded_thumb(client: Any, msg: Any, allow_stripped: bool = False) -> Optional[bytes]:
+    """The preview Telegram keeps with a file.  The tiny blurred one (a "stripped" size) only counts when ``allow_stripped``."""
+    thumbs = getattr(getattr(msg, "document", None), "thumbs", None) or []
+    if not any(not isinstance(t, types.PhotoStrippedSize) for t in thumbs) and not (allow_stripped and thumbs):
+        return None
+    try:
+        return await client.download_media(msg, file=bytes, thumb=-1) or None
+    except Exception as exc:  # noqa: BLE001
+        log.debug("could not download the embedded thumbnail: %s", exc)
+        return None
+
+
+@dataclass
+class StreamPlan:
+    """What a streamed rename sends along with the file, worked out without ever reading the file."""
+
+    attrs: List[Any]
+    as_document: bool
+    thumb: Optional[bytes]
+    duration: float
+
+
 class Renamer:
     def __init__(self, client: Any, cfg: Config, db: Database):
         self.client = client
@@ -111,6 +153,134 @@ class Renamer:
             errorlog.error_context.reset(context)
             shutil.rmtree(work, ignore_errors=True)
 
+    async def _custom_thumb(self, job: Job) -> Optional[bytes]:
+        """The thumbnail the user chose: the one of the format being used, else their default one."""
+        thumb = await self.db.get_format_thumb(job.user_id, job.fmt["id"]) if job.fmt else None
+        if thumb is None:
+            thumb = await self.db.get_thumb(job.user_id)
+        return thumb
+
+    async def _stream_plan(
+        self, job: Job, s: Dict[str, Any], msg: Any, ext: str, send_type: str, size: int
+    ) -> Optional[StreamPlan]:
+        """What a streamed rename would send, or ``None`` when the file has to be looked at first (it is then saved and uploaded as usual).
+
+        Streaming never saves the file, so nothing can read it: the video and audio details come from the ones Telegram keeps
+        with the original, the thumbnail from the user's setting or the original's own preview.  A rename that changes the
+        file itself (metadata and tracks) cannot be streamed either.
+        """
+        if not (self.cfg.fast_transfer and self.cfg.stream_transfer):
+            return None
+        video, audio = source_attributes(msg)
+        source = video if send_type == "video" else audio if send_type == "audio" else (video or audio)
+        duration = float(getattr(source, "duration", 0) or 0)
+        reason = None
+        if getattr(msg, "document", None) is None:
+            reason = "it is not a document"
+        elif size <= transfer.BIG_FILE_THRESHOLD:
+            reason = "it is small, so there is little to gain"
+        elif s["metadata_on"]:
+            reason = "metadata and tracks change the file"
+        elif send_type == "video" and video is None:
+            reason = "Telegram has no video details to reuse"
+        elif send_type == "audio" and audio is None:
+            reason = "Telegram has no audio details to reuse"
+        elif "{duration}" in (s["caption"] or "") and not duration:
+            reason = "the caption wants the duration and only the file can tell"
+        thumb = None
+        if reason is None:
+            thumb = await self._custom_thumb(job)
+            if thumb is None and ext.lower() in media.VIDEO_CONTAINERS:  # the usual way makes one from a frame of the file
+                thumb = await embedded_thumb(self.client, msg)
+                if thumb is None:
+                    reason = "there is no thumbnail to reuse"
+        if reason is not None:
+            log.info("%s is not streamed: %s", job.new_name, reason)
+            return None
+        attrs: List[Any] = [types.DocumentAttributeFilename(job.new_name)]
+        if send_type == "video":
+            attrs.append(types.DocumentAttributeVideo(duration=video.duration or 0, w=video.w or 1, h=video.h or 1, supports_streaming=True))
+        elif send_type == "audio":
+            attrs.append(
+                types.DocumentAttributeAudio(
+                    duration=int(audio.duration or 0), title=audio.title or None, performer=audio.performer or None
+                )
+            )
+        return StreamPlan(attrs=attrs, as_document=(send_type == "document"), thumb=thumb, duration=duration)
+
+    async def _stream(
+        self, job: Job, s: Dict[str, Any], msg: Any, orig_name: str, ext: str, send_type: str, size: int,
+        reporter: ProgressReporter, started: float,
+    ) -> bool:
+        """Rename by copying the file from the download straight into the upload: nothing is saved, so the whole takes about as long as
+        the slower of the two.  ``True`` when the file was sent; ``False`` when it cannot be streamed or streaming failed (the usual way follows)."""
+        plan = await self._stream_plan(job, s, msg, ext, send_type, size)
+        if plan is None:
+            return False
+        reporter.set_stage("⬇️⬆️ Streaming", size)
+        began = time.monotonic()
+        try:
+            input_file = await transfer.relay(
+                self.client, msg.document, job.new_name, connections=self.cfg.connections, progress=reporter, cancel=job.cancel
+            )
+        except Cancelled:
+            raise
+        except Exception as exc:  # noqa: BLE001
+            self._stream_failed(job, exc)
+            return False
+        seconds = time.monotonic() - began
+        if job.cancel.is_set():
+            raise Cancelled()
+        rates = f" · ⬇⬆ {human_size(size / seconds)}/s streamed" if seconds >= 1 else ""
+        caption = make_caption(s["caption"], new_name=job.new_name, orig_name=orig_name, ext=ext, size=size, duration=plan.duration)
+        try:
+            await self._deliver(
+                job, s, msg, input_file, caption=caption, attrs=plan.attrs, as_document=plan.as_document, thumb=plan.thumb,
+                final_size=size, warnings=list(job.notes), rates=rates, started=started, reporter=reporter,
+            )
+        except Exception as exc:  # noqa: BLE001
+            if not transfer.is_file_part_error(exc):  # only Telegram not accepting the parts is a streaming problem
+                raise
+            self._stream_failed(job, exc)
+            return False
+        transfer.stats.streamed += 1
+        return True
+
+    def _stream_failed(self, job: Job, exc: BaseException) -> None:
+        log.error("streaming %s failed (%s: %s); doing it the usual way", job.new_name, type(exc).__name__, exc, exc_info=exc)
+        transfer.stats.stream_failed(exc)
+        job.notes.append("Streaming didn't work for this file, so it was saved first and then uploaded.")
+
+    async def _deliver(
+        self, job: Job, s: Dict[str, Any], msg: Any, input_file: Any, *, caption: Optional[str], attrs: List[Any],
+        as_document: bool, thumb: Optional[bytes], final_size: int, warnings: List[str], rates: str, started: float,
+        reporter: ProgressReporter,
+    ) -> None:
+        """Send the uploaded file to its chat, then count it and say it is done."""
+        target = s.get("target_channel") or job.chat_id
+        kwargs: Dict[str, Any] = dict(
+            caption=caption, attributes=attrs, mime_type=guess_mime(job.new_name, getattr(msg.file, "mime_type", None)),
+            force_document=as_document, thumb=thumb, parse_mode="html",
+        )
+        if target == job.chat_id:
+            kwargs["reply_to"] = job.msg_id
+        try:
+            await self.client.send_file(target, input_file, **kwargs)
+        except (errors.EntityBoundsInvalidError, errors.MediaCaptionTooLongError, errors.MessageTooLongError):
+            # the caption template produced something Telegram rejects: send the file without a caption
+            kwargs.update(caption=None)
+            await self.client.send_file(target, input_file, **kwargs)
+            warnings.append("The caption was rejected by Telegram (check your caption template), so it was left out.")
+
+        await self.db.record_job(job.user_id, final_size)
+        job.state = "done"
+        took = time.monotonic() - started
+        note = "".join(f"\n⚠️ {esc(w)}" for w in warnings)
+        where = "" if target == job.chat_id else "\n📣 Sent to your channel."
+        await reporter.finish(
+            f"✅ <b>Done</b> in {human_duration(took)}\n<code>{esc(clip(job.new_name, 120))}</code>\n{human_size(final_size)}{rates}{where}{note}"
+        )
+
     async def _run(self, job: Job, s: Dict[str, Any], work: Path, reporter: ProgressReporter, started: float) -> None:
         client, cfg = self.client, self.cfg
         msg = await self._fresh_message(job)
@@ -120,6 +290,8 @@ class Renamer:
         size = int(msg.file.size or job.size or 0)
         src = work / f"src{ext}"
 
+        if await self._stream(job, s, msg, orig_name, ext, send_type, size, reporter, started):
+            return
         self._check_space(work, size, will_remux=bool(s["metadata_on"]))
 
         # ---- download ------------------------------------------------------------
@@ -177,22 +349,17 @@ class Renamer:
                 )
             )
 
-        thumb = await self.db.get_format_thumb(job.user_id, job.fmt["id"]) if job.fmt else None
-        if thumb is None:
-            thumb = await self.db.get_thumb(job.user_id)
+        thumb = await self._custom_thumb(job)
         if thumb is None and ext.lower() in media.VIDEO_CONTAINERS and info is not None and info.of("video"):
             thumb_path = work / "thumb.jpg"
             if await media.make_video_thumbnail(current, thumb_path, info.duration, job.cancel):
                 thumb = thumb_path.read_bytes()
 
         # ---- caption ---------------------------------------------------------------------
-        parsed = parse_info(split_ext(orig_name)[0])
-        values = build_values(parsed, split_ext(orig_name)[0], ext)
-        values.update(
-            filename=split_ext(job.new_name)[0], original=orig_name, ext=ext.lstrip("."),
-            size=human_size(final_size), duration=human_duration(info.duration) if info and info.duration else "",
+        caption = make_caption(
+            s["caption"], new_name=job.new_name, orig_name=orig_name, ext=ext, size=final_size,
+            duration=(info.duration if info else 0) or 0,
         )
-        caption = render_caption(s["caption"], values)[:1000] or None
 
         # ---- upload ---------------------------------------------------------------------------
         reporter.set_stage("⬆️ Uploading", final_size)
@@ -204,31 +371,12 @@ class Renamer:
         up_seconds = time.monotonic() - began
         if job.cancel.is_set():
             raise Cancelled()
-        target = s.get("target_channel") or job.chat_id
-        kwargs: Dict[str, Any] = dict(
-            caption=caption, attributes=attrs, mime_type=guess_mime(job.new_name, getattr(msg.file, "mime_type", None)),
-            force_document=(send_type == "document"), thumb=thumb, parse_mode="html",
-        )
-        if target == job.chat_id:
-            kwargs["reply_to"] = job.msg_id
-        try:
-            await client.send_file(target, input_file, **kwargs)
-        except (errors.EntityBoundsInvalidError, errors.MediaCaptionTooLongError, errors.MessageTooLongError):
-            # the caption template produced something Telegram rejects: send the file without a caption
-            kwargs.update(caption=None)
-            await client.send_file(target, input_file, **kwargs)
-            warnings.append("The caption was rejected by Telegram (check your caption template), so it was left out.")
-
-        await self.db.record_job(job.user_id, final_size)
-        job.state = "done"
-        took = time.monotonic() - started
-        note = "".join(f"\n⚠️ {esc(w)}" for w in warnings)
-        where = "" if target == job.chat_id else "\n📣 Sent to your channel."
         rates = "".join(  # how fast each way went; a transfer under a second says nothing
             f" · {arrow} {human_size(amount / seconds)}/s"
             for arrow, amount, seconds in (("⬇", size, down_seconds), ("⬆", final_size, up_seconds))
             if seconds >= 1 and amount > 0
         )
-        await reporter.finish(
-            f"✅ <b>Done</b> in {human_duration(took)}\n<code>{esc(clip(job.new_name, 120))}</code>\n{human_size(final_size)}{rates}{where}{note}"
+        await self._deliver(
+            job, s, msg, input_file, caption=caption, attrs=attrs, as_document=(send_type == "document"), thumb=thumb,
+            final_size=final_size, warnings=warnings, rates=rates, started=started, reporter=reporter,
         )

bot/sysinfo.py

@@ -140,8 +140,8 @@ def loop_backend() -> str:
     return "uvloop" if type(loop).__module__.split(".")[0] == "uvloop" else "asyncio"
 
 
-def speed_checks(parallel: bool, connections: int) -> List[Tuple[str, str, bool]]:
-    """``(short name, detail, fine)`` for the three things transfer speed depends on."""
+def speed_checks(parallel: bool, connections: int, stream: Optional[bool] = None) -> List[Tuple[str, str, bool]]:
+    """``(short name, detail, fine)`` for the things transfer speed depends on; ``stream`` (when given) adds the streaming rename."""
     crypto, loop = crypto_backend(), loop_backend()
     if crypto == "cryptg":
         encryption = ("cryptg", "cryptg", True)
@@ -151,28 +151,43 @@ def speed_checks(parallel: bool, connections: int) -> List[Tuple[str, str, bool]
         encryption = ("slow Python crypto", "pure Python, slow: cryptg is missing", False)
     event_loop = ("uvloop", "uvloop", True) if loop == "uvloop" else ("no uvloop", f"{loop} loop: uvloop is not running", False)
     transfers = (f"parallel ×{connections}", f"on, {connections} connections", True) if parallel else ("parallel off", "off (FAST_TRANSFER=0)", False)
-    return [encryption, event_loop, transfers]
+    checks = [encryption, event_loop, transfers]
+    if stream is not None:
+        if not parallel:
+            checks.append(("no streaming", "off: it needs the parallel transfer", False))
+        elif stream:
+            checks.append(("streaming", "on: big files are uploaded while they download", True))
+        else:
+            checks.append(("no streaming", "off (STREAM_RENAME=0)", False))
+    return checks
 
 
-def speed_summary(parallel: bool, connections: int) -> str:
-    """One line for the start-up notice: ``cryptg ✅ · uvloop ✅ · parallel ×8 ✅``."""
-    return " · ".join(f"{short} {'✅' if fine else '⚠️'}" for short, _detail, fine in speed_checks(parallel, connections))
+def speed_summary(parallel: bool, connections: int, stream: Optional[bool] = None) -> str:
+    """One line for the start-up notice: ``cryptg ✅ · uvloop ✅ · parallel ×8 ✅ · streaming ✅``."""
+    return " · ".join(f"{short} {'✅' if fine else '⚠️'}" for short, _detail, fine in speed_checks(parallel, connections, stream))
 
 
-def format_speed(parallel: bool, connections: int, stats: Any) -> str:
+def format_speed(parallel: bool, connections: int, stats: Any, stream: Optional[bool] = None) -> str:
     """The Speed section of /status: what is switched on, and how the transfers since the start went."""
-    labels = ("🔐 Encryption", "🔁 Event loop", "🚀 Parallel transfer")
+    labels = ("🔐 Encryption", "🔁 Event loop", "🚀 Parallel transfer", "⬇⬆ Streaming rename")
     lines = ["⚡ <b>Speed</b>"]
-    for label, (_short, detail, fine) in zip(labels, speed_checks(parallel, connections)):
+    for label, (_short, detail, fine) in zip(labels, speed_checks(parallel, connections, stream)):
         lines.append(f"{label}: {esc(detail)} {'✅' if fine else '⚠️'}")
     small = f"{BIG_FILE_THRESHOLD // 2 ** 20} MB"
     lines.append(f"⬇ <b>Downloads:</b> {stats.fast_down} parallel · {stats.standard_down} standard")
     lines.append(f"⬆ <b>Uploads:</b> {stats.fast_up} parallel · {stats.small_up} up to {small} · {stats.standard_up} standard")
+    if stream is not None:
+        lines.append(f"⬇⬆ <b>Streamed renames:</b> {stats.streamed}")
     if stats.fallbacks:
         lines.append(
             f"⚠️ The parallel transfer failed {stats.fallbacks}× and the standard, slower one took over. "
             f"Last: <code>{esc(clip(stats.last_fallback, 160))}</code>"
         )
+    if stats.stream_fallbacks:
+        lines.append(
+            f"⚠️ Streaming failed {stats.stream_fallbacks}× and those files were saved first and then uploaded. "
+            f"Last: <code>{esc(clip(stats.last_stream_fallback, 160))}</code>"
+        )
     return "\n".join(lines)
 
 

tests/test_fasttransfer.py

@@ -394,3 +394,219 @@ async def test_file_helper_without_positional_io(tmp_path):
     r._positional = False
     assert r.pread(10, 0) == b"helloworld"
     r.close()
+
+
+# ---------------------------------------------------------------------------
+# streamed copy: the download goes straight into the upload, nothing is saved
+# ---------------------------------------------------------------------------
+
+MIB = 1024 * 1024
+
+
+@pytest.mark.parametrize("size", [1, ft.UPLOAD_PART - 1, ft.UPLOAD_PART, ft.UPLOAD_PART + 1, MIB - 1, MIB, MIB + 1, 3 * MIB // 2 + 7, 5 * MIB + 12345])
+async def test_relay_copies_a_file_part_for_part(size):
+    blob = blob_of(size, seed=11)
+    store = {}
+    down = [FakeTransport(blob) for _ in range(3)]
+    up = [FakeTransport(store=store) for _ in range(3)]
+    seen = []
+    n = await ft.relay_parts(down, up, "loc", size, 4321, progress=lambda c, t: seen.append((c, t)))
+    expected = -(-size // ft.UPLOAD_PART)
+    assert n == expected and sorted(store) == list(range(expected))
+    assert all(fid == 4321 and total == expected for fid, total, _d in store.values())
+    sizes = [len(store[i][2]) for i in range(expected)]
+    assert all(s == ft.UPLOAD_PART for s in sizes[:-1])  # every part but the last is full size, whatever the download pieces were
+    assert b"".join(store[i][2] for i in range(expected)) == blob
+    assert seen[-1] == (size, size) and all(t == size for _c, t in seen)
+    assert [c for c, _t in seen] == sorted(c for c, _t in seen)  # progress only ever moves forward
+
+
+async def test_relay_fetches_every_part_once_and_both_sides_share_the_work():
+    size = 12 * MIB
+    blob = blob_of(size)
+    store = {}
+    down = [FakeTransport(blob) for _ in range(4)]
+    up = [FakeTransport(store=store) for _ in range(4)]
+    await ft.relay_parts(down, up, "loc", size, 1)
+    assert sorted(offset for t in down for offset, _limit in t.requests) == [i * MIB for i in range(12)]
+    assert sum(t.calls for t in up) == 24  # two upload parts for every downloaded one
+    assert sum(1 for t in down if t.calls) > 1 and sum(1 for t in up if t.calls) > 1
+
+
+async def test_relay_with_far_more_connections_than_parts_still_finishes():
+    blob = blob_of(MIB + 100)  # two downloaded parts, three upload parts
+    store = {}
+    down = [FakeTransport(blob) for _ in range(8)]
+    up = [FakeTransport(store=store) for _ in range(8)]
+    n = await asyncio.wait_for(ft.relay_parts(down, up, "loc", len(blob), 5), 5)
+    assert n == 3 and b"".join(store[i][2] for i in sorted(store)) == blob
+
+
+async def test_relay_does_not_run_ahead_of_a_slow_upload():
+    parts, fetched, completed, ahead = 40, 0, 0, 0
+    blob = blob_of(parts * MIB)
+
+    class Down(FakeTransport):
+        async def get_file(self, location, offset, limit):
+            nonlocal fetched, ahead
+            data = await super().get_file(location, offset, limit)
+            fetched += 1
+            ahead = max(ahead, fetched - completed)  # downloaded parts that are not sent yet
+            return data
+
+    class Up(FakeTransport):
+        async def save_part(self, file_id, part, total_parts, data):
+            nonlocal completed
+            await asyncio.sleep(0.005)
+            await super().save_part(file_id, part, total_parts, data)
+            if part % 2 == 1:  # the second half of a downloaded part: that part is sent
+                completed += 1
+
+    store = {}
+    down, up = [Down(blob) for _ in range(4)], [Up(store=store) for _ in range(2)]
+    await ft.relay_parts(down, up, "loc", parts * MIB, 1)
+    held_back = max(ft.RELAY_BUFFER_MIN, 2 * len(up))  # the buffer, plus a part in each hand
+    assert ahead <= held_back + len(down) + len(up) and ahead < parts // 2  # memory stays small however big the file is
+    assert b"".join(store[i][2] for i in sorted(store)) == blob
+
+
+async def test_relay_retries_failures_on_both_sides():
+    size = 3 * MIB + 7
+    blob = blob_of(size)
+    store = {}
+    down = [FakeTransport(blob, fail_first=2), FakeTransport(blob, fail_first=3, short_once=True)]
+    up = [FakeTransport(store=store, fail_first=2), FakeTransport(store=store)]
+    await ft.relay_parts(down, up, "loc", size, 1)
+    assert b"".join(store[i][2] for i in sorted(store)) == blob
+
+
+async def test_relay_gives_up_when_a_connection_keeps_failing():
+    blob = blob_of(2 * MIB)
+    broken = FakeTransport(store={}, fail_first=10_000)
+    with pytest.raises(ConnectionError):
+        await asyncio.wait_for(ft.relay_parts([FakeTransport(blob)], [broken], "loc", len(blob), 1), 10)
+    assert broken.calls == ft.MAX_RETRIES + 1
+
+
+@pytest.mark.parametrize("broken_side", ["down", "up"])
+async def test_relay_stops_both_sides_when_one_fails_for_good(broken_side):
+    class FileReferenceExpiredError(Exception):  # retrying cannot help
+        pass
+
+    class LateFatalUp(FakeTransport):
+        async def save_part(self, file_id, part, total_parts, data):
+            await asyncio.sleep(0.05)  # long enough for the downloads to fill the buffer and wait
+            raise FileReferenceExpiredError("expired")
+
+    blob = blob_of(40 * MIB)
+    if broken_side == "down":
+        down, up = [FakeTransport(blob, fatal=FileReferenceExpiredError("expired"))], [FakeTransport(store={}) for _ in range(2)]
+    else:
+        down, up = [FakeTransport(blob) for _ in range(3)], [LateFatalUp() for _ in range(2)]
+    with pytest.raises(FileReferenceExpiredError):  # and does not hang on a full or an empty buffer
+        await asyncio.wait_for(ft.relay_parts(down, up, "loc", len(blob), 1), 5)
+    assert sum(t.calls for t in down) < 40  # the download did not carry on to the end
+
+
+async def test_relay_cancel_stops_both_sides():
+    cancel = asyncio.Event()
+
+    class SlowDown(FakeTransport):
+        async def get_file(self, location, offset, limit):
+            await asyncio.sleep(0.005)
+            if offset >= 3 * MIB:
+                cancel.set()
+            return b"\0" * limit
+
+    up = [FakeTransport(store={}) for _ in range(3)]
+    with pytest.raises(Cancelled):
+        await asyncio.wait_for(ft.relay_parts([SlowDown() for _ in range(3)], up, "loc", 64 * MIB, 1, cancel=cancel), 5)
+    assert sum(t.calls for t in up) < 2 * 64
+    cancel.set()  # cancelled before it starts
+    untouched = FakeTransport(store={})
+    with pytest.raises(Cancelled):
+        await ft.relay_parts([FakeTransport(b"x" * MIB)], [untouched], "loc", MIB, 1, cancel=cancel)
+    assert untouched.calls == 0
+
+
+async def test_relay_refuses_what_it_cannot_do():
+    ok = [FakeTransport(b"x" * 10)]
+    with pytest.raises(ValueError):  # an empty file has no parts
+        await ft.relay_parts(ok, ok, "loc", 0, 1)
+    with pytest.raises(ValueError):  # no connection to fetch with, or none to send with
+        await ft.relay_parts([], ok, "loc", 10, 1)
+    with pytest.raises(ValueError):
+        await ft.relay_parts(ok, [], "loc", 10, 1)
+    with pytest.raises(ValueError):  # Telegram wants equal parts: a download part must be cut into whole ones
+        await ft.relay_parts(ok, ok, "loc", 10, 1, part_size=1000, send_size=512)
+
+
+class RelayClient:
+    class session:
+        dc_id = 2
+
+
+async def test_relay_fetches_from_the_files_dc_sends_to_ours_and_returns_an_input_file_big(monkeypatch):
+    from telethon.tl.types import InputFileBig
+
+    size = ft.BIG_FILE_THRESHOLD + 12345
+    blob = blob_of(size, seed=5)
+    store, opened = {}, []
+
+    async def fake_open(client, dc_id, connections):
+        group = [FakeTransport(blob, store=store, jitter=False) for _ in range(connections)]
+        opened.append((dc_id, group))
+        return group
+
+    monkeypatch.setattr(ft, "open_transports", fake_open)
+    seen = []
+    result = await ft.relay(RelayClient(), a_document(size, dc_id=4), "Big Name.mkv", connections=3, progress=lambda c, t: seen.append(c))
+    assert sorted(dc for dc, _group in opened) == [2, 4] and all(len(g) == 3 for _dc, g in opened)
+    assert isinstance(result, InputFileBig) and result.name == "Big Name.mkv"
+    assert result.parts == len(store) == -(-size // ft.UPLOAD_PART) and result.id == next(iter(store.values()))[0]
+    assert b"".join(store[i][2] for i in sorted(store)) == blob
+    assert all(t.closed for _dc, group in opened for t in group) and seen[-1] == size
+    assert ft.stats == ft.TransferStats()  # counting what was streamed is up to the caller, once the file is sent
+
+
+async def test_relay_closes_its_connections_when_it_fails_and_when_it_cannot_open_them(monkeypatch):
+    opened = []
+
+    async def failing_open(client, dc_id, connections):
+        group = [FakeTransport(b"", fail_first=10_000) for _ in range(connections)]
+        opened.append(group)
+        return group
+
+    monkeypatch.setattr(ft, "open_transports", failing_open)
+    with pytest.raises(ConnectionError):
+        await ft.relay(RelayClient(), a_document(ft.BIG_FILE_THRESHOLD + 1, dc_id=4), "x.mkv", connections=2)
+    assert len(opened) == 2 and all(t.closed for group in opened for t in group)
+
+    opened.clear()
+
+    async def half_open(client, dc_id, connections):
+        if dc_id == 4:
+            raise OSError("no route to DC 4")
+        group = [FakeTransport() for _ in range(connections)]
+        opened.append(group)
+        return group
+
+    monkeypatch.setattr(ft, "open_transports", half_open)
+    with pytest.raises(OSError):
+        await ft.relay(RelayClient(), a_document(ft.BIG_FILE_THRESHOLD + 1, dc_id=4), "x.mkv", connections=2)
+    assert len(opened) == 1 and all(t.closed for t in opened[0])  # what did open is not left connected
+
+
+def test_file_part_errors_are_told_apart_from_the_rest():
+    from telethon import errors
+
+    assert ft.is_file_part_error(errors.FilePartsInvalidError(request=None))
+    assert ft.is_file_part_error(errors.FilePartMissingError(request=None, capture=3))
+    assert not ft.is_file_part_error(errors.MediaCaptionTooLongError(request=None))
+    assert not ft.is_file_part_error(RuntimeError("FilePart"))
+
+
+def test_a_failed_stream_is_remembered_apart_from_the_parallel_failures():
+    stats = ft.TransferStats()
+    stats.stream_failed(ValueError("boom"))
+    assert (stats.stream_fallbacks, stats.last_stream_fallback, stats.fallbacks, stats.last_fallback) == (1, "ValueError: boom", 0, "")

tests/test_formats.py

@@ -128,14 +128,14 @@ def test_validate_name():
 
 
 def fmt_with(**kw):
-    base = {"id": 1, "name": "F", "has_thumb": False, "template": "T {title}",
+    base = {"id": 1, "name": "F", "has_thumb": False, "template": "T {title}", "caption": DEFAULT_FORMAT["caption"],
             "meta": dict(DEFAULT_FORMAT["meta"]), "keep_audio": [], "keep_subtitle": []}
     base.update(kw)
     return base
 
 
 def test_apply_format_lays_the_format_over_the_settings_without_mutating_them():
-    settings = dict(DEFAULT_SETTINGS, mode="caption", caption="<b>{filename}</b>", file_type="document", remove_words=["x"],
+    settings = dict(DEFAULT_SETTINGS, mode="caption", caption="<i>{original}</i>", file_type="document", remove_words=["x"],
                     target_channel=-100123, metadata_on=False, keep_audio=["french"])
     settings["meta"] = dict(DEFAULT_SETTINGS["meta"], title="GLOBAL TITLE")
     fmt = fmt_with(template="One Piece - E{episode}", meta=dict(DEFAULT_FORMAT["meta"], author="@Me"), keep_subtitle=["english"])
@@ -144,10 +144,22 @@ def test_apply_format_lays_the_format_over_the_settings_without_mutating_them():
     assert out["meta"]["author"] == "@Me" and out["meta"]["title"] == ""  # the format's own metadata, not the global one
     assert out["keep_audio"] == [] and out["keep_subtitle"] == ["english"]
     assert out["metadata_on"] is True
+    assert out["caption"] == "<b>{filename}</b>"  # the format's own, not the user's <i>{original}</i>
     # everything else is still the user's
-    assert (out["mode"], out["caption"], out["file_type"], out["remove_words"], out["target_channel"]) == (
-        "caption", "<b>{filename}</b>", "document", ["x"], -100123)
+    assert (out["mode"], out["file_type"], out["remove_words"], out["target_channel"]) == ("caption", "document", ["x"], -100123)
     assert settings["template"] == DEFAULT_SETTINGS["template"] and settings["meta"]["title"] == "GLOBAL TITLE"
+    assert settings["caption"] == "<i>{original}</i>"
+
+
+def test_a_format_brings_its_own_caption_and_never_borrows_the_users():
+    mine = dict(DEFAULT_SETTINGS, caption="<i>{original}</i> by me")
+    assert apply_format(mine, fmt_with(caption="<b>{title}</b> · {size}"))["caption"] == "<b>{title}</b> · {size}"
+    assert apply_format(mine, fmt_with())["caption"] == "<b>{filename}</b>"  # none of its own: the new file name in bold
+    odd = fmt_with()
+    del odd["caption"]  # a snapshot without the field
+    assert apply_format(mine, odd)["caption"] == "<b>{filename}</b>"
+    assert apply_format(mine, fmt_with(caption=""))["caption"] == "<b>{filename}</b>"
+    assert mine["caption"] == "<i>{original}</i> by me"
 
 
 def test_a_format_without_metadata_leaves_the_file_alone():
@@ -200,3 +212,65 @@ async def test_editing_a_deleted_format_raises_lookup_error(db):
         await apply_setting(store, U, "template", "x {title}")
     with pytest.raises(LookupError):
         await store.get_settings(U)
+
+
+# ---------------------------------------------------------------------------
+# a format's own caption
+# ---------------------------------------------------------------------------
+
+
+async def test_a_format_starts_with_the_file_name_in_bold_and_keeps_a_caption_of_its_own(db):
+    fmt = await db.create_format(U, "One Piece")
+    assert fmt["caption"] == "<b>{filename}</b>" == DEFAULT_SETTINGS["caption"]
+    out = await db.update_format(U, fmt["id"], caption="<i>{filename}</i> · {size}")
+    assert out["caption"] == "<i>{filename}</i> · {size}"
+    await db.update_format(U, fmt["id"], template="X {title}", meta={"title": "T"})  # other edits leave it alone
+    other = Database(db.url)
+    assert (await other.get_format(U, fmt["id"]))["caption"] == "<i>{filename}</i> · {size}"
+    await other.close()
+    assert (await db.list_formats(U))[0]["caption"] == "<i>{filename}</i> · {size}"
+    assert await db.get_settings(U) == DEFAULT_SETTINGS  # the user's own caption is a different thing
+
+
+async def test_formats_saved_before_captions_existed_get_the_default_one(db):
+    import json
+
+    from bot.db import Format
+
+    async with db._sessions() as session:
+        session.add(Format(user_id=U, name="Old", data=json.dumps({"template": "Old - {title}", "keep_audio": ["english"]})))
+        session.add(Format(user_id=U, name="Broken", data=json.dumps({"caption": 7})))
+        session.add(Format(user_id=U, name="Blank", data=json.dumps({"caption": "   "})))
+        await session.commit()
+    old, broken, blank = await db.list_formats(U)
+    assert old["template"] == "Old - {title}" and old["keep_audio"] == ["english"] and old["caption"] == "<b>{filename}</b>"
+    assert broken["caption"] == "<b>{filename}</b>" and blank["caption"] == "<b>{filename}</b>"
+
+
+async def test_a_copy_of_a_format_has_its_caption(db):
+    fmt = await db.create_format(U, "One Piece")
+    await db.update_format(U, fmt["id"], caption="<code>{filename}</code>")
+    copy = await db.copy_format(V, U, fmt["id"])
+    assert copy["caption"] == "<code>{filename}</code>"
+    await db.update_format(V, copy["id"], caption="<b>{title}</b>")
+    assert (await db.get_format(U, fmt["id"]))["caption"] == "<code>{filename}</code>"
+
+
+async def test_settings_ops_edit_a_formats_caption_with_the_checks_of_the_global_one(db):
+    fmt = await db.create_format(U, "One Piece")
+    store = FormatSettings(db, fmt["id"])
+    ok, msg = await apply_setting(store, U, "caption", "<b>{filename}</b> · {size} · {duration}")
+    assert ok and "Caption template saved" in msg and "Unknown" not in msg
+    assert (await db.get_format(U, fmt["id"]))["caption"] == "<b>{filename}</b> · {size} · {duration}"
+    ok, msg = await apply_setting(store, U, "caption", "<i>{nonsense}</i>")
+    assert ok and "Unknown variables" in msg and "{nonsense}" in msg  # kept as plain text, like for the global one
+    ok, msg = await apply_setting(store, U, "caption", "x" * 801)
+    assert not ok and "too long" in msg and (await db.get_format(U, fmt["id"]))["caption"] == "<i>{nonsense}</i>"
+    ok, msg = await apply_setting(store, U, "caption", "")
+    assert not ok and "can't be empty" in msg
+    for word in ("reset", "off", "none", "clear"):
+        await apply_setting(store, U, "caption", "<u>{filename}</u>")
+        ok, _ = await apply_setting(store, U, "caption", word)
+        assert ok and (await db.get_format(U, fmt["id"]))["caption"] == "<b>{filename}</b>", word  # the file name in bold again
+    assert (await store.get_settings(U))["caption"] == "<b>{filename}</b>"
+    assert await db.get_settings(U) == DEFAULT_SETTINGS

tests/test_formats_app.py

@@ -90,17 +90,18 @@ async def test_the_format_screen_shows_every_field_and_has_a_button_for_each(ctx
     fmt = await make_format(ctx, meta={"title": "One Piece", "video": "HD"}, keep_audio=["japanese", "2"])
     cb = await click(ctx, OWNER, f"fm:v:{fmt['id']}")
     text, buttons = cb.edits[-1]
-    for label in ("Template", "Thumbnail", "Title", "Author", "Artist", "Audio", "Subtitle", "Video", "Keep Audio", "Keep Subtitle"):
+    for label in ("Template", "Caption", "Thumbnail", "Title", "Author", "Artist", "Audio", "Subtitle", "Video", "Keep Audio", "Keep Subtitle"):
         assert label in text, label
     assert "One Piece - E{episode}" in text and "Japanese, 2" in text and "your default thumbnail" in text
     assert "<b>Title</b> ▸ <code>One Piece</code>" in text and "<b>Video</b> ▸ <code>HD</code>" in text  # values land on the right line
     assert "<b>Author</b> ▸ —" in text
     data = data_of(buttons)
     i = fmt["id"]
-    for field in ("name", "template", "title", "author", "artist", "audio", "subtitle", "video", "keep_audio", "keep_subtitle"):
+    for field in ("name", "template", "caption", "title", "author", "artist", "audio", "subtitle", "video", "keep_audio", "keep_subtitle"):
         assert f"fm:in:{i}:{field}" in data, field
     assert {f"fm:th:{i}", f"fm:del:{i}", "m:formats", "x:close"} <= set(data)
     assert f"fm:thd:{i}" not in data  # nothing to delete yet
+    assert f"fm:cr:{i}" not in data and "<b>Caption</b> ▸ <code>&lt;b&gt;{filename}&lt;/b&gt;</code> · default" in text  # nor to reset
 
 
 async def test_editing_each_field_goes_through_the_same_checks_as_the_settings(ctx):
@@ -108,6 +109,7 @@ async def test_editing_each_field_goes_through_the_same_checks_as_the_settings(c
     i = fmt["id"]
     edits = {
         "template": ("[S{season}E{episode}] {title}", lambda f: f["template"] == "[S{season}E{episode}] {title}"),
+        "caption": ("<i>{filename}</i> · {size}", lambda f: f["caption"] == "<i>{filename}</i> · {size}"),
         "title": ("One Piece", lambda f: f["meta"]["title"] == "One Piece"),
         "author": ("@Me", lambda f: f["meta"]["author"] == "@Me"),
         "artist": ("@Me", lambda f: f["meta"]["artist"] == "@Me"),
@@ -128,7 +130,8 @@ async def test_editing_each_field_goes_through_the_same_checks_as_the_settings(c
         chat, text, kw = ctx.client.edited[-1]
         assert kw["message"] == 700 + n and "🗂 One Piece" in text, field  # the format screen is back in the same message
     # none of this touched the user's own settings
-    assert (await ctx.db.get_settings(OWNER))["meta"]["title"] == "" and (await ctx.db.get_settings(OWNER))["keep_audio"] == []
+    mine = await ctx.db.get_settings(OWNER)
+    assert mine["meta"]["title"] == "" and mine["keep_audio"] == [] and mine["caption"] == "<b>{filename}</b>"
 
 
 async def test_invalid_values_keep_the_prompt_open_and_cancel_returns_to_the_format(ctx):
@@ -151,6 +154,42 @@ async def test_clearing_a_field_with_off(ctx):
     assert got["meta"]["title"] == "" and got["keep_audio"] == []
 
 
+async def test_a_formats_caption_is_its_own_and_can_be_reset(ctx):
+    await ctx.db.update_settings(OWNER, caption="<i>MY OWN {filename}</i>")
+    fmt = await make_format(ctx)
+    i = fmt["id"]
+    cb = await click(ctx, OWNER, f"fm:in:{i}:caption", msg_id=710)
+    assert cb.edits[-1][0] == panels.FORMAT_PROMPTS["caption"] and data_of(cb.edits[-1][1]) == [f"fm:cx:{i}"]
+    assert "reset" in cb.edits[-1][0]
+
+    ev = await say(ctx, OWNER, "x" * 801)  # the same checks as the global caption: too long keeps the prompt open
+    assert "too long" in ev.texts[0] and "Try again" in ev.texts[0] and OWNER in ctx.app.pending
+    assert (await ctx.db.get_format(OWNER, i))["caption"] == "<b>{filename}</b>"
+    ev = await say(ctx, OWNER, "<b>{filename}</b> · {size} · {nonsense}")
+    assert ev.deleted and OWNER not in ctx.app.pending
+    chat, text, kw = ctx.client.edited[-1]
+    assert kw["message"] == 710 and "Caption template saved" in text and "Unknown variables" in text and "{nonsense}" in text
+    assert "<b>Caption</b> ▸ <code>&lt;b&gt;{filename}&lt;/b&gt; · {size} · {nonsense}</code>" in text and "· default" not in text
+    assert f"fm:cr:{i}" in data_of(kw["buttons"])  # a custom caption can be reset with one tap
+
+    cb = await click(ctx, OWNER, f"fm:cr:{i}")
+    assert cb.toasts == ["Caption reset."] and (await ctx.db.get_format(OWNER, i))["caption"] == "<b>{filename}</b>"
+    assert "· default" in cb.edits[-1][0] and f"fm:cr:{i}" not in data_of(cb.edits[-1][1])
+    assert (await ctx.db.get_settings(OWNER))["caption"] == "<i>MY OWN {filename}</i>"  # the user's own caption was never touched
+
+    await ctx.db.update_format(OWNER, i, caption="<u>{filename}</u>")  # sending reset in the prompt does the same
+    await click(ctx, OWNER, f"fm:in:{i}:caption")
+    await say(ctx, OWNER, "reset")
+    assert (await ctx.db.get_format(OWNER, i))["caption"] == "<b>{filename}</b>" and OWNER not in ctx.app.pending
+
+
+async def test_the_reset_caption_button_of_a_deleted_format_says_so(ctx):
+    fmt = await make_format(ctx, caption="<u>{filename}</u>")
+    await ctx.db.delete_format(OWNER, fmt["id"])
+    cb = await click(ctx, OWNER, f"fm:cr:{fmt['id']}")
+    assert cb.alerts == ["That format no longer exists."] and "FORMATS" in cb.edits[-1][0]
+
+
 async def test_renaming_a_format(ctx):
     fmt = await make_format(ctx, "One Piece")
     await make_format(ctx, "Dragon Ball")
@@ -418,7 +457,7 @@ def test_the_format_screens_fit_in_a_caption_even_with_the_longest_values():
 
     longest = ["klingon-language-tag-x"] * 12
     fmt = {
-        "id": 12, "name": "N" * 30, "has_thumb": True, "template": "T" * 200,
+        "id": 12, "name": "N" * 30, "has_thumb": True, "template": "T" * 200, "caption": "C" * 800,
         "meta": {k: "M" * 200 for k in ("title", "author", "artist", "audio", "subtitle", "video")},
         "keep_audio": longest, "keep_subtitle": longest,
     }

tests/test_panels.py

@@ -102,7 +102,7 @@ def test_code_each_makes_one_escaped_block_per_item():
 
 def sample_format(**over):
     fmt = {"id": 12, "name": "One Piece", "has_thumb": False, "template": "[S{season}E{episode}] {title}",
-           "meta": {k: "" for k in ("title", "author", "artist", "audio", "subtitle", "video")}, "keep_audio": [], "keep_subtitle": []}
+           "caption": "<b>{filename}</b>", "meta": {k: "" for k in ("title", "author", "artist", "audio", "subtitle", "video")}, "keep_audio": [], "keep_subtitle": []}
     fmt.update(over)
     return fmt
 
@@ -224,3 +224,15 @@ def test_to_buttons_builds_callback_buttons():
 
 def test_known_languages_lists_names():
     assert "english" in panels.known_languages()
+
+
+def test_the_format_screen_shows_its_caption_and_offers_to_reset_a_custom_one():
+    plain = panels.format_view(sample_format())
+    assert "<b>Caption</b> ▸ <code>&lt;b&gt;{filename}&lt;/b&gt;</code> · default" in plain[0]
+    assert "fm:in:12:caption" in [d for row in plain[1] for _l, d in row] and "fm:cr:12" not in [d for row in plain[1] for _l, d in row]
+
+    custom = panels.format_view(sample_format(caption="<i>{filename}</i> · {size}"))
+    assert "<b>Caption</b> ▸ <code>&lt;i&gt;{filename}&lt;/i&gt; · {size}</code>" in custom[0] and "default" not in custom[0].split("Thumbnail")[0]
+    assert ("↩ Reset Caption", "fm:cr:12") in [button for row in custom[1] for button in row]
+    assert panels.without_inline(custom[1]) == panels.format_view(sample_format(caption="<i>{filename}</i> · {size}"), share_inline=False)[1]
+    assert "caption" in panels.FORMAT_PROMPTS["caption"].lower() and "<code>reset</code>" in panels.FORMAT_PROMPTS["caption"]

tests/test_pipeline.py

@@ -371,7 +371,43 @@ async def test_everything_else_still_comes_from_the_users_settings(env):
     job.new_name = "One Piece - E5.mkv"
     await renamer.process(job)
     (target, _f, kw), = client.sent
-    assert target == -100777 and kw["force_document"] is True and kw["caption"] == "<b>One Piece - E5</b> by me"
+    assert target == -100777 and kw["force_document"] is True  # the file type and the channel are still the user's
+    assert kw["caption"] == "<b>One Piece - E5</b>"  # the caption is not: this format has none, so it is the new name in bold
+
+
+@needs_ffmpeg
+async def test_a_format_sends_its_own_caption(env):
+    from bot.utils import human_size
+
+    await env.db.update_settings(42, caption="<i>THE USER'S CAPTION {filename}</i>")
+    fmt = await make_fmt(env, caption="<b>{filename}</b> · {original} · {size}")
+    client, renamer, job = make(env)
+    job.fmt = fmt
+    job.new_name = "One Piece - E5.mkv"
+    await renamer.process(job)
+    (_t, _f, kw), = client.sent
+    size = (env.media_dir / "sample.mkv").stat().st_size
+    assert kw["caption"] == f"<b>One Piece - E5</b> · {job.file_name} · {human_size(size)}" and kw["parse_mode"] == "html"
+
+
+@needs_ffmpeg
+async def test_a_format_without_a_caption_sends_the_new_name_in_bold_not_the_users_caption(env):
+    await env.db.update_settings(42, caption="<i>THE USER'S CAPTION {filename}</i>")
+    fmt = await make_fmt(env)
+    client, renamer, job = make(env)
+    job.fmt = fmt
+    job.new_name = "One Piece - E5.mkv"
+    await renamer.process(job)
+    assert client.sent[0][2]["caption"] == "<b>One Piece - E5</b>"
+
+
+@needs_ffmpeg
+async def test_a_rename_without_a_format_still_uses_the_users_caption(env):
+    await env.db.update_settings(42, caption="<i>{filename}</i> ({quality})")
+    await env.db.create_format(42, "Unused", template="X")  # having formats changes nothing for the normal renames
+    client, renamer, job = make(env)
+    await renamer.process(job)
+    assert client.sent[0][2]["caption"] == "<i>[S01E05] Tokyo Revengers [1080p]</i> (1080p)"
 
 
 @needs_ffmpeg

tests/test_share_app.py

@@ -54,8 +54,8 @@ async def test_the_format_screen_has_a_share_button_that_opens_the_chat_picker(c
 
 async def test_the_share_button_is_a_valid_telegram_button():
     """The format screen's rows go through Telethon's own conversion and serialise (no network needed)."""
-    fmt = {"id": 12, "name": "One Piece", "has_thumb": False, "template": "{title}", "meta": {k: "" for k in ("title", "author", "artist", "audio", "subtitle", "video")},
-           "keep_audio": [], "keep_subtitle": []}
+    fmt = {"id": 12, "name": "One Piece", "has_thumb": False, "template": "{title}", "caption": "<b>{filename}</b>",
+           "meta": {k: "" for k in ("title", "author", "artist", "audio", "subtitle", "video")}, "keep_audio": [], "keep_subtitle": []}
     _text, rows = panels.format_view(fmt, share_inline=True)
     client = TelegramClient(StringSession(), 1, "0123456789abcdef0123456789abcdef")
     markup = client.build_reply_markup(panels.to_buttons(rows))
@@ -210,6 +210,21 @@ async def test_opening_a_shared_format_shows_it_and_offers_to_add_it(ctx):
     assert await ctx.db.count_formats(ADMIN_CFG) == 0  # nothing is added until they say so
 
 
+async def test_a_shared_format_shows_its_caption_and_the_copy_keeps_it(ctx):
+    fmt, token = await share(ctx, caption="<i>{filename}</i> · {size}")
+    ev = await say(ctx, ADMIN_CFG, f"/start fmt_{token}")
+    assert "<b>Caption</b> ▸ <code>&lt;i&gt;{filename}&lt;/i&gt; · {size}</code>" in ev.texts[0] and "· default" not in ev.texts[0]
+    await click(ctx, ADMIN_CFG, f"fs:add:{token}")
+    (mine,) = await ctx.db.list_formats(ADMIN_CFG)
+    assert mine["caption"] == "<i>{filename}</i> · {size}"  # theirs from now on
+    await ctx.db.update_format(ADMIN_CFG, mine["id"], caption="<b>{title}</b>")
+    assert (await ctx.db.get_format(OWNER, fmt["id"]))["caption"] == "<i>{filename}</i> · {size}"
+
+    _plain, plain_token = await share(ctx, "Dragon Ball")  # without a caption of its own: the file name in bold
+    ev = await say(ctx, ADMIN_CFG, f"/start fmt_{plain_token}")
+    assert "<b>Caption</b> ▸ <code>&lt;b&gt;{filename}&lt;/b&gt;</code> · default" in ev.texts[0]
+
+
 async def test_a_plain_start_still_opens_the_menu(ctx):
     for text in ("/start", "/start hello", "/start fmt", "/start x_fmt_abc"):
         ev = await say(ctx, ADMIN_CFG, text)

tests/test_stream_rename.py

@@ -0,0 +1,442 @@
+"""Streaming renames: a big file goes straight from the download into the upload, and anything that needs the file takes the usual way."""
+import shutil
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from telethon import errors, types
+
+from bot import errorlog
+from bot import fasttransfer as ft
+from bot import pipeline
+from bot.config import load_config
+from bot.db import Database
+from bot.errors import Cancelled
+from bot.media import probe
+from bot.pipeline import Renamer, embedded_thumb, make_caption, source_attributes
+from bot.queue import Job
+from bot.utils import human_duration, human_size
+
+from .conftest import needs_ffmpeg
+
+MIB = 1024 * 1024
+BIG = 40 * MIB  # more than the 10 MB Telegram's big-file upload starts at; nothing this size is really moved: the engine is a stub here
+NAME = "[SubsPlease] Tokyo Revengers - 05 (1080p) [ABCD1234].mkv"
+NEW_NAME = "[S01E05] Tokyo Revengers [1080p].mkv"
+PREVIEW = b"\xff\xd8preview-of-the-original"
+
+
+def video_attr(duration=1400.5, w=1920, h=1080, **kw):
+    return types.DocumentAttributeVideo(duration=duration, w=w, h=h, supports_streaming=True, **kw)
+
+
+def audio_attr(duration=215.4, title="Song", performer="Me", **kw):
+    return types.DocumentAttributeAudio(duration=duration, title=title, performer=performer, **kw)
+
+
+def preview_sizes():
+    return [types.PhotoStrippedSize(type="i", bytes=b"\x01\x02"), types.PhotoSize(type="m", w=320, h=180, size=900)]
+
+
+class FakeMessage:
+    """A message with a document, the way Telethon shows it to the pipeline."""
+
+    def __init__(self, name=NAME, size=BIG, *, attributes=(), thumbs=None, mime="video/x-matroska", msg_id=7, document=True):
+        self.id = msg_id
+        self.file = SimpleNamespace(name=name, size=size, ext=Path(name or "").suffix, mime_type=mime)
+        self.document = SimpleNamespace(attributes=list(attributes), thumbs=thumbs, size=size, mime_type=mime) if document else None
+        video, audio = source_attributes(self)
+        self.video = self.document if video is not None else None  # like Telethon's own properties
+        self.audio = self.document if audio is not None else None
+
+
+class FakeClient:
+    def __init__(self, message, source):
+        self.message, self.source = message, source
+        self.sent, self.edits = [], []
+        self.relays, self.downloads, self.uploads, self.preview_downloads = [], [], [], 0
+        self.relay_error = None
+        self.send_errors = []
+
+    async def get_messages(self, chat_id, ids):
+        return self.message
+
+    async def download_media(self, message, file=None, thumb=None, progress_callback=None):
+        assert thumb is not None, "the whole file is fetched by transfer.download, which is stubbed"
+        self.preview_downloads += 1
+        return PREVIEW
+
+    async def send_file(self, target, input_file, **kwargs):
+        if self.send_errors:
+            raise self.send_errors.pop(0)
+        self.sent.append((target, input_file, kwargs))
+
+    async def edit_message(self, message, text, buttons=None, parse_mode=None):
+        self.edits.append((text, buttons))
+
+
+@pytest.fixture(autouse=True)
+def fresh_stats(monkeypatch):
+    monkeypatch.setattr(ft, "stats", ft.TransferStats())
+
+
+@pytest.fixture
+async def env(tmp_path, media_dir, monkeypatch):
+    def new_cfg(**environ):
+        return load_config({
+            "API_ID": "1", "API_HASH": "h", "BOT_TOKEN": "1:t", "OWNER_ID": "42", "TEMP_DIR": str(tmp_path / "work"),
+            "PROGRESS_INTERVAL": "2", **environ,  # fast transfer and streaming are on unless a test turns them off
+        })
+
+    db = Database(f"sqlite+aiosqlite:///{tmp_path}/s.db")
+    await db.init()
+    ns = SimpleNamespace(db=db, media_dir=media_dir, tmp=tmp_path, new_cfg=new_cfg, cfg=new_cfg())
+
+    async def relay(client, document, file_name, *, connections, progress=None, cancel=None):
+        client.relays.append((document, file_name, connections))
+        if client.relay_error is not None:
+            raise client.relay_error
+        progress(document.size // 2, document.size)
+        progress(document.size, document.size)
+        return types.InputFileBig(id=99, parts=-(-document.size // ft.UPLOAD_PART), name=file_name)
+
+    async def download(client, message, dest, *, fast, connections, progress=None, cancel=None, max_bytes=None):
+        client.downloads.append(Path(dest).name)
+        shutil.copy(client.source, dest)
+
+    async def upload(client, path, file_name, *, fast, connections, progress=None, cancel=None):
+        client.uploads.append(Path(path).read_bytes())
+        return types.InputFileBig(id=1, parts=1, name=file_name)
+
+    monkeypatch.setattr(ft, "relay", relay)
+    monkeypatch.setattr(ft, "download", download)
+    monkeypatch.setattr(ft, "upload", upload)
+    yield ns
+    await db.close()
+
+
+@pytest.fixture
+async def log_channel():
+    """What reaches the log channel while a test runs."""
+    reports = []
+
+    async def send(text):
+        reports.append(text)
+
+    handler = errorlog.install(send)
+    yield SimpleNamespace(reports=reports, flush=handler.aclose)
+    handler.detach()
+
+
+async def rename(env, message=None, *, cfg=None, new_name=NEW_NAME, fmt=None, source="sample.mkv"):
+    """Run one job; returns ``(client, job)``."""
+    message = message or FakeMessage(attributes=[video_attr()], thumbs=preview_sizes())
+    client = FakeClient(message, env.media_dir / source)
+    job = Job(id=1, user_id=42, chat_id=42, msg_id=7, file_name=message.file.name or "x", size=message.file.size, new_name=new_name,
+              status_msg=SimpleNamespace(id=99), fmt=fmt)
+    await Renamer(client, cfg or env.cfg, env.db).process(job)
+    return client, job
+
+
+def attributes_by_type(kwargs):
+    return {type(a): a for a in kwargs["attributes"]}
+
+
+# ---------------------------------------------------------------------------
+# the streamed way
+# ---------------------------------------------------------------------------
+
+
+async def test_a_big_plain_rename_is_streamed_with_the_details_telegram_already_has(env):
+    message = FakeMessage(attributes=[video_attr()], thumbs=preview_sizes())
+    client, job = await rename(env, message)
+
+    assert client.relays == [(message.document, NEW_NAME, env.cfg.connections)]
+    assert client.downloads == [] and client.uploads == []  # nothing was saved, and nothing uploaded from a file
+    (target, input_file, kw), = client.sent
+    assert target == 42 and kw["reply_to"] == 7 and input_file.name == NEW_NAME
+    by_type = attributes_by_type(kw)
+    assert by_type[types.DocumentAttributeFilename].file_name == NEW_NAME
+    video = by_type[types.DocumentAttributeVideo]
+    assert (video.duration, video.w, video.h, video.supports_streaming) == (1400.5, 1920, 1080, True)
+    assert kw["force_document"] is False and kw["mime_type"] == "video/x-matroska" and kw["parse_mode"] == "html"
+    assert kw["thumb"] == PREVIEW and client.preview_downloads == 1  # the original's own preview, as no frame can be cut from a file
+    assert kw["caption"] == "<b>[S01E05] Tokyo Revengers [1080p]</b>"
+    assert job.state == "done" and (await env.db.totals())[0] == 1
+    assert "Streaming" in client.edits[0][0]  # the progress message says what is going on
+    done, buttons = client.edits[-1]
+    assert "Done" in done and human_size(BIG) in done and buttons is None
+    assert ft.stats.streamed == 1 and ft.stats.stream_fallbacks == 0
+    assert not (Path(env.cfg.temp_dir) / "job1").exists()
+
+
+async def test_the_users_thumbnail_wins_and_the_original_preview_is_then_not_even_fetched(env):
+    await env.db.set_thumb(42, b"\xff\xd8mine")
+    client, _job = await rename(env)
+    assert client.sent[0][2]["thumb"] == b"\xff\xd8mine" and client.preview_downloads == 0 and len(client.relays) == 1
+
+    fmt = await env.db.create_format(42, "One Piece")
+    await env.db.set_format_thumb(42, fmt["id"], b"\xff\xd8format")
+    client, _job = await rename(env, fmt=await env.db.get_format(42, fmt["id"]))
+    assert client.sent[0][2]["thumb"] == b"\xff\xd8format" and len(client.relays) == 1
+
+
+async def test_a_format_streams_with_its_own_caption(env):
+    await env.db.update_settings(42, caption="<i>THE USER'S {filename}</i>")
+    fmt = await env.db.create_format(42, "One Piece", template="One Piece - E{episode}")
+    await env.db.update_format(42, fmt["id"], caption="<b>{filename}</b> · {size} · {duration}")
+    client, _job = await rename(env, new_name="One Piece - E5.mkv", fmt=await env.db.get_format(42, fmt["id"]))
+    (_t, _f, kw), = client.sent
+    assert len(client.relays) == 1
+    assert kw["caption"] == f"<b>One Piece - E5</b> · {human_size(BIG)} · {human_duration(1400.5)}"  # the duration is the one Telegram has
+
+
+async def test_a_file_that_gets_no_thumbnail_anyway_is_streamed_without_one(env):
+    message = FakeMessage("Archive.zip", BIG, mime="application/zip")
+    client, _job = await rename(env, message, new_name="Archive 2.zip")
+    (_t, _f, kw), = client.sent
+    assert len(client.relays) == 1 and kw["thumb"] is None and client.preview_downloads == 0
+    assert kw["force_document"] is True and kw["mime_type"] == "application/zip"
+    assert [type(a) for a in kw["attributes"]] == [types.DocumentAttributeFilename]
+
+
+async def test_an_original_sent_as_a_file_is_streamed_as_a_file(env):
+    await env.db.set_thumb(42, b"\xff\xd8mine")
+    client, _job = await rename(env, FakeMessage())  # an .mkv without any video details: Telegram sent it as a file
+    (_t, _f, kw), = client.sent
+    assert len(client.relays) == 1 and kw["force_document"] is True and kw["thumb"] == b"\xff\xd8mine"
+    assert types.DocumentAttributeVideo not in attributes_by_type(kw)
+
+
+async def test_audio_is_streamed_with_the_title_and_artist_telegram_has(env):
+    message = FakeMessage("old name.mp3", BIG, attributes=[audio_attr()], mime="audio/mpeg")
+    client, _job = await rename(env, message, new_name="New Song.mp3")
+    (_t, _f, kw), = client.sent
+    audio = attributes_by_type(kw)[types.DocumentAttributeAudio]
+    assert (audio.duration, audio.title, audio.performer) == (215, "Song", "Me") and not audio.voice
+    assert kw["force_document"] is False and kw["mime_type"] == "audio/mpeg" and kw["thumb"] is None and len(client.relays) == 1
+
+
+async def test_a_forced_type_is_followed_when_telegram_has_the_details_for_it(env):
+    await env.db.update_settings(42, file_type="document")
+    client, _job = await rename(env)  # sent as a video, but the user wants documents
+    (_t, _f, kw), = client.sent
+    assert len(client.relays) == 1 and kw["force_document"] is True and types.DocumentAttributeVideo not in attributes_by_type(kw)
+
+
+async def test_the_caption_may_use_the_duration_when_telegram_has_it(env):
+    await env.db.update_settings(42, caption="<b>{filename}</b> · {duration}")
+    client, _job = await rename(env)
+    assert client.sent[0][2]["caption"] == f"<b>[S01E05] Tokyo Revengers [1080p]</b> · {human_duration(1400.5)}" and len(client.relays) == 1
+
+
+async def test_a_streamed_file_goes_to_the_target_channel_like_any_other(env):
+    await env.db.update_settings(42, target_channel=-100123)
+    client, _job = await rename(env)
+    (target, _f, kw), = client.sent
+    assert target == -100123 and "reply_to" not in kw and "channel" in client.edits[-1][0]
+
+
+async def test_a_rejected_caption_is_dropped_for_a_streamed_file_too(env):
+    message = FakeMessage(attributes=[video_attr()], thumbs=preview_sizes())
+    client = FakeClient(message, env.media_dir / "sample.mkv")
+    client.send_errors = [errors.MediaCaptionTooLongError(request=None)]
+    job = Job(id=1, user_id=42, chat_id=42, msg_id=7, file_name=NAME, size=BIG, new_name=NEW_NAME, status_msg=SimpleNamespace(id=99))
+    await Renamer(client, env.cfg, env.db).process(job)
+    (_t, _f, kw), = client.sent
+    assert kw["caption"] is None and "caption was rejected" in client.edits[-1][0] and job.state == "done" and len(client.relays) == 1
+
+
+async def test_the_done_message_gives_the_speed_of_the_whole_stream(env, monkeypatch):
+    ticks = iter(range(0, 1000, 2))  # every look at the clock is two seconds later than the one before
+    monkeypatch.setattr(pipeline, "time", SimpleNamespace(monotonic=lambda: next(ticks)))
+    client, _job = await rename(env)
+    assert f"{human_size(BIG)} · ⬇⬆ {human_size(BIG / 2)}/s streamed" in client.edits[-1][0]
+
+
+async def test_a_stream_that_took_under_a_second_shows_no_speed(env):
+    client, _job = await rename(env)
+    assert "Done" in client.edits[-1][0] and "/s" not in client.edits[-1][0]
+
+
+# ---------------------------------------------------------------------------
+# when the file has to be looked at: the usual way (download, ffmpeg, upload)
+# ---------------------------------------------------------------------------
+
+
+def usual_way(client, job):
+    """The file was saved and uploaded the usual way, and sent once."""
+    assert client.relays == [] and len(client.downloads) == 1 and len(client.uploads) == 1
+    assert len(client.sent) == 1 and job.state == "done"
+    assert ft.stats.streamed == 0 and ft.stats.stream_fallbacks == 0  # not streaming is not a failure
+
+
+@needs_ffmpeg
+async def test_small_files_are_not_streamed(env):
+    message = FakeMessage(size=ft.BIG_FILE_THRESHOLD, attributes=[video_attr()], thumbs=preview_sizes())
+    client, job = await rename(env, message)
+    usual_way(client, job)
+
+
+@needs_ffmpeg
+async def test_metadata_and_tracks_change_the_file_so_it_is_not_streamed(env):
+    await env.db.update_settings(42, metadata_on=True, meta={"title": "Tokyo Revengers"})
+    client, job = await rename(env)
+    usual_way(client, job)
+
+
+@needs_ffmpeg
+async def test_streaming_can_be_switched_off_and_needs_the_parallel_transfer(env):
+    for environ in ({"STREAM_RENAME": "0"}, {"FAST_TRANSFER": "0"}):
+        client, job = await rename(env, cfg=env.new_cfg(**environ))
+        usual_way(client, job)
+
+
+@needs_ffmpeg
+async def test_a_video_without_a_thumbnail_to_reuse_is_saved_so_that_one_can_be_cut_from_it(env):
+    for thumbs in (None, [types.PhotoStrippedSize(type="i", bytes=b"\x01\x02")]):  # a blurred stripped preview is no thumbnail
+        message = FakeMessage(attributes=[video_attr()], thumbs=thumbs)
+        client, job = await rename(env, message)
+        usual_way(client, job)
+        assert client.sent[0][2]["thumb"][:2] == b"\xff\xd8" and client.sent[0][2]["thumb"] != PREVIEW  # a frame of the video
+
+
+@needs_ffmpeg
+async def test_a_forced_video_or_audio_type_needs_the_details_only_the_file_can_give(env):
+    await env.db.update_settings(42, file_type="video")
+    client, job = await rename(env, FakeMessage(thumbs=preview_sizes()))  # an .mkv sent as a file: no video details
+    usual_way(client, job)
+    assert types.DocumentAttributeVideo in attributes_by_type(client.sent[0][2])  # probed from the file instead
+
+    await env.db.update_settings(42, file_type="audio")
+    client, job = await rename(env, FakeMessage(attributes=[video_attr()], thumbs=preview_sizes()))
+    usual_way(client, job)
+
+
+@needs_ffmpeg
+async def test_a_caption_that_wants_the_duration_needs_the_file_when_telegram_does_not_know_it(env):
+    await env.db.update_settings(42, caption="<b>{filename}</b> · {duration}")
+    await env.db.set_thumb(42, b"\xff\xd8mine")
+    client, job = await rename(env, FakeMessage())  # sent as a file: no duration
+    usual_way(client, job)
+    duration = (await probe(env.media_dir / "sample.mkv")).duration
+    assert client.sent[0][2]["caption"] == f"<b>[S01E05] Tokyo Revengers [1080p]</b> · {human_duration(duration)}"  # read from the file
+
+
+@needs_ffmpeg
+async def test_a_message_without_a_document_is_never_streamed(env):
+    client, job = await rename(env, FakeMessage(document=False))
+    usual_way(client, job)
+
+
+# ---------------------------------------------------------------------------
+# when streaming goes wrong: the usual way takes over
+# ---------------------------------------------------------------------------
+
+
+@needs_ffmpeg
+async def test_a_failed_stream_falls_back_to_the_usual_way_and_says_so(env, log_channel):
+    client = FakeClient(FakeMessage(attributes=[video_attr()], thumbs=preview_sizes()), env.media_dir / "sample.mkv")
+    client.relay_error = RuntimeError("DC unreachable")
+    job = Job(id=1, user_id=42, chat_id=42, msg_id=7, file_name=NAME, size=BIG, new_name=NEW_NAME, status_msg=SimpleNamespace(id=99))
+    await Renamer(client, env.cfg, env.db).process(job)
+    await log_channel.flush()
+
+    assert len(client.relays) == 1 and len(client.downloads) == 1 and client.uploads == [(env.media_dir / "sample.mkv").read_bytes()]
+    (_t, _f, kw), = client.sent
+    assert kw["force_document"] is False and kw["caption"] == "<b>[S01E05] Tokyo Revengers [1080p]</b>"
+    assert job.state == "done" and (await env.db.totals())[0] == 1  # counted once
+    done = client.edits[-1][0]
+    assert "Done" in done and "Streaming didn't work for this file" in done
+    assert (ft.stats.streamed, ft.stats.stream_fallbacks, ft.stats.last_stream_fallback) == (0, 1, "RuntimeError: DC unreachable")
+    assert ft.stats.fallbacks == 0  # the parallel transfer itself did not fail
+    (report,) = log_channel.reports  # the log channel is told, with the traceback
+    assert "streaming" in report and "RuntimeError: DC unreachable" in report and "Traceback" in report
+    assert not (Path(env.cfg.temp_dir) / "job1").exists()
+
+
+@needs_ffmpeg
+async def test_telegram_refusing_the_streamed_parts_also_falls_back(env):
+    client = FakeClient(FakeMessage(attributes=[video_attr()], thumbs=preview_sizes()), env.media_dir / "sample.mkv")
+    client.send_errors = [errors.FilePartMissingError(request=None, capture=3)]
+    job = Job(id=1, user_id=42, chat_id=42, msg_id=7, file_name=NAME, size=BIG, new_name=NEW_NAME, status_msg=SimpleNamespace(id=99))
+    await Renamer(client, env.cfg, env.db).process(job)
+    assert len(client.relays) == 1 and len(client.downloads) == 1 and len(client.uploads) == 1 and len(client.sent) == 1
+    assert job.state == "done" and "Streaming didn't work" in client.edits[-1][0] and (await env.db.totals())[0] == 1
+    assert (ft.stats.streamed, ft.stats.stream_fallbacks) == (0, 1) and "FilePartMissingError" in ft.stats.last_stream_fallback
+
+
+async def test_any_other_refusal_after_streaming_fails_the_job_without_a_second_try(env):
+    client = FakeClient(FakeMessage(attributes=[video_attr()], thumbs=preview_sizes()), env.media_dir / "sample.mkv")
+    client.send_errors = [errors.ChatWriteForbiddenError(request=None)]
+    job = Job(id=1, user_id=42, chat_id=42, msg_id=7, file_name=NAME, size=BIG, new_name=NEW_NAME, status_msg=SimpleNamespace(id=99))
+    await Renamer(client, env.cfg, env.db).process(job)
+    assert job.state == "failed" and client.sent == [] and client.downloads == [] and len(client.relays) == 1  # no download/upload all over again
+    assert "Failed" in client.edits[-1][0] and ft.stats.stream_fallbacks == 0 and (await env.db.totals())[0] == 0
+
+
+async def test_cancelling_a_stream_cancels_the_job_and_does_not_fall_back(env):
+    client = FakeClient(FakeMessage(attributes=[video_attr()], thumbs=preview_sizes()), env.media_dir / "sample.mkv")
+    client.relay_error = Cancelled()
+    job = Job(id=1, user_id=42, chat_id=42, msg_id=7, file_name=NAME, size=BIG, new_name=NEW_NAME, status_msg=SimpleNamespace(id=99))
+    await Renamer(client, env.cfg, env.db).process(job)
+    assert job.state == "cancelled" and client.sent == [] and client.downloads == [] and "Cancelled" in client.edits[-1][0]
+    assert ft.stats.stream_fallbacks == 0 and ft.stats.streamed == 0
+
+
+async def test_a_cancel_that_arrives_as_the_stream_ends_sends_nothing(env, monkeypatch):
+    client = FakeClient(FakeMessage(attributes=[video_attr()], thumbs=preview_sizes()), env.media_dir / "sample.mkv")
+    job = Job(id=1, user_id=42, chat_id=42, msg_id=7, file_name=NAME, size=BIG, new_name=NEW_NAME, status_msg=SimpleNamespace(id=99))
+    real_relay = ft.relay
+
+    async def relay_then_cancel(*a, **kw):
+        result = await real_relay(*a, **kw)
+        job.cancel.set()
+        return result
+
+    monkeypatch.setattr(ft, "relay", relay_then_cancel)
+    await Renamer(client, env.cfg, env.db).process(job)
+    assert job.state == "cancelled" and client.sent == []
+
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+
+def test_source_attributes_ignore_round_videos_and_voice_notes():
+    plain, round_video = video_attr(), video_attr(round_message=True)
+    song, voice = audio_attr(), audio_attr(voice=True)
+    assert source_attributes(FakeMessage(attributes=[plain, song])) == (plain, song)
+    assert source_attributes(FakeMessage(attributes=[round_video, voice])) == (None, None)
+    assert source_attributes(FakeMessage(attributes=[types.DocumentAttributeFilename("a.mkv")])) == (None, None)
+    assert source_attributes(FakeMessage(document=False)) == (None, None) and source_attributes(object()) == (None, None)
+
+
+def test_make_caption_fills_in_what_it_is_given():
+    text = make_caption("<b>{filename}</b> · {original} · {size} · {duration} · {ext} · {nonsense}", new_name="New - E5.mkv",
+                        orig_name="[Grp] Show - 05 [720p].mkv", ext=".mkv", size=3 * MIB, duration=3725)
+    assert text == f"<b>New - E5</b> · [Grp] Show - 05 [720p].mkv · {human_size(3 * MIB)} · {human_duration(3725)} · mkv · {{nonsense}}"
+    assert make_caption("{duration}", new_name="a.mkv", orig_name="a.mkv", ext=".mkv", size=1, duration=0) is None  # empty: no caption
+    assert len(make_caption("x" * 5000, new_name="a.mkv", orig_name="a.mkv", ext=".mkv", size=1, duration=0)) == 1000
+
+
+async def test_the_embedded_preview_is_only_fetched_when_it_is_worth_having():
+    class Client:
+        def __init__(self, data=PREVIEW, error=None):
+            self.data, self.error, self.calls = data, error, 0
+
+        async def download_media(self, msg, file=None, thumb=None):
+            self.calls += 1
+            if self.error:
+                raise self.error
+            return self.data
+
+    stripped = [types.PhotoStrippedSize(type="i", bytes=b"\x01\x02")]
+    client = Client()
+    assert await embedded_thumb(client, FakeMessage(thumbs=preview_sizes())) == PREVIEW
+    assert await embedded_thumb(client, FakeMessage(thumbs=stripped)) is None and client.calls == 1  # blurred: not asked for
+    assert await embedded_thumb(client, FakeMessage(thumbs=stripped), allow_stripped=True) == PREVIEW
+    assert await embedded_thumb(client, FakeMessage(thumbs=None)) is None and await embedded_thumb(client, object()) is None
+    assert await embedded_thumb(Client(error=OSError("gone")), FakeMessage(thumbs=preview_sizes())) is None
+    assert await embedded_thumb(Client(data=b""), FakeMessage(thumbs=preview_sizes())) is None

tests/test_sysinfo.py

@@ -166,3 +166,26 @@ def test_format_status_adds_the_speed_section_only_when_given_one():
     assert "Speed" not in plain and plain.endswith("Telethon 1.45.0")
     with_speed = sysinfo.format_status(10, 1, "1.45.0", "⚡ <b>Speed</b>\nline")
     assert with_speed.startswith(plain) and with_speed.endswith("\n\n⚡ <b>Speed</b>\nline")
+
+
+def test_the_speed_summary_and_section_say_whether_streaming_is_on(monkeypatch):
+    from bot.fasttransfer import TransferStats
+
+    monkeypatch.setattr(sysinfo, "crypto_backend", lambda: "cryptg")
+    monkeypatch.setattr(sysinfo, "loop_backend", lambda: "uvloop")
+    assert sysinfo.speed_summary(True, 8, True) == "cryptg ✅ · uvloop ✅ · parallel ×8 ✅ · streaming ✅"
+    assert sysinfo.speed_summary(True, 8, False).endswith("parallel ×8 ✅ · no streaming ⚠️")
+    assert sysinfo.speed_summary(False, 8, True).endswith("parallel off ⚠️ · no streaming ⚠️")  # streaming needs the parallel transfer
+    assert sysinfo.speed_summary(True, 8) == "cryptg ✅ · uvloop ✅ · parallel ×8 ✅"  # nothing added when it is not asked for
+
+    stats = TransferStats(streamed=7)
+    text = sysinfo.format_speed(True, 8, stats, True)
+    assert "⬇⬆ Streaming rename: on: big files are uploaded while they download ✅" in text and "<b>Streamed renames:</b> 7" in text
+    assert "Streaming failed" not in text and "⚠️" not in text
+    assert "Streaming rename" not in sysinfo.format_speed(True, 8, stats) and "Streamed renames" not in sysinfo.format_speed(True, 8, stats)
+
+    stats.stream_failed(ValueError("<x>"))
+    text = sysinfo.format_speed(True, 8, stats, False)
+    assert "off (STREAM_RENAME=0) ⚠️" in text
+    assert "⚠️ Streaming failed 1× and those files were saved first and then uploaded. Last: <code>ValueError: &lt;x&gt;</code>" in text
+    assert "The parallel transfer failed" not in text  # a streaming failure is not a parallel-transfer failure