Move some parts of assistant to a different repo as plugins

This commit is contained in:
Lonami Exo 2019-05-11 20:41:10 +02:00
parent 4ca3517e22
commit 560d4bed09
2 changed files with 27 additions and 113 deletions

View File

@ -75,6 +75,11 @@ This example is the actual bot account [@TelethonianBot] running in the
[official Telethon's chat] to help people out. The file is a bit big and [official Telethon's chat] to help people out. The file is a bit big and
assumes some [`asyncio`] knowledge, but otherwise is easy to follow. assumes some [`asyncio`] knowledge, but otherwise is easy to follow.
In addition, it has optional plugins, which may be useful for your own code.
The plugins can be found at https://github.com/Lonami/TelethonianBotExt and
should be cloned into a `plugins` folder next to `assistant.py` for them to
work.
### [`interactive_telegram_client.py`] ### [`interactive_telegram_client.py`]
* Usable as: **user**. * Usable as: **user**.

View File

@ -1,5 +1,4 @@
import asyncio import asyncio
import difflib
import html import html
import logging import logging
import os import os
@ -9,7 +8,6 @@ import time
import urllib.parse import urllib.parse
from telethon import TelegramClient, events, types, custom, utils, errors from telethon import TelegramClient, events, types, custom, utils, errors
from telethon.extensions import markdown
logging.basicConfig(level=logging.WARNING) logging.basicConfig(level=logging.WARNING)
logging.getLogger('asyncio').setLevel(logging.ERROR) logging.getLogger('asyncio').setLevel(logging.ERROR)
@ -20,6 +18,7 @@ except ImportError:
aiohttp = None aiohttp = None
logging.warning('aiohttp module not available; #haste command disabled') logging.warning('aiohttp module not available; #haste command disabled')
def get_env(name, message, cast=str): def get_env(name, message, cast=str):
if name in os.environ: if name in os.environ:
return os.environ[name] return os.environ[name]
@ -41,13 +40,13 @@ bot = TelegramClient(NAME, API_ID, API_HASH)
# ============================== Constants ============================== # ============================== Constants ==============================
WELCOME = { WELCOME = {
-1001109500936: -1001109500936:
'Hi and welcome to the group. Before asking any questions, **please** ' 'Hi and welcome to the group. Before asking any questions, **please** '
'read [the docs](https://telethon.readthedocs.io/). Make sure you are ' 'read [the docs](https://telethon.readthedocs.io/). Make sure you are '
'using the latest version with `pip3 install -U telethon`, since most ' 'using the latest version with `pip3 install -U telethon`, since most '
'problems have already been fixed in newer versions.', 'problems have already been fixed in newer versions.',
-1001200633650: -1001200633650:
'Welcome to the off-topic group. Feel free to talk, ask or test anything ' 'Welcome to the off-topic group. Feel free to talk, ask or test anything '
'here, politely. Check the description if you need to test more spammy ' 'here, politely. Check the description if you need to test more spammy '
'"features" of your or other people\'s bots (sed commands too).' '"features" of your or other people\'s bots (sed commands too).'
@ -70,11 +69,6 @@ UPDATES = (
'Check out [Working with Updates](https://telethon.readthedocs.io' 'Check out [Working with Updates](https://telethon.readthedocs.io'
'/en/latest/basic/updates.html) in the documentation.' '/en/latest/basic/updates.html) in the documentation.'
) )
DOCS_CLIENT = 'https://telethon.readthedocs.io/en/latest/modules/client.html#'
DOCS_MESSAGE = (
'https://telethon.readthedocs.io/en/latest/'
'modules/custom.html#telethon.tl.custom.message.Message.'
)
SPAM = ( SPAM = (
"Telethon is free software. That means using it is a right: you are " "Telethon is free software. That means using it is a right: you are "
@ -219,39 +213,6 @@ async def handler(event):
]) ])
def get_docs_message(kind, query):
kind = kind.lower()
cls = {'client': TelegramClient, 'msg': custom.Message}[kind]
attr = search_attr(cls, query.lower())
if not attr:
return f'No such method "{query}" :/'
name = attr
if kind == 'client':
attr = attr_fullname(cls, attr)
url = DOCS_CLIENT
elif kind == 'msg':
name = f'Message.{name}'
url = DOCS_MESSAGE
else:
return f'No documentation for "{kind}"'
return f'Documentation for [{name}]({url}{attr})'
@bot.on(events.NewMessage(pattern='(?i)#(client|msg) (.+)', forwards=False))
async def handler(event):
"""#client or #msg query: Looks for the given attribute in RTD."""
await event.delete()
await event.respond(
get_docs_message(kind=event.pattern_match.group(1),
query=event.pattern_match.group(2)),
reply_to=event.reply_to_msg_id
)
@bot.on(events.NewMessage(pattern='(?i)#(ask|question)', forwards=False)) @bot.on(events.NewMessage(pattern='(?i)#(ask|question)', forwards=False))
async def handler(event): async def handler(event):
"""#ask or #question: Advices the user to ask a better question.""" """#ask or #question: Advices the user to ask a better question."""
@ -270,6 +231,7 @@ async def handler(event):
event.respond(SPAM, reply_to=event.reply_to_msg_id) event.respond(SPAM, reply_to=event.reply_to_msg_id)
]) ])
@bot.on(events.NewMessage(pattern='(?i)#(ot|offtopic)', forwards=False)) @bot.on(events.NewMessage(pattern='(?i)#(ot|offtopic)', forwards=False))
async def handler(event): async def handler(event):
"""#ot, #offtopic: Tells the user to move to @TelethonOffTopic.""" """#ot, #offtopic: Tells the user to move to @TelethonOffTopic."""
@ -397,81 +359,28 @@ async def handler(event):
text=GOOD_RESOURCES, text=GOOD_RESOURCES,
link_preview=False link_preview=False
) )
else:
m = re.match('(client|msg).(.+)', query)
if m:
text = get_docs_message(m.group(1), m.group(2))
query = markdown.parse(text)[0]
result = builder.article(query, text=text)
else:
m = re.match('ref.(.+)', query)
if m:
query = m.group(1)
text = DOCS.format(query, urllib.parse.quote(query))
result = builder.article(query, text=text)
await event.answer([result] if result else None) # NOTE: You should always answer, but we want plugins to be able to answer
# too (and we can only answer once), so we don't always answer here.
if result:
await event.answer([result])
# ============================== Inline ============================== # ============================== Inline ==============================
# ============================== AutoReply ==============================
@bot.on(events.NewMessage(pattern='(?i)how (.+?)[\W]*$', forwards=False))
@bot.on(events.NewMessage(pattern='(.+?)[\W]*?\?+', forwards=False))
async def handler(event):
words = event.pattern_match.group(1).split()
rates = [
search_attr(TelegramClient, ' '.join(words[-i:]), threshold=None)
for i in range(1, 4)
]
what = max(rates, key=lambda t: t[1])
if what[1] < 0.75:
return
name = what[0]
if len(name) < 4:
return # Short words trigger very commonly (such as "on")
attr = attr_fullname(TelegramClient, name)
await event.reply(
f'Documentation for [{name}]({DOCS_CLIENT}{attr})',
reply_to=event.reply_to_msg_id
)
# We have two @client.on, both could fire, stop that
raise events.StopPropagation
# ============================== AutoReply ==============================
# ============================== Helpers ==============================
def search_attr(cls, query, threshold=0.6):
seq = difflib.SequenceMatcher(b=query, autojunk=False)
scores = []
for n in dir(cls):
if not n.startswith('_'):
seq.set_seq1(n)
scores.append((n, seq.ratio()))
scores.sort(key=lambda t: t[1], reverse=True)
if threshold is None:
return scores[0]
else:
return scores[0][0] if scores[0][1] >= threshold else None
def attr_fullname(cls, n):
m = getattr(cls, n)
cls = sys.modules.get(m.__module__)
for name in m.__qualname__.split('.')[:-1]:
cls = getattr(cls, name)
return cls.__module__ + '.' + cls.__name__ + '.' + m.__name__
# ============================== Helpers ==============================
bot.start(bot_token=TOKEN) bot.start(bot_token=TOKEN)
# NOTE: This example has optional "plugins", which you can get by running:
#
# git clone https://github.com/Lonami/TelethonianBotExt plugins
#
# Into the same folder (so you would have `assistant.py` next to
# the now downloaded `plugins/` folder). We try importing them so
# that the example runs fine without them, but optionally load them.
try:
import plugins
plugins.init(bot)
except ImportError:
pass
bot.run_until_disconnected() bot.run_until_disconnected()