tessl/pypi-telethon
Full-featured Telegram client library for Python 3
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
utilities-helpers.mddocs/
Utilities & Helpers
Utility functions for entity resolution, input conversion, media type detection, and common Telegram operations in Telethon.
Capabilities
Entity Utilities
Convert between different entity types and resolve entities.
def get_display_name(entity) -> str:
"""
Get human-readable display name for an entity.
Parameters:
- entity: User, Chat, Channel, or similar object
Returns:
str: Display name (first_name last_name for users, title for chats)
Examples:
- User: "John Doe"
- Chat: "My Group Chat"
- Channel: "News Channel"
"""
def get_input_peer(
entity,
allow_self: bool = True,
check_hash: bool = True
) -> types.TypeInputPeer:
"""
Convert entity to InputPeer for API calls.
Parameters:
- entity: User, Chat, Channel, ID, username, or phone
- allow_self: Allow InputPeerSelf for current user
- check_hash: Verify access hash validity
Returns:
InputPeerUser, InputPeerChat, InputPeerChannel, or InputPeerSelf
Raises:
- ValueError: If entity cannot be resolved
- TypeNotFoundError: If entity type unknown
"""
def get_input_channel(entity) -> types.InputChannel:
"""
Convert entity to InputChannel.
Parameters:
- entity: Channel, supergroup, or their ID/username
Returns:
InputChannel object
Raises:
- ValueError: If not a channel/supergroup
"""
def get_input_user(entity) -> types.InputUser:
"""
Convert entity to InputUser.
Parameters:
- entity: User object, ID, username, or phone number
Returns:
InputUser object
Raises:
- ValueError: If not a user
"""
def get_input_dialog(dialog) -> types.InputDialogPeer:
"""
Convert dialog to InputDialogPeer.
Parameters:
- dialog: Dialog object or entity
Returns:
InputDialogPeer object for dialog operations
"""
def get_peer_id(peer) -> int:
"""
Extract numeric ID from any peer type.
Parameters:
- peer: Any peer object, InputPeer, or entity
Returns:
int: Numeric peer ID
Note:
Channel/supergroup IDs are returned as negative values
"""
def get_input_entity(entity) -> types.TypeInputPeer:
"""
Alias for get_input_peer with default parameters.
Parameters:
- entity: Entity to convert
Returns:
InputPeer object
"""Media Utilities
Detect and handle different media types and file formats.
def get_extension(media) -> str:
"""
Get file extension for media object.
Parameters:
- media: Media object (Document, Photo, etc.)
Returns:
str: File extension with dot (e.g., '.jpg', '.mp4', '.pdf')
"""
def is_image(file) -> bool:
"""
Check if file/media is an image.
Parameters:
- file: File path, media object, or message with media
Returns:
bool: True if image format
Supported formats: JPEG, PNG, GIF, WebP, BMP, TIFF, etc.
"""
def is_gif(file) -> bool:
"""
Check if file/media is a GIF animation.
Parameters:
- file: File path, media object, or message with media
Returns:
bool: True if GIF format
"""
def is_video(file) -> bool:
"""
Check if file/media is a video.
Parameters:
- file: File path, media object, or message with media
Returns:
bool: True if video format
Supported formats: MP4, AVI, MOV, MKV, WebM, etc.
"""
def is_audio(file) -> bool:
"""
Check if file/media is audio.
Parameters:
- file: File path, media object, or message with media
Returns:
bool: True if audio format
Supported formats: MP3, WAV, FLAC, OGG, M4A, etc.
"""
def get_input_document(document) -> types.InputDocument:
"""
Convert document to InputDocument.
Parameters:
- document: Document object or message with document
Returns:
InputDocument object for API calls
"""
def get_input_photo(photo) -> types.InputPhoto:
"""
Convert photo to InputPhoto.
Parameters:
- photo: Photo object or message with photo
Returns:
InputPhoto object for API calls
"""
def get_input_media(
file,
*,
thumb=None,
force_document: bool = False,
**kwargs
) -> types.TypeInputMedia:
"""
Convert file to appropriate InputMedia type.
Parameters:
- file: File path, bytes, or media object
- thumb: Thumbnail file
- force_document: Force InputMediaDocument instead of specific type
- **kwargs: Additional media attributes
Returns:
InputMediaPhoto, InputMediaDocument, or other InputMedia type
"""
def get_attributes(
file,
*,
mime_type: str = None,
attributes: List = None,
**kwargs
) -> List[types.TypeDocumentAttribute]:
"""
Generate document attributes for file.
Parameters:
- file: File path or file-like object
- mime_type: Override detected MIME type
- attributes: Custom attribute list
- **kwargs: Additional attribute parameters
Returns:
List of DocumentAttribute objects (filename, size, etc.)
"""Location Utilities
Handle geographical locations and file locations.
def get_input_geo(geo) -> types.InputGeoPoint:
"""
Convert location to InputGeoPoint.
Parameters:
- geo: GeoPoint object or (latitude, longitude) tuple
Returns:
InputGeoPoint object
"""
def get_input_location(location) -> types.TypeInputFileLocation:
"""
Convert media to InputFileLocation for downloads.
Parameters:
- location: Media object, Document, Photo, etc.
Returns:
InputFileLocation for downloading the file
"""
def get_input_chat_photo(photo) -> Union[types.InputPhoto, types.InputFile]:
"""
Convert photo for use as chat profile photo.
Parameters:
- photo: Photo object or file
Returns:
InputPhoto or InputFile for setting as chat photo
"""Message Utilities
Work with message objects and IDs.
def get_message_id(message) -> int:
"""
Extract message ID from message object or ID.
Parameters:
- message: Message object, ID, or InputMessage
Returns:
int: Message ID number
"""
def get_input_message(message) -> Union[int, types.InputMessage]:
"""
Convert message to InputMessage.
Parameters:
- message: Message object or ID
Returns:
InputMessage object or message ID
"""Text Processing
Handle text formatting and parsing.
def sanitize_parse_mode(mode) -> object:
"""
Convert parse mode string to parser object.
Parameters:
- mode: 'md', 'markdown', 'html', or parser object
Returns:
Parser object for text formatting
"""
def parse_phone(phone) -> str:
"""
Normalize phone number format.
Parameters:
- phone: Phone number string
Returns:
str: Normalized phone number (international format)
Examples:
- "+1234567890" -> "+1234567890"
- "123-456-7890" -> "+11234567890"
- "(123) 456-7890" -> "+11234567890"
"""
def parse_username(username) -> str:
"""
Normalize username format.
Parameters:
- username: Username string (with or without @)
Returns:
str: Username without @ prefix
Examples:
- "@username" -> "username"
- "username" -> "username"
- "t.me/username" -> "username"
"""Collection Utilities
Work with lists and iterables.
def chunks(iterable, size: int = 100):
"""
Split iterable into chunks of specified size.
Parameters:
- iterable: Any iterable object
- size: Maximum chunk size
Yields:
Lists of up to 'size' items
Usage:
for chunk in chunks(large_list, 50):
process_chunk(chunk)
"""
def is_list_like(obj) -> bool:
"""
Check if object behaves like a list.
Parameters:
- obj: Object to check
Returns:
bool: True if list-like (has __iter__ and __getitem__)
"""Call Utilities
Handle group call operations.
def get_input_group_call(call) -> types.InputGroupCall:
"""
Convert group call to InputGroupCall.
Parameters:
- call: GroupCall object or call ID
Returns:
InputGroupCall object for API calls
"""Usage Examples
Entity Resolution
import asyncio
from telethon import TelegramClient, utils
async def entity_examples():
client = TelegramClient('session', api_id, api_hash)
await client.start()
# Different ways to reference entities
entities = [
'username', # Username
'@username', # Username with @
'+1234567890', # Phone number
123456789, # User ID
-100123456789, # Channel/supergroup ID
'https://t.me/username' # Telegram URL
]
for entity_ref in entities:
try:
# Get the entity
entity = await client.get_entity(entity_ref)
# Use utility functions
display_name = utils.get_display_name(entity)
peer_id = utils.get_peer_id(entity)
input_peer = utils.get_input_peer(entity)
print(f"Entity: {entity_ref}")
print(f" Display name: {display_name}")
print(f" Peer ID: {peer_id}")
print(f" Input peer type: {type(input_peer).__name__}")
print("---")
except Exception as e:
print(f"Could not resolve {entity_ref}: {e}")
await client.disconnect()
asyncio.run(entity_examples())Media Type Detection
from telethon import utils
async def media_detection():
client = TelegramClient('session', api_id, api_hash)
await client.start()
chat = 'username'
# Check recent media messages
async for message in client.iter_messages(chat, limit=20):
if message.media:
print(f"Message {message.id}:")
# Detect media type
if utils.is_image(message.media):
print(" 📷 Image")
ext = utils.get_extension(message.media)
print(f" Extension: {ext}")
elif utils.is_video(message.media):
print(" 🎥 Video")
if utils.is_gif(message.media):
print(" (Animated GIF)")
elif utils.is_audio(message.media):
print(" 🎵 Audio")
else:
print(" 📄 Document")
# Get file info
if hasattr(message.media, 'document'):
doc = message.media.document
print(f" Size: {doc.size} bytes")
print(f" MIME: {doc.mime_type}")
await client.disconnect()Text Processing
from telethon import utils
def text_processing_examples():
# Phone number parsing
phones = [
"+1234567890",
"123-456-7890",
"(123) 456-7890",
"1.234.567.8900"
]
for phone in phones:
normalized = utils.parse_phone(phone)
print(f"Phone: {phone} -> {normalized}")
# Username parsing
usernames = [
"@username",
"username",
"t.me/username",
"https://t.me/username"
]
for username in usernames:
parsed = utils.parse_username(username)
print(f"Username: {username} -> {parsed}")
# Parse mode handling
modes = ['md', 'markdown', 'html', None]
for mode in modes:
parser = utils.sanitize_parse_mode(mode)
print(f"Parse mode: {mode} -> {type(parser).__name__}")
text_processing_examples()Collection Processing
from telethon import utils
async def collection_examples():
client = TelegramClient('session', api_id, api_hash)
await client.start()
# Get a large list of messages
messages = await client.get_messages('username', limit=500)
# Process in chunks to avoid memory issues
for chunk in utils.chunks(messages, 50):
print(f"Processing chunk of {len(chunk)} messages")
# Process each chunk
for message in chunk:
if message.text:
# Do something with message
pass
# Check if various objects are list-like
test_objects = [
messages, # TotalList
[1, 2, 3], # Regular list
(1, 2, 3), # Tuple
"string", # String (iterable but not list-like)
42 # Number
]
for obj in test_objects:
is_list = utils.is_list_like(obj)
print(f"{type(obj).__name__}: list-like = {is_list}")
await client.disconnect()Advanced Entity Operations
async def advanced_entity_ops():
client = TelegramClient('session', api_id, api_hash)
await client.start()
# Work with different entity types
entities = await client.get_dialogs(limit=10)
for dialog in entities:
entity = dialog.entity
try:
# Get various input types
input_peer = utils.get_input_peer(entity)
peer_id = utils.get_peer_id(entity)
display_name = utils.get_display_name(entity)
print(f"Entity: {display_name} (ID: {peer_id})")
# Check entity type and get specific input
if hasattr(entity, 'username'): # User or Channel
if entity.username:
print(f" Username: @{entity.username}")
if hasattr(entity, 'access_hash'): # Channel/User
if entity.megagroup or entity.broadcast:
input_channel = utils.get_input_channel(entity)
print(f" Input channel: {input_channel}")
elif not entity.bot:
input_user = utils.get_input_user(entity)
print(f" Input user: {input_user}")
# Get dialog input
input_dialog = utils.get_input_dialog(dialog)
print(f" Input dialog: {input_dialog}")
except Exception as e:
print(f"Error processing {display_name}: {e}")
print("---")
await client.disconnect()File and Media Utilities
import os
from telethon import utils
async def file_utilities():
client = TelegramClient('session', api_id, api_hash)
await client.start()
# Send various file types and analyze
files_to_send = [
'image.jpg',
'video.mp4',
'audio.mp3',
'document.pdf'
]
chat = 'username'
for file_path in files_to_send:
if os.path.exists(file_path):
# Get attributes before sending
attributes = utils.get_attributes(file_path)
print(f"File: {file_path}")
print(f" Attributes: {len(attributes)}")
# Send file
message = await client.send_file(chat, file_path)
# Analyze sent media
if message.media:
ext = utils.get_extension(message.media)
print(f" Extension: {ext}")
# Get input versions for API calls
if utils.is_image(message.media):
input_photo = utils.get_input_photo(message.media)
print(f" Input photo: {input_photo}")
elif hasattr(message.media, 'document'):
input_doc = utils.get_input_document(message.media)
print(f" Input document: {input_doc}")
# Get file location for downloading
input_location = utils.get_input_location(message.media)
print(f" Input location: {input_location}")
print("---")
await client.disconnect()Types
from typing import Union, List, Optional, Tuple, Any, Iterator
from telethon.tl import types
EntityLike = Union[int, str, types.User, types.Chat, types.Channel]
MediaLike = Union[types.Photo, types.Document, types.MessageMedia]
FileLike = Union[str, bytes, 'io.IOBase']
LocationLike = Union[types.GeoPoint, Tuple[float, float]]
PhoneNumber = str
Username = str
ParseMode = Union[str, object, None]
AttributeList = List[types.TypeDocumentAttribute]
ChunkIterator = Iterator[List[Any]]Install with Tessl CLI
npx tessl i tessl/pypi-telethon