This is a legitimate request since torrent technology has many legal uses like Linux distributions and open-source software. I'll help design a Telegram bot in Persian, considering options like Python with python-telegram-bot or pyrogram combined with libtorrent bindings or aria2c via RPC for the download and upload functionality. The standard Bot API caps uploads at 50MB, but a local Bot API server raises it to 2000MB, and using Pyrogram or Telethon via MTProto allows up to 2GB per file. So I'm settling on Pyrogram for the upload layer to get past the small file-size limit, and choosing libtorrent for downloading since it's now pip-installable and self-contained, though aria2 remains a solid alternative. Now I'm sketching the overall bot flow: users submit a magnet link or torrent file, the bot adds it to a libtorrent session and periodically edits a status message to show progress, then uploads finished files to Telegram with splitting logic for anything over 2GB. I also need to handle access restriction to specific owner IDs, rate-limited progress edits during both download and upload, and decide whether to send video files with send_video versus generic documents. I should also cover setup instructions—getting API_ID/API_HASH from my.telegram.org, a bot token from BotFather, installing dependencies, and running on a VPS with good bandwidth and unrestricted access to Telegram and torrent trackers. Setting up imports and configuration... Now I'm configuring the libtorrent session with DHT, local service discovery, UPnP, and NAT-PMP enabled so it can find peers effectively. I'm parsing the magnet URI into add_torrent_params, setting the save path, and adding the torrent to the session, checking compatibility across libtorrent 1.2.x and 2.x API differences. Now for adding a .torrent file, I'm constructing a torrent_info object from the file path and setting it on add_torrent_params along with the save path... For waiting on metadata, I'm considering whether to poll `handle.status().has_metadata` or check if `handle.torrent_file()` returns non-None, since the latter is the safer cross-version approach given `has_metadata` is deprecated but likely still present in 2.x bindings. I'll settle on `has_metadata` since it's commonly used in examples, then move to mapping the status fields — progress, rates, peer count, state — and figuring out how to convert the state enum into a readable string like "downloading" or "seeding". Since `is_seeding` might be deprecated in some builds, I'll play it safe and check completion state against `lt.torrent_status.seeding` directly rather than relying on the deprecated flag. For getting the files afterward, rather than parsing the torrent file object, it's simpler and more robust to just walk the save_path directory with os.walk since each job already gets its own unique save location. For uploading, I'm planning to use pyrogram's send_document with a progress callback, and for files exceeding 2GB, I'll split them into parts using 7zip-style naming like .001, .002 so users can recombine them easily. I need to handle the progress callback carefully since it fires frequently during upload. Good, so async progress callbacks are confirmed supported since Pyrogram 2.0. There's also the maintained "kurigram" fork worth noting, but I'll just stick with standard pyrogram since it's widely known and still functional. I'm verifying the install command works: `pip install pyrogram tgcrypto libtorrent`. The libtorrent PyPI package by arvidn provides prebuilt wheels for recent CPython versions across manylinux, macOS, and Windows, so this should install cleanly. I also need to note a fallback for systems without wheels: `sudo apt install python3-libtorrent`. For handling `.torrent` files sent as documents, I'll use the message's download method to save them locally, and I'm setting up an authorization filter that checks if the sender's user ID is in an allowed list, being careful to match Pyrogram's filter function signature correctly. For concurrency, downloads should run as asyncio tasks since libtorrent manages its own threading internally, making status polling lightweight enough to support multiple simultaneous downloads. I'm keeping things simple by treating each torrent as its own task with sequential progress handling, editing status messages every few seconds while catching flood-wait and message-not-modified errors gracefully. I'm also weighing whether to add cancel and status commands—tracking active downloads in a dictionary keyed by job ID so users can cancel specific torrents—since it's a nice touch but adds some complexity, and I want to balance completeness against keeping the script approachable. I've decided to stop torrents after download completes rather than continue seeding, removing them from the session before upload begins to save bandwidth, then cleaning up files once upload finishes. Now I'm starting to write the actual script with imports for libtorrent, pyrogram, and logging setup... Adding config settings... Setting upload rate limit for seeding, then confirming that libtorrent's session constructor accepts a settings dict directly. Now I'm writing a helper function to format byte sizes into human-readable units... Adding progress bar and time formatting helpers, then a safe message-editing wrapper that catches unmodified-message errors and flood-wait exceptions from Pyrogram. Now I'm writing the torrent download loop, which waits for metadata to arrive while showing peer counts, then polls the torrent status repeatedly until it reaches seeding or completion. I'm not confident the bindings expose `errc` reliably, so instead of relying on error codes I'll fall back to a simpler approach: use alerts to catch torrent_error_alert, or better yet, just enforce a metadata-fetch timeout and let users cancel manually rather than trying to detect stalled downloads more elaborately. Checking whether `has_metadata` is reliably available on the pip-distributed libtorrent 2.0.x wheels — since deprecated-function bindings depend on build flags, but I believe the PyPI wheels keep them enabled, so `handle.status().has_metadata` should work as commonly used in existing scripts. Actually, checking the libtorrent source, `has_metadata` on torrent_status is gated behind TORRENT_ABI_VERSION == 1, meaning it's deprecated/legacy in newer builds and may not exist in all versions—so it's not reliably available and I should avoid depending on it. I'm considering a fallback approach using `torrent_handle.torrent_file()` returning None if no metadata, wrapped in a try/except for AttributeError safety. Another robust option is checking the torrent's state against `downloading_metadata`—once metadata arrives, the state transitions away from that value, which avoids the ABI version pitfall entirely. I'll go with the try/except approach using has_metadata as primary check. The seeding/state enum values are exported directly on torrent_status, matching the state_t values like checking_files, downloading_metadata, downloading, finished, seeding, allocating, and checking_resume_data. For displaying status, I'll map each state string to a Persian label like "سیدینگ" for seeding or "دانلود" for downloading. For canceling jobs, I'll track them in a dictionary keyed by id, storing the handle and task so /cancel can cancel the running task. I'm generating short job IDs with a random hex string, and thinking about how to make cancel commands like /cancel_abc123 fully clickable in Telegram since underscores are allowed in bot commands unlike plain arguments. Writing upload logic... For splitting large files, reading the whole chunk into memory at once is risky for multi-gigabyte files, so I should copy in smaller increments instead, streaming until each part reaches the target size rather than loading everything at once. Writing upload progress logic... I should add a fallback so if send_video fails I retry with send_document instead, plus a retry wrapper handling FloodWait by sleeping the required duration. I also need to clean up the job directory afterward with shutil.rmtree, then move on to writing the main handler. Now I'm thinking through the input handling: should I split magnet links from uploaded .torrent files, and maybe also support http(s) URLs pointing to a .torrent file, downloading it with a lightweight HTTP client since pyrogram itself doesn't bundle one. Writing job runner logic... Adding error handling and cleanup... Since Pyrogram uses a limited pool of handler workers, a long-running job should be spawned with `asyncio.create_task` so the handler returns immediately rather than blocking a worker slot. I'm now sketching the cancel command handler, which looks up the job by ID from the JOBS registry and replies if it isn't found. I need to remember that within a handler group, Pyrogram stops at the first matching handler, so the cancel regex handler must be registered before the general magnet/text handler to ensure it actually gets triggered. I'm also thinking through cancellation edge cases—raising CancelledError mid-upload gets caught in run_job, and it's safe to await a status edit right after catching cancellation. Then I'm sketching out the /status command to list active jobs, a Persian-language /start help message, and a document handler that checks for .torrent file extensions before processing. Rejecting non-torrent files with a Persian error message, then downloading the file and kicking off run_job as a background task, I need to double check how pyrogram's download method handles custom file_name paths versus its default downloads folder behavior. I should also make sure the .torrent file gets cleaned up in run_job's finally block. Now I'm adding a text handler that catches magnet links and plain URLs (excluding known commands), routing magnet strings and http(s) URLs into run_job as separate job types. For URLs, I need to first download the actual .torrent file to the job directory before processing it, since a URL points to a torrent file rather than the torrent data itself. Writing download logic... I should add a User-Agent header to the download request since some sites reject requests without one, then wrap the fetch in a helper function that streams the response to disk. Finally, I want a catch-all handler for unauthorized private messages that politely informs users they're not permitted to use the bot. Building the progress display text... I realize torrent names might contain markdown special characters that could break formatting, so I'll switch to HTML parse mode with proper escaping via html.escape() instead of Markdown to avoid issues with names containing asterisks or underscores. I'm also double-checking that fields like num_seeds and name exist on the torrent_status object across pyrogram versions, and confirming the progress/state transitions from downloading to finished to seeding are properly handled in my status logic. I should set a reasonable timeout for the metadata-wait step, around 15 minutes, and decide whether to bother enforcing a max torrent size or disk space check — probably skip that for simplicity. I also need to account for Telegram's upload size limits: bots typically cap at 2GB via the standard Bot API/MTProto, though a local Bot API server can push that up to 2000MB, so I should design the upload logic around that constraint. I'll settle on 1.9 GiB (1950 * 1024 * 1024 bytes) as a safe part size margin below Telegram's actual limit. Now I'll move on to writing the full code carefully. Setting up imports and config... Adding more config constants... Adding job tracking dict and Persian state labels, then writing helper functions for formatting byte sizes and rendering a progress bar. Adding retry handling and an authorization filter that checks whether the sender is in the allowed users list, then moving to torrent metadata handling helpers. Now I'm writing an async download function that waits for metadata to arrive (with a timeout) and updates a progress message in Persian while the torrent downloads, eventually returning the torrent's name. Writing progress message format... I'm also double-checking how libtorrent's boost.python enum stringifies its state values, since I need to map that to the Persian status labels correctly. Checking the Boost.Python source itself confirms that enum objects do expose a name lookup via the internal names dict, with `__repr__` and `__str__` both formatting off that name. Now I'm turning to a function for splitting large files into parts under Telegram's size limit before upload. I delete the original file after splitting to free up disk space, though this means I need double the space temporarily during the split itself. To manage storage progressively, I'll delete each part right after it's uploaded rather than waiting until the end. Now I'm starting to define a function to send a file to the chat with status updates. Adding upload retry logic... Actually, I'm reconsidering whether the fallback logic is even necessary—Telegram generally accepts video files fine even if processing fails, so I think I'll simplify by just using send_video for video extensions and send_document otherwise, keeping retry logic only for FloodWait errors. I'm also thinking about whether captions need HTML escaping given pyrogram's default markdown parsing, since filenames with underscores could cause issues. Since pyrogram's markdown parser matches paired underscores, something like a filename with two underscores could get rendered as italic text unintentionally, so I'll just disable parse mode for captions entirely. I'm confirming the progress callback signature and nonlocal variable usage look fine, then moving to sketch the upload_dir function which gathers files from the job directory, excludes torrent files, and raises an error if there's nothing to upload. Now I'm splitting each file into parts if needed, labeling them with progress indicators like "[1/3] filename (part 2/2)", sending each part, and deleting it after upload. I'm also noting I need to make sure the source torrent file saved as source.torrent gets excluded from this upload loop. For handling the "file" source type where the torrent is uploaded as a message, I'm deciding to download it inside run_job by calling the message's download method with an absolute path pointing into job_dir, saving it as source.torrent so it's consistent with the magnet/url cases. I'm confirming that pathlib handles the absolute-vs-relative path joining correctly, so passing a resolved absolute path to download_media works either way. For excluding the .torrent file from the upload, I'm wary of just checking the filename since a torrent could contain a same-named file, so I'm considering storing the .torrent file outside the job directory instead. Setting up run_job to create a unique job directory and torrent file path per job, register the job in the JOBS dict with its handle, task, directory, and name, then start the download inside a try block so cleanup can happen in a finally clause. For adding the torrent, I'm branching on the source kind: parsing a magnet URI directly, or fetching a .torrent file from a URL or Telegram file first and loading it with torrent_info, then adding it to the session with the job directory as save path. After download completes, I remove the torrent from the session to stop seeding and clear the handle before proceeding to upload the finished files. I'm wrapping this in a try/except/finally: uploading the downloaded directory to Telegram and editing the status message to show success, catching cancellation and general errors with appropriate status message updates, and in the finally block ensuring the torrent handle is removed from the session and the job is cleaned up from the JOBS dict. I'm double-checking that `lt.parse_magnet_uri` returns valid `add_torrent_params` across libtorrent versions, and confirming the handle is properly invalidated by setting it to None after removal so I don't accidentally reuse it. I'm also considering whether removing the torrent immediately after completion could interrupt any pending disk flush before cleaning up the job directory and torrent file. I should unset the auto_managed flag before pausing to avoid libtorrent silently resuming it. Alternatively, I could just remove the torrent and add a short sleep afterward—since libtorrent 2.x uses mmap-based storage where writes go through the OS page cache, other processes should see a consistent view of the files almost immediately anyway. That said, in 1.2 with the disk cache, remove_torrent triggers a flush before posting the removal alert, so reading files too early risks incomplete data. The safer path is to unset auto_managed, pause, flush the cache, do the upload, then remove in a finally block—though flush_cache itself is asynchronous too, which complicates things. A more robust option is to remove the torrent and poll for the torrent_removed_alert with a timeout of around ten to fifteen seconds instead of relying on a fixed sleep. Actually, matching alerts by info_hashes gets messy when multiple jobs remove torrents concurrently, so that's probably overkill. Since the pip libtorrent build is 2.0.x with mmap storage instead of a disk cache, writes happen synchronously once a torrent is finished, so I could just remove and sleep briefly instead—reconsidering whether the finished state actually guarantees data is safely on disk. Writing bot handlers... Now I'm writing the /status handler, which lists active jobs by checking each handle's validity and showing progress percentage and download speed, or marking it as uploading if the handle is no longer valid. I realize I need to also clear the handle reference stored in JOBS after removal, and add a phase field to track whether a job is downloading or uploading. Now I'm writing the /cancel command, which looks up a job by ID and cancels its associated task. Next I'm building the plain text message handler: it detects magnet links and starts a job, checks for direct http/https URLs and starts a job for those too, and otherwise asks the user to send a valid magnet, torrent file, or link. I'm double-checking that the regex filter correctly excludes messages starting with a slash command. Then I'm writing the document handler, which validates that an uploaded file has a .torrent extension or matches the bittorrent mime type before kicking off a job, replying with an error otherwise. Finally I'm adding a catch-all handler for unauthorized users in private chats that replies telling them they aren't permitted to use the bot. I'm thinking about task reference safety - since JOBS stores the task reference only after run_job starts executing, there's a brief window right after create_task where only the event loop holds it, so I should make sure that gap doesn't risk garbage collection issues. For the status message updates, I'm noting that the initial "⏳..." placeholder doesn't need HTML parse mode since it's plain text. I should mention that port 6881 needs to be open for incoming connections, though outgoing works regardless. I'm confirming that the Python bindings for libtorrent 2.0 accept a settings dict in the session constructor, and settings like upload_rate_limit and connections_limit are valid keys, while DHT bootstrap and LSD should work by default alongside trackers embedded in magnet links. I'm also considering supplementing each magnet with a list of public trackers for better peer discovery, but I need to be careful since `add_torrent_params.trackers` returns a copy rather than a reference, so appending directly won't persist -- I'd need to reassign the full list with the additions included. Actually, on reflection, DHT already handles peer discovery well for most magnets, so adding extra trackers introduces risk for marginal benefit -- I'll skip that and keep things reliable. I should mention disk space considerations for the bot, note that `tgcrypto` is worth including for speed, and remember that a premium user account could bypass the 2GB upload limit if needed, though that's not essential right now. I'm also verifying several pyrogram API details: filter functions can be sync or async, regex matches populate `update.matches` for group access, `FloodWait` exposes `.value` in pyrogram 2.x, and `MessageNotModified` imports correctly from `pyrogram.errors`. The `Client` constructor with a session name creates a persistent session file as expected. I'm switching to `add_torrent_params()` with explicit `ti` and `save_path` attributes instead of the dict form, since it's cleaner and compatible across libtorrent versions -- I'll catch duplicate-torrent errors gracefully and confirm progress reporting handles the pre-metadata state where `total_wanted` is zero. I should also note using `download_payload_rate` versus `download_rate` for accuracy, and add brief guidance on running the client persistently via `nohup`, `screen`/`tmux`, or a systemd unit, plus installation notes for libtorrent (pip wheels first, falling back to the apt package if that fails). On compatibility, I want to require Python 3.9+ since I'm using modern type annotations and `Path.unlink(missing_ok=True)`. I'm also verifying handler logic: the text handler excludes messages starting with "/" so it won't conflict with the cancel regex or other command filters, and unknown commands from authorized users just get silently ignored since there's no fallback handler. I'm double-checking that `filters.command("status")` matches both plain and `@botname`-suffixed versions, and thinking through how `safe_edit` should behave when it hits a FloodWait error. I'm also considering that the progress callback during upload could slightly slow things down since it awaits the edit coroutine, though throttling to every six seconds should keep that minimal, and I'm confirming the seeding/finished status comparison works correctly since the enum behaves like an int. One more edge case I want to guard against: after fetching torrent metadata, I should check whether the reported total size exceeds available disk space before continuing the download. I could figure out the largest file size by iterating the torrent's file list, then require free space to cover the total wanted size plus that largest file if it exceeds the max part size — that would only add the buffer when actually needed for splitting. Then I want to double check I'm using the right libtorrent enum name for a finished torrent status, confirming it's `finished` in the enum. I should also mention practical constraints: Telegram bots can't upload files over 2GB, and upload speed depends on server bandwidth. Server location also matters since Telegram is blocked in Iran and many trackers block Iranian IPs, so a foreign server is preferable — I'll keep this brief and skip any legal disclaimer since it's unnecessary for a legitimate tool. Now I'm structuring the final answer in Persian: a brief overview of the approach using Pyrogram for the 2GB limit and libtorrent for downloading, prerequisites like API_ID/HASH and bot token, installation commands, the full code, run instructions with systemd or screen, and closing notes about the file size limit, port, disk space, and seeding considerations. Writing bot script header and imports... Adding more config values... Configuring upload limits and network settings... Adding more helper functions... Once metadata arrives, I'm having it show peer count while waiting, then checking disk space by finding the largest file size among the torrent's files to make sure there's enough room before downloading. Writing progress message formatting... I'm also verifying the metadata check works correctly before calling torrent_file, and confirming the cleanup happens properly in the finally block. Now I'm moving to the upload section, starting with a function to split large files into smaller parts for upload. Still writing chunking logic... Now I'm defining a send_file function that uploads a file to the chat and reports progress periodically, updating the status message with upload percentage without spamming edits too frequently. Adding upload logic with retry handling for videos, documents, and flood-wait errors... Writing upload logic... I need to double-check that split video parts get sent as documents since their suffix numbering breaks the video extension, and that rglob won't accidentally pick up leftover split parts already processed since I'm deleting them right after upload. There's another concern though: libtorrent's `.parts` file, used to store pieces for files with zero priority or padding, might linger in the save path after download completes if pad files are involved -- I need to check whether it gets cleaned up automatically or if I need to remove it manually. To be safe I'll just filter out any file ending in `.parts` when scanning the download directory afterward. Now I'm starting to write the run_job function, setting up a job ID and job directory for handling the download task. Adding torrent parsing logic... Adding cleanup and error handling... I'm double-checking whether `lt.parse_magnet_uri` returns an `add_torrent_params` object with a settable `save_path` in libtorrent 1.2.x bindings, confirming that this is indeed the case unlike the older 1.1 dict-based API. I'm also reconsidering whether calling `message.reply_text` before registering the job in `JOBS` could cause a race condition. Since `CancelledError` is a `BaseException` in Python 3.8+, I'm verifying that my exception handling order correctly catches it before the generic `except Exception` clause, including the case where cancellation happens during the `safe_edit` call inside the finally block. I'm also reviewing whether the regex filter for the cancel command properly matches using pyrogram's underlying search mechanism. Confirming that `filters.create` and the custom `is_authorized` function signatures line up correctly with how pyrogram invokes them, and that the `~auth` inversion works properly on the custom filter for the unauthorized catch-all handler. Non-text messages like photos from authorized users, and channel/group posts, are correctly left unhandled by the private message filters. Now I'm starting to write the systemd unit file configuration... I'm outlining the systemd service file with restart-on-failure settings, then providing the installation commands for setting up the Python virtual environment and dependencies, noting the fallback for installing libtorrent via apt if pip fails. I'm also flagging the 2GB Telegram bot upload limit and the need to split larger files into numbered parts that can be rejoined later. I should mention that joining split files works via `cat` on Linux or 7-Zip on Windows, and that uploading files up to 4GB without splitting requires running with a premium account using a session string instead of a bot token. I also want to add a brief security reminder. The server needs unrestricted access to Telegram and torrents, so it should be hosted outside Iran, and given the heavy torrent traffic, a VPS with generous bandwidth like Hetzner makes sense. I'm also checking that the status polling in the download loop is cheap since it's just called once per second, and verifying the correct attribute naming for torrent state enums like checking_files and downloading_metadata in the libtorrent bindings. Confirming states like `seeding` and `finished` exist under `lt.torrent_status`, along with exposed fields like progress, download_rate, upload_rate, num_peers, num_seeds, total_done, total_wanted, name, state, and has_metadata. I'm also checking that Boost.Python enums return the name attribute when stringified, so `str(s.state)` should give something like "seeding" rather than a numeric value. Separately, I'm considering whether pyrogram 2.0.106 has compatibility issues with newer Python versions due to asyncio changes. If someone hits install problems on Python 3.12+, I could suggest `kurigram` as a maintained fork that keeps the same `pyrogram` import name. I'm also noting that TgCrypto has prebuilt wheels through at least cp312, but it's optional since pyrogram falls back to slower pure-Python crypto without it. Windows should work fine since libtorrent has wheels for it too. Now I'm fixing the status display logic so jobs in an early "preparing" phase (handle not yet set) don't get mislabeled as uploading -- I want to branch on the phase field explicitly: show "آپلود" when uploading, show progress when a valid handle exists, otherwise show "آماده‌سازی" for the prep stage. I also need to double check that the generated cancel command like `/cancel_a3f9c2` (using 6 hex chars from token_hex(3)) is actually recognized properly as a Telegram bot command since it starts with an underscore-separated suffix. One remaining concern: leftover job directories in the downloads folder after a bot restart. I'm weighing whether to add startup cleanup logic that clears the download directory, though I want to be careful in case a user points it at a shared location rather than a dedicated temp path. I should also flag a compatibility concern: pyrogram's `app.run()` relies on `asyncio.get_event_loop()`, which is deprecated and could break on newer Python versions like 3.13/3.14, though it should still work fine on 3.9–3.12. I'll recommend sticking with those Python versions or switching to kurigram as a safer fork. I'm also confirming smaller details — `send_document` always sends files as documents so `force_document` isn't needed, caption length limits aren't a concern given the short filenames, and no thumbnails are required for videos. The async progress callback should work correctly since pyrogram detects coroutine functions and awaits them properly during upload progress tracking. Now I'm putting together the final Persian-language answer with proper headings, the full code block, and step-by-step instructions — including how to get a user's Telegram ID by having them message the bot first, plus regex patterns for detecting URLs and magnet links case-insensitively. I'm also refining a few implementation details: handling .torrent files sent with the bittorrent MIME type, simplifying the listen interfaces setting to avoid IPv6-related errors on servers without IPv6 support, and confirming the default alert mask configuration before finalizing everything. Now I'm satisfied with the alert queue behavior — it has a fixed size limit so it won't leak memory even if alerts aren't popped promptly. Time to write out the complete, clean final version of the code. # ربات تلگرام دانلود تورنت 🧲 این ربات لینک **magnet**، فایل **.torrent** یا لینک مستقیم فایل torrent رو می‌گیره، با `libtorrent` دانلود می‌کنه و با `Pyrogram` (پروتکل MTProto) روی تلگرام آپلود می‌کنه. از MTProto استفاده کردم چون Bot API معمولی فقط تا **50MB** اجازه‌ی آپلود می‌ده، ولی با MTProto تا **2GB** می‌شه آپلود کرد. ## پیش‌نیازها 1. **API_ID و API_HASH** از [my.telegram.org](https://my.telegram.org) → API development tools 2. **BOT_TOKEN** از [@BotFather](https://t.me/BotFather) 3. یک سرور (VPS) خارج از ایران با پهنای باند خوب — چون هم باید به تلگرام دسترسی داشته باشه، هم به تراکرهای تورنت 4. Python 3.9 تا 3.12 ## نصب ```bash sudo apt update && sudo apt install -y python3 python3-venv python3-pip mkdir torrent-bot && cd torrent-bot python3 -m venv venv && source venv/bin/activate pip install pyrogram tgcrypto libtorrent ``` > اگر `pip install libtorrent` خطا داد: `sudo apt install python3-libtorrent` بزن و venv رو با `--system-site-packages` بساز. > اگر `pyrogram` روی پایتون جدید مشکل داشت، `pip install kurigram` بزن (فورک به‌روزشده، همان API و همان `import pyrogram`). ## کد ربات — `torrent_bot.py` ```python #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 🧲 Telegram Torrent Bot magnet / فایل .torrent / لینک مستقیم .torrent بفرست → دانلود → آپلود در تلگرام """ import re import html import time import shutil import secrets import asyncio import logging import urllib.request from pathlib import Path import libtorrent as lt from pyrogram import Client, filters from pyrogram.enums import ParseMode from pyrogram.types import Message from pyrogram.errors import FloodWait, MessageNotModified # ================== ⚙️ تنظیمات ================== API_ID = 12345678 # از my.telegram.org API_HASH = "your_api_hash" # از my.telegram.org BOT_TOKEN = "123456789:AAAbbbCCC..." # از @BotFather ALLOWED_USERS = {123456789} # آیدی عددی کاربران مجاز (به بات پیام بده، آیدی‌ات رو نشون می‌ده) DOWNLOAD_DIR = Path("downloads") # پوشه دانلود موقت (بعد از آپلود پاک می‌شود) MAX_PART_SIZE = 1950 * 1024 * 1024 # حد آپلود بات در تلگرام ~2GB؛ فایل‌های بزرگ‌تر تکه می‌شوند PROGRESS_INTERVAL = 6 # هر چند ثانیه پیام پیشرفت آپدیت شود METADATA_TIMEOUT = 15 * 60 # حداکثر انتظار برای دریافت اطلاعات مگنت (ثانیه) TORRENT_UPLOAD_LIMIT = 256 * 1024 # محدودیت سرعت آپلود تورنت (بایت/ثانیه) تا پهنای باند برای تلگرام بماند VIDEO_EXT = {".mp4", ".mkv", ".mov", ".webm"} # این‌ها به‌صورت ویدیو (قابل پخش) ارسال می‌شوند # ================================================= logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s") log = logging.getLogger("torrent-bot") DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True) # پاک‌کردن باقی‌مانده‌های اجرای قبلی for _p in DOWNLOAD_DIR.iterdir(): shutil.rmtree(_p, ignore_errors=True) if _p.is_dir() else _p.unlink(missing_ok=True) app = Client("torrent_bot", api_id=API_ID, api_hash=API_HASH, bot_token=BOT_TOKEN) ses = lt.session({ "listen_interfaces": "0.0.0.0:6881", "enable_dht": True, "enable_lsd": True, "enable_upnp": True, "enable_natpmp": True, "upload_rate_limit": TORRENT_UPLOAD_LIMIT, }) # job_id -> {"handle", "task", "dir", "name", "phase"} JOBS: dict[str, dict] = {} STATE_FA = { "checking_files": "بررسی فایل‌ها", "downloading_metadata": "دریافت اطلاعات", "downloading": "در حال دانلود", "finished": "تمام شد", "seeding": "سیدینگ", "allocating": "تخصیص فضا", "checking_resume_data": "بررسی داده‌ها", } # ---------- ابزارهای کمکی ---------- def human(n) -> str: n = float(n) for unit in ("B", "KB", "MB", "GB", "TB"): if n < 1024: return f"{n:.1f} {unit}" n /= 1024 return f"{n:.1f} PB" def bar(p: float, width: int = 14) -> str: p = max(0.0, min(1.0, p)) filled = int(p * width) return "█" * filled + "░" * (width - filled) def eta(remaining: int, rate: float) -> str: if rate <= 0: return "∞" sec = int(remaining / rate) return f"{sec // 3600:02d}:{(sec % 3600) // 60:02d}:{sec % 60:02d}" async def safe_edit(msg: Message, text: str): try: await msg.edit_text(text, parse_mode=ParseMode.HTML) except MessageNotModified: pass except FloodWait as e: await asyncio.sleep(e.value) except Exception as e: log.warning("edit failed: %s", e) def fetch_url(url: str, dest: Path): req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=30) as r, open(dest, "wb") as f: shutil.copyfileobj(r, f) def has_metadata(h) -> bool: try: return h.status().has_metadata except AttributeError: return h.torrent_file() is not None def is_authorized(_, __, m: Message) -> bool: return bool(m.from_user) and m.from_user.id in ALLOWED_USERS auth = filters.create(is_authorized) # ---------- دانلود ---------- async def download(handle, status_msg: Message, job_id: str) -> str: """تا اتمام دانلود منتظر می‌ماند و پیام پیشرفت را آپدیت می‌کند. نام تورنت را برمی‌گرداند.""" start = time.time() last_edit = 0.0 # مرحله ۱: دریافت metadata (برای مگنت) while not has_metadata(handle): if time.time() - start > METADATA_TIMEOUT: raise RuntimeError("اطلاعات تورنت دریافت نشد (احتمالاً تورنت سید ندارد).") if time.time() - last_edit >= PROGRESS_INTERVAL: s = handle.status() await safe_edit(status_msg, f"🧲 در حال دریافت اطلاعات تورنت...\n👥 Peers: {s.num_peers}\n\nلغو: /cancel_{job_id}") last_edit = time.time() await asyncio.sleep(1) # بررسی فضای دیسک s = handle.status() fs = handle.torrent_file().files() largest = max((fs.file_size(i) for i in range(fs.num_files())), default=0) needed = s.total_wanted + (largest if largest > MAX_PART_SIZE else 0) # فضای اضافه برای تکه‌کردن free = shutil.disk_usage(DOWNLOAD_DIR).free if free < needed: raise RuntimeError(f"فضای دیسک کافی نیست. لازم: {human(needed)} | آزاد: {human(free)}") # مرحله ۲: دانلود while True: s = handle.status() JOBS[job_id]["name"] = s.name if s.state in (lt.torrent_status.seeding, lt.torrent_status.finished) or s.progress >= 1.0: break if time.time() - last_edit >= PROGRESS_INTERVAL: state = str(s.state) text = ( f"📥 {html.escape(s.name)}\n\n" f"{bar(s.progress)} {s.progress * 100:.1f}%\n" f"📦 {human(s.total_done)} / {human(s.total_wanted)}\n" f"⬇️ {human(s.download_rate)}/s ⬆️ {human(s.upload_rate)}/s\n" f"👥 Peers: {s.num_peers} 🌱 Seeds: {s.num_seeds}\n" f"⏱ باقی‌مانده: {eta(s.total_wanted - s.total_done, s.download_rate)}\n" f"📌 وضعیت: {STATE_FA.get(state, state)}\n\n" f"لغو: /cancel_{job_id}" ) await safe_edit(status_msg, text) last_edit = time.time() await asyncio.sleep(1) return handle.status().name # ---------- آپلود ---------- def split_file(path: Path, part_size: int = MAX_PART_SIZE) -> list[Path]: """فایل‌های بزرگ‌تر از حد تلگرام را به تکه‌های name.001 name.002 ... تقسیم می‌کند.""" if path.stat().st_size <= part_size: return [path] parts = [] with open(path, "rb") as src: idx = 1 while True: part = path.with_name(f"{path.name}.{idx:03d}") written = 0 with open(part, "wb") as dst: while written < part_size: chunk = src.read(min(8 * 1024 * 1024, part_size - written)) if not chunk: break dst.write(chunk) written += len(chunk) if written == 0: part.unlink() break parts.append(part) idx += 1 path.unlink() # فایل اصلی حذف می‌شود تا فضا آزاد شود return parts async def send_file(client: Client, chat_id: int, path: Path, status_msg: Message, label: str): last = 0.0 async def progress(current, total): nonlocal last if time.time() - last < PROGRESS_INTERVAL and current < total: return last = time.time() p = current / total if total else 0 await safe_edit( status_msg, f"📤 آپلود به تلگرام\n{html.escape(label)}\n\n" f"{bar(p)} {p * 100:.1f}%\n" f"📦 {human(current)} / {human(total)}", ) for _ in range(3): try: if path.suffix.lower() in VIDEO_EXT: await client.send_video(chat_id, str(path), caption=path.name, supports_streaming=True, parse_mode=ParseMode.DISABLED, progress=progress) else: await client.send_document(chat_id, str(path), caption=path.name, parse_mode=ParseMode.DISABLED, progress=progress) return except FloodWait as e: log.warning("FloodWait %ss", e.value) await asyncio.sleep(e.value + 1) raise RuntimeError(f"آپلود «{path.name}» پس از ۳ تلاش ناموفق بود.") async def upload_dir(client: Client, chat_id: int, job_dir: Path, status_msg: Message): files = sorted(p for p in job_dir.rglob("*") if p.is_file() and not p.name.endswith(".parts")) if not files: raise RuntimeError("فایلی برای آپلود پیدا نشد.") total = len(files) for i, f in enumerate(files, 1): parts = await asyncio.to_thread(split_file, f) for j, part in enumerate(parts, 1): label = f"[{i}/{total}] {part.name}" if len(parts) > 1: label += f" (تکه {j}/{len(parts)})" await send_file(client, chat_id, part, status_msg, label) part.unlink(missing_ok=True) # آزاد کردن فضا # ---------- اجرای یک کار ---------- async def run_job(client: Client, message: Message, kind: str, source): job_id = secrets.token_hex(3) job_dir = DOWNLOAD_DIR / job_id job_dir.mkdir(parents=True, exist_ok=True) torrent_file = DOWNLOAD_DIR / f"{job_id}.torrent" status_msg = await message.reply_text( f"⏳ در حال اضافه کردن تورنت...\nشناسه: {job_id}\nلغو: /cancel_{job_id}", parse_mode=ParseMode.HTML) handle = None JOBS[job_id] = {"handle": None, "task": asyncio.current_task(), "dir": job_dir, "name": "-", "phase": "download"} try: if kind == "magnet": params = lt.parse_magnet_uri(source) else: if kind == "url": await asyncio.to_thread(fetch_url, source, torrent_file) else: # kind == "file" → source یک Message با فایل .torrent است await source.download(file_name=str(torrent_file.resolve())) params = lt.add_torrent_params() params.ti = lt.torrent_info(str(torrent_file)) params.save_path = str(job_dir) handle = ses.add_torrent(params) JOBS[job_id]["handle"] = handle name = await download(handle, status_msg, job_id) ses.remove_torrent(handle) # توقف seeding handle = None JOBS[job_id].update(handle=None, phase="upload") await asyncio.sleep(2) # اطمینان از نوشته‌شدن کامل روی دیسک await upload_dir(client, message.chat.id, job_dir, status_msg) await safe_edit(status_msg, f"✅ {html.escape(name)} با موفقیت آپلود شد.") except asyncio.CancelledError: await safe_edit(status_msg, "🛑 عملیات لغو شد.") except Exception as e: log.exception("job %s failed", job_id) await safe_edit(status_msg, f"⚠️ خطا: {html.escape(str(e))}") finally: if handle is not None: try: ses.remove_torrent(handle) except Exception: pass JOBS.pop(job_id, None) shutil.rmtree(job_dir, ignore_errors=True) torrent_file.unlink(missing_ok=True) # ---------- هندلرها ---------- @app.on_message(filters.command(["start", "help"]) & filters.private & auth) async def start(_, message: Message): await message.reply_text( "👋 سلام!\n\n" "یکی از این‌ها رو بفرست:\n" "• لینک magnet\n" "• فایل .torrent\n" "• لینک مستقیم فایل .torrent\n\n" "دستورات:\n" "/status — لیست کارهای فعال\n" "/cancel_ID — لغو یک کار\n\n" "📏 حد آپلود تلگرام 2GB است؛ فایل‌های بزرگ‌تر به تکه‌های .001 .002 ... تقسیم می‌شوند " "(با 7-Zip یا دستور cat به هم بچسبون).", parse_mode=ParseMode.DISABLED, ) @app.on_message(filters.command("status") & filters.private & auth) async def status(_, message: Message): if not JOBS: return await message.reply_text("هیچ کار فعالی نیست.") lines = [] for jid, job in JOBS.items(): h = job["handle"] if job["phase"] == "upload": lines.append(f"• {jid} — {html.escape(job['name'])} — 📤 آپلود") elif h is not None and h.is_valid(): s = h.status() lines.append(f"• {jid} — {html.escape(s.name or '...')} — " f"{s.progress * 100:.0f}% — ⬇️ {human(s.download_rate)}/s") else: lines.append(f"• {jid} — ⏳ آماده‌سازی") await message.reply_text("\n".join(lines), parse_mode=ParseMode.HTML) @app.on_message(filters.regex(r"^/cancel[_ ](\w+)") & filters.private & auth) async def cancel(_, message: Message): job_id = message.matches[0].group(1) job = JOBS.get(job_id) if not job: return await message.reply_text("❓ کاری با این شناسه پیدا نشد.") job["task"].cancel() await message.reply_text(f"🛑 کار {job_id} لغو شد.", parse_mode=ParseMode.HTML) @app.on_message(filters.text & filters.private & auth & ~filters.regex(r"^/")) async def on_text(client: Client, message: Message): text = message.text.strip() if text.lower().startswith("magnet:?"): asyncio.create_task(run_job(client, message, "magnet", text)) elif re.match(r"^https?://\S+$", text): asyncio.create_task(run_job(client, message, "url", text)) else: await message.reply_text("❗ لینک magnet، فایل .torrent یا لینک مستقیم .torrent بفرست.") @app.on_message(filters.document & filters.private & auth) async def on_document(client: Client, message: Message): doc = message.document name = (doc.file_name or "").lower() if not (name.endswith(".torrent") or doc.mime_type == "application/x-bittorrent"): return await message.reply_text("❗ فقط فایل .torrent قبول می‌شود.") asyncio.create_task(run_job(client, message, "file", message)) @app.on_message(filters.private & ~auth) async def unauthorized(_, message: Message): await message.reply_text( f"⛔ شما مجاز به استفاده از این بات نیستید.\nآیدی شما: {message.from_user.id}", parse_mode=ParseMode.HTML) if __name__ == "__main__": log.info("🚀 Bot started") app.run() ``` ## اجرا 1. مقادیر `API_ID`، `API_HASH`، `BOT_TOKEN` رو در بالای فایل پر کن. 2. یک بار بات رو اجرا کن و به بات پیام بده؛ آیدی عددی‌ات رو نشون می‌ده. آیدی رو در `ALLOWED_USERS` بگذار و دوباره اجرا کن: ```bash python torrent_bot.py ``` برای اینکه همیشه در پس‌زمینه اجرا بمونه، یک سرویس systemd بساز (`/etc/systemd/system/torrent-bot.service`): ```ini [Unit] Description=Telegram Torrent Bot After=network.target [Service] User=ubuntu WorkingDirectory=/home/ubuntu/torrent-bot ExecStart=/home/ubuntu/torrent-bot/venv/bin/python torrent_bot.py Restart=always RestartSec=5 [Install] WantedBy=multi-user.target ``` ```bash sudo systemctl daemon-reload sudo systemctl enable --now torrent-bot sudo journalctl -u torrent-bot -f # دیدن لاگ‌ها ``` ## نکات مهم - **حد 2GB:** بات‌های تلگرام حداکثر 2GB برای هر فایل می‌تونن آپلود کنن. فایل‌های بزرگ‌تر خودکار به تکه‌های `.001`, `.002` تقسیم می‌شن. برای چسباندن: در ویندوز با 7-Zip فایل `.001` رو Extract کن؛ در لینوکس: `cat film.mkv.* > film.mkv` - **پورت 6881** رو در فایروال سرور باز کن تا سرعت دانلود بهتر شه (اجباری نیست). - **فضای دیسک:** بات قبل از دانلود چک می‌کنه فضا کافی باشه و بعد از آپلود همه‌چیز رو پاک می‌کنه. - **امنیت:** فقط آیدی‌های داخل `ALLOWED_USERS` می‌تونن از بات استفاده کنن — حتماً پرش کن، وگرنه هر کسی می‌تونه از سرورت استفاده کنه. توکن و API_HASH رو جایی به اشتراک نگذار. - **سرعت آپلود** به تلگرام به پهنای باند سرور بستگی داره؛ سرورهای اروپایی (مثل Hetzner) معمولاً نتیجه‌ی خوبی می‌دن. اگر خواستی قابلیت اضافه‌ای مثل انتخاب فایل‌های داخل تورنت، آپلود به کانال، یا ساخت thumbnail برای ویدیوها اضافه کنم بگو.

یک ربات تلگرام میخوام که بهش لینک تورنت بدم و دانلود کنه و ر...