diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..cb79ddf --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,35 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.10", "3.11", "3.12", "3.13"] + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements-dev.txt + pip install -e . + + - name: Lint + run: | + ruff check . + ruff format --check . + + - name: Test + run: pytest diff --git a/.gitignore b/.gitignore index 3c56d01..6bdb45b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,28 @@ -/venv/ +# Secrets +.env +*.env +!example.env + +# Downloaded stickers +downloads/ + +# Virtual environments +venv/ .venv/ -.env \ No newline at end of file +env/ + +# Python build & cache artefacts +__pycache__/ +*.py[cod] +*.egg-info/ +build/ +dist/ +.pytest_cache/ +.ruff_cache/ +.coverage +htmlcov/ + +# Editors / OS +.idea/ +.vscode/ +.DS_Store diff --git a/README.md b/README.md index 9aa450c..9e3bcf0 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,227 @@ # telegram-sticker-downloader -a python telegram sticker downloader +Download Telegram sticker packs to your disk — one folder per pack, in the file +formats you ask for. -with more info \ No newline at end of file +- Downloads static (`.webp`), animated (`.tgs`), video (`.webm`) stickers and PNG + thumbnails, filtered to whichever types you want. +- Downloads files **in parallel**, with retries and Telegram rate-limit handling. +- **Resumable**: files already on disk are skipped without spending API calls. +- Writes a `pack.json` per pack with each sticker's emoji, size and file names. +- Works as a scriptable CLI *and* as the original interactive prompt. + +## Requirements + +- Python 3.10 or newer +- A Telegram **bot token** (free, takes a minute — see below) + +## Install + +```bash +git clone https://git.hiddenden.cafe/Hiddenden/telegram-sticker-downloader +cd telegram-sticker-downloader + +python3 -m venv .venv +source .venv/bin/activate # Windows: .venv\Scripts\activate + +pip install -e . # installs the `tg-stickers` command +# or, without installing the package: +pip install -r requirements.txt +``` + +## Get a bot token + +1. Open Telegram and message [@BotFather](https://t.me/BotFather). +2. Send `/newbot` and follow the prompts. +3. Copy the token it gives you. +4. Save it next to the project: + +```bash +cp example.env .env +# then edit .env: +# TELEGRAM_BOT_TOKEN=123456:ABC-DEF... +``` + +The token can also come from the `TELEGRAM_BOT_TOKEN` environment variable or +from `--token`. You do not need to add the bot to any chat — reading public +sticker packs is enough. + +## Usage + +```bash +# One pack +tg-stickers https://t.me/addstickers/SomePack + +# Several packs, only the static images and thumbnails +tg-stickers SomePack AnotherPack --types webp,png + +# Everything listed in urls.txt, 16 files at a time +tg-stickers --from-file urls.txt --concurrency 16 + +# From a pipe +printf 'PackOne\nPackTwo\n' | tg-stickers --from-file - + +# See what would happen without writing anything +tg-stickers --from-file urls.txt --dry-run + +# No arguments: the original interactive prompts +tg-stickers +``` + +Without installing the package, the same thing works as +`python -m sticker_downloader ...` or `python app.py ...`. + +### Pack references + +Anything recognisable is accepted, so you can paste straight from Telegram: + +``` +https://t.me/addstickers/SomePack t.me/addstickers/SomePack +https://telegram.me/addstickers/Pack tg://addstickers?set=SomePack +https://t.me/addemoji/SomePack SomePack +``` + +In a `--from-file` list, blank lines and `#` comments are ignored, and duplicate +packs are downloaded only once. + +### Options + +| Option | Description | +| --- | --- | +| `PACK ...` | One or more pack URLs / names | +| `-f, --from-file PATH` | Read references from a file, one per line (`-` for stdin) | +| `-t, --types LIST` | `webp`, `tgs`, `webm`, `png`, comma-separated, or `all` (default: `all`) | +| `-o, --output DIR` | Where packs are written (default: `downloads`) | +| `-c, --concurrency N` | Files downloaded in parallel per pack (default: `8`) | +| `--retries N` | Attempts per network call (default: `3`) | +| `--overwrite` | Re-download files that already exist | +| `--emoji-names` | Put each sticker's emoji in its file name | +| `--no-metadata` | Skip writing `pack.json` | +| `--no-pack-thumbnail` | Skip the pack's own cover image | +| `--dry-run` | Report what would be downloaded, write nothing | +| `--token TOKEN` | Bot token (overrides the environment) | +| `--env-file PATH` | dotenv file to read the token from (default: `.env`) | +| `-v, --verbose` | Log every file individually | +| `-q, --quiet` | Only report fatal errors | +| `--version` | Print the version | + +### File types + +| Type | What it is | Notes | +| --- | --- | --- | +| `webp` | Static stickers | Regular image, opens anywhere | +| `tgs` | Animated stickers | Gzipped Lottie JSON, not a video file | +| `webm` | Video stickers | VP9 with alpha | +| `png` | Thumbnails | Small preview Telegram generates per sticker | + +A sticker has exactly one primary type — `--types tgs` on a pack of static +stickers downloads nothing but the thumbnails you asked for. + +## Output + +``` +downloads/ +└── SomePack/ + ├── 001.webp + ├── 001_thumb.png + ├── 002.webp + ├── 002_thumb.png + ├── _pack_thumbnail.webp + └── pack.json +``` + +Files are numbered in pack order. With `--emoji-names` they become +`001_🦊.webp`. `pack.json` records what was downloaded: + +```json +{ + "pack": { + "reference": "https://t.me/addstickers/SomePack", + "name": "SomePack", + "title": "Some Pack", + "sticker_type": "regular" + }, + "generated_at": "2026-07-27T12:00:00+00:00", + "downloader_version": "1.0.0", + "requested_file_types": ["webp", "png"], + "sticker_count": 2, + "stickers": [ + { + "index": 1, + "emoji": "🦊", + "file_unique_id": "AgADBAADwqbcCw", + "type": "regular", + "width": 512, + "height": 512, + "is_animated": false, + "is_video": false, + "files": ["001.webp", "001_thumb.png"] + } + ] +} +``` + +## How it works + +Packs are processed one at a time so progress stays readable; the files inside a +pack download concurrently, bounded by `--concurrency`. + +- **Resume.** Existing files are detected *before* any API call, so re-running a + large batch costs almost nothing. Use `--overwrite` to force a refresh. +- **Atomic writes.** Files land via a temporary `.part` file, so an interrupted + run never leaves a half-written sticker that a later run would skip. +- **Retries.** Timeouts, connection errors and HTTP 429/5xx are retried with + exponential backoff; Telegram's own `retry_after` hint is respected. Genuine + errors (unknown pack, bad token) fail immediately instead of being retried. +- **Partial failure is not fatal.** A file that cannot be downloaded is reported + and the rest of the pack continues; the run's exit code reflects it. + +### Exit codes + +| Code | Meaning | +| --- | --- | +| `0` | Everything downloaded | +| `1` | One or more packs or files failed | +| `2` | Bad arguments, missing/rejected token, or aborted | + +## Development + +```bash +pip install -r requirements-dev.txt + +pytest # test suite +ruff check . # lint +ruff format . # format +``` + +The tests use fakes for the Telegram bot, plus a real local HTTP server for the +download layer, so the whole suite runs offline in well under a second. + +`.gitea/workflows/ci.yml` runs lint and tests on Python 3.10–3.13. It needs +Gitea Actions enabled on the repository and a registered `act_runner`; without a +runner the file is simply ignored. + +Layout: + +| Module | Responsibility | +| --- | --- | +| `sticker_downloader/cli.py` | Argument parsing, interactive prompts, wiring | +| `sticker_downloader/downloader.py` | Planning and downloading a pack | +| `sticker_downloader/urls.py` | Pack-reference parsing | +| `sticker_downloader/config.py` | Settings and file-type handling | +| `sticker_downloader/fetcher.py` | HTTP layer behind a small protocol | +| `sticker_downloader/retry.py` | Retry policy and backoff | +| `sticker_downloader/progress.py` | Console reporting | +| `sticker_downloader/results.py` | Result and totals objects | + +## Limitations + +- The Bot API only serves files up to 20 MB. Stickers are far smaller, so this + is not normally a concern. +- `.tgs` and `.webm` are downloaded as-is; no conversion to GIF/APNG is done. +- Private or deleted packs cannot be read, and are reported as + `sticker pack not found`. + +## License + +MIT — see [LICENSE](LICENSE). diff --git a/app.py b/app.py index 50c6fc0..6688dc2 100644 --- a/app.py +++ b/app.py @@ -1,160 +1,12 @@ -import asyncio -import os -from dotenv import load_dotenv -import aiohttp -from telegram import Bot -from telegram.error import TelegramError -from telegram.ext import Application +#!/usr/bin/env python3 +"""Backwards-compatible entry point. -async def download_single_pack(bot: Bot, session: aiohttp.ClientSession, sticker_url: str, allowed_types: list): - """ - Asynchronously downloads specific file types from a sticker pack into its own folder. - Includes improved logging for skipped files. - """ - master_download_dir = "downloads" - try: - if 't.me/addstickers/' not in sticker_url: - print(f"--> Invalid URL format: {sticker_url}. Skipping.") - return +Kept so ``python app.py`` keeps working. The real implementation lives in the +``sticker_downloader`` package; prefer ``python -m sticker_downloader`` or the +``tg-stickers`` console script, both of which also accept command-line flags. +""" - pack_name = sticker_url.split('/')[-1] - print(f"\n--- Processing pack: {pack_name} ---") +from sticker_downloader.cli import main - pack_path = os.path.join(master_download_dir, pack_name) - os.makedirs(pack_path, exist_ok=True) - print(f"Saving to directory: ./{pack_path}/") - - sticker_set = await bot.get_sticker_set(pack_name) - total_stickers = len(sticker_set.stickers) - print(f"Found {total_stickers} stickers. Filtering for types: {', '.join(allowed_types)}") - - download_count = 0 - skipped_count = 0 - for i, sticker in enumerate(sticker_set.stickers): - was_skipped = True # Ga ervan uit dat de sticker wordt overgeslagen, tenzij we iets downloaden - - # 1. Check voor primaire sticker (webp, tgs, webm) - primary_extension = ".webp" - if sticker.is_animated: - primary_extension = ".tgs" - elif sticker.is_video: - primary_extension = ".webm" - - primary_type = primary_extension.strip('.') - if primary_type in allowed_types: - was_skipped = False - file_to_download = await bot.get_file(sticker.file_id) - file_name = f"{i+1:03d}{primary_extension}" - file_path = os.path.join(pack_path, file_name) - - async with session.get(file_to_download.file_path) as response: - if response.status == 200: - content = await response.read() - with open(file_path, 'wb') as f: - f.write(content) - download_count += 1 - - # 2. Check voor PNG thumbnail - if 'png' in allowed_types and sticker.thumbnail: - was_skipped = False - thumb_to_download = await bot.get_file(sticker.thumbnail.file_id) - thumb_name = f"{i+1:03d}_thumb.png" - thumb_path = os.path.join(pack_path, thumb_name) - - async with session.get(thumb_to_download.file_path) as response: - if response.status == 200: - content = await response.read() - with open(thumb_path, 'wb') as f: - f.write(content) - download_count += 1 - - # NIEUW: Verbeterde logging voor overgeslagen bestanden - if was_skipped: - skipped_count += 1 - - # Update de status op de console - print(f"\rProcessing... (sticker {i+1}/{total_stickers}, downloaded: {download_count}, skipped: {skipped_count})", end="", flush=True) - - print(f"\n✅ Download complete for pack: {pack_name}. Total files downloaded: {download_count}") - - except TelegramError as e: - print(f"\nError for pack '{pack_name}': {e}") - except aiohttp.ClientError as e: - print(f"\nNetwork error while downloading for '{pack_name}': {e}") - except Exception as e: - print(f"\nAn unexpected error occurred with pack '{pack_name}': {e}") - -async def main(): - """ - Main asynchronous function to run the sticker downloader script. - """ - print("--- Telegram Sticker Pack Downloader ---") - - load_dotenv() - bot_token = os.getenv("TELEGRAM_BOT_TOKEN") - - if not bot_token: - print("\n[ERROR] Telegram Bot Token niet gevonden!") - print("Maak een '.env' bestand aan en voeg 'TELEGRAM_BOT_TOKEN=jouw_token' toe.") - return - - application = Application.builder().token(bot_token).build() - bot = application.bot - - async with aiohttp.ClientSession() as session: - try: - bot_user = await bot.get_me() - print(f"Bot '{bot_user.first_name}' initialized successfully.") - except Exception as e: - print(f"Error initializing bot. Is je token in .env correct? Details: {e}") - return - - print("\nAvailable file types: webp (static), tgs (animated), webm (video), png (thumbnail)") - type_input = input("Enter desired file types (comma-separated), or type 'all': ").strip().lower() - - # GEWIJZIGD: Correcte logica voor 'all' of lege input - if not type_input or type_input == 'all': - allowed_types = ['webp', 'tgs', 'webm', 'png'] - print("--> Downloading all available types.") - else: - allowed_types = [t.strip() for t in type_input.split(',')] - print(f"--> Selected types: {', '.join(allowed_types)}") - - print("\nChoose an option:") - print("1: Download a single sticker pack from a URL") - print("2: Download multiple packs from a 'urls.txt' file") - - choice = input("Enter your choice (1 or 2): ").strip() - - if choice == '1': - sticker_url = input("Enter the Sticker Pack URL: ").strip() - if not sticker_url: - print("Error: Sticker URL cannot be empty.") - else: - await download_single_pack(bot, session, sticker_url, allowed_types) - - elif choice == '2': - try: - with open('urls.txt', 'r') as f: - urls = [line.strip() for line in f if line.strip()] - - if not urls: - print("'urls.txt' is empty.") - return - - print(f"\nFound {len(urls)} URLs. Starting batch download...") - for url in urls: - await download_single_pack(bot, session, url, allowed_types) - - except FileNotFoundError: - print("\nError: 'urls.txt' not found.") - except Exception as e: - print(f"An error occurred while reading the file: {e}") - - else: - print("Invalid choice. Please run the script again and enter 1 or 2.") - - print("\n--- Script finished ---") - -if __name__ == '__main__': - asyncio.run(main()) \ No newline at end of file +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/example.env b/example.env index f8df979..94b3e31 100644 --- a/example.env +++ b/example.env @@ -1 +1,3 @@ -TELEGRAM_BOT_TOKEN=your_token \ No newline at end of file +# Copy this file to .env and fill in your own bot token. +# Create a bot by messaging @BotFather on Telegram; it replies with the token. +TELEGRAM_BOT_TOKEN=your_token_here diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..cae6ff3 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,68 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "telegram-sticker-downloader" +description = "Download Telegram sticker packs into per-pack folders." +readme = "README.md" +requires-python = ">=3.10" +license = "MIT" +license-files = ["LICENSE"] +authors = [{ name = "Latte" }] +keywords = ["telegram", "stickers", "downloader", "cli"] +dynamic = ["version"] +classifiers = [ + "Development Status :: 4 - Beta", + "Environment :: Console", + "Intended Audience :: End Users/Desktop", + "Programming Language :: Python :: 3", + "Topic :: Communications :: Chat", + "Topic :: Utilities", +] +dependencies = [ + "python-telegram-bot>=21,<23", + "aiohttp>=3.9", + "python-dotenv>=1.0", +] + +[project.optional-dependencies] +dev = ["pytest>=8", "pytest-asyncio>=0.24", "ruff>=0.6"] + +[project.scripts] +tg-stickers = "sticker_downloader.cli:run" + +[project.urls] +Homepage = "https://git.hiddenden.cafe/Hiddenden/telegram-sticker-downloader" + +[tool.setuptools.dynamic] +version = { attr = "sticker_downloader.__version__" } + +[tool.setuptools.packages.find] +include = ["sticker_downloader*"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["."] +asyncio_mode = "auto" +asyncio_default_fixture_loop_scope = "function" +addopts = "-q --strict-markers" +filterwarnings = ["error", "default::DeprecationWarning"] + +[tool.ruff] +line-length = 90 +target-version = "py310" + +[tool.ruff.lint] +select = [ + "E", # pycodestyle + "F", # pyflakes + "I", # isort + "UP", # pyupgrade + "B", # bugbear + "SIM", # simplify + "RUF", +] + +[tool.ruff.lint.per-file-ignores] +"tests/*" = ["B008"] diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..5e9d764 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,4 @@ +-r requirements.txt +pytest>=8 +pytest-asyncio>=0.24 +ruff>=0.6 diff --git a/requirements.txt b/requirements.txt index 3671bcc..a2c9da6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,19 +1,5 @@ -aiohappyeyeballs==2.6.1 -aiohttp==3.12.14 -aiosignal==1.4.0 -anyio==4.9.0 -attrs==25.3.0 -certifi==2025.7.14 -dotenv==0.9.9 -frozenlist==1.7.0 -h11==0.16.0 -httpcore==1.0.9 -httpx==0.28.1 -idna==3.10 -multidict==6.6.3 -propcache==0.3.2 -python-dotenv==1.1.1 -python-telegram-bot==22.2 -sniffio==1.3.1 -typing_extensions==4.14.1 -yarl==1.20.1 +# Runtime dependencies. Transitive packages are resolved by pip. +# For development tooling see requirements-dev.txt. +python-telegram-bot>=21,<23 +aiohttp>=3.9 +python-dotenv>=1.0 diff --git a/sticker_downloader/__init__.py b/sticker_downloader/__init__.py new file mode 100644 index 0000000..11bdcd3 --- /dev/null +++ b/sticker_downloader/__init__.py @@ -0,0 +1,27 @@ +"""Download Telegram sticker packs from the command line.""" + +from sticker_downloader.config import ALL_FILE_TYPES, DownloadConfig, parse_file_types +from sticker_downloader.downloader import FileOutcome, PackResult, StickerDownloader +from sticker_downloader.errors import ( + FetchError, + InvalidPackReference, + StickerDownloaderError, +) +from sticker_downloader.urls import parse_pack_name, read_pack_references + +__version__ = "1.0.0" + +__all__ = [ + "ALL_FILE_TYPES", + "DownloadConfig", + "FetchError", + "FileOutcome", + "InvalidPackReference", + "PackResult", + "StickerDownloader", + "StickerDownloaderError", + "__version__", + "parse_file_types", + "parse_pack_name", + "read_pack_references", +] diff --git a/sticker_downloader/__main__.py b/sticker_downloader/__main__.py new file mode 100644 index 0000000..9c0c33a --- /dev/null +++ b/sticker_downloader/__main__.py @@ -0,0 +1,6 @@ +"""Allow ``python -m sticker_downloader``.""" + +from sticker_downloader.cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sticker_downloader/cli.py b/sticker_downloader/cli.py new file mode 100644 index 0000000..c189a02 --- /dev/null +++ b/sticker_downloader/cli.py @@ -0,0 +1,349 @@ +"""Command-line interface.""" + +from __future__ import annotations + +import argparse +import asyncio +import os +import sys +from collections.abc import Callable, Sequence +from pathlib import Path + +import aiohttp +from dotenv import load_dotenv +from telegram import Bot +from telegram.error import InvalidToken, TelegramError + +from sticker_downloader import __version__ +from sticker_downloader.config import ( + ALL_FILE_TYPES, + DEFAULT_CONCURRENCY, + DEFAULT_OUTPUT_DIR, + DEFAULT_RETRIES, + DEFAULT_URL_FILE, + FILE_TYPE_HELP, + DownloadConfig, + parse_file_types, +) +from sticker_downloader.downloader import StickerDownloader +from sticker_downloader.errors import StickerDownloaderError +from sticker_downloader.fetcher import AiohttpFetcher +from sticker_downloader.progress import ConsoleProgress, NullProgress, ProgressReporter +from sticker_downloader.results import PackResult +from sticker_downloader.urls import parse_pack_reference_lines, read_pack_references + +EXIT_OK = 0 +EXIT_FAILURES = 1 +EXIT_USAGE = 2 + +TOKEN_ENV_VAR = "TELEGRAM_BOT_TOKEN" + +_TOKEN_INSTRUCTIONS = f"""\ +Create a bot with @BotFather on Telegram, then either: + * put {TOKEN_ENV_VAR}= in a .env file next to this project, or + * export {TOKEN_ENV_VAR}=, or + * pass --token .""" + +_TOKEN_MISSING = f"No Telegram bot token found.\n\n{_TOKEN_INSTRUCTIONS}" +_TOKEN_REJECTED = f"Telegram rejected the bot token.\n\n{_TOKEN_INSTRUCTIONS}" + + +class UsageError(StickerDownloaderError): + """Raised for bad invocations; reported without a traceback.""" + + +def build_parser() -> argparse.ArgumentParser: + types_list = ", ".join(f"{name} ({FILE_TYPE_HELP[name]})" for name in ALL_FILE_TYPES) + parser = argparse.ArgumentParser( + prog="tg-stickers", + description="Download Telegram sticker packs into per-pack folders.", + epilog=( + "examples:\n" + " tg-stickers https://t.me/addstickers/SomePack\n" + " tg-stickers SomePack AnotherPack --types webp,png\n" + f" tg-stickers --from-file {DEFAULT_URL_FILE} --concurrency 16\n" + " tg-stickers --from-file - < my-packs.txt\n" + " tg-stickers # interactive prompts\n" + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + "packs", + nargs="*", + metavar="PACK", + help="sticker pack URLs, tg:// links or bare pack names", + ) + parser.add_argument( + "-f", + "--from-file", + metavar="PATH", + help=f"read references from a file, one per line ('-' for stdin); " + f"blank lines and # comments are ignored (default: {DEFAULT_URL_FILE} " + f"when chosen interactively)", + ) + parser.add_argument( + "-t", + "--types", + metavar="LIST", + help=f"comma-separated file types to save, or 'all'. Supported: {types_list}", + ) + parser.add_argument( + "-o", + "--output", + metavar="DIR", + type=Path, + default=DEFAULT_OUTPUT_DIR, + help=f"directory to write packs into (default: {DEFAULT_OUTPUT_DIR})", + ) + parser.add_argument( + "-c", + "--concurrency", + type=int, + default=DEFAULT_CONCURRENCY, + metavar="N", + help=f"files to download in parallel per pack (default: {DEFAULT_CONCURRENCY})", + ) + parser.add_argument( + "--retries", + type=int, + default=DEFAULT_RETRIES, + metavar="N", + help=f"attempts per network call (default: {DEFAULT_RETRIES})", + ) + parser.add_argument( + "--overwrite", + action="store_true", + help="re-download files that are already on disk (default: skip them)", + ) + parser.add_argument( + "--emoji-names", + action="store_true", + help="include each sticker's emoji in its file name", + ) + parser.add_argument( + "--no-metadata", + dest="write_metadata", + action="store_false", + help="do not write pack.json alongside the stickers", + ) + parser.add_argument( + "--no-pack-thumbnail", + dest="pack_thumbnail", + action="store_false", + help="do not download the pack's own cover thumbnail", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="list what would be downloaded without writing anything", + ) + parser.add_argument( + "--token", + metavar="TOKEN", + help=f"bot token (default: ${TOKEN_ENV_VAR}, also read from --env-file)", + ) + parser.add_argument( + "--env-file", + metavar="PATH", + default=".env", + help="dotenv file to load the token from (default: .env)", + ) + verbosity = parser.add_mutually_exclusive_group() + verbosity.add_argument( + "-v", "--verbose", action="store_true", help="log every file individually" + ) + verbosity.add_argument( + "-q", "--quiet", action="store_true", help="only report fatal errors" + ) + parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") + return parser + + +def _resolve_references( + args: argparse.Namespace, progress: ProgressReporter +) -> list[str]: + """Work out which packs to download, prompting only when nothing was given.""" + if args.packs and args.from_file: + raise UsageError("pass either PACK arguments or --from-file, not both") + + if args.packs: + references, problems = parse_pack_reference_lines(args.packs, source="argument") + for problem in problems: + progress.warn(problem.split(": ", 1)[-1]) + if not references: + raise UsageError("no valid sticker pack references given") + return references + + if args.from_file: + return _references_from_file(args.from_file, progress) + + return [] + + +def _references_from_file(source: str, progress: ProgressReporter) -> list[str]: + if source == "-": + references, problems = parse_pack_reference_lines( + sys.stdin.read().splitlines(), source="" + ) + else: + path = Path(source) + if not path.is_file(): + raise UsageError(f"{path} not found") + references, problems = read_pack_references(path) + + for problem in problems: + progress.warn(problem) + if not references: + raise UsageError(f"no valid sticker pack references in {source}") + return references + + +def collect_interactively( + *, + types_given: str | None, + url_file: Path = DEFAULT_URL_FILE, + ask: Callable[[str], str] = input, + say: Callable[[str], None] = print, +) -> tuple[str | None, list[str]]: + """Ask for the file types and pack references. + + Returns ``(types_text, references)``; ``types_text`` is ``None`` when the + caller already supplied ``--types``. + """ + say(f"--- Telegram Sticker Pack Downloader {__version__} ---") + + if types_given is None: + say("") + say("Available file types:") + for name in ALL_FILE_TYPES: + say(f" {name:<5} {FILE_TYPE_HELP[name]}") + types_given = ask("File types (comma-separated, or 'all') [all]: ").strip() + + # Validate now so a typo does not surface after the bot has connected. + parse_file_types(types_given) + + say("") + say("Choose an option:") + say(" 1: download one or more packs from URLs you type") + say(f" 2: download every pack listed in {url_file}") + choice = ask("Your choice (1 or 2) [1]: ").strip() or "1" + + if choice == "1": + answer = ask("Sticker pack URL(s), separated by spaces: ").strip() + references = answer.split() + if not references: + raise UsageError("no sticker pack URL given") + return types_given, references + + if choice == "2": + if not url_file.is_file(): + raise UsageError(f"{url_file} not found") + references, problems = read_pack_references(url_file) + for problem in problems: + say(f"warning: {problem}") + if not references: + raise UsageError(f"no valid sticker pack references in {url_file}") + say(f"Found {len(references)} pack(s) in {url_file}.") + return types_given, references + + raise UsageError(f"invalid choice {choice!r}; expected 1 or 2") + + +def _make_progress(args: argparse.Namespace) -> ProgressReporter: + if args.quiet: + return NullProgress() + return ConsoleProgress(sys.stderr, verbose=args.verbose) + + +def _resolve_token(args: argparse.Namespace) -> str: + env_file = Path(args.env_file) + if env_file.is_file(): + load_dotenv(env_file) + else: + load_dotenv() + token = (args.token or os.getenv(TOKEN_ENV_VAR) or "").strip() + if not token: + raise UsageError(_TOKEN_MISSING) + return token + + +async def _download( + token: str, + references: Sequence[str], + config: DownloadConfig, + progress: ProgressReporter, +) -> list[PackResult]: + timeout = aiohttp.ClientTimeout(total=300, sock_connect=30, sock_read=60) + async with Bot(token) as bot: + me = await bot.get_me() + progress.info(f"Connected as @{me.username}.") + async with aiohttp.ClientSession(timeout=timeout) as session: + downloader = StickerDownloader( + bot=bot, + fetcher=AiohttpFetcher(session), + config=config, + progress=progress, + ) + return await downloader.download_all(references) + + +def main(argv: Sequence[str] | None = None) -> int: + """Run the CLI. Returns the process exit code.""" + args = build_parser().parse_args(argv) + progress = _make_progress(args) + + try: + references = _resolve_references(args, progress) + types_text = args.types + + if not references: + if not sys.stdin.isatty(): + raise UsageError( + "no packs given. Pass PACK arguments, --from-file PATH, " + "or run interactively from a terminal." + ) + types_text, references = collect_interactively(types_given=args.types) + + config = DownloadConfig( + file_types=parse_file_types(types_text), + output_dir=args.output, + concurrency=args.concurrency, + retries=args.retries, + overwrite=args.overwrite, + write_metadata=args.write_metadata, + pack_thumbnail=args.pack_thumbnail, + emoji_names=args.emoji_names, + dry_run=args.dry_run, + ) + token = _resolve_token(args) + except (UsageError, StickerDownloaderError, ValueError) as error: + print(f"error: {error}", file=sys.stderr) + return EXIT_USAGE + except (EOFError, KeyboardInterrupt): + print("\nAborted.", file=sys.stderr) + return EXIT_USAGE + + if config.dry_run: + progress.info("Dry run: nothing will be written to disk.") + + try: + results = asyncio.run(_download(token, references, config, progress)) + except InvalidToken: + print(f"error: {_TOKEN_REJECTED}", file=sys.stderr) + return EXIT_USAGE + except TelegramError as error: + print(f"error: could not reach Telegram: {error}", file=sys.stderr) + return EXIT_FAILURES + except KeyboardInterrupt: + print("\nInterrupted.", file=sys.stderr) + return EXIT_FAILURES + + return EXIT_OK if all(result.ok for result in results) else EXIT_FAILURES + + +def run() -> int: # pragma: no cover - thin console-script wrapper + return main() + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/sticker_downloader/config.py b/sticker_downloader/config.py new file mode 100644 index 0000000..f489704 --- /dev/null +++ b/sticker_downloader/config.py @@ -0,0 +1,88 @@ +"""Download settings and file-type handling.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path + +from sticker_downloader.errors import InvalidFileType + +STATIC = "webp" +ANIMATED = "tgs" +VIDEO = "webm" +THUMBNAIL = "png" + +#: Every type the downloader knows how to save, in a stable display order. +ALL_FILE_TYPES: tuple[str, ...] = (STATIC, ANIMATED, VIDEO, THUMBNAIL) + +FILE_TYPE_HELP: dict[str, str] = { + STATIC: "static stickers", + ANIMATED: "animated stickers (Lottie)", + VIDEO: "video stickers", + THUMBNAIL: "PNG thumbnails", +} + +DEFAULT_OUTPUT_DIR = Path("downloads") +DEFAULT_URL_FILE = Path("urls.txt") +DEFAULT_CONCURRENCY = 8 +DEFAULT_RETRIES = 3 +METADATA_FILE_NAME = "pack.json" +PACK_THUMBNAIL_STEM = "_pack_thumbnail" + + +@dataclass(frozen=True) +class DownloadConfig: + """Everything that shapes a download run.""" + + file_types: frozenset[str] = field(default_factory=lambda: frozenset(ALL_FILE_TYPES)) + output_dir: Path = DEFAULT_OUTPUT_DIR + concurrency: int = DEFAULT_CONCURRENCY + retries: int = DEFAULT_RETRIES + overwrite: bool = False + write_metadata: bool = True + pack_thumbnail: bool = True + emoji_names: bool = False + dry_run: bool = False + + def __post_init__(self) -> None: + unknown = sorted(self.file_types - set(ALL_FILE_TYPES)) + if unknown: + raise InvalidFileType(f"unknown file type(s): {', '.join(unknown)}") + if not self.file_types: + raise InvalidFileType("at least one file type is required") + if self.concurrency < 1: + raise ValueError("concurrency must be at least 1") + if self.retries < 1: + raise ValueError("retries must be at least 1") + + @property + def sorted_file_types(self) -> list[str]: + """The requested types in canonical order, for stable output.""" + return [file_type for file_type in ALL_FILE_TYPES if file_type in self.file_types] + + def wants(self, file_type: str) -> bool: + return file_type in self.file_types + + +def parse_file_types(raw: str | None) -> frozenset[str]: + """Turn user input such as ``"webp, png"``, ``"all"`` or ``""`` into a type set. + + Empty input and ``"all"`` both mean "every supported type", matching what the + interactive prompt advertises. + """ + text = (raw or "").strip().lower() + if not text or text == "all": + return frozenset(ALL_FILE_TYPES) + + requested = [part.strip().lstrip(".") for part in text.replace(" ", ",").split(",")] + selected = {part for part in requested if part} + if not selected: + raise InvalidFileType("no file types given") + + unknown = sorted(selected - set(ALL_FILE_TYPES)) + if unknown: + raise InvalidFileType( + f"unknown file type(s): {', '.join(unknown)}. " + f"Supported: {', '.join(ALL_FILE_TYPES)}, or 'all'." + ) + return frozenset(selected) diff --git a/sticker_downloader/downloader.py b/sticker_downloader/downloader.py new file mode 100644 index 0000000..5c68871 --- /dev/null +++ b/sticker_downloader/downloader.py @@ -0,0 +1,359 @@ +"""The async sticker-pack downloader.""" + +from __future__ import annotations + +import asyncio +import json +import os +import re +from collections.abc import Awaitable, Callable, Iterable, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, TypeVar + +from telegram.error import BadRequest, TelegramError + +from sticker_downloader.config import ( + ALL_FILE_TYPES, + ANIMATED, + METADATA_FILE_NAME, + PACK_THUMBNAIL_STEM, + STATIC, + THUMBNAIL, + VIDEO, + DownloadConfig, +) +from sticker_downloader.errors import InvalidPackReference, StickerDownloaderError +from sticker_downloader.fetcher import Fetcher +from sticker_downloader.progress import NullProgress, ProgressReporter +from sticker_downloader.results import FileOutcome, FileStatus, PackResult +from sticker_downloader.retry import with_retries +from sticker_downloader.urls import parse_pack_name + +T = TypeVar("T") + +_UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]') +_PART_SUFFIX = ".part" + + +@dataclass(frozen=True) +class _FileJob: + """One file to put on disk.""" + + file_id: str + stem: Path + #: ``None`` means "derive the extension from Telegram's own file path", + #: which is how pack thumbnails are handled (they can be webp, tgs or webm). + suffix: str | None + + +class StickerDownloader: + """Downloads sticker packs concurrently into per-pack directories.""" + + def __init__( + self, + bot: Any, + fetcher: Fetcher, + config: DownloadConfig | None = None, + progress: ProgressReporter | None = None, + *, + sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, + now: Callable[[], datetime] | None = None, + ) -> None: + self._bot = bot + self._fetcher = fetcher + self._config = config or DownloadConfig() + self._progress: ProgressReporter = progress or NullProgress() + self._sleep = sleep + self._now = now or (lambda: datetime.now(timezone.utc)) + + @property + def config(self) -> DownloadConfig: + return self._config + + # -- public API ------------------------------------------------------- + + async def download_all(self, references: Iterable[str]) -> list[PackResult]: + """Download every reference in turn. + + Packs run one after another so progress stays readable; the files inside a + pack are downloaded concurrently. + """ + results: list[PackResult] = [] + for reference in references: + results.append(await self.download_pack(reference)) + self._progress.run_finished(results) + return results + + async def download_pack(self, reference: str) -> PackResult: + """Download a single pack, never raising for an expected failure.""" + try: + pack_name = parse_pack_name(reference) + except InvalidPackReference as error: + result = PackResult(reference=reference, error=str(error)) + self._progress.pack_finished(result) + return result + + try: + sticker_set = await self._call(lambda: self._bot.get_sticker_set(pack_name)) + except Exception as error: + result = PackResult( + reference=reference, name=pack_name, error=_describe(error) + ) + self._progress.pack_finished(result) + return result + + stickers: Sequence[Any] = list(getattr(sticker_set, "stickers", None) or []) + name = getattr(sticker_set, "name", None) or pack_name + directory = self._config.output_dir / name + + jobs = self._plan_jobs(sticker_set, stickers, directory) + self._progress.pack_started(reference, name, len(stickers), len(jobs)) + + if not self._config.dry_run: + await asyncio.to_thread(directory.mkdir, parents=True, exist_ok=True) + + semaphore = asyncio.Semaphore(self._config.concurrency) + outcomes = list( + await asyncio.gather(*(self._run_job(job, semaphore) for job in jobs)) + ) + + result = PackResult( + reference=reference, + name=name, + title=getattr(sticker_set, "title", None), + directory=directory, + total_stickers=len(stickers), + outcomes=outcomes, + ) + + if self._config.write_metadata and not self._config.dry_run: + try: + await self._write_metadata(sticker_set, stickers, directory, result) + except OSError as error: + self._progress.warn(f"could not write {METADATA_FILE_NAME}: {error}") + + self._progress.pack_finished(result) + return result + + # -- planning --------------------------------------------------------- + + def _plan_jobs( + self, sticker_set: Any, stickers: Sequence[Any], directory: Path + ) -> list[_FileJob]: + jobs: list[_FileJob] = [] + + set_thumbnail = getattr(sticker_set, "thumbnail", None) + if self._config.pack_thumbnail and set_thumbnail is not None: + jobs.append( + _FileJob( + file_id=set_thumbnail.file_id, + stem=directory / PACK_THUMBNAIL_STEM, + suffix=None, + ) + ) + + for index, sticker in enumerate(stickers, start=1): + base = self._sticker_stem(index, sticker) + + primary = primary_file_type(sticker) + if self._config.wants(primary): + jobs.append( + _FileJob( + file_id=sticker.file_id, + stem=directory / base, + suffix=f".{primary}", + ) + ) + + thumbnail = getattr(sticker, "thumbnail", None) + if self._config.wants(THUMBNAIL) and thumbnail is not None: + jobs.append( + _FileJob( + file_id=thumbnail.file_id, + stem=directory / f"{base}_thumb", + suffix=f".{THUMBNAIL}", + ) + ) + + return jobs + + def _sticker_stem(self, index: int, sticker: Any) -> str: + base = f"{index:03d}" + if not self._config.emoji_names: + return base + emoji = sanitize_filename_part(getattr(sticker, "emoji", None) or "") + return f"{base}_{emoji}" if emoji else base + + # -- execution -------------------------------------------------------- + + async def _run_job(self, job: _FileJob, semaphore: asyncio.Semaphore) -> FileOutcome: + async with semaphore: + outcome = await self._process_job(job) + self._progress.file_finished(outcome) + return outcome + + async def _process_job(self, job: _FileJob) -> FileOutcome: + planned_path = self._planned_path(job) + try: + existing = None if self._config.overwrite else self._existing_path(job) + if existing is not None: + return FileOutcome(existing, FileStatus.SKIPPED_EXISTING) + + if self._config.dry_run: + return FileOutcome(planned_path, FileStatus.PLANNED) + + file = await self._call(lambda: self._bot.get_file(job.file_id)) + remote_path = getattr(file, "file_path", None) + if not remote_path: + raise StickerDownloaderError("Telegram returned no file path") + + destination = self._resolve_path(job, remote_path) + data = await self._call(lambda: self._fetcher.fetch(remote_path)) + await asyncio.to_thread(_write_atomically, destination, data) + return FileOutcome(destination, FileStatus.DOWNLOADED, size=len(data)) + except Exception as error: + return FileOutcome(planned_path, FileStatus.FAILED, error=_describe(error)) + + async def _call(self, operation: Callable[[], Awaitable[T]]) -> T: + """Run one network operation with this run's retry policy.""" + return await with_retries( + operation, + attempts=self._config.retries, + on_retry=self._on_retry, + sleep=self._sleep, + ) + + def _on_retry(self, error: BaseException, attempt: int, delay: float) -> None: + self._progress.warn( + f"{_describe(error)} — retrying in {delay:.1f}s (attempt {attempt + 1})" + ) + + # -- paths ------------------------------------------------------------ + + def _planned_path(self, job: _FileJob) -> Path: + """Best guess at the destination, used for dry runs and error messages.""" + return self._resolve_path(job, f"unknown.{STATIC}") + + def _resolve_path(self, job: _FileJob, remote_path: str) -> Path: + suffix = job.suffix if job.suffix is not None else _suffix_from_url(remote_path) + return job.stem.parent / f"{job.stem.name}{suffix}" + + def _existing_path(self, job: _FileJob) -> Path | None: + """The already-downloaded file for ``job``, if there is one. + + Checking before calling ``get_file`` means resuming a large batch costs no + API calls for the parts that are already on disk. + """ + if job.suffix is not None: + candidate = job.stem.parent / f"{job.stem.name}{job.suffix}" + return candidate if candidate.is_file() else None + for file_type in ALL_FILE_TYPES: + candidate = job.stem.parent / f"{job.stem.name}.{file_type}" + if candidate.is_file(): + return candidate + return None + + # -- metadata --------------------------------------------------------- + + async def _write_metadata( + self, + sticker_set: Any, + stickers: Sequence[Any], + directory: Path, + result: PackResult, + ) -> None: + present = { + outcome.path + for outcome in result.outcomes + if outcome.status in (FileStatus.DOWNLOADED, FileStatus.SKIPPED_EXISTING) + } + + entries = [] + for index, sticker in enumerate(stickers, start=1): + base = self._sticker_stem(index, sticker) + files = sorted( + path.name + for path in present + if path.stem == base or path.stem == f"{base}_thumb" + ) + entries.append( + { + "index": index, + "emoji": getattr(sticker, "emoji", None), + "file_unique_id": getattr(sticker, "file_unique_id", None), + "type": _enum_value(getattr(sticker, "type", None)), + "width": getattr(sticker, "width", None), + "height": getattr(sticker, "height", None), + "is_animated": bool(getattr(sticker, "is_animated", False)), + "is_video": bool(getattr(sticker, "is_video", False)), + "files": files, + } + ) + + from sticker_downloader import __version__ + + payload = { + "pack": { + "reference": result.reference, + "name": result.name, + "title": result.title, + "sticker_type": _enum_value(getattr(sticker_set, "sticker_type", None)), + }, + "generated_at": self._now().isoformat(), + "downloader_version": __version__, + "requested_file_types": self._config.sorted_file_types, + "sticker_count": len(stickers), + "stickers": entries, + } + + text = json.dumps(payload, indent=2, ensure_ascii=False) + "\n" + await asyncio.to_thread( + (directory / METADATA_FILE_NAME).write_text, text, encoding="utf-8" + ) + + +# -- module-level helpers ------------------------------------------------- + + +def primary_file_type(sticker: Any) -> str: + """The file type Telegram stores this sticker as.""" + if getattr(sticker, "is_animated", False): + return ANIMATED + if getattr(sticker, "is_video", False): + return VIDEO + return STATIC + + +def sanitize_filename_part(text: str) -> str: + """Strip characters that are illegal or awkward inside a file name.""" + cleaned = _UNSAFE_FILENAME_CHARS.sub("", text).strip(" .") + return cleaned[:32] + + +def _suffix_from_url(remote_path: str) -> str: + suffix = Path(remote_path.split("?")[0]).suffix.lower() + return suffix if suffix else f".{STATIC}" + + +def _write_atomically(destination: Path, data: bytes) -> None: + """Write via a temporary file so an interrupted run leaves no partial file.""" + destination.parent.mkdir(parents=True, exist_ok=True) + temporary = destination.with_name(destination.name + _PART_SUFFIX) + temporary.write_bytes(data) + os.replace(temporary, destination) + + +def _enum_value(value: Any) -> Any: + return getattr(value, "value", value) + + +def _describe(error: BaseException) -> str: + """A short, user-facing description of a failure.""" + if isinstance(error, BadRequest) and "stickerset_invalid" in str(error).lower(): + return "sticker pack not found" + if isinstance(error, TelegramError): + return f"{type(error).__name__}: {error}" + message = str(error).strip() + return message or type(error).__name__ diff --git a/sticker_downloader/errors.py b/sticker_downloader/errors.py new file mode 100644 index 0000000..ad22f17 --- /dev/null +++ b/sticker_downloader/errors.py @@ -0,0 +1,31 @@ +"""Exception types used across the package.""" + +from __future__ import annotations + + +class StickerDownloaderError(Exception): + """Base class for every error raised by this package.""" + + +class InvalidPackReference(StickerDownloaderError): + """Raised when a string cannot be read as a sticker pack reference.""" + + +class InvalidFileType(StickerDownloaderError): + """Raised when the user asks for a file type we cannot download.""" + + +class FetchError(StickerDownloaderError): + """Raised when the CDN returns a non-success status for a file.""" + + def __init__(self, url: str, status: int, reason: str = "") -> None: + detail = f" ({reason})" if reason else "" + super().__init__(f"HTTP {status}{detail} for {url}") + self.url = url + self.status = status + self.reason = reason + + @property + def retriable(self) -> bool: + """Rate limits and server-side faults are worth another attempt.""" + return self.status == 429 or self.status >= 500 diff --git a/sticker_downloader/fetcher.py b/sticker_downloader/fetcher.py new file mode 100644 index 0000000..3591b26 --- /dev/null +++ b/sticker_downloader/fetcher.py @@ -0,0 +1,29 @@ +"""Fetching file bytes from Telegram's CDN.""" + +from __future__ import annotations + +from typing import Protocol, runtime_checkable + +import aiohttp + +from sticker_downloader.errors import FetchError + + +@runtime_checkable +class Fetcher(Protocol): + """Minimal interface the downloader needs to read a file off the network.""" + + async def fetch(self, url: str) -> bytes: ... + + +class AiohttpFetcher: + """A :class:`Fetcher` backed by a shared :class:`aiohttp.ClientSession`.""" + + def __init__(self, session: aiohttp.ClientSession) -> None: + self._session = session + + async def fetch(self, url: str) -> bytes: + async with self._session.get(url) as response: + if response.status >= 400: + raise FetchError(url, response.status, response.reason or "") + return await response.read() diff --git a/sticker_downloader/progress.py b/sticker_downloader/progress.py new file mode 100644 index 0000000..4bd8205 --- /dev/null +++ b/sticker_downloader/progress.py @@ -0,0 +1,158 @@ +"""Console reporting for a download run.""" + +from __future__ import annotations + +import sys +from typing import IO, Protocol + +from sticker_downloader.results import ( + FileOutcome, + FileStatus, + PackResult, + RunTotals, + format_size, +) + + +class ProgressReporter(Protocol): + """Everything the downloader and CLI report as a run unfolds.""" + + def info(self, message: str) -> None: ... + def warn(self, message: str) -> None: ... + def error(self, message: str) -> None: ... + def pack_started( + self, reference: str, name: str, total_stickers: int, total_files: int + ) -> None: ... + def file_finished(self, outcome: FileOutcome) -> None: ... + def pack_finished(self, result: PackResult) -> None: ... + def run_finished(self, results: list[PackResult]) -> None: ... + + +class NullProgress: + """Reports nothing. Used by ``--quiet`` and by tests.""" + + def info(self, message: str) -> None: ... + def warn(self, message: str) -> None: ... + def error(self, message: str) -> None: ... + def pack_started( + self, reference: str, name: str, total_stickers: int, total_files: int + ) -> None: ... + def file_finished(self, outcome: FileOutcome) -> None: ... + def pack_finished(self, result: PackResult) -> None: ... + def run_finished(self, results: list[PackResult]) -> None: ... + + +class ConsoleProgress: + """Writes human-friendly progress to a stream. + + On a terminal the per-pack counters are redrawn in place; when the output is + piped to a file each pack reports a single summary line instead, so logs stay + readable. + """ + + def __init__( + self, + stream: IO[str] | None = None, + *, + verbose: bool = False, + use_ansi: bool | None = None, + ) -> None: + self._stream = stream if stream is not None else sys.stderr + self._verbose = verbose + if use_ansi is None: + use_ansi = bool(getattr(self._stream, "isatty", lambda: False)()) + self._use_ansi = use_ansi + self._live_width = 0 + self._pack_total = 0 + self._pack_done = 0 + + # -- plain messages --------------------------------------------------- + + def info(self, message: str) -> None: + self._write_line(message) + + def warn(self, message: str) -> None: + self._write_line(f"warning: {message}") + + def error(self, message: str) -> None: + self._write_line(f"error: {message}") + + # -- run lifecycle ---------------------------------------------------- + + def pack_started( + self, reference: str, name: str, total_stickers: int, total_files: int + ) -> None: + self._pack_total = total_files + self._pack_done = 0 + self._write_line(f"\n{name}: {total_stickers} stickers, {total_files} files") + + def file_finished(self, outcome: FileOutcome) -> None: + self._pack_done += 1 + if outcome.status is FileStatus.FAILED: + self.error(f"{outcome.name}: {outcome.error}") + return + if self._verbose: + detail = ( + f" ({format_size(outcome.size)})" + if outcome.status is FileStatus.DOWNLOADED + else "" + ) + self._write_line(f" {outcome.status.value}: {outcome.name}{detail}") + return + total = self._pack_total or self._pack_done + self._draw_live(f" {self._pack_done}/{total} files") + + def pack_finished(self, result: PackResult) -> None: + self._clear_live() + if result.error is not None: + self.error(f"{result.reference}: {result.error}") + return + location = f" -> {result.directory}" if result.directory else "" + self._write_line(f" {result.summary()}{location}") + + def run_finished(self, results: list[PackResult]) -> None: + self._clear_live() + totals = RunTotals.from_results(results) + if totals.packs == 0: + self._write_line("\nNothing to do.") + return + written = format_size(totals.bytes_written) + lines = [ + "", + f"Done: {totals.packs_ok}/{totals.packs} packs", + f" files downloaded : {totals.downloaded} ({written})", + ] + if totals.planned: + lines.append(f" files planned : {totals.planned}") + if totals.skipped: + lines.append(f" already present : {totals.skipped}") + if totals.failed: + lines.append(f" files failed : {totals.failed}") + if totals.packs_failed: + lines.append(f" packs failed : {totals.packs_failed}") + for result in results: + if not result.ok: + reason = result.error or f"{result.failed} file(s) failed" + lines.append(f" - {result.reference}: {reason}") + self._write_line("\n".join(lines)) + + # -- internals -------------------------------------------------------- + + def _draw_live(self, text: str) -> None: + if not self._use_ansi: + return + padding = " " * max(0, self._live_width - len(text)) + self._stream.write(f"\r{text}{padding}") + self._stream.flush() + self._live_width = len(text) + + def _clear_live(self) -> None: + if self._use_ansi and self._live_width: + self._stream.write("\r" + " " * self._live_width + "\r") + self._stream.flush() + self._live_width = 0 + + def _write_line(self, text: str) -> None: + self._clear_live() + self._stream.write(f"{text}\n") + self._stream.flush() diff --git a/sticker_downloader/results.py b/sticker_downloader/results.py new file mode 100644 index 0000000..a3d2b25 --- /dev/null +++ b/sticker_downloader/results.py @@ -0,0 +1,118 @@ +"""Result objects describing what a download run did.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path + + +class FileStatus(str, Enum): + """Outcome of a single file within a pack.""" + + DOWNLOADED = "downloaded" + SKIPPED_EXISTING = "skipped_existing" + FAILED = "failed" + PLANNED = "planned" # --dry-run only + + +@dataclass(frozen=True) +class FileOutcome: + """What happened to one sticker file.""" + + path: Path + status: FileStatus + size: int = 0 + error: str | None = None + + @property + def name(self) -> str: + return self.path.name + + +@dataclass +class PackResult: + """Aggregated outcome for one sticker pack.""" + + reference: str + name: str | None = None + title: str | None = None + directory: Path | None = None + total_stickers: int = 0 + outcomes: list[FileOutcome] = field(default_factory=list) + error: str | None = None + + @property + def downloaded(self) -> int: + return sum(1 for o in self.outcomes if o.status is FileStatus.DOWNLOADED) + + @property + def skipped(self) -> int: + return sum(1 for o in self.outcomes if o.status is FileStatus.SKIPPED_EXISTING) + + @property + def planned(self) -> int: + return sum(1 for o in self.outcomes if o.status is FileStatus.PLANNED) + + @property + def failed(self) -> int: + return sum(1 for o in self.outcomes if o.status is FileStatus.FAILED) + + @property + def bytes_written(self) -> int: + return sum(o.size for o in self.outcomes if o.status is FileStatus.DOWNLOADED) + + @property + def ok(self) -> bool: + """True when the pack was processed without a fatal or per-file error.""" + return self.error is None and self.failed == 0 + + def summary(self) -> str: + if self.error is not None: + return f"failed: {self.error}" + parts = [f"{self.downloaded} downloaded"] + if self.planned: + parts.append(f"{self.planned} planned") + if self.skipped: + parts.append(f"{self.skipped} already present") + if self.failed: + parts.append(f"{self.failed} failed") + return ", ".join(parts) + + +@dataclass(frozen=True) +class RunTotals: + """Totals across every pack in a run.""" + + packs: int + packs_ok: int + packs_failed: int + downloaded: int + skipped: int + planned: int + failed: int + bytes_written: int + + @classmethod + def from_results(cls, results: list[PackResult]) -> RunTotals: + return cls( + packs=len(results), + packs_ok=sum(1 for r in results if r.ok), + packs_failed=sum(1 for r in results if not r.ok), + downloaded=sum(r.downloaded for r in results), + skipped=sum(r.skipped for r in results), + planned=sum(r.planned for r in results), + failed=sum(r.failed for r in results), + bytes_written=sum(r.bytes_written for r in results), + ) + + +def format_size(num_bytes: int) -> str: + """Human-readable byte count, e.g. ``1.4 MiB``.""" + size = float(num_bytes) + for unit in ("B", "KiB", "MiB", "GiB"): + if size < 1024 or unit == "GiB": + precision = 0 if unit == "B" else 1 + return f"{size:.{precision}f} {unit}" + size /= 1024 + raise AssertionError("unreachable") # pragma: no cover diff --git a/sticker_downloader/retry.py b/sticker_downloader/retry.py new file mode 100644 index 0000000..cfee4d0 --- /dev/null +++ b/sticker_downloader/retry.py @@ -0,0 +1,84 @@ +"""Retry helper shared by the Telegram API calls and the file downloads.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from typing import TypeVar + +import aiohttp +from telegram.error import BadRequest, Forbidden, InvalidToken, NetworkError, RetryAfter + +from sticker_downloader.errors import FetchError + +T = TypeVar("T") + +#: Transient failures. ``BadRequest`` subclasses ``NetworkError`` in +#: python-telegram-bot, so it has to be excluded explicitly below — retrying a +#: "sticker set is invalid" response would just waste three round-trips. +_RETRIABLE = (NetworkError, aiohttp.ClientError, asyncio.TimeoutError, OSError) +_NEVER_RETRIABLE = (BadRequest, Forbidden, InvalidToken) + + +def is_retriable(error: BaseException) -> bool: + """Whether another attempt could plausibly succeed.""" + if isinstance(error, RetryAfter): + return True + if isinstance(error, _NEVER_RETRIABLE): + return False + if isinstance(error, FetchError): + return error.retriable + return isinstance(error, _RETRIABLE) + + +def retry_delay(error: BaseException, attempt: int, base_delay: float) -> float: + """Seconds to wait before ``attempt`` (1-based) is retried. + + Honours Telegram's own ``retry_after`` hint when it sends one, and otherwise + backs off exponentially. + """ + if isinstance(error, RetryAfter): + retry_after = getattr(error, "retry_after", None) + if isinstance(retry_after, (int, float)): + return float(retry_after) + # Newer python-telegram-bot versions may hand back a timedelta. + seconds = getattr(retry_after, "total_seconds", None) + if callable(seconds): + return float(seconds()) + return base_delay * (2 ** (attempt - 1)) + + +async def with_retries( + operation: Callable[[], Awaitable[T]], + *, + attempts: int = 3, + base_delay: float = 0.5, + on_retry: Callable[[BaseException, int, float], None] | None = None, + sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, +) -> T: + """Await ``operation``, retrying transient failures up to ``attempts`` times. + + Args: + operation: A zero-argument coroutine function; called afresh per attempt. + attempts: Total number of tries, including the first one. + base_delay: Seed for the exponential backoff, in seconds. + on_retry: Called with ``(error, attempt, delay)`` before each retry. + sleep: Injectable sleep, so tests do not have to wait. + + Raises: + The last error, once the attempts are exhausted or it is not retriable. + """ + last_error: BaseException + for attempt in range(1, attempts + 1): + try: + return await operation() + except Exception as error: + last_error = error + if attempt == attempts or not is_retriable(error): + raise + delay = retry_delay(error, attempt, base_delay) + if on_retry is not None: + on_retry(error, attempt, delay) + await sleep(delay) + + raise last_error # pragma: no cover — loop always returns or raises diff --git a/sticker_downloader/urls.py b/sticker_downloader/urls.py new file mode 100644 index 0000000..0d515a7 --- /dev/null +++ b/sticker_downloader/urls.py @@ -0,0 +1,125 @@ +"""Turn the many ways of naming a sticker pack into a bare pack name.""" + +from __future__ import annotations + +import re +from collections.abc import Iterable +from pathlib import Path +from urllib.parse import parse_qs, urlparse + +from sticker_downloader.errors import InvalidPackReference + +#: Telegram pack names are ASCII letters, digits and underscores. +_PACK_NAME_RE = re.compile(r"^[A-Za-z0-9_]{1,64}$") + +#: Hosts that serve ``/addstickers/`` links. +_KNOWN_HOSTS = frozenset({"t.me", "telegram.me", "telegram.dog", "telesco.pe"}) + +_ADD_STICKERS_PATHS = frozenset({"addstickers", "addemoji"}) + + +def parse_pack_name(reference: str) -> str: + """Extract the pack name from anything a user is likely to paste. + + Accepts full URLs (``https://t.me/addstickers/Foo``), scheme-less links + (``t.me/addstickers/Foo``), ``tg://addstickers?set=Foo`` deep links, a bare + ``addstickers/Foo`` path, and a plain pack name. + + Raises: + InvalidPackReference: if no pack name can be recovered. + """ + text = (reference or "").strip().strip('"').strip("'") + if not text: + raise InvalidPackReference("empty sticker pack reference") + + if text.lower().startswith("tg://"): + return _validate(_name_from_deep_link(text), reference) + + if "/" not in text: + return _validate(text, reference) + + return _validate(_name_from_url(text, reference), reference) + + +def _name_from_deep_link(text: str) -> str: + parsed = urlparse(text) + query = parse_qs(parsed.query) + for key in ("set", "name"): + values = query.get(key) + if values and values[0].strip(): + return values[0].strip() + return "" + + +def _name_from_url(text: str, reference: str) -> str: + candidate = text if "//" in text else f"https://{text}" + parsed = urlparse(candidate) + segments = [segment for segment in parsed.path.split("/") if segment] + + host = parsed.netloc.lower().removeprefix("www.").split(":")[0] + if host in _ADD_STICKERS_PATHS: + # A path-only reference such as "addstickers/Foo". + return segments[0] if segments else "" + + if host not in _KNOWN_HOSTS: + raise InvalidPackReference( + f"{reference!r} is not a Telegram sticker link " + "(expected something like https://t.me/addstickers/PackName)" + ) + + if len(segments) < 2 or segments[0].lower() not in _ADD_STICKERS_PATHS: + raise InvalidPackReference(f"{reference!r} is not an /addstickers/ link") + return segments[1] + + +def _validate(name: str, reference: str) -> str: + name = name.strip() + if not _PACK_NAME_RE.match(name): + raise InvalidPackReference( + f"{reference!r} does not contain a valid pack name " + "(letters, digits and underscores only)" + ) + return name + + +def parse_pack_reference_lines( + lines: Iterable[str], source: str = "input" +) -> tuple[list[str], list[str]]: + """Parse pack references from lines of text. + + Blank lines and ``#`` comments are ignored. Duplicate references collapse to + the first occurrence so a pack is never downloaded twice in one run. + + Returns: + ``(references, problems)`` — valid references in input order, plus a + human-readable message for every line that could not be parsed. + """ + references: list[str] = [] + problems: list[str] = [] + seen: set[str] = set() + + for line_number, raw_line in enumerate(lines, start=1): + line = raw_line.strip() + if not line or line.startswith("#"): + continue + try: + name = parse_pack_name(line) + except InvalidPackReference as error: + problems.append(f"{source}:{line_number}: {error}") + continue + if name.lower() in seen: + continue + seen.add(name.lower()) + references.append(line) + + return references, problems + + +def read_pack_references(path: Path | str) -> tuple[list[str], list[str]]: + """Read pack references from a text file, one per line. + + See :func:`parse_pack_reference_lines` for the accepted syntax. + """ + file_path = Path(path) + lines = file_path.read_text(encoding="utf-8").splitlines() + return parse_pack_reference_lines(lines, source=str(file_path)) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..465553d --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,142 @@ +"""Test doubles for the Telegram bot and the CDN fetcher.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +from typing import Any + +import pytest +from telegram.error import BadRequest + +from sticker_downloader.config import DownloadConfig + + +@dataclass +class FakeFile: + file_path: str + + +@dataclass +class FakeThumbnail: + file_id: str + + +@dataclass +class FakeSticker: + file_id: str + file_unique_id: str = "uniq" + emoji: str | None = "😀" + is_animated: bool = False + is_video: bool = False + thumbnail: FakeThumbnail | None = None + width: int = 512 + height: int = 512 + type: str = "regular" + + +@dataclass +class FakeStickerSet: + name: str + title: str = "A Pack" + sticker_type: str = "regular" + stickers: list[FakeSticker] = field(default_factory=list) + thumbnail: FakeThumbnail | None = None + + +class FakeBot: + """Implements just the two coroutines the downloader calls.""" + + def __init__( + self, + sets: dict[str, FakeStickerSet] | None = None, + *, + extensions: dict[str, str] | None = None, + ) -> None: + self._sets = sets or {} + #: file_id -> extension served by the fake CDN URL. + self._extensions = extensions or {} + self.get_sticker_set_calls: list[str] = [] + self.get_file_calls: list[str] = [] + + async def get_sticker_set(self, name: str) -> FakeStickerSet: + self.get_sticker_set_calls.append(name) + try: + return self._sets[name] + except KeyError: + raise BadRequest("Stickerset_invalid") from None + + async def get_file(self, file_id: str) -> FakeFile: + self.get_file_calls.append(file_id) + extension = self._extensions.get(file_id, "webp") + return FakeFile(file_path=f"https://cdn.example/file/{file_id}.{extension}") + + +class FakeFetcher: + """Returns deterministic bytes and can be told to fail for some URLs.""" + + def __init__( + self, + *, + payload: bytes = b"sticker-bytes", + failures: dict[str, Exception] | None = None, + ) -> None: + self._payload = payload + self._failures = failures or {} + self.urls: list[str] = [] + self.max_concurrent = 0 + self._in_flight = 0 + + async def fetch(self, url: str) -> bytes: + self.urls.append(url) + self._in_flight += 1 + self.max_concurrent = max(self.max_concurrent, self._in_flight) + try: + await asyncio.sleep(0) + for needle, error in self._failures.items(): + if needle in url: + raise error + return self._payload + finally: + self._in_flight -= 1 + + +async def no_sleep(_delay: float) -> None: + """Drop-in for ``asyncio.sleep`` so retry tests run instantly.""" + return None + + +def make_sticker(index: int, **overrides: Any) -> FakeSticker: + """A static sticker with a PNG thumbnail, unless overridden.""" + defaults: dict[str, Any] = { + "file_id": f"file{index}", + "file_unique_id": f"uniq{index}", + "thumbnail": FakeThumbnail(file_id=f"thumb{index}"), + } + defaults.update(overrides) + return FakeSticker(**defaults) + + +@pytest.fixture +def pack() -> FakeStickerSet: + return FakeStickerSet( + name="TestPack", + title="Test Pack", + stickers=[make_sticker(1), make_sticker(2)], + thumbnail=FakeThumbnail(file_id="packthumb"), + ) + + +@pytest.fixture +def bot(pack: FakeStickerSet) -> FakeBot: + return FakeBot({pack.name: pack}) + + +@pytest.fixture +def fetcher() -> FakeFetcher: + return FakeFetcher() + + +@pytest.fixture +def config(tmp_path) -> DownloadConfig: + return DownloadConfig(output_dir=tmp_path / "downloads") diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..5d78721 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,344 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from sticker_downloader import cli +from sticker_downloader.cli import ( + EXIT_FAILURES, + EXIT_OK, + EXIT_USAGE, + UsageError, + build_parser, + collect_interactively, + main, +) +from sticker_downloader.results import FileOutcome, FileStatus, PackResult + + +@pytest.fixture(autouse=True) +def isolated_cwd(tmp_path, monkeypatch): + """Run each test in a clean directory with no ambient token.""" + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False) + return tmp_path + + +@pytest.fixture +def captured_run(monkeypatch): + """Replace the network layer and record what the CLI asked it to do.""" + recorded: dict = {} + + def _install(results: list[PackResult]) -> dict: + async def fake_download(token, references, config, progress): + recorded["token"] = token + recorded["references"] = list(references) + recorded["config"] = config + return results + + monkeypatch.setattr(cli, "_download", fake_download) + return recorded + + return _install + + +def ok_result(name: str = "Pack") -> PackResult: + return PackResult( + reference=name, + name=name, + outcomes=[FileOutcome(Path(f"{name}/001.webp"), FileStatus.DOWNLOADED, size=10)], + ) + + +def failed_result(name: str = "Pack") -> PackResult: + return PackResult(reference=name, name=name, error="sticker pack not found") + + +# -- parser --------------------------------------------------------------- + + +def test_parser_defaults(): + args = build_parser().parse_args([]) + assert args.packs == [] + assert args.from_file is None + assert args.types is None + assert args.output == Path("downloads") + assert args.concurrency == 8 + assert args.retries == 3 + assert args.write_metadata is True + assert args.pack_thumbnail is True + assert args.overwrite is False + assert args.dry_run is False + + +def test_version_exits_cleanly(capsys): + with pytest.raises(SystemExit) as excinfo: + main(["--version"]) + assert excinfo.value.code == EXIT_OK + assert "1.0.0" in capsys.readouterr().out + + +def test_verbose_and_quiet_are_mutually_exclusive(): + with pytest.raises(SystemExit): + build_parser().parse_args(["-v", "-q", "Pack"]) + + +# -- argument validation -------------------------------------------------- + + +def test_packs_and_from_file_conflict(capsys): + assert main(["Pack", "--from-file", "urls.txt"]) == EXIT_USAGE + assert "not both" in capsys.readouterr().err + + +def test_missing_from_file_is_a_usage_error(capsys): + assert main(["--from-file", "nope.txt"]) == EXIT_USAGE + assert "not found" in capsys.readouterr().err + + +def test_invalid_file_type_is_reported_before_connecting(capsys): + assert main(["Pack", "--types", "gif"]) == EXIT_USAGE + assert "gif" in capsys.readouterr().err + + +def test_all_references_invalid_is_a_usage_error(capsys): + assert main(["https://example.com/x"]) == EXIT_USAGE + assert "no valid sticker pack references" in capsys.readouterr().err + + +def test_missing_token_explains_how_to_set_one(capsys): + assert main(["Pack"]) == EXIT_USAGE + assert "BotFather" in capsys.readouterr().err + + +def test_no_packs_and_no_tty_is_a_usage_error(capsys, monkeypatch): + monkeypatch.setattr(cli.sys.stdin, "isatty", lambda: False) + assert main([]) == EXIT_USAGE + assert "no packs given" in capsys.readouterr().err + + +# -- wiring --------------------------------------------------------------- + + +def test_successful_run_returns_zero(captured_run, monkeypatch): + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "token-from-env") + recorded = captured_run([ok_result()]) + + assert main(["https://t.me/addstickers/Pack", "--types", "webp,png"]) == EXIT_OK + assert recorded["token"] == "token-from-env" + assert recorded["references"] == ["https://t.me/addstickers/Pack"] + assert recorded["config"].file_types == frozenset({"webp", "png"}) + + +def test_failing_pack_returns_one(captured_run, monkeypatch): + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "t") + captured_run([ok_result("A"), failed_result("B")]) + assert main(["A", "B"]) == EXIT_FAILURES + + +def test_token_flag_beats_environment(captured_run, monkeypatch): + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "from-env") + recorded = captured_run([ok_result()]) + assert main(["Pack", "--token", "from-flag"]) == EXIT_OK + assert recorded["token"] == "from-flag" + + +def test_token_is_read_from_env_file(captured_run, isolated_cwd): + (isolated_cwd / "custom.env").write_text("TELEGRAM_BOT_TOKEN=from-file\n") + recorded = captured_run([ok_result()]) + assert main(["Pack", "--env-file", "custom.env"]) == EXIT_OK + assert recorded["token"] == "from-file" + + +def test_flags_reach_the_config(captured_run, monkeypatch, tmp_path): + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "t") + recorded = captured_run([ok_result()]) + + exit_code = main( + [ + "Pack", + "--output", + str(tmp_path / "out"), + "--concurrency", + "4", + "--retries", + "5", + "--overwrite", + "--emoji-names", + "--no-metadata", + "--no-pack-thumbnail", + "--dry-run", + ] + ) + + assert exit_code == EXIT_OK + config = recorded["config"] + assert config.output_dir == tmp_path / "out" + assert config.concurrency == 4 + assert config.retries == 5 + assert config.overwrite is True + assert config.emoji_names is True + assert config.write_metadata is False + assert config.pack_thumbnail is False + assert config.dry_run is True + + +def test_default_types_are_all_when_not_interactive(captured_run, monkeypatch): + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "t") + recorded = captured_run([ok_result()]) + assert main(["Pack"]) == EXIT_OK + assert recorded["config"].sorted_file_types == ["webp", "tgs", "webm", "png"] + + +def test_references_come_from_file(captured_run, monkeypatch, isolated_cwd): + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "t") + (isolated_cwd / "packs.txt").write_text( + "# mine\nhttps://t.me/addstickers/A\n\nB\n", encoding="utf-8" + ) + recorded = captured_run([ok_result()]) + + assert main(["--from-file", "packs.txt"]) == EXIT_OK + assert recorded["references"] == ["https://t.me/addstickers/A", "B"] + + +def test_references_come_from_stdin(captured_run, monkeypatch): + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "t") + monkeypatch.setattr(cli.sys, "stdin", _FakeStdin("A\nB\n")) + recorded = captured_run([ok_result()]) + + assert main(["--from-file", "-"]) == EXIT_OK + assert recorded["references"] == ["A", "B"] + + +def test_invalid_lines_are_warned_about_but_do_not_stop_the_run( + captured_run, monkeypatch, isolated_cwd, capsys +): + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "t") + (isolated_cwd / "packs.txt").write_text("Good\nme/addstickers/Broken\n") + captured_run([ok_result()]) + + assert main(["--from-file", "packs.txt", "--quiet"]) == EXIT_OK + + +class _FakeStdin: + def __init__(self, text: str) -> None: + self._text = text + + def read(self) -> str: + return self._text + + def isatty(self) -> bool: + return False + + +# -- interactive mode ----------------------------------------------------- + + +def _scripted(answers: list[str]): + remaining = list(answers) + + def ask(_prompt: str) -> str: + return remaining.pop(0) + + return ask + + +def test_interactive_single_url(): + types, references = collect_interactively( + types_given=None, + ask=_scripted(["webp,png", "1", "https://t.me/addstickers/A t.me/addstickers/B"]), + say=lambda _msg: None, + ) + assert types == "webp,png" + assert references == ["https://t.me/addstickers/A", "t.me/addstickers/B"] + + +def test_interactive_defaults_to_all_types_and_option_one(): + types, references = collect_interactively( + types_given=None, + ask=_scripted(["", "", "A"]), + say=lambda _msg: None, + ) + assert types == "" + assert references == ["A"] + + +def test_interactive_skips_the_type_prompt_when_given(tmp_path): + types, references = collect_interactively( + types_given="webp", + ask=_scripted(["1", "A"]), + say=lambda _msg: None, + ) + assert types == "webp" + assert references == ["A"] + + +def test_interactive_reads_the_url_file(tmp_path): + url_file = tmp_path / "urls.txt" + url_file.write_text("https://t.me/addstickers/A\nB\n", encoding="utf-8") + + _types, references = collect_interactively( + types_given="all", + url_file=url_file, + ask=_scripted(["2"]), + say=lambda _msg: None, + ) + assert references == ["https://t.me/addstickers/A", "B"] + + +def test_interactive_missing_url_file(tmp_path): + with pytest.raises(UsageError): + collect_interactively( + types_given="all", + url_file=tmp_path / "absent.txt", + ask=_scripted(["2"]), + say=lambda _msg: None, + ) + + +def test_interactive_rejects_a_bad_choice(): + with pytest.raises(UsageError): + collect_interactively( + types_given="all", ask=_scripted(["9"]), say=lambda _msg: None + ) + + +def test_interactive_rejects_an_empty_url(): + with pytest.raises(UsageError): + collect_interactively( + types_given="all", ask=_scripted(["1", " "]), say=lambda _msg: None + ) + + +def test_interactive_validates_types_before_anything_else(): + with pytest.raises(Exception, match="gif"): + collect_interactively( + types_given=None, ask=_scripted(["gif"]), say=lambda _msg: None + ) + + +def test_interactive_path_is_used_when_stdin_is_a_tty( + captured_run, monkeypatch, tmp_path +): + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "t") + monkeypatch.setattr(cli.sys.stdin, "isatty", lambda: True) + monkeypatch.setattr( + cli, "collect_interactively", lambda **_kwargs: ("webp", ["Pack"]) + ) + recorded = captured_run([ok_result()]) + + assert main([]) == EXIT_OK + assert recorded["references"] == ["Pack"] + assert recorded["config"].file_types == frozenset({"webp"}) + + +def test_ctrl_d_during_prompts_aborts_cleanly(monkeypatch, capsys): + monkeypatch.setattr(cli.sys.stdin, "isatty", lambda: True) + + def raise_eof(**_kwargs): + raise EOFError + + monkeypatch.setattr(cli, "collect_interactively", raise_eof) + assert main([]) == EXIT_USAGE + assert "Aborted" in capsys.readouterr().err diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..e255d80 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import pytest + +from sticker_downloader.config import ALL_FILE_TYPES, DownloadConfig, parse_file_types +from sticker_downloader.errors import InvalidFileType + + +@pytest.mark.parametrize("raw", ["", " ", None, "all", "ALL", " All "]) +def test_blank_and_all_mean_everything(raw: str | None) -> None: + assert parse_file_types(raw) == frozenset(ALL_FILE_TYPES) + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("webp", {"webp"}), + ("webp,png", {"webp", "png"}), + (" WEBP , PNG ", {"webp", "png"}), + ("webp png", {"webp", "png"}), + (".webp,.tgs", {"webp", "tgs"}), + ("png,png", {"png"}), + ], +) +def test_parses_explicit_type_lists(raw: str, expected: set[str]) -> None: + assert parse_file_types(raw) == frozenset(expected) + + +@pytest.mark.parametrize("raw", ["gif", "webp,gif", "jpeg", ",,,"]) +def test_rejects_unsupported_types(raw: str) -> None: + with pytest.raises(InvalidFileType): + parse_file_types(raw) + + +def test_sorted_file_types_is_canonical_order() -> None: + config = DownloadConfig(file_types=frozenset({"png", "webp", "tgs"})) + assert config.sorted_file_types == ["webp", "tgs", "png"] + + +def test_wants_reports_membership() -> None: + config = DownloadConfig(file_types=frozenset({"webp"})) + assert config.wants("webp") + assert not config.wants("png") + + +def test_rejects_empty_type_set() -> None: + with pytest.raises(InvalidFileType): + DownloadConfig(file_types=frozenset()) + + +def test_rejects_unknown_type_in_config() -> None: + with pytest.raises(InvalidFileType): + DownloadConfig(file_types=frozenset({"gif"})) + + +@pytest.mark.parametrize(("concurrency", "retries"), [(0, 3), (-1, 3), (1, 0)]) +def test_rejects_nonsensical_limits(concurrency: int, retries: int) -> None: + with pytest.raises(ValueError): + DownloadConfig(concurrency=concurrency, retries=retries) diff --git a/tests/test_downloader.py b/tests/test_downloader.py new file mode 100644 index 0000000..1ec4d8f --- /dev/null +++ b/tests/test_downloader.py @@ -0,0 +1,270 @@ +from __future__ import annotations + +import json +from dataclasses import replace + +import pytest +from telegram.error import TimedOut + +from sticker_downloader.config import METADATA_FILE_NAME, DownloadConfig +from sticker_downloader.downloader import ( + StickerDownloader, + primary_file_type, + sanitize_filename_part, +) +from sticker_downloader.errors import FetchError +from sticker_downloader.results import FileStatus +from tests.conftest import ( + FakeBot, + FakeFetcher, + FakeSticker, + FakeStickerSet, + FakeThumbnail, + make_sticker, + no_sleep, +) + + +def build(bot, fetcher, config, **kwargs) -> StickerDownloader: + return StickerDownloader( + bot=bot, fetcher=fetcher, config=config, sleep=no_sleep, **kwargs + ) + + +async def test_downloads_stickers_thumbnails_and_pack_cover(bot, fetcher, config): + result = await build(bot, fetcher, config).download_pack( + "https://t.me/addstickers/TestPack" + ) + + assert result.ok + assert result.name == "TestPack" + assert result.title == "Test Pack" + assert result.total_stickers == 2 + + directory = config.output_dir / "TestPack" + written = sorted(p.name for p in directory.iterdir()) + assert written == [ + "001.webp", + "001_thumb.png", + "002.webp", + "002_thumb.png", + "_pack_thumbnail.webp", + METADATA_FILE_NAME, + ] + assert (directory / "001.webp").read_bytes() == b"sticker-bytes" + assert result.downloaded == 5 + assert result.bytes_written == 5 * len(b"sticker-bytes") + + +async def test_file_type_filter_limits_what_is_written(bot, fetcher, config): + config = replace(config, file_types=frozenset({"webp"}), pack_thumbnail=False) + result = await build(bot, fetcher, config).download_pack("TestPack") + + directory = config.output_dir / "TestPack" + assert sorted(p.name for p in directory.iterdir()) == [ + "001.webp", + "002.webp", + METADATA_FILE_NAME, + ] + assert result.downloaded == 2 + # No thumbnails were even looked up. + assert bot.get_file_calls == ["file1", "file2"] + + +async def test_extension_follows_sticker_kind(config, fetcher): + animated = FakeSticker(file_id="anim", is_animated=True) + video = FakeSticker(file_id="vid", is_video=True) + pack = FakeStickerSet(name="Mixed", stickers=[animated, video]) + bot = FakeBot({"Mixed": pack}, extensions={"anim": "tgs", "vid": "webm"}) + + await build(bot, fetcher, config).download_pack("Mixed") + + directory = config.output_dir / "Mixed" + assert (directory / "001.tgs").is_file() + assert (directory / "002.webm").is_file() + + +async def test_pack_thumbnail_extension_comes_from_telegram(config, fetcher): + pack = FakeStickerSet( + name="Vid", stickers=[make_sticker(1)], thumbnail=FakeThumbnail("cover") + ) + bot = FakeBot({"Vid": pack}, extensions={"cover": "webm"}) + + await build(bot, fetcher, config).download_pack("Vid") + + assert (config.output_dir / "Vid" / "_pack_thumbnail.webm").is_file() + + +async def test_existing_files_are_skipped_without_api_calls(bot, fetcher, config): + directory = config.output_dir / "TestPack" + directory.mkdir(parents=True) + (directory / "001.webp").write_bytes(b"old") + + result = await build(bot, fetcher, config).download_pack("TestPack") + + assert result.skipped == 1 + assert result.downloaded == 4 + assert (directory / "001.webp").read_bytes() == b"old" + assert "file1" not in bot.get_file_calls + + +async def test_overwrite_replaces_existing_files(bot, fetcher, config): + directory = config.output_dir / "TestPack" + directory.mkdir(parents=True) + (directory / "001.webp").write_bytes(b"old") + + config = replace(config, overwrite=True) + result = await build(bot, fetcher, config).download_pack("TestPack") + + assert result.skipped == 0 + assert (directory / "001.webp").read_bytes() == b"sticker-bytes" + + +async def test_dry_run_writes_nothing(bot, fetcher, config): + config = replace(config, dry_run=True) + result = await build(bot, fetcher, config).download_pack("TestPack") + + assert result.planned == 5 + assert result.downloaded == 0 + assert not config.output_dir.exists() + assert fetcher.urls == [] + assert bot.get_file_calls == [] + + +async def test_invalid_reference_never_touches_the_network(bot, fetcher, config): + result = await build(bot, fetcher, config).download_pack("https://example.com/nope") + + assert not result.ok + assert result.error is not None + assert bot.get_sticker_set_calls == [] + + +async def test_unknown_pack_reports_a_readable_error(bot, fetcher, config): + result = await build(bot, fetcher, config).download_pack("NoSuchPack") + + assert not result.ok + assert result.error == "sticker pack not found" + assert not config.output_dir.exists() + + +async def test_one_bad_file_does_not_sink_the_pack(bot, config): + fetcher = FakeFetcher(failures={"file1": FetchError("file1", 404, "Not Found")}) + result = await build(bot, fetcher, config).download_pack("TestPack") + + assert not result.ok + assert result.failed == 1 + assert result.downloaded == 4 + failure = next(o for o in result.outcomes if o.status is FileStatus.FAILED) + assert failure.path.name == "001.webp" + assert "404" in (failure.error or "") + + +async def test_transient_failures_are_retried(bot, config): + calls = {"n": 0} + real_fetch = FakeFetcher().fetch + + class FlakyFetcher: + async def fetch(self, url: str) -> bytes: + calls["n"] += 1 + if calls["n"] == 1: + raise TimedOut + return await real_fetch(url) + + result = await build(bot, FlakyFetcher(), config).download_pack("TestPack") + + assert result.ok + assert result.downloaded == 5 + assert calls["n"] == 6 # five files plus the one retry + + +async def test_no_partial_files_are_left_behind(bot, config): + fetcher = FakeFetcher(failures={"file2": FetchError("file2", 500)}) + config = replace(config, retries=1) + await build(bot, fetcher, config).download_pack("TestPack") + + directory = config.output_dir / "TestPack" + assert [p.name for p in directory.glob("*.part")] == [] + + +async def test_concurrency_is_bounded(config, fetcher): + stickers = [make_sticker(i) for i in range(1, 11)] + bot = FakeBot({"Big": FakeStickerSet(name="Big", stickers=stickers)}) + config = replace(config, concurrency=3) + + await build(bot, fetcher, config).download_pack("Big") + + assert fetcher.max_concurrent <= 3 + assert len(fetcher.urls) == 20 + + +async def test_emoji_names_are_used_when_requested(config, fetcher): + pack = FakeStickerSet(name="Emo", stickers=[FakeSticker(file_id="a", emoji="🦊")]) + bot = FakeBot({"Emo": pack}) + config = replace(config, emoji_names=True, pack_thumbnail=False) + + await build(bot, fetcher, config).download_pack("Emo") + + assert (config.output_dir / "Emo" / "001_🦊.webp").is_file() + + +async def test_metadata_describes_the_pack(bot, fetcher, config): + await build(bot, fetcher, config).download_pack("https://t.me/addstickers/TestPack") + + payload = json.loads( + (config.output_dir / "TestPack" / METADATA_FILE_NAME).read_text(encoding="utf-8") + ) + assert payload["pack"]["name"] == "TestPack" + assert payload["pack"]["title"] == "Test Pack" + assert payload["pack"]["reference"] == "https://t.me/addstickers/TestPack" + assert payload["sticker_count"] == 2 + assert payload["requested_file_types"] == ["webp", "tgs", "webm", "png"] + first = payload["stickers"][0] + assert first["index"] == 1 + assert first["emoji"] == "😀" + assert first["files"] == ["001.webp", "001_thumb.png"] + + +async def test_metadata_can_be_disabled(bot, fetcher, config): + config = replace(config, write_metadata=False) + await build(bot, fetcher, config).download_pack("TestPack") + assert not (config.output_dir / "TestPack" / METADATA_FILE_NAME).exists() + + +async def test_download_all_reports_every_pack(fetcher, config): + packs = { + "One": FakeStickerSet(name="One", stickers=[make_sticker(1)]), + "Two": FakeStickerSet(name="Two", stickers=[make_sticker(1)]), + } + bot = FakeBot(packs) + + results = await build(bot, fetcher, config).download_all(["One", "Two", "Missing"]) + + assert [r.reference for r in results] == ["One", "Two", "Missing"] + assert [r.ok for r in results] == [True, True, False] + + +def test_primary_file_type_prefers_animated_over_video(): + assert primary_file_type(FakeSticker(file_id="a")) == "webp" + assert primary_file_type(FakeSticker(file_id="a", is_animated=True)) == "tgs" + assert primary_file_type(FakeSticker(file_id="a", is_video=True)) == "webm" + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("🦊", "🦊"), + ("a/b", "ab"), + ("..", ""), + ('x<>:"|?*y', "xy"), + ("x" * 50, "x" * 32), + ], +) +def test_sanitize_filename_part(raw: str, expected: str): + assert sanitize_filename_part(raw) == expected + + +async def test_output_directory_is_created_lazily(bot, fetcher, tmp_path): + nested = tmp_path / "a" / "b" / "c" + config = DownloadConfig(output_dir=nested) + await build(bot, fetcher, config).download_pack("TestPack") + assert (nested / "TestPack" / "001.webp").is_file() diff --git a/tests/test_fetcher.py b/tests/test_fetcher.py new file mode 100644 index 0000000..d59953e --- /dev/null +++ b/tests/test_fetcher.py @@ -0,0 +1,79 @@ +"""Exercises AiohttpFetcher against a real local HTTP server.""" + +from __future__ import annotations + +from collections.abc import AsyncIterator + +import aiohttp +import pytest +import pytest_asyncio +from aiohttp import web + +from sticker_downloader.errors import FetchError +from sticker_downloader.fetcher import AiohttpFetcher, Fetcher + + +async def _handle(request: web.Request) -> web.Response: + status = int(request.match_info["status"]) + if status == 200: + return web.Response(body=b"file-bytes", content_type="image/webp") + return web.Response(status=status, text="nope") + + +@pytest_asyncio.fixture +async def server() -> AsyncIterator[str]: + app = web.Application() + app.router.add_get("/{status}", _handle) + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + port = runner.addresses[0][1] + try: + yield f"http://127.0.0.1:{port}" + finally: + await runner.cleanup() + + +@pytest_asyncio.fixture +async def session() -> AsyncIterator[aiohttp.ClientSession]: + async with aiohttp.ClientSession() as client: + yield client + + +async def test_returns_the_body_on_success(server: str, session) -> None: + fetcher = AiohttpFetcher(session) + assert await fetcher.fetch(f"{server}/200") == b"file-bytes" + + +@pytest.mark.parametrize("status", [404, 429, 500, 503]) +async def test_raises_fetch_error_with_the_status( + server: str, session, status: int +) -> None: + fetcher = AiohttpFetcher(session) + with pytest.raises(FetchError) as excinfo: + await fetcher.fetch(f"{server}/{status}") + assert excinfo.value.status == status + assert str(status) in str(excinfo.value) + + +async def test_only_server_side_statuses_are_retriable(server: str, session) -> None: + fetcher = AiohttpFetcher(session) + with pytest.raises(FetchError) as not_found: + await fetcher.fetch(f"{server}/404") + with pytest.raises(FetchError) as rate_limited: + await fetcher.fetch(f"{server}/429") + + assert not_found.value.retriable is False + assert rate_limited.value.retriable is True + + +async def test_connection_failures_surface_as_client_errors(session) -> None: + fetcher = AiohttpFetcher(session) + # Port 1 is reserved and refuses connections. + with pytest.raises(aiohttp.ClientError): + await fetcher.fetch("http://127.0.0.1:1/200") + + +def test_aiohttp_fetcher_satisfies_the_protocol(session) -> None: + assert isinstance(AiohttpFetcher(session), Fetcher) diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..d58aa4f --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,139 @@ +"""End-to-end test: the real downloader and HTTP layer against a local server. + +Only the Telegram Bot API itself is faked; everything else — planning, the +concurrency limit, retries, atomic writes and metadata — is the real code path. +""" + +from __future__ import annotations + +import json +from collections.abc import AsyncIterator + +import aiohttp +import pytest +import pytest_asyncio +from aiohttp import web + +from sticker_downloader.config import METADATA_FILE_NAME, DownloadConfig +from sticker_downloader.downloader import StickerDownloader +from sticker_downloader.fetcher import AiohttpFetcher +from tests.conftest import ( + FakeFile, + FakeStickerSet, + FakeThumbnail, + make_sticker, + no_sleep, +) + +#: file_id -> (extension, body). "flaky1" fails once before succeeding. +FILES = { + "file1": ("webp", b"one"), + "thumb1": ("png", b"one-thumb"), + "file2": ("webp", b"two"), + "thumb2": ("png", b"two-thumb"), + "packthumb": ("webp", b"cover"), +} + + +class _Server: + def __init__(self) -> None: + self.base = "" + self.hits: list[str] = [] + self.transient_failures = {"file2"} + + async def handle(self, request: web.Request) -> web.Response: + file_id = request.match_info["file_id"] + self.hits.append(file_id) + if file_id in self.transient_failures: + # Fail the first attempt only, so the retry path runs for real. + self.transient_failures.discard(file_id) + return web.Response(status=503, text="try again") + extension, body = FILES[file_id] + return web.Response(body=body, content_type=f"image/{extension}") + + +@pytest_asyncio.fixture +async def server() -> AsyncIterator[_Server]: + state = _Server() + app = web.Application() + app.router.add_get("/{file_id}.{extension}", state.handle) + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + state.base = f"http://127.0.0.1:{runner.addresses[0][1]}" + try: + yield state + finally: + await runner.cleanup() + + +class ServingBot: + """A Bot stand-in whose file paths point at the local test server.""" + + def __init__(self, base_url: str, sets: dict[str, FakeStickerSet]) -> None: + self._base_url = base_url + self._sets = sets + + async def get_sticker_set(self, name: str) -> FakeStickerSet: + return self._sets[name] + + async def get_file(self, file_id: str) -> FakeFile: + extension, _body = FILES[file_id] + return FakeFile(file_path=f"{self._base_url}/{file_id}.{extension}") + + +@pytest.fixture +def pack_of_two() -> FakeStickerSet: + return FakeStickerSet( + name="RealPack", + title="Real Pack", + stickers=[make_sticker(1), make_sticker(2)], + thumbnail=FakeThumbnail(file_id="packthumb"), + ) + + +async def test_full_download_over_http(server, pack_of_two, tmp_path): + config = DownloadConfig(output_dir=tmp_path / "out", concurrency=2) + async with aiohttp.ClientSession() as session: + downloader = StickerDownloader( + bot=ServingBot(server.base, {"RealPack": pack_of_two}), + fetcher=AiohttpFetcher(session), + config=config, + sleep=no_sleep, + ) + result = await downloader.download_pack("https://t.me/addstickers/RealPack") + + assert result.ok, result.error + assert result.downloaded == 5 + + directory = config.output_dir / "RealPack" + assert (directory / "001.webp").read_bytes() == b"one" + assert (directory / "001_thumb.png").read_bytes() == b"one-thumb" + assert (directory / "002.webp").read_bytes() == b"two" + assert (directory / "_pack_thumbnail.webp").read_bytes() == b"cover" + assert list(directory.glob("*.part")) == [] + + # The 503 on file2 was retried, so that file was requested twice. + assert server.hits.count("file2") == 2 + + metadata = json.loads((directory / METADATA_FILE_NAME).read_text(encoding="utf-8")) + assert metadata["sticker_count"] == 2 + assert metadata["stickers"][1]["files"] == ["002.webp", "002_thumb.png"] + + +async def test_second_run_skips_everything(server, pack_of_two, tmp_path): + config = DownloadConfig(output_dir=tmp_path / "out") + bot = ServingBot(server.base, {"RealPack": pack_of_two}) + + async with aiohttp.ClientSession() as session: + downloader = StickerDownloader( + bot=bot, fetcher=AiohttpFetcher(session), config=config, sleep=no_sleep + ) + await downloader.download_pack("RealPack") + hits_after_first = len(server.hits) + second = await downloader.download_pack("RealPack") + + assert second.downloaded == 0 + assert second.skipped == 5 + assert len(server.hits) == hits_after_first # no new requests at all diff --git a/tests/test_progress.py b/tests/test_progress.py new file mode 100644 index 0000000..d7d021e --- /dev/null +++ b/tests/test_progress.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +import io +from pathlib import Path + +import pytest + +from sticker_downloader.progress import ConsoleProgress, NullProgress +from sticker_downloader.results import ( + FileOutcome, + FileStatus, + PackResult, + RunTotals, + format_size, +) + + +def outcome(name: str, status: FileStatus, size: int = 0) -> FileOutcome: + return FileOutcome(Path("downloads/Pack") / name, status, size=size) + + +def reporter(**kwargs) -> tuple[ConsoleProgress, io.StringIO]: + stream = io.StringIO() + return ConsoleProgress(stream, use_ansi=False, **kwargs), stream + + +def test_pack_lifecycle_is_reported(): + progress, stream = reporter() + progress.pack_started("Pack", "Pack", 2, 3) + progress.file_finished(outcome("001.webp", FileStatus.DOWNLOADED, 1024)) + result = PackResult( + reference="Pack", + name="Pack", + directory=Path("downloads/Pack"), + total_stickers=2, + outcomes=[outcome("001.webp", FileStatus.DOWNLOADED, 1024)], + ) + progress.pack_finished(result) + + text = stream.getvalue() + assert "Pack: 2 stickers, 3 files" in text + assert "1 downloaded" in text + assert "downloads/Pack" in text + + +def test_quiet_files_are_not_listed_without_verbose(): + progress, stream = reporter() + progress.file_finished(outcome("001.webp", FileStatus.DOWNLOADED, 10)) + assert stream.getvalue() == "" + + +def test_verbose_lists_each_file_with_its_size(): + progress, stream = reporter(verbose=True) + progress.file_finished(outcome("001.webp", FileStatus.DOWNLOADED, 2048)) + assert "downloaded: 001.webp (2.0 KiB)" in stream.getvalue() + + +def test_failures_are_always_reported(): + progress, stream = reporter() + progress.file_finished( + FileOutcome(Path("001.webp"), FileStatus.FAILED, error="HTTP 404") + ) + assert "error: 001.webp: HTTP 404" in stream.getvalue() + + +def test_pack_error_is_reported_instead_of_a_summary(): + progress, stream = reporter() + progress.pack_finished(PackResult(reference="Nope", error="sticker pack not found")) + assert "error: Nope: sticker pack not found" in stream.getvalue() + + +def test_run_summary_lists_failing_packs(): + progress, stream = reporter() + good = PackResult( + reference="A", + name="A", + outcomes=[outcome("001.webp", FileStatus.DOWNLOADED, 1024)], + ) + bad = PackResult(reference="B", name="B", error="sticker pack not found") + progress.run_finished([good, bad]) + + text = stream.getvalue() + assert "Done: 1/2 packs" in text + assert "files downloaded : 1 (1.0 KiB)" in text + assert "packs failed : 1" in text + assert "- B: sticker pack not found" in text + + +def test_empty_run_says_so(): + progress, stream = reporter() + progress.run_finished([]) + assert "Nothing to do." in stream.getvalue() + + +def test_live_line_is_redrawn_on_a_terminal(): + stream = io.StringIO() + progress = ConsoleProgress(stream, use_ansi=True) + progress.pack_started("Pack", "Pack", 1, 2) + progress.file_finished(outcome("001.webp", FileStatus.DOWNLOADED, 1)) + progress.file_finished(outcome("002.webp", FileStatus.DOWNLOADED, 1)) + assert "\r 1/2 files" in stream.getvalue() + assert "\r 2/2 files" in stream.getvalue() + + +def test_null_progress_accepts_every_call(): + progress = NullProgress() + progress.info("x") + progress.warn("x") + progress.error("x") + progress.pack_started("a", "b", 1, 1) + progress.file_finished(outcome("001.webp", FileStatus.DOWNLOADED)) + progress.pack_finished(PackResult(reference="a")) + progress.run_finished([]) + + +def test_run_totals_aggregate_every_status(): + results = [ + PackResult( + reference="A", + outcomes=[ + outcome("001.webp", FileStatus.DOWNLOADED, 100), + outcome("002.webp", FileStatus.SKIPPED_EXISTING), + outcome("003.webp", FileStatus.FAILED), + ], + ), + PackResult(reference="B", outcomes=[outcome("001.webp", FileStatus.PLANNED)]), + ] + totals = RunTotals.from_results(results) + assert (totals.packs, totals.packs_ok, totals.packs_failed) == (2, 1, 1) + assert (totals.downloaded, totals.skipped, totals.planned, totals.failed) == ( + 1, + 1, + 1, + 1, + ) + assert totals.bytes_written == 100 + + +@pytest.mark.parametrize( + ("num_bytes", "expected"), + [(0, "0 B"), (512, "512 B"), (2048, "2.0 KiB"), (5 * 1024**2, "5.0 MiB")], +) +def test_format_size(num_bytes: int, expected: str): + assert format_size(num_bytes) == expected diff --git a/tests/test_retry.py b/tests/test_retry.py new file mode 100644 index 0000000..6855158 --- /dev/null +++ b/tests/test_retry.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +import asyncio +from datetime import timedelta + +import aiohttp +import pytest +from telegram.error import BadRequest, InvalidToken, NetworkError, RetryAfter, TimedOut + +from sticker_downloader.errors import FetchError +from sticker_downloader.retry import is_retriable, retry_delay, with_retries +from tests.conftest import no_sleep + + +async def test_returns_first_success() -> None: + calls = 0 + + async def operation() -> str: + nonlocal calls + calls += 1 + return "ok" + + assert await with_retries(operation, sleep=no_sleep) == "ok" + assert calls == 1 + + +async def test_retries_transient_failure_then_succeeds() -> None: + attempts = 0 + + async def operation() -> str: + nonlocal attempts + attempts += 1 + if attempts < 3: + raise TimedOut + return "ok" + + assert await with_retries(operation, attempts=3, sleep=no_sleep) == "ok" + assert attempts == 3 + + +async def test_raises_after_exhausting_attempts() -> None: + attempts = 0 + + async def operation() -> str: + nonlocal attempts + attempts += 1 + raise NetworkError("down") + + with pytest.raises(NetworkError): + await with_retries(operation, attempts=2, sleep=no_sleep) + assert attempts == 2 + + +async def test_does_not_retry_bad_request() -> None: + """BadRequest subclasses NetworkError in python-telegram-bot: it must not retry.""" + attempts = 0 + + async def operation() -> str: + nonlocal attempts + attempts += 1 + raise BadRequest("Stickerset_invalid") + + with pytest.raises(BadRequest): + await with_retries(operation, attempts=5, sleep=no_sleep) + assert attempts == 1 + + +async def test_reports_each_retry() -> None: + seen: list[tuple[str, int, float]] = [] + + async def operation() -> str: + raise TimedOut + + def on_retry(error: BaseException, attempt: int, delay: float) -> None: + seen.append((type(error).__name__, attempt, delay)) + + with pytest.raises(TimedOut): + await with_retries( + operation, attempts=3, base_delay=1.0, on_retry=on_retry, sleep=no_sleep + ) + assert seen == [("TimedOut", 1, 1.0), ("TimedOut", 2, 2.0)] + + +async def test_awaits_the_requested_delays() -> None: + slept: list[float] = [] + + async def record(delay: float) -> None: + slept.append(delay) + + async def operation() -> str: + raise TimedOut + + with pytest.raises(TimedOut): + await with_retries(operation, attempts=3, base_delay=0.5, sleep=record) + assert slept == [0.5, 1.0] + + +@pytest.mark.parametrize( + ("error", "expected"), + [ + (TimedOut(), True), + (NetworkError("x"), True), + (aiohttp.ClientError(), True), + (asyncio.TimeoutError(), True), + (RetryAfter(3), True), + (FetchError("u", 500), True), + (FetchError("u", 429), True), + (FetchError("u", 404), False), + (FetchError("u", 403), False), + (BadRequest("nope"), False), + (InvalidToken(), False), + (ValueError("nope"), False), + ], +) +def test_retriability_classification(error: BaseException, expected: bool) -> None: + assert is_retriable(error) is expected + + +def test_retry_after_delay_uses_telegram_hint() -> None: + assert retry_delay(RetryAfter(7), attempt=1, base_delay=0.5) == pytest.approx(7.0) + + +def test_retry_after_delay_accepts_a_timedelta() -> None: + """python-telegram-bot will switch ``retry_after`` to a timedelta in a future + major version; handle both shapes.""" + + class FutureRetryAfter(RetryAfter): + @property + def retry_after(self) -> timedelta: # type: ignore[override] + return timedelta(seconds=12) + + error = FutureRetryAfter(1) + assert retry_delay(error, attempt=1, base_delay=0.5) == pytest.approx(12.0) + + +def test_backoff_is_exponential() -> None: + delays = [retry_delay(TimedOut(), attempt=n, base_delay=0.5) for n in (1, 2, 3)] + assert delays == [0.5, 1.0, 2.0] diff --git a/tests/test_urls.py b/tests/test_urls.py new file mode 100644 index 0000000..d8ef77c --- /dev/null +++ b/tests/test_urls.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import pytest + +from sticker_downloader.errors import InvalidPackReference +from sticker_downloader.urls import ( + parse_pack_name, + parse_pack_reference_lines, + read_pack_references, +) + + +@pytest.mark.parametrize( + "reference", + [ + "https://t.me/addstickers/MyPack", + "http://t.me/addstickers/MyPack", + "https://www.t.me/addstickers/MyPack", + "t.me/addstickers/MyPack", + "https://telegram.me/addstickers/MyPack", + "https://telegram.dog/addstickers/MyPack", + "https://t.me/addstickers/MyPack/", + "https://t.me/addstickers/MyPack?utm_source=x", + "tg://addstickers?set=MyPack", + "addstickers/MyPack", + "MyPack", + " MyPack ", + '"https://t.me/addstickers/MyPack"', + ], +) +def test_accepts_every_reasonable_form(reference: str) -> None: + assert parse_pack_name(reference) == "MyPack" + + +def test_accepts_custom_emoji_links() -> None: + assert parse_pack_name("https://t.me/addemoji/MyPack") == "MyPack" + + +@pytest.mark.parametrize( + "reference", + [ + "", + " ", + # The original urls.txt had this typo on its first line. + "me/addstickers/LokisStickers", + "https://example.com/addstickers/MyPack", + "https://t.me/MyPack", + "https://t.me/joinchat/MyPack", + "https://t.me/addstickers/", + "My-Pack!", + "tg://addstickers?other=MyPack", + ], +) +def test_rejects_non_sticker_references(reference: str) -> None: + with pytest.raises(InvalidPackReference): + parse_pack_name(reference) + + +def test_reference_lines_skip_comments_blanks_and_duplicates() -> None: + references, problems = parse_pack_reference_lines( + [ + "# a comment", + "", + " ", + "https://t.me/addstickers/One", + "https://t.me/addstickers/Two", + # Same pack, different spelling: only the first survives. + "One", + "t.me/addstickers/one", + ] + ) + assert references == ["https://t.me/addstickers/One", "https://t.me/addstickers/Two"] + assert problems == [] + + +def test_reference_lines_report_bad_lines_with_line_numbers() -> None: + references, problems = parse_pack_reference_lines( + ["https://t.me/addstickers/Good", "not a link at all"], source="urls.txt" + ) + assert references == ["https://t.me/addstickers/Good"] + assert len(problems) == 1 + assert problems[0].startswith("urls.txt:2:") + + +def test_read_pack_references_from_file(tmp_path) -> None: + path = tmp_path / "urls.txt" + path.write_text( + "# packs\nhttps://t.me/addstickers/Alpha\nbroken line\n", encoding="utf-8" + ) + references, problems = read_pack_references(path) + assert references == ["https://t.me/addstickers/Alpha"] + assert len(problems) == 1 + assert str(path) in problems[0] diff --git a/urls.txt b/urls.txt index b916344..acb48f4 100644 --- a/urls.txt +++ b/urls.txt @@ -1,4 +1,10 @@ -me/addstickers/LokisStickers +# One sticker pack per line. Blank lines and #-comments are ignored. +# Accepted forms: https://t.me/addstickers/Name, t.me/addstickers/Name, +# tg://addstickers?set=Name, or just the bare pack name. +# +# Run with: tg-stickers --from-file urls.txt + +https://t.me/addstickers/LokisStickers https://t.me/addstickers/KelixFox https://t.me/addstickers/KingLewd https://t.me/addstickers/WildeFoxy3 @@ -94,4 +100,143 @@ https://t.me/addstickers/Stuffiepack https://t.me/addstickers/DLWRubi https://t.me/addstickers/KazSubstances https://t.me/addstickers/MoneroSticker0 -https://t.me/addstickers/DuckyDalmatian \ No newline at end of file +https://t.me/addstickers/DuckyDalmatian +https://t.me/addstickers/Bomadensfw_by_fStikBot +https://t.me/addstickers/FavoritosKato +https://t.me/addstickers/FapyFurs_by_fStikBot +https://t.me/addstickers/bowiebucky +https://t.me/addstickers/PetNala +https://t.me/addstickers/BenjiLewdStickers +https://t.me/addstickers/Psyox +https://t.me/addstickers/Chase68381NaL +https://t.me/addstickers/AnimatedAftersoon3_by_fStikBot +https://t.me/addstickers/TkaiNSFW +https://t.me/addstickers/dennicat +https://t.me/addstickers/saturnschmoovin +https://t.me/addstickers/ButterscotchWIP +https://t.me/addstickers/kittyvio +https://t.me/addstickers/AzulaBun +https://t.me/addstickers/FromAIC0 +https://t.me/addstickers/lolopoolopolool_by_fStikBot +https://t.me/addstickers/furrimilota_by_Camila_QT_by_TgEmodziBot +https://t.me/addstickers/simkitty +https://t.me/addstickers/sp10cc96bf559dd5ed10aea377f6370efd_by_stckrRobot +https://t.me/addstickers/StormysFavStickers_by_fStikBot +https://t.me/addstickers/Testing_5_by_fStikBot +https://t.me/addstickers/RolLemon2_by_fStikBot +https://t.me/addstickers/Jacobsstuff_by_fStikBot +https://t.me/addstickers/NSFWfurry2_by_fStikBot +https://t.me/addstickers/sp519a082666b43af388b3a1e8ac65925b_by_stckrRobot +https://t.me/addstickers/lewdyhio +https://t.me/addstickers/Zimty +https://t.me/addstickers/Frostyst +https://t.me/addstickers/SpaxeAustral +https://t.me/addstickers/KodiiPack +https://t.me/addstickers/ScissorsMalamute +https://t.me/addstickers/PentaButt +https://t.me/addstickers/spdea0591d563be46ecde20181a1ee9c64_by_stckrRobot +https://t.me/addstickers/BambiStickersbyJeniak +https://t.me/addstickers/TheCurlAnimatedNSFW +https://t.me/addstickers/WuiskilPawlowski78886NaL +https://t.me/addstickers/NoteHusky +https://t.me/addstickers/Horny_Owo_by_fStikBot +https://t.me/addstickers/DogRockets_by_fStikBot +https://t.me/addstickers/Horny_Stikers_6_by_fStikBot +https://t.me/addstickers/TheCommonGroundsByGlopossum +https://t.me/addstickers/CappuccinosStickers +https://t.me/addstickers/sp604d9c59ee1e47beec3119fb0bbe68fc_by_stckrRobot +https://t.me/addstickers/FurryPacksTop_by_fStikBot +https://t.me/addstickers/Matix03734NL +https://t.me/addstickers/Orionsticks +https://t.me/addstickers/flspack +https://t.me/addstickers/Freyaboar +https://t.me/addstickers/NightfireZephyrPureNSFW +https://t.me/addstickers/furryhentvideo +https://t.me/addstickers/dexthefolf_animated_sfw +https://t.me/addstickers/JosephTK +https://t.me/addstickers/Senips_Stickers +https://t.me/addstickers/hiostickerpack +https://t.me/addstickers/SudoBun +https://t.me/addstickers/RileyBullyBunny +https://t.me/addstickers/moshfox +https://t.me/addstickers/LoonaNudes_NSFW +https://t.me/addstickers/VilkatheMarshmallow +https://t.me/addstickers/MerciabyCocoLine +https://t.me/addstickers/Nikogifs_by_fStikBot +https://t.me/addstickers/GautreauNSFWSD +https://t.me/addstickers/CruiseTheDingo +https://t.me/addstickers/sp362b3639b3710246e74b6d82e8578144_by_stckrRobot +https://t.me/addstickers/Rollemon3_by_fStikBot +https://t.me/addstickers/yiffgifffff +https://t.me/addstickers/NSFWfurryStickers2_by_fStikBot +https://t.me/addstickers/hornyyyyyyyyyyyyyyy_by_fStikBot +https://t.me/addstickers/Keyyho +https://t.me/addstickers/Merlistickers125_by_fStikBot +https://t.me/addstickers/DakotaFluff +https://t.me/addstickers/AussieDOC +https://t.me/addstickers/VappySauce +https://t.me/addstickers/KrautibyDLWanimated +https://t.me/addstickers/BoBooty +https://t.me/addstickers/Mewglestickerpack +https://t.me/addstickers/konatals1 +https://t.me/addstickers/xenia_by_jf049 +https://t.me/addstickers/GottGotty +https://t.me/addstickers/Tolf82010NaL +https://t.me/addstickers/Umbreon_pack +https://t.me/addstickers/AcePanda +https://t.me/addstickers/MiksStickers2 +https://t.me/addstickers/bc455155_97cf_4e0b_9d51_419345164053_by_sticat_bot +https://t.me/addstickers/protobeans +https://t.me/addstickers/Wafelpack +https://t.me/addstickers/ZuriPack1 +https://t.me/addstickers/Coconut90321NaL +https://t.me/addstickers/One_sex +https://t.me/addstickers/animeluckystar_by_fStikBot +https://t.me/addstickers/foxemotesanim2 +https://t.me/addstickers/BlepGoesShep +https://t.me/addstickers/wypher_by_fStikBot +https://t.me/addstickers/fruitflavored +https://t.me/addstickers/ChesterVR +https://t.me/addstickers/Atlasrac +https://t.me/addstickers/biteyrix2 +https://t.me/addstickers/Blahaj +https://t.me/addstickers/ProtogenBadger +https://t.me/addstickers/Rusty10075NaL +https://t.me/addstickers/RipleyUmbreon +https://t.me/addstickers/foxona_fox +https://t.me/addstickers/Wypher_2 +https://t.me/addstickers/sp0cdbcec6ae20ffc0e24737fb810f8cf6_by_stckrRobot +https://t.me/addstickers/Toby03391NaL +https://t.me/addstickers/DuckyDalmatian +https://t.me/addstickers/RainerByZempy +https://t.me/addstickers/Filthypawkisser +https://t.me/addstickers/sp589d3e5778bde2076e5cd1ed94cb7473_by_stckrRobot +https://t.me/addstickers/Meesk +https://t.me/addstickers/EdgyCatboy +https://t.me/addstickers/Benji_Dog +https://t.me/addstickers/EHlla144 +https://t.me/addstickers/Flipymotes +https://t.me/addstickers/FlipysComms2 +https://t.me/addstickers/Jake10632NaL +https://t.me/addstickers/GatoFalopaa +https://t.me/addstickers/MechaFox29921NaL +https://t.me/addstickers/cool_gifs_4_by_fStikBot +https://t.me/addstickers/mowsh +https://t.me/addstickers/catahouligann_winston +https://t.me/addstickers/Bluezeru +https://t.me/addstickers/cliffdoggo +https://t.me/addstickers/Leon07572NaL +https://t.me/addstickers/Mir55783NaL +https://t.me/addstickers/PossSticks2 +https://t.me/addstickers/saturnhuskyuwu +https://t.me/addstickers/Kelix91488NaL +https://t.me/addstickers/Boomer49491NaL +https://t.me/addstickers/spf73c0a02b79e7af5bc087966a4cdfe67_by_stckrRobot +https://t.me/addstickers/Kaz89592NaL +https://t.me/addstickers/Koyodi +https://t.me/addstickers/Bitingowo +https://t.me/addstickers/Momochi85822NaL +https://t.me/addstickers/biteyrix +https://t.me/addstickers/scvPrototypeSuzi_by_fStikBot +https://t.me/addstickers/claytja +https://t.me/addstickers/CinoCat