Reputation: 421
I am using telethon in python, and I want to try to keep track of the number o new members who join a channel over time.
My first attempt was using the telethon.tl.functions.channels.GetFullChannelRequest
function:
async def save_subs(channel_id):
client = TelegramClient('anon', TELEGRAM_API_ID, TELEGRAM_API_HASH)
await client.start()
async with client:
previous_count = None
number_new_members = 0
while True:
res = await client(
GetFullChannelRequest(
channel=channel_id
)
)
channel_info = res.to_dict()
count = channel_info['full_chat']['participants_count']
print(f'current count: {count}')
if previous_count is not None and count > previous_count:
number_new_members += count - previous_count
print(f'new subs: {number_new_members}')
previous_count = count
time.sleep(1)
def start(channel_id):
asyncio.run(save_subs())
However, it seems that every 1 second it too much, and every 4 requests there is a delay of a few seconds. Perhaps this is being rate limited?
Then I tried using the events.ChatAction
event because the docs say that a new user joining a channel should trigger, but nothing is getting triggered:
def check_new_user(channel_id):
telegram_client = TelegramClient('anon', TELEGRAM_API_ID, TELEGRAM_API_HASH)
@telegram_client.on(events.ChatAction)
async def my_event_handler(event):
# this never gets printed
print('event caught')
telegram_client.start()
telegram_client.run_until_disconnected()
def start(channel_id):
check_events(channel_id)
Upvotes: 2
Views: 1110
Reputation: 7141
Many requests to Telegram include a count
, accessible via Telethon's .total
attribute on the returned lists:
participants = await client.get_participants(channel, 1)
print(participants.total)
This will fetch a single participant (you must have permission to do so), and as a byproduct, Telegram fills-in the total.
Upvotes: 3
Reputation: 3860
I'd advise a more straightforward approach. You can do it using Telegram's t.me
application.
DEFAULT_TIMEOUT = 10
def scrap_telegram_members_count(slug: str) -> int:
def parse_int(s: str) -> int:
return int(re.sub(r"[^\d]+", "", s))
r = requests.get(
f"https://t.me/{slug}",
timeout=DEFAULT_TIMEOUT,
)
r.raise_for_status()
text = r.text
m = re.search(r'<div class="tgme_page_extra">([^,]+),([^<]+)</div>', text)
if not m:
raise RuntimeError("Cannot find members count")
members_count = parse_int(m.group(1))
return members_count
From my experience, now you can send not less than ~1k requests per day this way.
Upvotes: 2