Compare commits
1
Commits
main
...
big-update
| Author | SHA1 | Date | |
|---|---|---|---|
 Latte
|
1ff979f1e5 |
@@ -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
|
||||
+27
-2
@@ -1,3 +1,28 @@
|
||||
/venv/
|
||||
# Secrets
|
||||
.env
|
||||
*.env
|
||||
!example.env
|
||||
|
||||
# Downloaded stickers
|
||||
downloads/
|
||||
|
||||
# Virtual environments
|
||||
venv/
|
||||
.venv/
|
||||
.env
|
||||
env/
|
||||
|
||||
# Python build & cache artefacts
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*.egg-info/
|
||||
build/
|
||||
dist/
|
||||
.pytest_cache/
|
||||
.ruff_cache/
|
||||
.coverage
|
||||
htmlcov/
|
||||
|
||||
# Editors / OS
|
||||
.idea/
|
||||
.vscode/
|
||||
.DS_Store
|
||||
|
||||
@@ -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
|
||||
- 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).
|
||||
|
||||
@@ -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())
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
||||
+3
-1
@@ -1 +1,3 @@
|
||||
TELEGRAM_BOT_TOKEN=your_token
|
||||
# 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
|
||||
|
||||
@@ -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"]
|
||||
@@ -0,0 +1,4 @@
|
||||
-r requirements.txt
|
||||
pytest>=8
|
||||
pytest-asyncio>=0.24
|
||||
ruff>=0.6
|
||||
+5
-19
@@ -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
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
@@ -0,0 +1,6 @@
|
||||
"""Allow ``python -m sticker_downloader``."""
|
||||
|
||||
from sticker_downloader.cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -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}=<your token> in a .env file next to this project, or
|
||||
* export {TOKEN_ENV_VAR}=<your token>, or
|
||||
* pass --token <your 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="<stdin>"
|
||||
)
|
||||
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())
|
||||
@@ -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)
|
||||
@@ -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__
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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/<name>`` 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))
|
||||
@@ -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")
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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()
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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]
|
||||
@@ -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]
|
||||
@@ -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
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user