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

Package update notifier completed

Browse Source
This commit is contained in:
Davte
2020-04-27 13:45:25 +02:00
parent 16014c5a42
commit 1e2316e748
6 changed files with 211 additions and 141 deletions
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"

116
davtelepot/administration_tools.py
View File

@@ -18,7 +18,7 @@ import logging
from sqlalchemy.exc import ResourceClosedError
# Project modules
from . import bot as davtelepot_bot, messages, __version__ as version
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,
@@ -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,64 +793,74 @@ async def get_version():
last_commit = f"{e}"
if last_commit.startswith("fatal: not a git repository"):
last_commit = "-"
davtelepot_version = 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
@@ -858,6 +868,7 @@ async def notify_new_version(bot):
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',
@@ -872,15 +883,42 @@ async def get_package_updates(bot: davtelepot_bot,
new_version = web_page['info']['version']
current_version = package.__version__
if new_version != current_version:
print(f"New version of {package}: "
f"<code>{current_version}</code> > "
f"<code>{new_version}</code>")
# TODO notify administrators
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):
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
@@ -902,8 +940,6 @@ def init(telegram_bot, talk_messages=None, admin_messages=None):
for command in ['stop', 'restart', 'maintenance']
]
asyncio.ensure_future(get_package_updates(telegram_bot))
@telegram_bot.additional_task(when='BEFORE')
async def load_talking_sessions():
sessions = []

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())