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

Whenever an email is removed from the white list, the linked Telegram account will be kicked out.

Browse Source
This commit is contained in:
Davte 2021-06-19 18:07:56 +02:00
parent 706de9acf5
commit 650b531ad4
Signed by: Davte
GPG Key ID: 209AE674A0007425
5 changed files with 154 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
README.md
View File

@ -3,8 +3,29 @@
Breaking Italy Club bot
## Instructions
```bash
git clone https://gogs.davte.it/Davte/bic_bot.git
cd bic_bot
bash ./run_me.sh
```
* Get a Telegram bot API token ([@botfather](https://t.me/botfather)), allow groups and set the bot [group privacy](https://core.telegram.org/bots#privacy-mode) to off.
* Install bic_bot
```bash
git clone https://gogs.davte.it/Davte/bic_bot.git
cd bic_bot
bash ./run_me.sh
```
* Add the bot to the BIC chat as administrator
* Write `/set_chat` in the chat
* Send a csv file to the bot named "patrons.csv" with patreons emails
```csv
person@example.com,
someone@gmail.com,
[...]
```
* [NOT YET IMPLEMENTED] Using Patreon's API, the patrons list will be automatically downloaded from the server
* Users will be able to link their email to their Telegram account automatically. A confirmation email will help to verify the association
## Functioning
* Whenever a person joins the chat, the bot will check whether their Telegram account is linked to an email listed in the patrons list. If not, the user will be kicked out.
* Whenever an email is removed from the white list, the linked Telegram account will be kicked out.
* [NOT YET IMPLEMENTED] Any user can automatically link their email to their Telegram account. If their email is white-listed, they can join the chat.
## Credits
This [davtelepot-based](https://gogs.davte.it/davte/davtelepot) bot has been developed by [@Davte](https://www.davte.it).
No official endorsement has been given by Breaking Italy.

38
bic_bot/authorization.py
View File

@ -14,10 +14,46 @@ async def elevate_to_admin(bot: davtelepot.bot.Bot, language: str, user_record:
return bot.get_message('elevation', 'denied', language=language)
async def set_chat(bot: davtelepot.bot.Bot, update: dict, language: str):
await bot.delete_message(update=update)
if 'chat' in update and 'id' in update['chat']:
chat_id = update['chat']['id']
if chat_id > 0:
return bot.get_message('elevation',
'chat_not_set',
language=language)
bot.db['information'].upsert(
dict(
name='bic_chat_id',
value=chat_id
),
['name']
)
bot.shared_data['bic_chat_id'] = chat_id
await bot.send_message(chat_id=update['from']['id'],
text=bot.get_message('elevation',
'chat_set',
chat_id=chat_id,
language=language))
def init(telegram_bot: davtelepot.bot.Bot):
if 'information' not in telegram_bot.db.tables:
table = telegram_bot.db.create_table('information')
table.create_column('name', telegram_bot.db.types.string(25))
table.create_column('value', telegram_bot.db.types.string(100))
telegram_bot.messages['elevation'] = elevation_messages
bic_chat_id = telegram_bot.db['information'].find_one(name='bic_chat_id')
if bic_chat_id is not None:
bic_chat_id = int(bic_chat_id['value'])
telegram_bot.shared_data['bic_chat_id'] = bic_chat_id
@telegram_bot.command(command='elevate',
authorization_level='everybody')
async def _elevate_to_admin(bot, language, user_record):
return await elevate_to_admin(bot=bot, language=language, user_record=user_record)
return await elevate_to_admin(bot=bot, language=language, user_record=user_record)
@telegram_bot.command(command='set_chat',
authorization_level='admin')
async def _set_chat(bot, update, language):
return await set_chat(bot=bot, update=update, language=language)

3
bic_bot/bot.py
View File

@ -6,7 +6,7 @@ import davtelepot.bot
from davtelepot.messages import (default_unknown_command_message as unknown_command_message,
default_authorization_denied_message as authorization_denied_message)
from . import authorization
from . import authorization, patreon
from .messages import language_messages, supported_languages
current_path = os.path.dirname(
@ -88,6 +88,7 @@ def run():
davtelepot.authorization.init(telegram_bot=bic_bot)
authorization.init(telegram_bot=bic_bot)
davtelepot.administration_tools.init(telegram_bot=bic_bot)
patreon.init(telegram_bot=bic_bot)
davtelepot.languages.init(telegram_bot=bic_bot,
language_messages=language_messages,
supported_languages=supported_languages)

21
bic_bot/messages.py
View File

@ -1,4 +1,12 @@
elevation_messages = {
'chat_not_set': {
'en': "This chat cannot be set as BIC chat! ❌",
'it': "Questa chat non può essere impostata come chat del BIC ❌",
},
'chat_set': {
'en': "BIC chat ID set ✅\n🏷 Chat ID: {chat_id}",
'it': "ID della chat del BIC salvato ✅\n🏷 ID chat: {chat_id}",
},
'denied': {
'en': "You have no right elevate yourself to Founder! 🚫",
'it': "Non hai diritto a ottenere i privilegi di Fondatore! 🚫",
@ -46,6 +54,17 @@ language_messages = {
}
}
patreon_messages = {
'join_chat': {
'en': "Thank you for your Patreon subscription! You may enter ",
'it': "",
},
'list_updated': {
'en': "Patrons white list updated ✅",
'it': "Lista dei patron aggiornata ✅",
},
}
supported_languages = {
'en': {
'flag': '🇬🇧',
@ -55,4 +74,4 @@ supported_languages = {
'flag': '🇮🇹',
'name': 'Italiano'
}
}
}

69
bic_bot/patreon.py Normal file
View File

@ -0,0 +1,69 @@
import asyncio
import logging
import os
import davtelepot
from .messages import patreon_messages
def is_patrons_list_file(update: dict):
return (update['document']['mime_type'] == 'text/csv'
and 'patron' in update['document']['file_name'])
async def kick_unlisted_patrons(bot: davtelepot.bot.Bot):
for record in bot.db.query("SELECT u.telegram_id telegram_id, p.id patron_id FROM patrons p "
"LEFT JOIN users u ON u.id = p.user_id "
"WHERE p.tier IS NULL AND p.is_in_chat = 1"):
try:
await bot.kickChatMember(chat_id=bot.shared_data['bic_chat_id'],
user_id=record['telegram_id'])
bot.db.query(f"UPDATE patrons SET is_in_chat = 0 WHERE id = {record['patron_id']}")
except Exception as e:
logging.error(e)
async def handle_patrons_list_file(bot: davtelepot.bot.Bot, update: dict, language: str):
file_name = davtelepot.utilities.get_secure_key(length=12)
while os.path.isfile(f'{bot.path}/data/{file_name}'):
file_name = davtelepot.utilities.get_secure_key(length=12)
await bot.download_file(file_id=update['document']['file_id'],
file_name=file_name,
path=f'{bot.path}/data')
patrons_list = davtelepot.utilities.csv_read(f'{bot.path}/data/{file_name}')
os.remove(f'{bot.path}/data/{file_name}')
with bot.db as db: # Unique SQL transaction
db.query("UPDATE patrons SET tier = NULL")
for patron in patrons_list:
db['patrons'].upsert(
dict(tier=1,
email=patron['email']),
['email']
)
asyncio.ensure_future(kick_unlisted_patrons(bot=bot))
return bot.get_message('patreon', 'list_updated', language=language)
def init(telegram_bot: davtelepot.bot.Bot):
telegram_bot.messages['patreon'] = patreon_messages
if 'patrons' not in telegram_bot.db.tables:
table = telegram_bot.db.create_table('patrons')
else:
table = telegram_bot.db.get_table('patrons')
if 'email' not in table.columns:
table.create_column('email', telegram_bot.db.types.string(320))
if 'tier' not in table.columns:
table.create_column('tier', telegram_bot.db.types.integer)
if 'user_id' not in table.columns:
table.create_column('user_id', telegram_bot.db.types.integer)
if 'is_in_chat' not in table.columns:
table.create_column('is_in_chat', telegram_bot.db.types.boolean)
@telegram_bot.document_handler(condition=is_patrons_list_file,
authorization_level='administrator')
async def _handle_patrons_list_file(bot: davtelepot.bot.Bot,
update: dict,
language: str):
return await handle_patrons_list_file(bot=bot, update=update,
language=language)