LonamiWebs/Telethon
1
1
Fork 0
mirror of https://github.com/LonamiWebs/Telethon.git synced 2025-02-09 08:00:53 +03:00
Code Issues Packages Projects Releases Wiki Activity

Update examples

Browse Source
This commit is contained in:
Lonami Exo 2018-02-16 21:02:47 +01:00
parent 83d9d1d78e
commit 3c6f34fe6a
2 changed files with 55 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
telethon_examples/print_updates.py
View File

@ -1,46 +1,36 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# A simple script to print all updates received # A simple script to print all updates received
from getpass import getpass
from os import environ from os import environ
# environ is used to get API information from environment variables # environ is used to get API information from environment variables
# You could also use a config file, pass them as arguments, # You could also use a config file, pass them as arguments,
# or even hardcode them (not recommended) # or even hardcode them (not recommended)
from telethon import TelegramClient from telethon import TelegramClient
from telethon.errors import SessionPasswordNeededError
def main(): def main():
session_name = environ.get('TG_SESSION', 'session') session_name = environ.get('TG_SESSION', 'session')
user_phone = environ['TG_PHONE']
client = TelegramClient(session_name, client = TelegramClient(session_name,
int(environ['TG_API_ID']), int(environ['TG_API_ID']),
environ['TG_API_HASH'], environ['TG_API_HASH'],
proxy=None, proxy=None,
update_workers=4) update_workers=4,
spawn_read_thread=False)
print('INFO: Connecting to Telegram Servers...', end='', flush=True) if 'TG_PHONE' in environ:
client.connect() client.start(phone=environ['TG_PHONE'])
print('Done!') else:
client.start()
if not client.is_user_authorized():
print('INFO: Unauthorized user')
client.send_code_request(user_phone)
code_ok = False
while not code_ok:
code = input('Enter the auth code: ')
try:
code_ok = client.sign_in(user_phone, code)
except SessionPasswordNeededError:
password = getpass('Two step verification enabled. Please enter your password: ')
code_ok = client.sign_in(password=password)
print('INFO: Client initialized succesfully!')
client.add_update_handler(update_handler) client.add_update_handler(update_handler)
input('Press Enter to stop this!\n') print('(Press Ctrl+C to stop this)')
client.idle()
def update_handler(update): def update_handler(update):
print(update) print(update)
print('Press Enter to stop this!')
if __name__ == '__main__': if __name__ == '__main__':
main() main()

133
telethon_examples/replier.py
View File

