Automated sync from GitHub

.gitattributes

@@ -1,8 +1,8 @@
+*.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
 *.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

bot/app.py

@@ -20,7 +20,7 @@ from . import fasttransfer as transfer
 from . import media, panels, sysinfo
 from .config import Config
 from .db import Database
-from .parser import build_manual_name, compute_new_name
+from .parser import build_manual_name, compute_new_name, missing_variables
 from .pipeline import Renamer, original_name_of, source_ext
 from .queue import Job, JobQueue
 from .settings_ops import META_COMMANDS, apply_setting, parse_channel, preview_template
@@ -55,6 +55,7 @@ BACK = {
     "keep_audio": "tool", "keep_subtitle": "tool",
 }
 PANEL_NAMES = {"home", "rename", "cap", "meta", "tool", "ftype", "help", "admin"}
+QUEUE_SETTLE = 1.5  # seconds an idle worker waits so files sent together are queued in message order
 TRACKS_PROBE_BYTES = 12 * 1024 * 1024
 TRACKS_FULL_LIMIT = 700 * 1024 * 1024
 
@@ -98,7 +99,7 @@ class App:
         self.admins: Set[int] = set(cfg.admin_ids)
         self.pending: Dict[int, Pending] = {}
         self.renamer = Renamer(client, cfg, db, self._log_send)
-        self.queue = JobQueue(self.renamer.process, workers=cfg.max_jobs, on_cancelled=self._on_job_cancelled)
+        self.queue = JobQueue(self.renamer.process, workers=cfg.max_jobs, on_cancelled=self._on_job_cancelled, settle=QUEUE_SETTLE)
         self._warned: Set[int] = set()
 
     # ------------------------------------------------------------------ auth
@@ -471,21 +472,29 @@ class App:
         s = await self.db.get_settings(uid)
         orig = original_name_of(msg)
         ext = source_ext(msg, orig)
+        notes: List[str] = []
         if manual:
             new_name = build_manual_name(manual, ext)
         else:
