tdjson_example.py

#!/usr/bin/env python3
# Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com),
# Pellegrino Prevete (pellegrinoprevete@gmail.com)  2014-2025
#
# Distributed under the Boost Software License, Version 1.0. (See accompanying
# file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

import json
import os
import sys
from ctypes import CDLL, CFUNCTYPE, c_char_p, c_double, c_int
from ctypes.util import find_library
from typing import Any, Dict, Optional


class TdExample:
    """A Python client for the Telegram API using TDLib."""

    def __init__(self, api_id: int = None, api_hash: str = None):
        """Initialize a Telegram client.

        Args:
            api_id: Telegram API ID (get from https://my.telegram.org)
            api_hash: Telegram API hash (get from https://my.telegram.org)
        """
        self.api_id = api_id
        self.api_hash = api_hash
        self._load_library()
        self._setup_functions()
        self._setup_logging()
        self.client_id = self._td_create_client_id()

    def _load_library(self) -> None:
        """Load the TDLib shared library."""
        tdjson_path = find_library("tdjson")
        if tdjson_path is None:
            if os.name == "nt":
                tdjson_path = os.path.join(os.path.dirname(__file__), "tdjson.dll")
            else:
                sys.exit(
                    "Error: Can't find 'tdjson' library. Make sure it's installed correctly."
                )

        try:
            self.tdjson = CDLL(tdjson_path)
        except Exception as e:
            sys.exit(f"Error loading TDLib: {e}")

    def _setup_functions(self) -> None:
        """Set up function signatures for TDLib calls."""
        # Create client ID
        self._td_create_client_id = self.tdjson.td_create_client_id
        self._td_create_client_id.restype = c_int
        self._td_create_client_id.argtypes = []

        # Receive updates
        self._td_receive = self.tdjson.td_receive
        self._td_receive.restype = c_char_p
        self._td_receive.argtypes = [c_double]

        # Send requests
        self._td_send = self.tdjson.td_send
        self._td_send.restype = None
        self._td_send.argtypes = [c_int, c_char_p]

        # Execute synchronous requests
        self._td_execute = self.tdjson.td_execute
        self._td_execute.restype = c_char_p
        self._td_execute.argtypes = [c_char_p]

        # Set log callback
        self.log_message_callback_type = CFUNCTYPE(None, c_int, c_char_p)
        self._td_set_log_message_callback = self.tdjson.td_set_log_message_callback
        self._td_set_log_message_callback.restype = None
        self._td_set_log_message_callback.argtypes = [
            c_int,
            self.log_message_callback_type,
        ]

    def _setup_logging(self, verbosity_level: int = 1) -> None:
        """Configure TDLib logging.

        Args:
            verbosity_level: 0-fatal, 1-errors, 2-warnings, 3+-debug
        """

        @self.log_message_callback_type
        def on_log_message_callback(verbosity_level, message):
            if verbosity_level == 0:
                sys.exit(f"TDLib fatal error: {message.decode('utf-8')}")

        self._td_set_log_message_callback(2, on_log_message_callback)
        self.execute(
            {"@type": "setLogVerbosityLevel", "new_verbosity_level": verbosity_level}
        )

    def execute(self, query: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """Execute a synchronous TDLib request.

        Args:
            query: The request to execute

        Returns:
            Response from TDLib or None
        """
        query_json = json.dumps(query).encode("utf-8")
        result = self._td_execute(query_json)
        if result:
            return json.loads(result.decode("utf-8"))
        return None

    def send(self, query: Dict[str, Any]) -> None:
        """Send an asynchronous request to TDLib.

        Args:
            query: The request to send
        """
        query_json = json.dumps(query).encode("utf-8")
        self._td_send(self.client_id, query_json)

    def receive(self, timeout: float = 1.0) -> Optional[Dict[str, Any]]:
        """Receive a response or update from TDLib.

        Args:
            timeout: Maximum number of seconds to wait

        Returns:
            An update or response from TDLib, or None if nothing received
        """
        result = self._td_receive(timeout)
        if result:
            return json.loads(result.decode("utf-8"))
        return None

    def login(self) -> None:
        """Start the authentication process."""
        self.send({"@type": "getOption", "name": "version"})

        print("Starting Telegram authentication flow...")
        print("Press Ctrl+C to cancel at any time.")

        try:
            self._handle_authentication()
        except KeyboardInterrupt:
            print("\nAuthentication canceled by user.")
            sys.exit(0)

    def _handle_authentication(self) -> None:
        """Handle the TDLib authentication flow."""
        while True:
            event = self.receive()
            if not event:
                continue

            # Print all updates for debugging
            event_type = event["@type"]
            if event_type != "updateAuthorizationState":
                print(f"Receive: {json.dumps(event, indent=2)}")

            # Process authorization states
            if event_type == "updateAuthorizationState":
                auth_state = event["authorization_state"]
                auth_type = auth_state["@type"]

                if auth_type == "authorizationStateClosed":
                    print("Authorization state closed.")
                    break

                elif auth_type == "authorizationStateWaitTdlibParameters":
                    if not self.api_id or not self.api_hash:
                        print(
                            "\nYou MUST obtain your own api_id and api_hash at https://my.telegram.org"
                        )
                        self.api_id = int(input("Please enter your API ID: "))
                        self.api_hash = input("Please enter your API hash: ")

                    print("Setting TDLib parameters...")
                    self.send(
                        {
                            "@type": "setTdlibParameters",
                            "database_directory": "tdlib_data",
                            "use_message_database": True,
                            "use_secret_chats": True,
                            "api_id": self.api_id,
                            "api_hash": self.api_hash,
                            "system_language_code": "en",
                            "device_model": "Python TDLib Client",
                            "application_version": "1.1",
                        }
                    )

                elif auth_type == "authorizationStateWaitPhoneNumber":
                    phone_number = input(
                        "Please enter your phone number (international format): "
                    )
                    self.send(
                        {
                            "@type": "setAuthenticationPhoneNumber",
                            "phone_number": phone_number,
                        }
                    )

                elif auth_type == "authorizationStateWaitPremiumPurchase":
                    print("Telegram Premium subscription is required.")
                    return

                elif auth_type == "authorizationStateWaitEmailAddress":
                    email_address = input("Please enter your email address: ")
                    self.send(
                        {
                            "@type": "setAuthenticationEmailAddress",
                            "email_address": email_address,
                        }
                    )

                elif auth_type == "authorizationStateWaitEmailCode":
                    code = input(
                        "Please enter the email authentication code you received: "
                    )
                    self.send(
                        {
                            "@type": "checkAuthenticationEmailCode",
                            "code": {
                                "@type": "emailAddressAuthenticationCode",
                                "code": code,
                            },
                        }
                    )

                elif auth_type == "authorizationStateWaitCode":
                    code = input("Please enter the authentication code you received: ")
                    self.send({"@type": "checkAuthenticationCode", "code": code})

                elif auth_type == "authorizationStateWaitRegistration":
                    first_name = input("Please enter your first name: ")
                    last_name = input("Please enter your last name: ")
                    self.send(
                        {
                            "@type": "registerUser",
                            "first_name": first_name,
                            "last_name": last_name,
                        }
                    )

                elif auth_type == "authorizationStateWaitPassword":
                    password = input("Please enter your password: ")
                    self.send(
                        {"@type": "checkAuthenticationPassword", "password": password}
                    )

                elif auth_type == "authorizationStateReady":
                    print("Authorization complete! You are now logged in.")
                    return


def main():
    """Main function to demonstrate client usage."""
    # Example API credentials - DO NOT USE THESE
    # Get your own from https://my.telegram.org
    DEFAULT_API_ID = 94575
    DEFAULT_API_HASH = "a3406de8d171bb422bb6ddf3bbd800e2"

    print("TDLib Python Client")
    print("===================")
    print(
        "IMPORTANT: You should obtain your own api_id and api_hash at https://my.telegram.org"
    )
    print("         The default values are for demonstration only.\n")

    use_default = (
        input("Use default API credentials for testing? (y/n): ").lower() == "y"
    )

    if use_default:
        client = TdExample(DEFAULT_API_ID, DEFAULT_API_HASH)
    else:
        client = TdExample()

    # Test execute method
    print("\nTesting TDLib execute method...")
    result = client.execute(
        {
            "@type": "getTextEntities",
            "text": "@telegram /test_command https://telegram.org telegram.me",
        }
    )
    print(f"Text entities: {json.dumps(result, indent=2)}")

    # Start login process
    client.login()

    # Main event loop
    print("\nEntering main event loop. Press Ctrl+C to exit.")
    try:
        while True:
            event = client.receive()
            if event:
                print(json.dumps(event, indent=2))
    except KeyboardInterrupt:
        print("\nExiting...")


if __name__ == "__main__":
    main()