Davte/davtelepot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Package updates checker implemented

Browse Source 
Notify administrators when new versions are available for PyPi packages in `bot.packages`.
This commit is contained in:
Davte 2020-04-27 13:48:33 +02:00
parent 5a70dcfeb2
commit edb2201773
8 changed files with 247 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
davtelepot/__init__.py
View File

@ -14,7 +14,7 @@ __author__ = "Davide Testa"
__email__ = "davide@davte.it"
__credits__ = ["Marco Origlia", "Nick Lee @Nickoala"]
__license__ = "GNU General Public License v3.0"
__version__ = "2.4.25"
__version__ = "2.5.0"
__maintainer__ = "Davide Testa"
__contact__ = "t.me/davte"

150
davtelepot/administration_tools.py
View File

@ -12,17 +12,19 @@ davtelepot.admin_tools.init(my_bot)
import asyncio
import datetime
import json
import logging
# Third party modules
import davtelepot
from davtelepot import messages
from davtelepot.utilities import (
async_wrapper, Confirmator, extract, get_cleaned_text, get_user,
escape_html_chars, line_drawing_unordered_list, make_button,
from sqlalchemy.exc import ResourceClosedError
# Project modules
from . import bot as davtelepot_bot, messages, __version__
from .utilities import (
async_wrapper, CachedPage, Confirmator, extract, get_cleaned_text,
get_user, escape_html_chars, line_drawing_unordered_list, make_button,
make_inline_keyboard, remove_html_tags, send_part_of_text_file,
send_csv_file
)
from sqlalchemy.exc import ResourceClosedError
async def _forward_to(update, bot, sender, addressee, is_admin=False):
@ -179,11 +181,9 @@ def get_talk_panel(bot, update, user_record=None, text=''):
{
key: val
for key, val in user.items()
if key in (
'first_name',
if key in ('first_name',
'last_name',
'username'
)
'username')
}
)
),
@ -779,7 +779,7 @@ def get_maintenance_exception_criterion(bot, allowed_command):
return criterion
async def get_version():
async def get_last_commit():
"""Get last commit hash and davtelepot version."""
try:
_subprocess = await asyncio.create_subprocess_exec(
@ -793,70 +793,132 @@ async def get_version():
last_commit = f"{e}"
if last_commit.startswith("fatal: not a git repository"):
last_commit = "-"
davtelepot_version = davtelepot.__version__
return last_commit, davtelepot_version
return last_commit
async def _version_command(bot, update, user_record):
last_commit, davtelepot_version = await get_version()
last_commit = await get_last_commit()
return bot.get_message(
'admin', 'version_command', 'result',
last_commit=last_commit,
davtelepot_version=davtelepot_version,
davtelepot_version=__version__,
update=update, user_record=user_record
)
async def notify_new_version(bot):
async def notify_new_version(bot: davtelepot_bot):
"""Notify `bot` administrators about new versions.
Notify admins when last commit and/or davtelepot version change.
"""
last_commit, davtelepot_version = await get_version()
last_commit = await get_last_commit()
old_record = bot.db['version_history'].find_one(
order_by=['-id']
)
current_versions = {
f"{package.__name__}_version": package.__version__
for package in bot.packages
}
current_versions['last_commit'] = last_commit
if old_record is None:
old_record = dict(
updated_at=datetime.datetime.min,
last_commit=None,
davtelepot_version=None
)
if (
old_record['last_commit'] != last_commit
or old_record['davtelepot_version'] != davtelepot_version
for name in current_versions.keys():
if name not in old_record:
old_record[name] = None
if any(
old_record[name] != current_version
for name, current_version in current_versions.items()
):
new_record = dict(
updated_at=datetime.datetime.now(),
last_commit=last_commit,
davtelepot_version=davtelepot_version
)
bot.db['version_history'].insert(
new_record
dict(
updated_at=datetime.datetime.now(),
**current_versions
)
)
for admin in bot.administrators:
text = bot.get_message(
'admin', 'new_version', 'title',
user_record=admin
) + '\n\n'
if last_commit != old_record['last_commit']:
text += bot.get_message(
'admin', 'new_version', 'last_commit',
old_record=old_record,
new_record=current_versions,
user_record=admin
) + '\n\n'
text += '\n'.join(
f"<b>{name[:-len('_version')]}</b>: "
f"<code>{old_record[name]}</code> —> "
f"<code>{current_version}</code>"
for name, current_version in current_versions.items()
if name not in ('last_commit', )
and current_version != old_record[name]
)
for admin in bot.db['users'].find(privileges=[1, 2]):
await bot.send_message(
chat_id=admin['telegram_id'],
disable_notification=True,
text='\n\n'.join(
bot.get_message(
'admin', 'new_version', field,
old_record=old_record,
new_record=new_record,
user_record=admin
)
for field in filter(
lambda x: (x not in old_record
or old_record[x] != new_record[x]),
('title', 'last_commit', 'davtelepot_version')
)
)
text=text
)
return
def init(telegram_bot, talk_messages=None, admin_messages=None):
async def get_package_updates(bot: davtelepot_bot,
monitoring_interval: int = 60 * 60):
while 1:
news = dict()
for package in bot.packages:
package_web_page = CachedPage.get(
f'https://pypi.python.org/pypi/{package.__name__}/json',
cache_time=2,
mode='json'
)
web_page = await package_web_page.get_page()
if web_page is None or isinstance(web_page, Exception):
logging.error(f"Cannot get updates for {package.__name__}, "
"skipping...")
continue
new_version = web_page['info']['version']
current_version = package.__version__
if new_version != current_version:
news[package.__name__] = {
'current': current_version,
'new': new_version
}
if news:
for admin in bot.administrators:
text = bot.get_message(
'admin', 'updates_available', 'header',
user_record=admin
) + '\n\n'
text += '\n'.join(
f"<b>{package}</b>: "
f"<code>{versions['current']}</code> —> "
f"<code>{versions['new']}</code>"
for package, versions in news.items()
)
await bot.send_message(
chat_id=admin['telegram_id'],
disable_notification=True,
text=text
)
await asyncio.sleep(monitoring_interval)
def init(telegram_bot,
talk_messages=None,
admin_messages=None,
packages=None):
"""Assign parsers, commands, buttons and queries to given `bot`."""
if packages is None:
packages = []
telegram_bot.packages.extend(
filter(lambda package: package not in telegram_bot.packages,
packages)
)
asyncio.ensure_future(get_package_updates(telegram_bot))
if talk_messages is None:
talk_messages = messages.default_talk_messages
telegram_bot.messages['talk'] = talk_messages
@ -1045,7 +1107,7 @@ def init(telegram_bot, talk_messages=None, admin_messages=None):
'help_section',)
},
show_in_keyboard=False,
authorization_level='admin',)
authorization_level='admin')
async def version_command(bot, update, user_record):
return await _version_command(bot=bot,
update=update,

8
davtelepot/api.py
View File

@ -1,4 +1,4 @@
"""This module provides a glow-like middleware for Telegram bot API.
"""This module provides a python mirror for Telegram bot API.
All methods and parameters are the same as the original json API.
A simple aiohttp asynchronous web client is used to make requests.
@ -10,11 +10,11 @@ import datetime
import json
import logging
# Third party modules
from typing import Union, List
# Third party modules
import aiohttp
from aiohttp import web
import aiohttp.web
class TelegramError(Exception):
@ -82,7 +82,7 @@ class TelegramBot:
"""
loop = asyncio.get_event_loop()
app = web.Application()
app = aiohttp.web.Application()
sessions_timeouts = {
'getUpdates': dict(
timeout=35,

98
davtelepot/authorization.py
View File

@ -2,9 +2,11 @@
# Standard library modules
from collections import OrderedDict
from typing import Callable, Union
# Project modules
from .bot import Bot
from .messages import default_authorization_messages
from .utilities import (
Confirmator, get_cleaned_text, get_user, make_button, make_inline_keyboard
)
@ -256,85 +258,10 @@ def get_authorization_function(bot):
return is_authorized
deafult_authorization_messages = {
'auth_command': {
'description': {
'en': "Edit user permissions. To select a user, reply to "
"a message of theirs or write their username",
'it': "Cambia il grado di autorizzazione di un utente "
"(in risposta o scrivendone lo username)"
},
'unhandled_case': {
'en': "<code>Unhandled case :/</code>",
'it': "<code>Caso non previsto :/</code>"
},
'instructions': {
'en': "Reply with this command to a user or write "
"<code>/auth username</code> to edit their permissions.",
'it': "Usa questo comando in risposta a un utente "
"oppure scrivi <code>/auth username</code> per "
"cambiarne il grado di autorizzazione."
},
'unknown_user': {
'en': "Unknown user.",
'it': "Utente sconosciuto."
},
'choose_user': {
'en': "{n} users match your query. Please select one.",
'it': "Ho trovato {n} utenti che soddisfano questi criteri.\n"
"Per procedere selezionane uno."
},
'no_match': {
'en': "No user matches your query. Please try again.",
'it': "Non ho trovato utenti che soddisfino questi criteri.\n"
"Prova di nuovo."
}
},
'ban_command': {
'description': {
'en': "Reply to a user with /ban to ban them",
'it': "Banna l'utente (da usare in risposta)"
}
},
'auth_button': {
'description': {
'en': "Edit user permissions",
'it': "Cambia il grado di autorizzazione di un utente"
},
'confirm': {
'en': "Are you sure?",
'it': "Sicuro sicuro?"
},
'back_to_user': {
'en': "Back to user",
'it': "Torna all'utente"
},
'permission_denied': {
'user': {
'en': "You cannot appoint this user!",
'it': "Non hai l'autorità di modificare i permessi di questo "
"utente!"
},
'role': {
'en': "You're not allowed to appoint someone to this role!",
'it': "Non hai l'autorità di conferire questo permesso!"
}
},
'no_change': {
'en': "No change suggested!",
'it': "È già così!"
},
'appointed': {
'en': "Permission granted",
'it': "Permesso conferito"
}
},
}
async def _authorization_command(bot, update, user_record):
text = get_cleaned_text(bot=bot, update=update, replace=['auth'])
reply_markup = None
# noinspection PyUnusedLocal
result = bot.get_message(
'authorization', 'auth_command', 'unhandled_case',
update=update, user_record=user_record
@ -509,7 +436,17 @@ async def _ban_command(bot, update, user_record):
return
def init(telegram_bot: Bot, roles=None, authorization_messages=None):
def default_get_administrators_function(bot: Bot):
return list(
bot.db['users'].find(privileges=[1,2])
)
def init(telegram_bot: Bot,
roles: Union[list, OrderedDict] = None,
authorization_messages=None,
get_administrators_function: Callable[[object],
list] = None):
"""Set bot roles and assign role-related commands.
Pass an OrderedDict of `roles` to get them set.
@ -537,8 +474,11 @@ def init(telegram_bot: Bot, roles=None, authorization_messages=None):
telegram_bot.set_authorization_function(
get_authorization_function(telegram_bot)
)
if authorization_messages is None:
authorization_messages = deafult_authorization_messages
get_administrators_function = (get_administrators_function
or default_get_administrators_function)
telegram_bot.set_get_administrator_function(get_administrators_function)
authorization_messages = (authorization_messages
or default_authorization_messages)
telegram_bot.messages['authorization'] = authorization_messages
@telegram_bot.command(command='/auth', aliases=[], show_in_keyboard=False,

28
davtelepot/bot.py
View File

@ -34,13 +34,16 @@ Usage
# Standard library modules
import asyncio
from collections import OrderedDict
import datetime
import io
import inspect
import logging
import os
import re
import sys
from collections import OrderedDict
from typing import Callable
# Third party modules
from aiohttp import web
@ -210,6 +213,8 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
if 'chat' in update
else None
)
# Function to get updated list of bot administrators
self._get_administrators = lambda bot: []
# Message to be returned if user is not allowed to call method
self._authorization_denied_message = None
# Default authorization function (always return True)
@ -223,6 +228,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
self.placeholder_requests = dict()
self.shared_data = dict()
self.Role = None
self.packages = [sys.modules['davtelepot']]
# Add `users` table with its fields if missing
if 'users' not in self.db.tables:
table = self.db.create_table(
@ -553,6 +559,26 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
default_inline_query_answer
)
def set_get_administrator_function(self,
new_function: Callable[[object],
list]):
"""Set a new get_administrators function.
This function should take bot as argument and return an updated list
of its administrators.
Example:
```python
def get_administrators(bot):
admins = bot.db['users'].find(privileges=2)
return list(admins)
```
"""
self._get_administrators = new_function
@property
def administrators(self):
return self._get_administrators(self)
async def message_router(self, update, user_record):
"""Route Telegram `message` update to appropriate message handler."""
for key, value in update.items():

91
davtelepot/messages.py
View File

@ -134,14 +134,6 @@ default_admin_messages = {
'it': "Vecchio commit: <code>{old_record[last_commit]}</code>\n"
"Nuovo commit: <code>{new_record[last_commit]}</code>",
},
'davtelepot_version': {
'en': "davtelepot version: "
"<code>{old_record[davtelepot_version]}</code> —> "
"<code>{new_record[davtelepot_version]}</code>",
'it': "Versione di davtelepot: "
"<code>{old_record[davtelepot_version]}</code> —> "
"<code>{new_record[davtelepot_version]}</code>",
},
},
'query_button': {
'error': {
@ -256,6 +248,14 @@ default_admin_messages = {
"sessione"
}
},
'updates_available': {
'header': {
'en': "🔔 Updates available! ⬇️\n\n"
"Click to /restart bot",
'it': "🔔 Aggiornamenti disponibili! ⬇\n\n"
"Clicka qui per fare il /restart",
},
},
'version_command': {
'reply_keyboard_button': {
'en': "Version #️⃣",
@ -275,6 +275,81 @@ default_admin_messages = {
},
}
default_authorization_messages = {
'auth_command': {
'description': {
'en': "Edit user permissions. To select a user, reply to "
"a message of theirs or write their username",
'it': "Cambia il grado di autorizzazione di un utente "
"(in risposta o scrivendone lo username)"
},
'unhandled_case': {
'en': "<code>Unhandled case :/</code>",
'it': "<code>Caso non previsto :/</code>"
},
'instructions': {
'en': "Reply with this command to a user or write "
"<code>/auth username</code> to edit their permissions.",
'it': "Usa questo comando in risposta a un utente "
"oppure scrivi <code>/auth username</code> per "
"cambiarne il grado di autorizzazione."
},
'unknown_user': {
'en': "Unknown user.",
'it': "Utente sconosciuto."
},
'choose_user': {
'en': "{n} users match your query. Please select one.",
'it': "Ho trovato {n} utenti che soddisfano questi criteri.\n"
"Per procedere selezionane uno."
},
'no_match': {
'en': "No user matches your query. Please try again.",
'it': "Non ho trovato utenti che soddisfino questi criteri.\n"
"Prova di nuovo."
}
},
'ban_command': {
'description': {
'en': "Reply to a user with /ban to ban them",
'it': "Banna l'utente (da usare in risposta)"
}
},
'auth_button': {
'description': {
'en': "Edit user permissions",
'it': "Cambia il grado di autorizzazione di un utente"
},
'confirm': {
'en': "Are you sure?",
'it': "Sicuro sicuro?"
},
'back_to_user': {
'en': "Back to user",
'it': "Torna all'utente"
},
'permission_denied': {
'user': {
'en': "You cannot appoint this user!",
'it': "Non hai l'autorità di modificare i permessi di questo "
"utente!"
},
'role': {
'en': "You're not allowed to appoint someone to this role!",
'it': "Non hai l'autorità di conferire questo permesso!"
}
},
'no_change': {
'en': "No change suggested!",
'it': "È già così!"
},
'appointed': {
'en': "Permission granted",
'it': "Permesso conferito"
}
},
}
default_authorization_denied_message = {
'en': "You are not allowed to use this command, sorry.",
'it': "Non disponi di autorizzazioni sufficienti per questa richiesta, spiacente.",

17
davtelepot/suggestions.py
View File

@ -149,15 +149,6 @@ async def _suggestions_button(bot: davtelepot.bot.Bot, update, user_record, data
when = datetime.datetime.now()
with bot.db as db:
registered_user = db['users'].find_one(telegram_id=user_id)
admins = [
x['telegram_id']
for x in db['users'].find(
privileges=[
bot.Role.get_role_by_name('admin').code,
bot.Role.get_role_by_name('founder').code
]
)
]
db['suggestions'].update(
dict(
id=suggestion_id,
@ -176,11 +167,11 @@ async def _suggestions_button(bot: davtelepot.bot.Bot, update, user_record, data
bot=bot,
update=update, user_record=user_record,
)
for admin in admins:
for admin in bot.administrators:
when += datetime.timedelta(seconds=1)
asyncio.ensure_future(
bot.send_message(
chat_id=admin,
chat_id=admin['telegram_id'],
text=suggestion_message,
parse_mode='HTML'
)
@ -248,8 +239,10 @@ async def _see_suggestions(bot: davtelepot.bot.Bot, update, user_record):
)
def init(telegram_bot: davtelepot.bot.Bot, suggestion_messages=default_suggestion_messages):
def init(telegram_bot: davtelepot.bot.Bot, suggestion_messages=None):
"""Set suggestion handling for `bot`."""
if suggestion_messages is None:
suggestion_messages = default_suggestion_messages
telegram_bot.messages['suggestions'] = suggestion_messages
suggestion_prefixes = (
list(suggestion_messages['suggestions_command']['reply_keyboard_button'].values())

1
setup.py
View File

@ -59,6 +59,7 @@ setuptools.setup(
'bs4',
'dataset',
],
python_requires='>=3.5',
classifiers=[
"Development Status :: 5 - Production/Stable",
"Environment :: Console",