-            new_name, _info = compute_new_name(
+            new_name, info = compute_new_name(
                 template=s["template"], mode=s["mode"], original_name=orig, original_ext=ext,
                 caption=msg.raw_text or "", remove_words=s["remove_words"],
             )
+            gaps = missing_variables(s["template"], info)
+            if gaps:
+                where_from = "caption" if s["mode"] == "caption" else "file name"
+                notes.append(
+                    "Couldn't find " + ", ".join("{%s}" % g for g in gaps) + f" in the {where_from}, so it is left empty. "
+                    "Cancel and use /rename, or change the template."
+                )
         job = Job(
             id=self.queue.new_id(), user_id=uid, chat_id=event.chat_id, msg_id=msg.id,
-            file_name=orig, size=size, new_name=new_name, manual=bool(manual),
+            file_name=orig, size=size, new_name=new_name, manual=bool(manual), notes=notes, order=event.id,
         )
         ahead = self.queue.running + self.queue.waiting
         where = "starting soon" if ahead == 0 else f"{ahead} ahead of it"
         job.status_msg = await event.reply(
-            f"📥 <b>Queued</b> ({where})\n<code>{esc(clip(new_name, 120))}</code>",
+            f"📥 <b>Queued</b> ({where})\n<code>{esc(clip(new_name, 120))}</code>" + "".join(f"\n⚠️ {esc(n)}" for n in notes),
             buttons=panels.to_buttons([[("✖ Cancel", f"job:x:{job.id}")]]),
             parse_mode="html",
         )

bot/db.py

@@ -13,6 +13,8 @@ from typing import Any, Dict, List, Optional, Tuple
 from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
 
 from sqlalchemy import BigInteger, DateTime, Integer, LargeBinary, String, Text, func, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
@@ -117,9 +119,11 @@ class Database:
     async def _get_or_create(self, session, uid: int) -> User:
         row = await session.get(User, uid)
         if row is None:
-            row = User(id=uid, settings="{}")
-            session.add(row)
-            await session.flush()
+            # A forwarded batch is handled by several tasks at once (and two instances can overlap briefly while the
+            # Space restarts), so let the database settle who creates the row instead of failing on the unique key.
+            insert = pg_insert if self.engine.dialect.name == "postgresql" else sqlite_insert
+            await session.execute(insert(User).values(id=uid, settings="{}").on_conflict_do_nothing(index_elements=["id"]))
+            row = await session.get(User, uid)
         return row
 
     async def get_settings(self, uid: int) -> Dict[str, Any]:

bot/parser.py

@@ -102,6 +102,13 @@ _EP_WORD_RE = re.compile(_Q + r"(?:Episode|Ep)[ ._\-]?(\d{1,4})(?!\d)", re.I)
 _E_RE = re.compile(_Q + r"E(\d{1,4})(?![A-Za-z0-9])", re.I)
 _DASH_EP_RE = re.compile(r"\s-\s?(\d{1,4})(?:v\d+)?(?=[\s\[\(._\-]|$)")
 _HASH_EP_RE = re.compile(r"#(\d{1,4})(?!\d)")
+# episode written first: "1147 - One Piece" / "1147-One Piece". A 1-2 digit number needs a space beside
+# the dash, so titles such as "3-gatsu no Lion" keep their number.
+_LEAD_EP_RE = re.compile(
+    r"(?:(?P<long>\d{3,4})(?:v\d+)?\s*[-–—]\s*|(?P<short>\d{1,2})(?:v\d+)?(?:\s+[-–—]\s*|\s*[-–—]\s+))(?=\S)"
+)
+_BRACKET_EP_RE = re.compile(r"\[(\d{1,4})(?:v\d+)?\]")  # "[1147]"
+_NUM_TOKEN_RE = re.compile(_Q + r"(\d{3,4})(?:v\d+)?(?![A-Za-z0-9])")  # "One Piece 1147 720p"
 _YEAR_PAREN_RE = re.compile(r"[\(\[]((?:19|20)\d{2})[\)\]]")
 _YEAR_BARE_RE = re.compile(_Q + r"((?:19|20)\d{2})(?![A-Za-z0-9])")
 _CODEC_RE = re.compile(_Q + r"(HEVC|[xh]\.?26[45]|AV1|AVC|VP9)(?![A-Za-z0-9])", re.I)
@@ -136,6 +143,44 @@ def _fmt_num(value: str) -> str:
     return f"{int(value):02d}"
 
 
+def _is_year(value: str) -> bool:
+    return 1900 <= int(value) <= 2099
+
+
+def _leading_or_bracket_episode(norm: str) -> Optional[Tuple[str, Tuple[int, int]]]:
+    """Episode numbers written first ("1147 - One Piece") or alone in brackets ("[1147]")."""
+    start = len(norm) - len(_LEADING_GROUPS_RE.sub("", norm))  # skip "[Group] " prefixes
+    m = _LEAD_EP_RE.match(norm, start)
+    if m:
+        number = m.group("long") or m.group("short")
+        if not _is_year(number):
+            return number, (start, m.end())
+    for m in _BRACKET_EP_RE.finditer(norm):
+        if not _is_year(m.group(1)):
+            return m.group(1), m.span()
+    return None
+
+
+def _bare_episode(norm: str, spans: List[Tuple[int, int]]) -> Optional[Tuple[str, Tuple[int, int]]]:
+    """A lone 3-4 digit number between the title and the release tags ("One Piece 1147 720p")."""
+    marks = [a for a, _b in spans if a > 0]
+    first = min(marks) if marks else len(norm)
+    found = None
+    for m in _NUM_TOKEN_RE.finditer(norm):
+        if m.start() == 0 or m.start() >= first or _is_year(m.group(1)):
+            continue
+        if any(a <= m.start() < b for a, b in spans):
+            continue
+        found = m
+    if found is None:
+        return None
+    if not marks:  # no release tags to anchor on: the number must end the title
+        tail = norm[found.end():].strip()
+        if tail and tail[0] not in "[(-–—":
+            return None
+    return found.group(1), found.span()
+
+
 def parse_info(text: str) -> Info:
     """Extract title / season / episode / quality / year / codec from a release name."""
     raw = _strip_markup(text or "")
@@ -168,12 +213,11 @@ def parse_info(text: str) -> Info:
                     episode = me.group(1)
                     spans.append(me.span())
                     break
-    if episode and not season:
-        season = "1"
-    if episode:
-        info.episode = _fmt_num(episode)
-    if season:
-        info.season = _fmt_num(season)
+            if not episode:
+                found = _leading_or_bracket_episode(norm)
+                if found:
+                    episode = found[0]
+                    spans.append(found[1])
 
     # --- quality --------------------------------------------------------------
     mq = _QUALITY_RE.search(norm)
@@ -222,6 +266,19 @@ def parse_info(text: str) -> Info:
     if ms2:
         spans.append(ms2.span())
 
+    # --- episode without any marker: "One Piece 1147 720p" (skipped for names with a year: movie titles) ---
+    if not episode and not info.year:
+        found = _bare_episode(norm, spans)
+        if found:
+            episode = found[0]
+            spans.append(found[1])
+    if episode and not season:
+        season = "1"
+    if episode:
+        info.episode = _fmt_num(episode)
+    if season:
+        info.season = _fmt_num(season)
+
     # --- title ------------------------------------------------------------------------
     body = _LEADING_GROUPS_RE.sub("", norm)
     shift = len(norm) - len(body)
@@ -247,6 +304,10 @@ def parse_info(text: str) -> Info:
     mb = re.search(r"\s[\[\(]", body)
     if mb and mb.start() < cut:
         cut = mb.start()
+    # so does a channel tag: "One Piece @Anime_X_Universe"
+    mh = re.search(r"\s@\w", body)
+    if mh and mh.start() < cut:
+        cut = mh.start()
     info.title = _clean_title(body[:cut])
     return info
 
@@ -310,6 +371,22 @@ def render_template(template: str, values: Dict[str, str]) -> str:
     return _MULTISPACE_RE.sub(" ", rendered).strip(" -._")
 
 
+_GROUP_RE = re.compile(r"\[[^\[\]]*\]|\([^()]*\)")
+
+
+def missing_variables(template: str, info: Info) -> List[str]:
+    """Variables the template uses *outside* [..]/(..) groups that are empty for this file.
+
+    A bare ``{episode}`` that could not be found leaves a visible hole in the name ("One Piece - E.mkv"),
+    unlike a bracket group, which is dropped on purpose; the caller warns the user about these.
+    """
+    out: List[str] = []
+    for name in _TOKEN_RE.findall(_GROUP_RE.sub("", template or "")):
+        if name in ("season", "episode", "quality", "year", "codec") and not getattr(info, name) and name not in out:
+            out.append(name)
+    return out
+
+
 def apply_remove_words(text: str, words: Iterable[str]) -> str:
     for word in words:
         word = word.strip()

bot/pipeline.py

@@ -141,7 +141,7 @@ class Renamer:
                 info = await media.probe(src, job.cancel)
             except MediaError as exc:
                 log.info("ffprobe could not read %s: %s", orig_name, exc)
-        warnings = []
+        warnings = list(job.notes)
         plan = media.build_plan(info, s, ext) if (info is not None and s["metadata_on"]) else None
         if s["metadata_on"] and plan is None and info is not None and ext.lower() not in (media.VIDEO_CONTAINERS | media.AUDIO_TAG_EXTS):
             warnings.append(f"Metadata is not supported for {ext or 'this file type'}, so it was skipped.")

bot/queue.py

@@ -21,6 +21,8 @@ class Job:
     size: int
     new_name: str
     manual: bool = False
+    notes: List[str] = field(default_factory=list)  # warnings shown with the queued / finished message
+    order: int = 0  # id of the message that triggered the job: files are processed oldest message first
     status_msg: Any = None
     state: str = "queued"  # queued | running | done | failed | cancelled
     cancel: asyncio.Event = field(default_factory=asyncio.Event)
@@ -30,13 +32,28 @@ class Job:
 
 Runner = Callable[[Job], Awaitable[None]]
 
+SETTLE_CAP = 4  # an idle worker never waits longer than this many settle periods in a row
+
+
+def _order_key(job: Job) -> tuple:
+    return (job.order, job.id)
+
 
 class JobQueue:
-    def __init__(self, runner: Runner, workers: int = 1, on_cancelled: Optional[Runner] = None):
+    """Jobs run oldest-message-first by a small pool of workers.
+
+    Telethon handles messages that arrive together concurrently, so jobs are not necessarily added in the
+    order the files were sent.  Picking by message id (not by insertion order) and letting an idle worker wait
+    until nothing new was queued for ``settle`` seconds (at most ``SETTLE_CAP`` times that) keeps a forwarded
+    batch in its original order.
+    """
+
+    def __init__(self, runner: Runner, workers: int = 1, on_cancelled: Optional[Runner] = None, settle: float = 0.0):
         self._runner = runner
         self._on_cancelled = on_cancelled
         self._workers = max(1, workers)
-        self._queue: "asyncio.Queue[Job]" = asyncio.Queue()
+        self._settle = max(0.0, settle)
+        self._wake = asyncio.Event()
         self._ids = itertools.count(1)
         self._tasks: List[asyncio.Task] = []
         self.jobs: Dict[int, Job] = {}
@@ -60,7 +77,7 @@ class JobQueue:
         """Queue a job; returns how many jobs are ahead of it (running + waiting)."""
         ahead = self.running + self.waiting
         self.jobs[job.id] = job
-        self._queue.put_nowait(job)
+        self._wake.set()
         return ahead
 
     @property
@@ -75,14 +92,14 @@ class JobQueue:
         """0 = running now, 1 = next up, ..."""
         if job.state == "running":
             return 0
-        waiting = sorted((j for j in self.jobs.values() if j.state == "queued"), key=lambda j: j.id)
+        waiting = sorted((j for j in self.jobs.values() if j.state == "queued"), key=_order_key)
         try:
             return waiting.index(job) + 1
         except ValueError:
             return 0
 
     def for_user(self, user_id: int) -> List[Job]:
-        return sorted((j for j in self.jobs.values() if j.user_id == user_id), key=lambda j: j.id)
+        return sorted((j for j in self.jobs.values() if j.user_id == user_id), key=_order_key)
 
     def cancel(self, job_id: int, user_id: Optional[int] = None) -> bool:
         job = self.jobs.get(job_id)
@@ -113,24 +130,48 @@ class JobQueue:
         except Exception:  # noqa: BLE001
             log.exception("job callback failed")
 
+    def _claim(self) -> Optional[Job]:
+        """Take the oldest queued job (no await in between, so two workers never take the same one)."""
+        queued = [j for j in self.jobs.values() if j.state == "queued"]
+        if not queued:
+            return None
+        job = min(queued, key=_order_key)
+        job.state = "running"
+        job.started = time.time()
+        return job
+
+    async def _settle_batch(self) -> None:
+        """The rest of a forwarded batch may still be on its way: wait until nothing new was queued for a while."""
+        if not self._settle:
+            return
+        deadline = time.monotonic() + self._settle * SETTLE_CAP
+        while time.monotonic() < deadline:
+            self._wake.clear()
+            try:
+                await asyncio.wait_for(self._wake.wait(), timeout=self._settle)
+            except asyncio.TimeoutError:
+                return
+
+    async def _next(self) -> Job:
+        while True:
+            job = self._claim()
+            if job is not None:
+                return job
+            self._wake.clear()
+            await self._wake.wait()
+            await self._settle_batch()
+
     async def _worker(self) -> None:
         while True:
-            job = await self._queue.get()
+            job = await self._next()
             try:
-                if job.state != "queued":  # cancelled while waiting
-                    continue
-                job.state = "running"
-                job.started = time.time()
-                try:
-                    await self._runner(job)
-                except asyncio.CancelledError:
-                    raise
-                except Exception:  # noqa: BLE001 - the runner reports its own failures
-                    log.exception("job %s crashed", job.id)
-                    job.state = "failed"
-                finally:
-                    if job.state == "running":
-                        job.state = "done"
-                    self.jobs.pop(job.id, None)
+                await self._runner(job)
+            except asyncio.CancelledError:
+                raise
+            except Exception:  # noqa: BLE001 - the runner reports its own failures
+                log.exception("job %s crashed", job.id)
+                job.state = "failed"
             finally:
-                self._queue.task_done()
+                if job.state == "running":
+                    job.state = "done"
+                self.jobs.pop(job.id, None)

tests/test_app.py

@@ -16,6 +16,7 @@ from bot import panels
 from bot.app import COMMANDS, OWNER_COMMANDS, PANEL_NAMES, App, Pending, cmd_pattern, is_renameable
 from bot.config import load_config
 from bot.db import Database
+from bot.queue import JobQueue
 
 from .conftest import needs_ffmpeg
 from .tgfakes import (
@@ -393,6 +394,79 @@ async def test_remove_words_are_applied_to_the_new_name(ctx):
     assert "SomeChannel" not in job.new_name and "Tokyo Revengers" in job.new_name
 
 
+async def test_the_one_piece_style_name_gets_its_episode_number(ctx):
+    await say(ctx, OWNER, "/template One Piece - E{episode}")
+    ev = await send_doc(ctx, OWNER, "1147 - One Piece [Dual] 720p @Anime_X_Universe.mkv", mid=21)
+    (job,) = ctx.app.queue.jobs.values()
+    assert job.new_name == "One Piece - E1147.mkv" and job.notes == []
+    assert "One Piece - E1147.mkv" in ev.texts[0] and "⚠️" not in ev.texts[0]
+
+
+async def test_a_template_variable_that_cannot_be_filled_is_flagged(ctx):
+    await say(ctx, OWNER, "/template One Piece - E{episode}")
+    ev = await send_doc(ctx, OWNER, "Plain Movie.mkv")
+    (job,) = ctx.app.queue.jobs.values()
+    assert job.new_name == "One Piece - E.mkv"
+    assert "⚠️" in ev.texts[0] and "{episode}" in ev.texts[0] and "file name" in ev.texts[0] and "/rename" in ev.texts[0]
+    assert job.notes and "{episode}" in job.notes[0]
+
+    await say(ctx, OWNER, "/mode caption")
+    ev = await send_doc(ctx, OWNER, "Plain Movie.mkv", mid=22)
+    assert "in the caption" in ev.texts[0]
+
+
+async def test_default_template_and_manual_names_never_warn(ctx):
+    ev = await send_doc(ctx, OWNER, "Plain Movie.mkv")
+    assert "⚠️" not in ev.texts[0]  # bracket groups vanish on purpose
+    await say(ctx, OWNER, "/template One Piece - E{episode}")
+    ev = await say(ctx, OWNER, "/rename My Name", reply=document_message("Plain Movie.mkv", mid=40), mid=41)
+    assert "⚠️" not in ev.texts[0]
+
+
+class SlowEvent(FakeEvent):
+    """A message whose reply takes a while, like a slow round trip to Telegram."""
+
+    def __init__(self, *a, delay=0.0, **kw):
+        super().__init__(*a, **kw)
+        self.delay = delay
+
+    async def reply(self, text=None, **kw):
+        await asyncio.sleep(self.delay)
+        return await super().respond(text, **kw)
+
+
+async def test_jobs_are_ordered_by_the_message_that_triggered_them(ctx):
+    await send_doc(ctx, OWNER, "B.mkv", mid=31)
+    await send_doc(ctx, OWNER, "A.mkv", mid=30)
+    await say(ctx, OWNER, "/rename X", reply=document_message("old.mkv", 1024, mid=5), mid=34)
+    assert sorted(j.order for j in ctx.app.queue.jobs.values()) == [30, 31, 34]
+    assert [j.file_name for j in ctx.app.queue.for_user(OWNER)] == ["A.mkv", "B.mkv", "old.mkv"]
+
+
+async def test_a_forwarded_batch_comes_out_in_the_order_it_was_sent(ctx):
+    """Telethon handles messages that arrive together concurrently; the results must still be in order."""
+    ran = []
+
+    async def runner(job):
+        ran.append(job.file_name)
+
+    ctx.app.queue = JobQueue(runner, workers=1, settle=0.3)
+    ctx.app.queue.start()
+    await asyncio.sleep(0.02)  # the worker is idle, waiting for files
+    names = {
+        1145: "1145 - One Piece [Dual] 720p @Anime_X_Universe.mkv",
+        1146: "1146 - One Piece [Dual] 720p @Anime_X_Universe.mkv",
+        1147: "1147 - One Piece [Dual] 720p @Anime_X_Universe.mkv",
+    }
+    delays = {1145: 0.10, 1146: 0.00, 1147: 0.05}  # the replies finish in a different order than the files arrived
+    events = [
+        SlowEvent(OWNER, message=document_message(name, 1024, mid=mid), delay=delays[mid]) for mid, name in names.items()
+    ]
+    await asyncio.gather(*(dispatch(ctx.client, ev) for ev in reversed(events)))
+    await asyncio.sleep(0.7)
+    assert ran == [names[1145], names[1146], names[1147]]
+
+
 async def test_rename_command_uses_the_given_name_and_keeps_the_extension(ctx):
     reply = document_message("old.name.mkv", 10 * 1024 * 1024, mid=33)
     ev = await say(ctx, OWNER, "/rename My Custom Name", reply=reply, mid=34)

tests/test_db.py

@@ -74,3 +74,14 @@ async def test_admins_and_kv(db):
     assert await db.kv_get("session") == "def"
     await db.kv_delete("session")
     assert await db.kv_get("session") is None
+
+
+async def test_first_use_by_many_concurrent_handlers_does_not_collide(db):
+    """A forwarded batch is handled by several tasks at once; all of them look the new user up together."""
+    import asyncio
+
+    results = await asyncio.gather(*(db.get_settings(77) for _ in range(8)))
+    assert all(r == DEFAULT_SETTINGS for r in results)
+    # the same goes for the other writers that create the row on first use
+    await asyncio.gather(*(db.record_job(78, 10) for _ in range(5)), *(db.set_thumb(79, b"x") for _ in range(5)))
+    assert await db.count_users() == 3

tests/test_parser.py

@@ -6,6 +6,7 @@ from bot.parser import (
     build_manual_name,
     compute_new_name,
     finalize_name,
+    missing_variables,
     parse_caption,
     parse_info,
     render_template,
@@ -200,3 +201,94 @@ def test_manual_name_sanitised():
 def test_sanitize_and_remove_words():
     assert sanitize("  ..hidden:name  ") == "hidden name"
     assert apply_remove_words("Show [Group] 720p", ["[group]"]) == "Show 720p"
+
+
+# ---- episode numbers written first / bare / in brackets (e.g. "1147 - One Piece [Dual] 720p @Channel") ----
+
+
+@pytest.mark.parametrize(
+    "name,title,episode,quality",
+    [
+        ("1147 - One Piece [Dual] 720p @Anime_X_Universe.mkv", "One Piece", "1147", "720p"),
+        ("1145 - One Piece [Dual] 720p @Anime_X_Universe.mkv", "One Piece", "1145", "720p"),
+        ("1147 - One Piece @Anime_X_Universe.mkv", "One Piece", "1147", ""),
+        ("1147-One Piece [720p].mkv", "One Piece", "1147", "720p"),
+        ("1147 – One Piece 720p.mkv", "One Piece", "1147", "720p"),
+        ("1147_-_One_Piece_[Dual]_720p.mkv", "One Piece", "1147", "720p"),
+        ("[Group] 1147 - One Piece 720p.mkv", "One Piece", "1147", "720p"),
+        ("1147v2 - One Piece 720p.mkv", "One Piece", "1147", "720p"),
+        ("05 - Naruto [720p].mkv", "Naruto", "05", "720p"),
+        ("1080 - One Piece [Dual] 720p.mkv", "One Piece", "1080", "720p"),  # 1080 is an episode here, not a resolution
+        ("One Piece 1147 720p.mkv", "One Piece", "1147", "720p"),
+        ("One.Piece.1147.720p.WEB-DL.mkv", "One Piece", "1147", "720p"),
+        ("One Piece 1147 Dual Audio 720p.mkv", "One Piece", "1147", "720p"),
+        ("One Piece 1147 [Dual].mkv", "One Piece", "1147", ""),
+        ("Naruto Shippuden 500 - Name.mkv", "Naruto Shippuden", "500", ""),
+        ("[1147] One Piece 720p.mkv", "One Piece", "1147", "720p"),
+        ("One Piece [1147] [720p].mkv", "One Piece", "1147", "720p"),
+        ("[Group] Title [05] [720p].mkv", "Title", "05", "720p"),
+    ],
+)
+def test_episode_without_a_marker_word(name, title, episode, quality):
+    info = parse_info(name)
+    assert (info.title, info.season, info.episode, info.quality) == (title, "01", episode, quality)
+
+
+@pytest.mark.parametrize(
+    "name,title",
+    [
+        ("3-gatsu no Lion 720p.mkv", "3-gatsu no Lion"),
+        ("5-toubun no Hanayome [1080p].mkv", "5-toubun no Hanayome"),
+        ("86-Eighty Six 720p.mkv", "86-Eighty Six"),
+        ("2001 - A Space Odyssey (1968).mkv", "2001 - A Space Odyssey"),
+        ("Blade Runner 2049 1080p BluRay.mkv", "Blade Runner"),
+        ("12 Angry Men (1957) 1080p BluRay.mkv", "12 Angry Men"),
+        ("300 (2006) 1080p BluRay.mkv", "300"),
+        ("Room 237 (2012) 1080p.mkv", "Room 237"),
+        ("Apollo 13 (1995) 1080p.mkv", "Apollo 13"),
+        ("Spider-Man 3 1080p WEB-DL.mkv", "Spider-Man 3"),
+        ("Toy Story 4 720p.mkv", "Toy Story 4"),
+        ("Oppenheimer.2023.1080p.BluRay.x265.10bit.mkv", "Oppenheimer"),
+        ("Title [2019] [720p].mkv", "Title"),
+        ("Some Movie 1920x1080.mkv", "Some Movie"),
+        ("Title 10bit 720p.mkv", "Title 10bit"),
+    ],
+)
+def test_numbers_that_belong_to_titles_are_not_episodes(name, title):
+    info = parse_info(name)
+    assert info.episode == "" and info.title == title
+
+
+def test_an_explicit_marker_beats_a_leading_number():
+    info = parse_info("86 - Eighty Six - 05 (1080p).mkv")
+    assert (info.title, info.episode, info.quality) == ("86 - Eighty Six", "05", "1080p")
+
+
+def test_channel_tags_end_the_title():
+    assert parse_info("One Piece @Anime_X_Universe 720p.mkv").title == "One Piece"
+    assert parse_info("Some @ Title.mkv").title == "Some @ Title"
+
+
+def test_the_users_one_piece_files_get_their_episode_number():
+    for number in ("1145", "1146", "1147"):
+        name, _info = compute_new_name(
+            template="One Piece - E{episode}", mode="filename", original_ext=".mkv",
+            original_name=f"{number} - One Piece [Dual] 720p @Anime_X_Universe.mkv",
+        )
+        assert name == f"One Piece - E{number}.mkv"
+    name, _ = compute_new_name(
+        template=DEFAULT_TEMPLATE, mode="filename", original_ext=".mkv",
+        original_name="1147 - One Piece [Dual] 720p @Anime_X_Universe.mkv",
+    )
+    assert name == "[S01E1147] One Piece [720p].mkv"
+
+
+def test_missing_variables_only_reports_holes_outside_bracket_groups():
+    info = parse_info("Plain Movie.mkv")
+    assert missing_variables("One Piece - E{episode}", info) == ["episode"]
+    assert missing_variables("S{season}E{episode} {title}", info) == ["season", "episode"]
+    assert missing_variables("{title} {quality} {quality}", info) == ["quality"]
+    assert missing_variables(DEFAULT_TEMPLATE, info) == []  # bracket groups vanish on purpose
+    assert missing_variables("{title} ({year})", info) == []
+    assert missing_variables("{title} {nope}", info) == []
+    assert missing_variables("One Piece - E{episode}", parse_info("1147 - One Piece 720p.mkv")) == []

tests/test_pipeline.py

@@ -114,6 +114,15 @@ async def test_plain_rename_sends_video_with_thumbnail(env):
     assert any("Tokyo Revengers" in t for t in env.logs)
 
 
+@needs_ffmpeg
+async def test_notes_from_queueing_appear_in_the_done_message(env):
+    client, renamer, job = make(env)
+    job.notes = ["Couldn't find {episode} in the file name, so it is left empty."]
+    await renamer.process(job)
+    assert job.state == "done"
+    assert "⚠️ Couldn't find {episode} in the file name" in client.edits[-1][0]
+
+
 @needs_ffmpeg
 async def test_metadata_and_track_selection_are_applied_before_upload(env):
     await env.db.update_settings(

tests/test_queue_progress.py

@@ -1,4 +1,5 @@
 import asyncio
+import time
 from types import SimpleNamespace
 
 from telethon import errors
@@ -141,6 +142,142 @@ async def test_runner_crash_does_not_kill_the_worker():
     await q.stop()
 
 
+def ordered_job(queue, order, user=1):
+    return Job(id=queue.new_id(), user_id=user, chat_id=user, msg_id=order, file_name=f"{order}.mkv", size=1,
+               new_name=f"{order}.mkv", order=order)
+
+
+async def idle_queue(runner, **kw):
+    """A started queue whose worker is already waiting for work, like the real bot between files."""
+    q = JobQueue(runner, workers=1, **kw)
+    q.start()
+    await asyncio.sleep(0.02)
+    return q
+
+
+async def test_a_batch_queued_out_of_order_runs_in_message_order():
+    ran = []
+
+    async def runner(j):
+        ran.append(j.order)
+
+    q = await idle_queue(runner, settle=0.1)
+    for order in (1146, 1147, 1145):  # the order in which three concurrent handlers happened to finish
+        q.add(ordered_job(q, order))
+        await asyncio.sleep(0.01)
+    await asyncio.sleep(0.4)
+    assert ran == [1145, 1146, 1147]
+    await q.stop()
+
+
+async def test_without_a_settle_time_the_first_queued_job_starts_at_once():
+    ran = []
+
+    async def runner(j):
+        ran.append(j.order)
+
+    q = await idle_queue(runner)
+    for order in (1146, 1147, 1145):
+        q.add(ordered_job(q, order))
+        await asyncio.sleep(0.01)
+    await asyncio.sleep(0.1)
+    assert ran[0] == 1146  # which is why the bot uses a settle time
+    await q.stop()
+
+
+async def test_jobs_queued_while_busy_are_picked_oldest_message_first():
+    ran, gate = [], asyncio.Event()
+
+    async def runner(j):
+        ran.append(j.order)
+        if j.order == 1:
+            await gate.wait()
+
+    q = await idle_queue(runner)
+    q.add(ordered_job(q, 1))
+    await asyncio.sleep(0.02)
+    for order in (30, 20, 25):
+        q.add(ordered_job(q, order))
+    gate.set()
+    await asyncio.sleep(0.1)
+    assert ran == [1, 20, 25, 30]
+    await q.stop()
+
+
+async def test_an_idle_worker_waits_for_the_settle_time_before_starting():
+    ran = []
+
+    async def runner(j):
+        ran.append(j.id)
+
+    q = await idle_queue(runner, settle=0.3)
+    q.add(job(q))
+    await asyncio.sleep(0.1)
+    assert ran == [] and q.waiting == 1
+    await asyncio.sleep(0.5)
+    assert ran == [1]
+    await q.stop()
+
+
+async def test_every_new_job_extends_the_wait_so_a_slow_batch_is_still_grouped():
+    ran = []
+
+    async def runner(j):
+        ran.append(j.order)
+
+    q = await idle_queue(runner, settle=0.3)
+    for order in (3, 2, 1):  # each arrives 0.2 s after the previous one: more than one settle period in total
+        q.add(ordered_job(q, order))
+        await asyncio.sleep(0.2)
+    await asyncio.sleep(0.6)
+    assert ran == [1, 2, 3]
+    await q.stop()
+
+
+async def test_a_steady_stream_cannot_hold_the_worker_back_forever():
+    started_at, last_add = [], []
+
+    async def runner(j):
+        started_at.append(time.monotonic())
+
+    q = await idle_queue(runner, settle=0.15)  # capped at SETTLE_CAP * 0.15 s
+    for order in range(30, 0, -1):
+        q.add(ordered_job(q, order))
+        last_add.append(time.monotonic())
+        await asyncio.sleep(0.05)
+    assert started_at and started_at[0] < last_add[-1] - 0.3  # it started well before the stream stopped
+    await q.stop()
+
+
+async def test_position_and_listing_follow_message_order():
+    async def runner(j):
+        pass
+
+    q = JobQueue(runner, workers=1)  # never started: everything stays queued
+    a, b, c = ordered_job(q, 30), ordered_job(q, 10), ordered_job(q, 20)
+    for j in (a, b, c):
+        q.add(j)
+    assert [q.position(j) for j in (b, c, a)] == [1, 2, 3]
+    assert [j.order for j in q.for_user(1)] == [10, 20, 30]
+
+
+async def test_cancelling_during_the_settle_time_skips_the_job():
+    ran = []
+
+    async def runner(j):
+        ran.append(j.id)
+
+    q = await idle_queue(runner, settle=0.2)
+    a, b = job(q), job(q)
+    q.add(a)
+    q.add(b)
+    await asyncio.sleep(0.05)
+    assert q.cancel(a.id)
+    await asyncio.sleep(0.6)
+    assert ran == [b.id]
+    await q.stop()
+
+
 # ---------------------------------------------------------------------------
 # progress reporter
 # ---------------------------------------------------------------------------