Processor

StorageDriver

Bases: str, Enum

Available storage drivers for TDL.

LEGACY

LEGACY = 'legacy'

BOLT

BOLT = 'bolt'

FILE

FILE = 'file'

TDLCommand

Bases: str, Enum

Available TDL commands.

BACKUP

BACKUP = 'backup'

LOGIN

LOGIN = 'login'

MIGRATE

MIGRATE = 'migrate'

RECOVER

RECOVER = 'recover'

CHAT

CHAT = 'chat'

DOWNLOAD

DOWNLOAD = 'download'

EXTENSION

EXTENSION = 'extension'

FORWARD

FORWARD = 'forward'

UPLOAD

UPLOAD = 'upload'

COMPLETION

COMPLETION = 'completion'

VERSION

VERSION = 'version'

TDLResult

Bases: BaseModel

Result of TDL command execution.

success

success: bool = Field(..., description='Whether the command executed successfully')

return_code

return_code: int = Field(..., description='Process return code')

stdout

stdout: str = Field(default='', description='Standard output from the command')

stderr

stderr: str = Field(default='', description='Standard error from the command')

command

command: list[str] = Field(..., description='The executed command')

TDLConfig

Bases: BaseModel

Configuration for TDL processor.

debug

debug: bool = Field(default=False, description='Enable debug mode')

delay

delay: timedelta | None = Field(default=None, description='Delay between each task')

limit

limit: int = Field(default=2, description='Max number of concurrent tasks')

namespace

namespace: str = Field(default='default', description='Namespace for Telegram session')

ntp

ntp: str | None = Field(default=None, description='NTP server host')

pool

pool: int = Field(default=8, description='Size of the DC pool, zero means infinity')

proxy

proxy: str | None = Field(
    default=None, description="Proxy address, format: protocol://username:password@host:port"
)

reconnect_timeout

reconnect_timeout: timedelta = Field(
    default=timedelta(minutes=5), description="Telegram client reconnection backoff timeout"
)

storage

storage: dict[str, str] = Field(
    default_factory=lambda: {"type": "bolt", "path": f"{home()}/.tdl/data"},
    description="Storage options",
)

threads

threads: int = Field(default=4, description='Max threads for transfer one item')

TelegramDownloader

Bases: BaseModel

Enhanced Telegram Downloader with full TDL CLI support.

Methods:

Name	Description
login	Login to Telegram.
backup	Backup your data.
migrate	Migrate your current data to another storage.
recover	Recover your data.
download	Download anything from Telegram (protected) chat.
upload	Upload anything to Telegram.
forward	Forward messages with automatic fallback and message routing.
chat_list	List all chats.
chat_export	Export chat messages.
extension_list	List installed extensions.
extension_install	Install an extension.
extension_remove	Remove an extension.
get_version	Get TDL version information.
generate_completion	Generate autocompletion script for specified shell.

output_folder

output_folder: Path = Field(..., description='The output directory for the downloaded files')

config

config: TDLConfig = Field(default_factory=TDLConfig, description='TDL configuration options')

tdl_binary

tdl_binary: str

Get the path to TDL binary based on platform.

login

login() -> TDLResult

Login to Telegram.

Source code in src/core/processor.py

async def login(self) -> TDLResult:
    """Login to Telegram."""
    command = [*self._build_base_command(), TDLCommand.LOGIN]
    return await self._execute_command(command)

backup

backup(destination: str | None = None) -> TDLResult

Backup your data.

Source code in src/core/processor.py

async def backup(self, destination: str | None = None) -> TDLResult:
    """Backup your data."""
    command = [*self._build_base_command(), TDLCommand.BACKUP]
    if destination:
        command.extend(["--dest", destination])
    return await self._execute_command(command)

migrate

migrate(from_storage: str, to_storage: str) -> TDLResult

Migrate your current data to another storage.

Source code in src/core/processor.py

async def migrate(self, from_storage: str, to_storage: str) -> TDLResult:
    """Migrate your current data to another storage."""
    command = [
        *self._build_base_command(),
        TDLCommand.MIGRATE,
        "--from",
        from_storage,
        "--to",
        to_storage,
    ]
    return await self._execute_command(command)

recover

recover(source: str) -> TDLResult

Recover your data.

Source code in src/core/processor.py

async def recover(self, source: str) -> TDLResult:
    """Recover your data."""
    command = [*self._build_base_command(), TDLCommand.RECOVER, "--source", source]
    return await self._execute_command(command)

