Skip to content

Update tdjson_example.py to a new clean structure. #3276

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Apr 12, 2025
Merged

Update tdjson_example.py to a new clean structure. #3276

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
de62c96
Update tdjson_example.py
9v2 Mar 16, 2025
22821a4
Update tdjson_example.py
9v2 Mar 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
381 changes: 245 additions & 136 deletions example/python/tdjson_example.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,145 +1,254 @@
#
#!/usr/bin/env python3
# Copyright Aliaksei Levin ([email protected]), Arseny Smirnov ([email protected]),
# Pellegrino Prevete ([email protected]) 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)
#
from ctypes.util import find_library
from ctypes import *

import json
import os
import sys

# load 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')
from ctypes import CDLL, CFUNCTYPE, c_char_p, c_double, c_int
from ctypes.util import find_library
from typing import Any, Dict, Optional, Union


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 cancelled 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
if event.get('@type') != 'updateAuthorizationState':
print(f"Received: {json.dumps(event, indent=2)}")

# Process authorization states
if event.get('@type') == 'updateAuthorizationState':
auth_state = event['authorization_state']
auth_type = auth_state.get('@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 == '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:
sys.exit("Can't find 'tdjson' library")
tdjson = CDLL(tdjson_path)

# load TDLib functions from shared library
_td_create_client_id = tdjson.td_create_client_id
_td_create_client_id.restype = c_int
_td_create_client_id.argtypes = []

_td_receive = tdjson.td_receive
_td_receive.restype = c_char_p
_td_receive.argtypes = [c_double]

_td_send = tdjson.td_send
_td_send.restype = None
_td_send.argtypes = [c_int, c_char_p]

_td_execute = tdjson.td_execute
_td_execute.restype = c_char_p
_td_execute.argtypes = [c_char_p]

log_message_callback_type = CFUNCTYPE(None, c_int, c_char_p)

_td_set_log_message_callback = tdjson.td_set_log_message_callback
_td_set_log_message_callback.restype = None
_td_set_log_message_callback.argtypes = [c_int, log_message_callback_type]

# initialize TDLib log with desired parameters
@log_message_callback_type
def on_log_message_callback(verbosity_level, message):
if verbosity_level == 0:
sys.exit('TDLib fatal error: %r' % message)

def td_execute(query):
query = json.dumps(query).encode('utf-8')
result = _td_execute(query)
if result:
result = json.loads(result.decode('utf-8'))
return result

_td_set_log_message_callback(2, on_log_message_callback)

# setting TDLib log verbosity level to 1 (errors)
print(str(td_execute({'@type': 'setLogVerbosityLevel', 'new_verbosity_level': 1, '@extra': 1.01234})).encode('utf-8'))


# create client
client_id = _td_create_client_id()

# simple wrappers for client usage
def td_send(query):
query = json.dumps(query).encode('utf-8')
_td_send(client_id, query)

def td_receive():
result = _td_receive(1.0)
if result:
result = json.loads(result.decode('utf-8'))
return result

# another test for TDLib execute method
print(str(td_execute({'@type': 'getTextEntities', 'text': '@telegram /test_command https://telegram.org telegram.me', '@extra': ['5', 7.0, 'a']})).encode('utf-8'))

# start the client by sending a request to it
td_send({'@type': 'getOption', 'name': 'version', '@extra': 1.01234})

# main events cycle
while True:
event = td_receive()
if event:
# process authorization states
if event['@type'] == 'updateAuthorizationState':
auth_state = event['authorization_state']

# if client is closed, we need to destroy it and create new client
if auth_state['@type'] == 'authorizationStateClosed':
break

# set TDLib parameters
# you MUST obtain your own api_id and api_hash at https://my.telegram.org
# and use them in the setTdlibParameters call
if auth_state['@type'] == 'authorizationStateWaitTdlibParameters':
td_send({'@type': 'setTdlibParameters',
'database_directory': 'tdlib',
'use_message_database': True,
'use_secret_chats': True,
'api_id': 94575,
'api_hash': 'a3406de8d171bb422bb6ddf3bbd800e2',
'system_language_code': 'en',
'device_model': 'Desktop',
'application_version': '1.0'})

# enter phone number to log in
if auth_state['@type'] == 'authorizationStateWaitPhoneNumber':
phone_number = input('Please enter your phone number: ')
td_send({'@type': 'setAuthenticationPhoneNumber', 'phone_number': phone_number})

# enter email address to log in
if auth_state['@type'] == 'authorizationStateWaitEmailAddress':
email_address = input('Please enter your email address: ')
td_send({'@type': 'setAuthenticationEmailAddress', 'email_address': email_address})

# wait for email authorization code
if auth_state['@type'] == 'authorizationStateWaitEmailCode':
code = input('Please enter the email authentication code you received: ')
td_send({'@type': 'checkAuthenticationEmailCode',
'code': {'@type': 'emailAddressAuthenticationCode', 'code' : code}})

# wait for authorization code
if auth_state['@type'] == 'authorizationStateWaitCode':
code = input('Please enter the authentication code you received: ')
td_send({'@type': 'checkAuthenticationCode', 'code': code})

# wait for first and last name for new users
if auth_state['@type'] == 'authorizationStateWaitRegistration':
first_name = input('Please enter your first name: ')
last_name = input('Please enter your last name: ')
td_send({'@type': 'registerUser', 'first_name': first_name, 'last_name': last_name})

# wait for password if present
if auth_state['@type'] == 'authorizationStateWaitPassword':
password = input('Please enter your password: ')
td_send({'@type': 'checkAuthenticationPassword', 'password': password})

# handle an incoming update or an answer to a previously sent request
print(str(event).encode('utf-8'))
sys.stdout.flush()
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()