@ -9,17 +9,12 @@ file, including TG_API_ID, TG_API_HASH, TG_PHONE and optionally TG_SESSION.
This script assumes that you have certain files on the working directory, This script assumes that you have certain files on the working directory,
such as "xfiles.m4a" or "anytime.png" for some of the automated replies. such as "xfiles.m4a" or "anytime.png" for some of the automated replies.
""" """
from getpass import getpass import re
from collections import defaultdict from collections import defaultdict
from datetime import datetime, timedelta from datetime import datetime, timedelta
from os import environ from os import environ
import re from telethon import TelegramClient, events, utils
from telethon import TelegramClient
from telethon.errors import SessionPasswordNeededError
from telethon.tl.types import UpdateNewChannelMessage, UpdateShortMessage, MessageService
from telethon.tl.functions.messages import EditMessageRequest
"""Uncomment this for debugging """Uncomment this for debugging
import logging import logging
@ -35,103 +30,57 @@ REACTS = {'emacs': 'Needs more vim',
recent_reacts = defaultdict(list) recent_reacts = defaultdict(list)
def update_handler(update): if __name__ == '__main__':
global recent_reacts # TG_API_ID and TG_API_HASH *must* exist or this won't run!
try: session_name = environ.get('TG_SESSION', 'session')
msg = update.message client = TelegramClient(
except AttributeError: session_name, int(environ['TG_API_ID']), environ['TG_API_HASH'],
# print(update, 'did not have update.message') spawn_read_thread=False, proxy=None, update_workers=4
return )
if isinstance(msg, MessageService):
print(msg, 'was service msg')
return
# React to messages in supergroups and PMs @client.on(events.NewMessage)
if isinstance(update, UpdateNewChannelMessage): def my_handler(event: events.NewMessage.Event):
words = re.split('\W+', msg.message) global recent_reacts
# This utils function gets the unique identifier from peers (to_id)
to_id = utils.get_peer_id(event.message.to_id)
# Through event.raw_text we access the text of messages without format
words = re.split('\W+', event.raw_text)
# Try to match some reaction
for trigger, response in REACTS.items(): for trigger, response in REACTS.items():
if len(recent_reacts[msg.to_id.channel_id]) > 3: if len(recent_reacts[to_id]) > 3:
# Silently ignore triggers if we've recently sent 3 reactions # Silently ignore triggers if we've recently sent 3 reactions
break break
if trigger in words: if trigger in words:
# Remove recent replies older than 10 minutes # Remove recent replies older than 10 minutes
recent_reacts[msg.to_id.channel_id] = [ recent_reacts[to_id] = [
a for a in recent_reacts[msg.to_id.channel_id] if a for a in recent_reacts[to_id] if
datetime.now() - a < timedelta(minutes=10) datetime.now() - a < timedelta(minutes=10)
] ]
# Send a reaction # Send a reaction as a reply (otherwise, event.respond())
client.send_message(msg.to_id, response, reply_to=msg.id) event.reply(response)
# Add this reaction to the list of recent actions # Add this reaction to the list of recent actions
recent_reacts[msg.to_id.channel_id].append(datetime.now()) recent_reacts[to_id].append(datetime.now())
if isinstance(update, UpdateShortMessage):
words = re.split('\W+', msg)
for trigger, response in REACTS.items():
if len(recent_reacts[update.user_id]) > 3:
# Silently ignore triggers if we've recently sent 3 reactions
break
if trigger in words:
# Send a reaction
client.send_message(update.user_id, response, reply_to=update.id)
# Add this reaction to the list of recent reactions
recent_reacts[update.user_id].append(datetime.now())
# Automatically send relevant media when we say certain things # Automatically send relevant media when we say certain things
# When invoking requests, get_input_entity needs to be called manually # When invoking requests, get_input_entity needs to be called manually
if isinstance(update, UpdateNewChannelMessage) and msg.out: if event.out:
if msg.message.lower() == 'x files theme': if event.raw_text.lower() == 'x files theme':
client.send_voice_note(msg.to_id, 'xfiles.m4a', reply_to=msg.id) client.send_voice_note(event.message.to_id, 'xfiles.m4a',
if msg.message.lower() == 'anytime': reply_to=event.message.id)
client.send_file(msg.to_id, 'anytime.png', reply_to=msg.id) if event.raw_text.lower() == 'anytime':
if '.shrug' in msg.message: client.send_file(event.message.to_id, 'anytime.png',
client(EditMessageRequest( reply_to=event.message.id)
client.get_input_entity(msg.to_id), msg.id, if '.shrug' in event.text:
message=msg.message.replace('.shrug', r'¯\_(ツ)_/¯') event.edit(event.text.replace('.shrug', r'¯\_(ツ)_/¯'))
))
if isinstance(update, UpdateShortMessage) and update.out: if 'TG_PHONE' in environ:
if msg.lower() == 'x files theme': client.start(phone=environ['TG_PHONE'])
client.send_voice_note(update.user_id, 'xfiles.m4a', reply_to=update.id) else:
if msg.lower() == 'anytime': client.start()
client.send_file(update.user_id, 'anytime.png', reply_to=update.id)
if '.shrug' in msg:
client(EditMessageRequest(
client.get_input_entity(update.user_id), update.id,
message=msg.replace('.shrug', r'¯\_(ツ)_/¯')
))
print('(Press Ctrl+C to stop this)')
if __name__ == '__main__': client.idle()
session_name = environ.get('TG_SESSION', 'session')
user_phone = environ['TG_PHONE']
client = TelegramClient(
session_name, int(environ['TG_API_ID']), environ['TG_API_HASH'],
proxy=None, update_workers=4
)
try:
print('INFO: Connecting to Telegram Servers...', end='', flush=True)
client.connect()
print('Done!')
if not client.is_user_authorized():
print('INFO: Unauthorized user')
client.send_code_request(user_phone)
code_ok = False
while not code_ok:
code = input('Enter the auth code: ')
try:
code_ok = client.sign_in(user_phone, code)
except SessionPasswordNeededError:
password = getpass('Two step verification enabled. '
'Please enter your password: ')
code_ok = client.sign_in(password=password)
print('INFO: Client initialized successfully!')
client.add_update_handler(update_handler)
input('Press Enter to stop this!\n')
except KeyboardInterrupt:
pass
finally:
client.disconnect()