hoelee/Telethon
1
0
Fork 0
mirror of https://github.com/LonamiWebs/Telethon.git synced 2025-08-05 19:42:29 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Update examples to be async

Browse Source
This commit is contained in:
Lonami Exo
2018-06-24 12:04:23 +02:00
parent 58031b3adf
commit 026c0c4f9d
5 changed files with 134 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
telethon_examples/replier.py
View File

@@ -10,11 +10,12 @@ 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.
"""
import re
import asyncio
from collections import defaultdict
from datetime import datetime, timedelta
from os import environ
from telethon import TelegramClient, events, utils
from telethon import TelegramClient, events
"""Uncomment this for debugging
import logging
@@ -30,57 +31,54 @@ REACTS = {'emacs': 'Needs more vim',
recent_reacts = defaultdict(list)
# TG_API_ID and TG_API_HASH *must* exist or this won't run!
session_name = environ.get('TG_SESSION', 'session')
client = TelegramClient(
session_name, int(environ['TG_API_ID']), environ['TG_API_HASH'],
proxy=None
)
@client.on(events.NewMessage)
async def my_handler(event: events.NewMessage.Event):
global recent_reacts
# 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():
if len(recent_reacts[event.chat_id]) > 3:
# Silently ignore triggers if we've recently sent 3 reactions
break
if trigger in words:
# Remove recent replies older than 10 minutes
recent_reacts[event.chat_id] = [
a for a in recent_reacts[event.chat_id] if
datetime.now() - a < timedelta(minutes=10)
]
# Send a reaction as a reply (otherwise, event.respond())
await event.reply(response)
# Add this reaction to the list of recent actions
recent_reacts[event.chat_id].append(datetime.now())
# Automatically send relevant media when we say certain things
# When invoking requests, get_input_entity needs to be called manually
if event.out:
chat = await event.get_input_chat()
if event.raw_text.lower() == 'x files theme':
await client.send_file(chat, 'xfiles.m4a',
reply_to=event.message.id, voice_note=True)
if event.raw_text.lower() == 'anytime':
await client.send_file(chat, 'anytime.png',
reply_to=event.message.id)
if '.shrug' in event.text:
await event.edit(event.text.replace('.shrug', r'¯\_(ツ)_/¯'))
if __name__ == '__main__':
# TG_API_ID and TG_API_HASH *must* exist or this won't run!
session_name = environ.get('TG_SESSION', 'session')
client = TelegramClient(
session_name, int(environ['TG_API_ID']), environ['TG_API_HASH'],
spawn_read_thread=False, proxy=None, update_workers=4
)
@client.on(events.NewMessage)
def my_handler(event: events.NewMessage.Event):
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():
if len(recent_reacts[to_id]) > 3:
# Silently ignore triggers if we've recently sent 3 reactions
break
if trigger in words:
# Remove recent replies older than 10 minutes
recent_reacts[to_id] = [
a for a in recent_reacts[to_id] if
datetime.now() - a < timedelta(minutes=10)
]
# Send a reaction as a reply (otherwise, event.respond())
event.reply(response)
# Add this reaction to the list of recent actions
recent_reacts[to_id].append(datetime.now())
# Automatically send relevant media when we say certain things
# When invoking requests, get_input_entity needs to be called manually
if event.out:
if event.raw_text.lower() == 'x files theme':
client.send_voice_note(event.message.to_id, 'xfiles.m4a',
reply_to=event.message.id)
if event.raw_text.lower() == 'anytime':
client.send_file(event.message.to_id, 'anytime.png',
reply_to=event.message.id)
if '.shrug' in event.text:
event.edit(event.text.replace('.shrug', r'¯\_(ツ)_/¯'))
if 'TG_PHONE' in environ:
client.start(phone=environ['TG_PHONE'])
else:
client.start()
loop = asyncio.get_event_loop()
loop.run_until_complete(client.start(phone=environ.get('TG_PHONE')))
print('(Press Ctrl+C to stop this)')
client.idle()
client.run_until_disconnected()