Package that wraps the python-telegram-bot library to make broadcasting easier.
Project description
Python Telegram Broadcast
This is a simple Python package that build on top of python-telegram-bot to make broadcasting easier.
Features
- Broadcast message to multiple users
- Support media type:
- Text
- Photo
- Document
- Video
- Support broadcast strategy:
- Asynchronous Sequentially
- Asynchronous Process Pool
- Asynchronous Multiprocessing Pool
Dependencies
- python-telegram-bot=21.1.1
- mypy
Installation
pip3 install python_telegram_broadcast
Example
Get File ID from Telegram
import asyncio
from python_telegram_broadcast import (
get_file_id, select_broadcast_method, BroadcastMethodType
)
def wrapper(token, method, target_id, payload):
caption = ""
seconds = 0.0 # 0.0 means no timeout
max_retry = 5 # Maximum retry
loop = asyncio.get_event_loop()
return loop.run_until_complete(
get_file_id(
token, method, target_id, payload, caption, seconds, max_retry
)
)
if __name__ == "__main__":
bot_token: str = "TELEGRAM_BOT_TOKEN" # << Change to your bot token
user_telegram_id: int = 123456789 # << Change to your telegram id
file_path: str = "" # << Change to your file path or the URL of your file
broadcast_method = select_broadcast_method(BroadcastMethodType.PHOTO)
# If file_path is a URL
if "://" in file_path:
file_id = wrapper(bot_token, broadcast_method, user_telegram_id, file_path)
else:
# Otherwise
file_id = wrapper(bot_token, broadcast_method, user_telegram_id, open(file_path, "rb"))
print(file_id)
Broadcast Photo
from typing import Tuple
from python_telegram_broadcast import (
handle_broadcast,
select_broadcast_strategy, BroadcastStrategyType,
write_sent_result
)
def broadcast_wrapper(token, method, stg, slist, payload):
loop = asyncio.get_event_loop()
return loop.run_until_complete(
handle_broadcast(
slist, token, method, stg, payload, "...",
use_multiproc=False, use_nproc=1, seconds=0.0, max_retry=5
)
)
# Use bot_token, broadcast_method, file_path from the previous example!!!
export_path = ""
broadcast_strategy = select_broadcast_strategy(BroadcastStrategyType.ASYNCIO_SEQUENTIAL)
# Read subscriber list from file, but you can also read from database of your choice
subscriber_list: list[Tuple[int, str]] = []
with open("./asset/subscriber.txt", "r") as file:
header = file.readline()
while True:
line = file.readline()
if not line:
break
telegram_id, username = line.strip().split(",")
subscriber_list.append((int(telegram_id), username))
s, f = broadcast_wrapper(
bot_token, broadcast_method, broadcast_strategy,
subscriber_list,
file_id
)
for S in s:
print(f"Success: {s}")
for F in f:
print(f"Failed: {f}")
if len(f) > 0:
write_sent_result(f"{export_path}/result_{file_path}.txt", f, file_path)
Tests
TODO
Source Code
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Close
Hashes for python_telegram_broadcast-0.8.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 4e1ac8b05b335d24ea9cd45ed4935962750925d239269ed6a32ec84ed244bd8a |
|
MD5 | a6877903305db2c8ab3a2647b571baf4 |
|
BLAKE2b-256 | b73e79cb308a26c7a6d1287181a9e5dfa2ebab132624bc02db3db16c81b9193e |
Close
Hashes for python_telegram_broadcast-0.8.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | d4b6c502f1f2fb570db7ba518c2e81d43d8b07f1b9eea3c48691860e1ec4cd19 |
|
MD5 | 1165e43f894ee49e67796edcc9bd72ac |
|
BLAKE2b-256 | fd4c7efe80117263fbf060effda8bc91aa529097a3148b0a332bb2df63f3d3cc |