Spaces:
Running
Running
| """/sequence + /sort: put files in episode order and report which episodes are missing or repeated.""" | |
| from __future__ import annotations | |
| import asyncio | |
| from dataclasses import dataclass, field | |
| from typing import Dict, List, Optional, Sequence, Tuple | |
| from .parser import parse_caption, parse_info, split_ext | |
| from .utils import clip, esc | |
| MAX_LISTED = 12 # names shown per section of the report | |
| class SeqItem: | |
| msg_id: int | |
| name: str | |
| caption: str = "" | |
| season: Optional[int] = None | |
| episode: Optional[int] = None | |
| class SequenceState: | |
| """What /sequence has collected so far for one user.""" | |
| items: List[SeqItem] = field(default_factory=list) | |
| status_id: Optional[int] = None # the "N files collected" message, replaced after each burst of files | |
| task: Optional["asyncio.Future[None]"] = None # the debounced status update that is waiting for the burst to end | |
| version: int = 0 # bumped for every file, so the status task can tell whether more are still arriving | |
| closed: bool = False # /sort or /cancel ended the collection | |
| full_noted: bool = False # the "limit reached" notice was shown | |
| def detect(name: str, caption: str, mode: str) -> Tuple[Optional[int], Optional[int]]: | |
| """``(season, episode)`` read from the file name and the caption, the one the rename mode names first. | |
| The season is 1 when only an episode number is found; both are ``None`` when there is no episode number. | |
| """ | |
| from_name = lambda: parse_info(split_ext(name)[0]) # noqa: E731 | |
| from_caption = lambda: parse_caption(caption) # noqa: E731 | |
| for read in ((from_caption, from_name) if mode == "caption" else (from_name, from_caption)): | |
| info = read() | |
| if info.episode.isdigit(): | |
| return (int(info.season) if info.season.isdigit() else 1), int(info.episode) | |
| return None, None | |
| class Analysis: | |
| ordered: List[SeqItem] # numbered files by season / episode, then the others in the order they were sent | |
| unknown: List[SeqItem] # no episode number found | |
| missing: Dict[int, List[int]] # season -> episode numbers absent between its first and last | |
| repeated: Dict[Tuple[int, int], List[SeqItem]] # (season, episode) -> the files that share it | |
| first: Optional[Tuple[int, int]] = None | |
| last: Optional[Tuple[int, int]] = None | |
| def analyse(items: Sequence[SeqItem], mode: str) -> Analysis: | |
| for item in items: | |
| item.season, item.episode = detect(item.name, item.caption, mode) | |
| numbered = sorted((i for i in items if i.episode is not None), key=lambda i: (i.season, i.episode, i.msg_id)) | |
| unknown = sorted((i for i in items if i.episode is None), key=lambda i: i.msg_id) | |
| by_season: Dict[int, List[int]] = {} | |
| groups: Dict[Tuple[int, int], List[SeqItem]] = {} | |
| for item in numbered: | |
| by_season.setdefault(item.season, []).append(item.episode) | |
| groups.setdefault((item.season, item.episode), []).append(item) | |
| missing: Dict[int, List[int]] = {} | |
| for season, episodes in by_season.items(): | |
| have = set(episodes) | |
| gaps = [n for n in range(min(have), max(have) + 1) if n not in have] | |
| if gaps: | |
| missing[season] = gaps | |
| return Analysis( | |
| ordered=numbered + unknown, | |
| unknown=unknown, | |
| missing=missing, | |
| repeated={k: v for k, v in groups.items() if len(v) > 1}, | |
| first=(numbered[0].season, numbered[0].episode) if numbered else None, | |
| last=(numbered[-1].season, numbered[-1].episode) if numbered else None, | |
| ) | |
| def _ep(n: int) -> str: | |
| return f"E{n:02d}" | |
| def ranges(numbers: Sequence[int], limit: int = 0) -> str: | |
| """``[3, 7, 8, 9]`` -> ``E03, E07–E09``; with ``limit`` the text is cut after that many characters.""" | |
| spans: List[Tuple[int, int]] = [] | |
| for n in sorted(set(numbers)): | |
| if spans and n == spans[-1][1] + 1: | |
| spans[-1] = (spans[-1][0], n) | |
| else: | |
| spans.append((n, n)) | |
| texts = [_ep(a) if a == b else f"{_ep(a)}–{_ep(b)}" for a, b in spans] | |
| if limit and len(", ".join(texts)) > limit: | |
| kept: List[str] = [] | |
| used = 0 | |
| for i, text in enumerate(texts): | |
| if used + len(text) + 2 > limit - 24: | |
| kept.append(f"… and {sum(b - a + 1 for a, b in spans[i:])} more") | |
| break | |
| kept.append(text) | |
| used += len(text) + 2 | |
| return ", ".join(kept) | |
| return ", ".join(texts) | |
| def _label(season: int, episode: int, multi: bool) -> str: | |
| return f"S{season:02d}{_ep(episode)}" if multi else _ep(episode) | |
| def _bullets(lines: Sequence[str]) -> List[str]: | |
| out = [f"• {line}" for line in lines[:MAX_LISTED]] | |
| if len(lines) > MAX_LISTED: | |
| out.append(f"… and {len(lines) - MAX_LISTED} more") | |
| return out | |
| def report_sections(a: Analysis, total: int, sent: int, failed: Sequence[SeqItem] = (), note: str = "") -> List[str]: | |
| """The pieces of the message shown after the files were sent (HTML), each short enough on its own.""" | |
| seasons = {i.season for i in a.ordered if i.season is not None} | |
| multi = len(seasons) > 1 | |
| head = f"✅ <b>Sorted {sent} of {total} files</b>" if sent != total else f"✅ <b>Sorted {total} file{'s' if total != 1 else ''}</b>" | |
| if a.first and a.last: | |
| head += f" · {_label(*a.first, multi)} → {_label(*a.last, multi)}" | |
| parts = [head] | |
| if note: | |
| parts.append(note) | |
| if a.missing: | |
| lines = [] | |
| for season in sorted(a.missing)[:MAX_LISTED]: | |
| text = ranges(a.missing[season], limit=300) | |
| lines.append(f"S{season:02d}: {text}" if multi else text) | |
| if len(a.missing) > MAX_LISTED: | |
| lines.append(f"… and {len(a.missing) - MAX_LISTED} more seasons") | |
| parts.append("⚠️ <b>Missing episodes</b>\n" + "\n".join(lines)) | |
| if a.repeated: | |
| lines = [] | |
| for (season, episode), files in sorted(a.repeated.items()): | |
| names = ", ".join(f"<code>{esc(clip(f.name, 40))}</code>" for f in files[:3]) | |
| more = f" +{len(files) - 3} more" if len(files) > 3 else "" | |
| lines.append(f"{_label(season, episode, multi)} ×{len(files)}: {names}{more}") | |
| parts.append("🔁 <b>Repeated episodes</b>\n" + "\n".join(_bullets(lines))) | |
| if a.unknown: | |
| lines = [f"<code>{esc(clip(i.name, 50))}</code>" for i in a.unknown] | |
| parts.append("❓ <b>No episode number found</b> (sent last)\n" + "\n".join(_bullets(lines))) | |
| if failed: | |
| lines = [f"<code>{esc(clip(i.name, 50))}</code>" for i in failed] | |
| parts.append("🚫 <b>Couldn't be sent</b>\n" + "\n".join(_bullets(lines))) | |
| if a.first and not (a.missing or a.repeated): | |
| parts.append("No missing or repeated episodes.") | |
| return parts | |
| def format_report( | |
| a: Analysis, total: int, sent: int, failed: Sequence[SeqItem] = (), note: str = "", limit: int = 3600 | |
| ) -> List[str]: | |
| """The report as one message, or a few when the lists are long (Telegram allows about 4000 characters).""" | |
| messages: List[str] = [] | |
| current = "" | |
| for part in report_sections(a, total, sent, failed, note): | |
| if current and len(current) + 2 + len(part) > limit: | |
| messages.append(current) | |
| current = part | |
| else: | |
| current = f"{current}\n\n{part}" if current else part | |
| messages.append(current) | |
| return messages | |