Unless otherwise specified, the following applies to all snippets:
from telethon import TelegramClient, sync, errors
from telethon.tl import functions as f, types as t
client = TelegramClient(...).start() # successfully logged in
👍1
Counts the amount of photos above/below a given size threshold
threshold = 75 * 1024 # 75kb#count #media #photos
below = 0
above = 0
chat = 'mychat'
filter = t.InputMessagesFilterPhotos
for m in client.iter_messages(chat, filter=filter):
if m.photo.sizes[-1].size < threshold:
below += 1
else:
above += 1
print('Below: {}\nAbove: {}\nTotal: {}'.format(below, above, below + above))
👍1
Reply to a message with "haste" to upload to hastebin and edit your message to the URL
Note: does not handle failure of http request.
#hastebin #haste
Note: does not handle failure of http request.
import requests
@client.on(events.NewMessage(outgoing=True, pattern="(?i)^haste$"))
def haste(e):
if e.is_reply and e.get_reply_message().message:
text = e.get_reply_message().text
e.edit("[Haste link to the above](https://hastebin.com/{}) ".format(requests.post("https://hastebin.com/documents", data=text).json()['key']))
#hastebin #haste
👍1
Block, report and clear all new PMs. Use with caution
#block #PM #report
whitelist = {d.id for d in client.get_dialogs() if d.id > 0}
@client.on(events.NewMessage)
async def handler(event):
if not event.is_private or event.chat_id in whitelist:
return
who = await event.get_input_chat()
await client(f.messages.ReportSpamRequest(who))
await client(f.contacts.BlockRequest(who))
await client(f.messages.DeleteHistoryRequest(who, 0))#block #PM #report
👍1
Get % of people who participated in the last 5000 messages
#analytics #stats #participation
group = 'somegroup'
people = {}
total = client.get_participants(group, limit=0).total
for m in client.iter_messages(group, limit=5000):
if not m.sender.bot:
people[m.sender.id] = m.sender.first_name
print(len(people)/total)
#analytics #stats #participation
👍1
Get the usage count of the top 50 words in the chat
#analytics #stats #words
class custom(dict):
def __missing__(self, key): return 0
words = custom()
progress = event.reply("Processed 0 messages")
total = 0
for msg in client.iter_messages(group):
total += 1
if total % 200 == 0:
progress.edit("Processed {} messages".format(total))
if msg.text:
for word in msg.text.split():
words[word.lower()] += 1
global freq
freq = sorted(words, key=words.get, reverse=True)
out = ""
for i in range(51):
out += "{}. {}:{}\n".format(i+1, words[freq[i]], freq[i])
progress.edit(out, parse_mode=None)#analytics #stats #words
👍1
Helper function for determining how long ago a user was online.
Sample usage:
#filter #status #user
Sample usage:
within_6_days = [x for x in client.iter_participants(group) if online_within(x, 6)]import datetime
def online_within(participant, days):
status = participant.status
if isinstance(status, t.UserStatusOnline):
return True
last_seen = status.was_online if isinstance(status, t.UserStatusOffline) else None
if last_seen:
now = datetime.datetime.now(tz=datetime.timezone.utc)
diff = now - last_seen
return diff <= datetime.timedelta(days=days)
if isinstance(status, t.UserStatusRecently) and days >= 1 \
or isinstance(status, t.UserStatusLastWeek) and days >= 7 \
or isinstance(status, t.UserStatusLastMonth) and days >= 30:
return True
return False
#filter #status #user
❤1
Example on how to send a file multiple times with only one upload:
#file #upload #cache
Note that this only works in master branch at time of writing. Expect implementation in v1.6
file = client.upload_file(path_to_photo)
#send as photo
client.send_file(chat, file)
#send as doc
client.send_file(chat, file, force_document=True)
#file #upload #cache
Note that this only works in master branch at time of writing. Expect implementation in v1.6
🔥3🥰1