download

download(
    urls: list[str] | str,
    include: list[str] | None = None,
    exclude: list[str] | None = None,
    restart: bool = False,
    skip_same: bool = False,
) -> TDLResult

Download anything from Telegram (protected) chat.

Source code in src/core/processor.py

async def download(
    self,
    urls: list[str] | str,
    include: list[str] | None = None,
    exclude: list[str] | None = None,
    restart: bool = False,
    skip_same: bool = False,
) -> TDLResult:
    """Download anything from Telegram (protected) chat."""
    if isinstance(urls, list):
        urls = ",".join(urls)

    command = [
        *self._build_base_command(),
        TDLCommand.DOWNLOAD,
        "--dir",
        self.output_folder.as_posix(),
        "--url",
        urls,
    ]

    if include:
        command.extend(["--include", ",".join(include)])

    if exclude:
        command.extend(["--exclude", ",".join(exclude)])

    if restart:
        command.append("--restart")

    if skip_same:
        command.append("--skip-same")

    return await self._execute_command(command, timeout=3600)  # 1 hour timeout

upload

upload(path: str, to: str, remove_after: bool = False) -> TDLResult

Upload anything to Telegram.

Source code in src/core/processor.py

async def upload(self, path: str, to: str, remove_after: bool = False) -> TDLResult:
    """Upload anything to Telegram."""
    command = [*self._build_base_command(), TDLCommand.UPLOAD, "--path", path, "--to", to]

    if remove_after:
        command.append("--rm")

    return await self._execute_command(command, timeout=3600)  # 1 hour timeout

forward

forward(from_chat: str, to_chat: str, filter_text: str | None = None) -> TDLResult

Forward messages with automatic fallback and message routing.

Source code in src/core/processor.py

async def forward(
    self, from_chat: str, to_chat: str, filter_text: str | None = None
) -> TDLResult:
    """Forward messages with automatic fallback and message routing."""
    command = [
        *self._build_base_command(),
        TDLCommand.FORWARD,
        "--from",
        from_chat,
        "--to",
        to_chat,
    ]

    if filter_text:
        command.extend(["--filter", filter_text])

    return await self._execute_command(command)

chat_list

chat_list() -> TDLResult

List all chats.

Source code in src/core/processor.py

async def chat_list(self) -> TDLResult:
    """List all chats."""
    command = [*self._build_base_command(), TDLCommand.CHAT, "ls"]
    return await self._execute_command(command)

chat_export

chat_export(chat: str, output: str | None = None) -> TDLResult

Export chat messages.

Source code in src/core/processor.py

async def chat_export(self, chat: str, output: str | None = None) -> TDLResult:
    """Export chat messages."""
    command = [*self._build_base_command(), TDLCommand.CHAT, "export", "--chat", chat]

    if output:
        command.extend(["--output", output])

    return await self._execute_command(command, timeout=3600)  # 1 hour timeout

extension_list

extension_list() -> TDLResult

List installed extensions.

Source code in src/core/processor.py

async def extension_list(self) -> TDLResult:
    """List installed extensions."""
    command = [*self._build_base_command(), TDLCommand.EXTENSION, "ls"]
    return await self._execute_command(command)

extension_install

extension_install(extension: str) -> TDLResult

Install an extension.

Source code in src/core/processor.py

async def extension_install(self, extension: str) -> TDLResult:
    """Install an extension."""
    command = [*self._build_base_command(), TDLCommand.EXTENSION, "install", extension]
    return await self._execute_command(command)

extension_remove

extension_remove(extension: str) -> TDLResult

Remove an extension.

Source code in src/core/processor.py

async def extension_remove(self, extension: str) -> TDLResult:
    """Remove an extension."""
    command = [*self._build_base_command(), TDLCommand.EXTENSION, "rm", extension]
    return await self._execute_command(command)

get_version

get_version() -> TDLResult

Get TDL version information.

Source code in src/core/processor.py

async def get_version(self) -> TDLResult:
    """Get TDL version information."""
    command = [*self._build_base_command(), TDLCommand.VERSION]
    return await self._execute_command(command)

generate_completion

generate_completion(shell: str) -> TDLResult

Generate autocompletion script for specified shell.

Source code in src/core/processor.py

async def generate_completion(self, shell: str) -> TDLResult:
    """Generate autocompletion script for specified shell."""
    command = [*self._build_base_command(), TDLCommand.COMPLETION, shell]
    return await self._execute_command(command)