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

Compare commits

base: Davte:1b0acad7b48ad1f890635b87f34068c7bbb3a2d1
Branches Tags
Davte:main
..
compare: Davte:8fa1737faf6f95a7e51a4f5724d879fb61713ee1
Branches Tags
Davte:main

No commits in common. "1b0acad7b48ad1f890635b87f34068c7bbb3a2d1" and "8fa1737faf6f95a7e51a4f5724d879fb61713ee1" have entirely different histories.
1b0acad7b4 ... 8fa1737faf

12 changed files with 2 additions and 815 deletions
Show Stats Expand all files Collapse all files

68
.gitignore vendored
View File

@ -1,68 +0,0 @@
# local_* files
local_*
my_config.sh
# Data folder
bic_bot/data/*
!bic_bot/data/__init__.py
# ---> Python
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
.ipynb_checkpoints/
# C extensions
*.so
# Distribution / packaging
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
*.egg-info/
.installed.cfg
*.egg
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*,cover
# Translations
*.mo
*.pot
# Django stuff:
*.log
# Sphinx documentation
docs/_build/
# PyBuilder
target/

39
README.md
View File

@ -1,38 +1,3 @@
# bic_bot
# bibot
Breaking Italy Club bot
## User manual
Basic users just need to follow these simple instructions.
* Start the bot and write `/email your_email@example.com`
* Click the verification link in the mail (also check the SPAM folder!)
* Wait for the invitation link. It will be available as soon as your patron status is verified by the staff
## Installation
This needs to be done only one time server-side.
* 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 that chat
* Send the bot a csv file named "patrons.csv" with patrons' emails
```
person@example.com,
someone@gmail.com,
[...]
```
The csv file must be sent periodically.
* [NOT YET IMPLEMENTED] Using Patreon's API, the patrons list will be automatically downloaded from the server
## 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.
* 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](http://breakingitaly.club) staff.
Breaking Italy bot

0
bic_bot/__init__.py
View File

3
bic_bot/__main__.py
View File

@ -1,3 +0,0 @@
from . import bot
bot.run()

63
bic_bot/authorization.py
View File

@ -1,63 +0,0 @@
import davtelepot
from .messages import elevation_messages
async def elevate_to_admin(bot: davtelepot.bot.Bot, language: str, user_record: dict):
if len(bot.administrators) < 1:
user_record['privileges'] = 1
bot.db['users'].upsert(
user_record,
['id']
)
return bot.get_message('elevation', 'granted', language=language)
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')
else:
table = telegram_bot.db.get_table('information')
if 'name' not in table.columns:
table.create_column('name', telegram_bot.db.types.string(25))
if 'value' not in table.columns:
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)
@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)

98
bic_bot/bot.py
View File

@ -1,98 +0,0 @@
import logging
import os
import sys
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, email_verification, patreon
from .messages import language_messages, supported_languages
current_path = os.path.dirname(
os.path.abspath(
__file__
)
)
def append_to_passwords_file(line_to_append):
with open(f'{current_path}/data/passwords.py', 'a') as passwords_file:
passwords_file.write(line_to_append)
try:
from .data.passwords import telegram_token
if not telegram_token:
raise ImportError
except ImportError as e:
try:
telegram_token = input("Enter telegram bot API token:\n"
"For more information: https://core.telegram.org/bots/\n\t\t")
append_to_passwords_file(f'telegram_token = "{telegram_token}"\n')
except KeyboardInterrupt:
logging.error("Telegram bot token not provided, aborting...")
sys.exit(1)
bic_bot = davtelepot.bot.Bot(token=telegram_token,
database_url=f'bic_bot/data/bot.db')
def run():
try:
from .data.config import log_file_name
except ImportError:
log_file_name = 'bic_bot.log'
try:
from .data.config import errors_file_name
except ImportError:
errors_file_name = 'bic_bot.errors.log'
log_file = f"{current_path}/data/{log_file_name}"
errors_file = f"{current_path}/data/{errors_file_name}"
# Outputs the log in console, log_file and errors_file
# Log formatter: datetime, module name (filled with spaces up to 15
# characters), logging level name (filled to 8), message
# noinspection SpellCheckingInspection
log_formatter = logging.Formatter(
"%(asctime)s [%(module)-15s %(levelname)-8s] %(message)s",
style='%'
)
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)
file_handler = logging.FileHandler(log_file, mode="a", encoding="utf-8")
file_handler.setFormatter(log_formatter)
file_handler.setLevel(logging.DEBUG)
root_logger.addHandler(file_handler)
file_handler = logging.FileHandler(errors_file, mode="a", encoding="utf-8")
file_handler.setFormatter(log_formatter)
file_handler.setLevel(logging.ERROR)
root_logger.addHandler(file_handler)
console_handler = logging.StreamHandler()
console_handler.setFormatter(log_formatter)
console_handler.setLevel(logging.DEBUG)
root_logger.addHandler(console_handler)
bic_bot.set_path(current_path)
bic_bot.set_class_log_file_name(log_file_name)
bic_bot.set_class_errors_file_name(errors_file_name)
bic_bot.set_unknown_command_message(
unknown_command_message
)
bic_bot.set_authorization_denied_message(
authorization_denied_message
)
davtelepot.authorization.init(telegram_bot=bic_bot)
authorization.init(telegram_bot=bic_bot)
davtelepot.administration_tools.init(telegram_bot=bic_bot)
email_verification.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)
davtelepot.helper.init(telegram_bot=bic_bot)
exit_code = bic_bot.run()
sys.exit(exit_code)

0
bic_bot/data/__init__.py
View File

218
bic_bot/email_verification.py
View File

@ -1,218 +0,0 @@
import asyncio
import datetime
import logging
import os
import re
import string
from email.mime.text import MIMEText
import aiosmtplib
import davtelepot
email_regex = re.compile(r'[A-z\-_0-9]{1,65}@[A-z\-_0-9]{1,320}\.[A-z]{2,24}', flags=re.I)
validity_timedelta = datetime.timedelta(minutes=15)
current_path = os.path.dirname(
os.path.abspath(
__file__
)
)
def append_to_passwords_file(line_to_append):
with open(f'{current_path}/data/passwords.py', 'a') as passwords_file:
passwords_file.write(line_to_append)
try:
from .data.passwords import mail_server_address
except ImportError:
mail_server_address = input("Enter the mail server address:\n\t\t")
append_to_passwords_file(f'mail_server_address = "{mail_server_address}"\n')
try:
from .data.passwords import mail_username
except ImportError:
mail_username = input("Enter the mail server username:\n\t\t")
append_to_passwords_file(f'mail_username = "{mail_username}"\n')
try:
from .data.passwords import mail_password
except ImportError:
mail_password = input("Enter the mail server password:\n\t\t")
append_to_passwords_file(f'mail_password = "{mail_password}"\n')
async def invite_new_patrons(bot: davtelepot.bot.Bot):
invite_link = await get_invite_link(bot=bot)
for record in bot.db.query("SELECT u.*, p.id patron_id, c.id confirm_id FROM patrons p "
"LEFT JOIN users u ON u.id = p.user_id "
"LEFT JOIN confirmation_codes c ON c.user_id = u.id "
"WHERE p.tier AND (p.is_in_chat IS NULL OR p.is_in_chat = 0) AND NOT c.notified"):
try:
await bot.send_message(
chat_id=record['telegram_id'],
text=bot.get_message('patreon', 'confirmation_email', 'notification',
invite_link=invite_link,
user_record=record)
)
bot.db['confirmation_codes'].update(
dict(id=record['confirm_id'], notified=True),
['id']
)
except Exception as e:
logging.error(e)
async def send_confirmation_email(recipient_email: str,
message_text: str,
message_subject: str = None):
message = MIMEText(message_text, 'html')
message['To'] = recipient_email
message['From'] = mail_username
message['Subject'] = message_subject
await aiosmtplib.send(message, hostname=mail_server_address,
port=465, use_tls=True,
sender=mail_username,
username=mail_username, password=mail_password)
async def link_email(bot: davtelepot.bot.Bot, update: dict, user_record: dict, language: str):
patron_record = bot.db['patrons'].find_one(user_id=user_record['id'],
order_by=['-id'])
if patron_record: # If the user is already verified, send invite link
return await verify(bot=bot, update=update, user_record=user_record, language=language)
email_address = email_regex.search(update['text'].lower())
telegram_account = davtelepot.utilities.get_user(record=user_record)
confirmation_code = davtelepot.utilities.get_secure_key(
allowed_chars=(string.ascii_uppercase + string.ascii_lowercase + string.digits),
length=12
)
if email_address is None:
return bot.get_message('patreon', 'confirmation_email', 'invalid_email_address')
email_address = email_address.group()
bot.db['confirmation_codes'].upsert(
dict(
user_id=user_record['id'],
email=email_address,
code=confirmation_code,
expiry=datetime.datetime.now() + validity_timedelta,
times_tried=0,
notified=0
),
['email']
)
message_text = bot.get_message('patreon', 'confirmation_email', 'text',
confirmation_link=f"https://t.me/{bot.name}?start=00verify_{confirmation_code}",
confirmation_code=confirmation_code,
telegram_account=telegram_account,
bot=bot,
language=language)
message_subject = bot.get_message('patreon', 'confirmation_email', 'subject',
language=language)
await send_confirmation_email(recipient_email=email_address, message_text=message_text,
message_subject=message_subject)
return bot.get_message('patreon', 'confirmation_email', 'sent',
language=language)
async def verify(bot: davtelepot.bot.Bot, update: dict, user_record: dict, language: str):
if not ('chat' in update and update['chat']['id'] > 0): # Ignore public messages
return
patron_record = bot.db['patrons'].find_one(user_id=user_record['id'],
order_by=['-id'])
text = update['text']
confirmation_code = re.findall(r'(?:00verif.{1,3}_)?([A-z0-9]{12})', text)
confirmation_record = bot.db['confirmation_codes'].find_one(user_id=user_record['id'],
order_by=['-id'])
invite_link = None
if patron_record is not None and patron_record['tier']:
invite_link = await bot.shared_data['get_invite_link'](bot=bot)
message_fields = ['patreon', 'confirmation_email', 'send_invite_link']
elif patron_record is not None:
message_fields = ['patreon', 'confirmation_email', 'wait_for_invite_link']
elif (confirmation_record
and davtelepot.utilities.str_to_datetime(confirmation_record['expiry'])
< datetime.datetime.now()) or (confirmation_record and confirmation_record['times_tried'] > 2):
message_fields = ['patreon', 'confirmation_email', 'expired_code']
elif confirmation_code and confirmation_record is not None and confirmation_code[0] == confirmation_record['code']:
bot.db['patrons'].upsert(
dict(
email=confirmation_record['email'],
user_id=user_record['id']
),
'email'
)
message_fields = ['patreon', 'confirmation_email', 'wait_for_invite_link']
else:
message_fields = ['patreon', 'confirmation_email', 'confirmation_failed']
confirmation_record['times_tried'] += 1
bot.db['confirmation_codes'].update(
confirmation_record,
['id']
)
return bot.get_message(*message_fields, invite_link=invite_link, language=language)
async def change_invite_link(bot: davtelepot.bot.Bot, old_link: str, sleep_time: int = 0):
await asyncio.sleep(sleep_time)
if old_link == bot.db['information'].find_one(name='invite_link')['value']:
await bot.exportChatInviteLink(chat_id=bot.shared_data['bic_chat_id'])
async def get_invite_link(bot: davtelepot.bot.Bot):
invite_link_expiry = bot.db['information'].find_one(name='invite_link_expiry')
if (invite_link_expiry is None
or davtelepot.utilities.str_to_datetime(invite_link_expiry['value'])
<= datetime.datetime.now()):
invite_link = await bot.exportChatInviteLink(chat_id=bot.shared_data['bic_chat_id'])
bot.db['information'].upsert(
dict(name='invite_link_expiry',
value=datetime.datetime.now() + datetime.timedelta(hours=1),
),
['name']
)
bot.db['information'].upsert(
dict(name='invite_link',
value=invite_link),
['name']
)
invite_link = bot.db['information'].find_one(name='invite_link')['value']
# Inactivate the invite link after 60 minutes
asyncio.ensure_future(change_invite_link(bot=bot, old_link=invite_link, sleep_time=3600))
return invite_link
def init(telegram_bot: davtelepot.bot.Bot):
telegram_bot.shared_data['get_invite_link'] = get_invite_link
asyncio.ensure_future(get_invite_link(bot=telegram_bot))
asyncio.ensure_future(invite_new_patrons(bot=telegram_bot))
if 'confirmation_codes' not in telegram_bot.db.tables:
table = telegram_bot.db.create_table('confirmation_codes')
else:
table = telegram_bot.db.get_table('confirmation_codes')
if 'user_id' not in table.columns:
table.create_column('user_id', telegram_bot.db.types.integer)
if 'email' not in table.columns:
table.create_column('email', telegram_bot.db.types.string(320))
if 'code' not in table.columns:
table.create_column('code', telegram_bot.db.types.string(12))
if 'times_tried' not in table.columns:
table.create_column('times_tried', telegram_bot.db.types.integer)
if 'expiry' not in table.columns:
table.create_column('expiry', telegram_bot.db.types.datetime)
if 'notified' not in table.columns:
table.create_column('notified', telegram_bot.db.types.boolean)
@telegram_bot.command('/email',
authorization_level='everybody')
async def _link_email(bot, update, user_record, language):
return await link_email(bot=bot, update=update,
user_record=user_record, language=language)
@telegram_bot.command('/verify',
aliases=['verifica', '/verifica', '00verifica', '00verify'],
authorization_level='everybody')
async def _verify(bot, update, user_record, language):
return await verify(bot=bot, update=update,
user_record=user_record, language=language)

147
bic_bot/messages.py
View File

@ -1,147 +0,0 @@
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! 🚫",
},
'granted': {
'en': "You have been elevated to Founder! 👑",
'it': "Ora sei Fondatore! 👑",
},
}
language_messages = {
'language_command': {
'name': {
'en': "/language",
'it': "/lingua"
},
'reply_keyboard_button': {
'en': "Language 🗣",
'it': "Lingua 🗣"
},
'alias': {
'en': "Language 🗣",
'it': "Lingua 🗣"
},
'description': {
'en': "Change language settings",
'it': "Cambia le impostazioni della lingua"
}
},
'language_button': {
'description': {
'en': "Change language settings",
'it': "Cambia le impostazioni della lingua"
},
'language_set': {
'en': "Selected language: English 🇬🇧",
'it': "Lingua selezionata: Italiano 🇮🇹"
}
},
'language_panel': {
'text': {
'en': "<b>Choose a language</b>",
'it': "<b>Seleziona una lingua</b>"
}
}
}
patreon_messages = {
'confirmation_email': {
'confirmation_failed': {
'en': "Wrong confirmation code. Check your email and try again: /verify",
'it': "Codice di verifica errato. Controlla la mail e prova di nuovo: /verifica",
},
'expired_code': {
'en': "The code has expired. Please try again: /email",
'it': "Codice di verifica scaduto. Ottienine un altro: /email",
},
'invalid_email_address': {
'en': "The email you provided is invalid. Please try again.",
'it': "La mail inserita non è valida. Per favore riprova.",
},
'notification': {
'en': "🔔 Your status as patron has been verified.\n"
"You can now join the chat clicking this temporary link:\n"
"{invite_link}\n\n"
"If the link has expired, click /verify to get a new one.",
'it': "🔔 Il tuo ruolo di patron è stato confermato.\n"
"Ora puoi unirti alla chat usando questo link temporaneo:\n"
"{invite_link}\n\n"
"Se il link è scaduto, clicka /verifica per ottenerne uno nuovo.",
},
'send_invite_link': {
'en': "You can join the chat clicking this temporary link:\n"
"{invite_link}\n\n"
"If the link has expired, click /verify to get a new one.",
'it': "Puoi unirti alla chat usando questo link temporaneo:\n"
"{invite_link}\n\n"
"Se il link è scaduto, clicka /verifica per ottenerne uno nuovo.",
},
'sent': {
'en': "✉️ Check your email, including the SPAM folder 🗑",
'it': "✉️ Controlla la mail, compresa la cartella SPAM 🗑",
},
'subject': {
'en': "Confirm your email",
'it': "Conferma la tua email",
},
'text': {
'en': "<h1>Welcome!</h1>\n"
"<p>Click <a href=\"{confirmation_link}\">this link</a> or send <a href=\"https://t.me/{"
"bot.name}\">@{bot.name}</a> <code>/verify {confirmation_code}</code> to link this email address to "
"the telegram account {telegram_account}.</p> <p>As soon as your email will be listed among "
"patrons, you will receive the invite link to the Breaking Italy Club chat! Just wait a few days "
"=)</p>",
'it': "<h1>Bevenutə!</h1>\n"
"<p>Clicka su <a href=\"{confirmation_link}\">questo link</a> oppure scrivi <code>/verifica {"
"confirmation_code}</code> a <a href=\"https://t.me/{bot.name}\">@{bot.name}</a> per associare "
"questo indirizzo email all'account telegram {telegram_account}.</p> "
"<p>Appena la tua email sarà inserita nell'elenco dei patron riceverai il link di invito nella chat "
"del Breaking Italy Club! Pazienta qualche giorno =)</p>",
},
'wait_for_invite_link': {
'en': "Your email has been linked with this Telegram account.\n"
"However, it is not included in the patrons list yet.\n"
"Please wait a few days, I will notify you when you can join the chat. 🔔\n"
"When you are listed among patrons, you can get a temporary invite link to join the chat at any "
"time: just click /verify",
'it': "La tua mail è stata associata a questo account Telegram.\n"
"Tuttavia, ancora non compare nella lista dei patron.\n"
"Pazienta qualche giorno, ti invierò un messaggio non appena potrai unirti alla chat. 🔔\n"
"Quando comparirai nella lista dei patron, potrai ricevere un link temporaneo di invito per "
"aggiungerti alla chat in qualiasi momento: basterà clickare /verifica",
},
},
'join_chat': {
'en': "Thank you for your Patreon subscription! You may enter ",
'it': "",
},
'list_updated': {
'en': "Patrons whitelist updated ✅",
'it': "Lista dei patron aggiornata ✅",
},
'whitelist_updated': {
'en': "Patrons old whitelist updated ✅",
'it': "Vecchia lista dei patron aggiornata ✅",
},
}
supported_languages = {
'en': {
'flag': '🇬🇧',
'name': 'English'
},
'it': {
'flag': '🇮🇹',
'name': 'Italiano'
}
}

150
bic_bot/patreon.py
View File

@ -1,150 +0,0 @@
import asyncio
import logging
import os
import davtelepot
from .email_verification import invite_new_patrons
from .messages import patreon_messages
def is_whitelist_file(update: dict):
return (update['document']['mime_type'] == 'text/csv'
and 'white' in update['document']['file_name'])
async def handle_whitelist_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')
whitelist = 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("DELETE FROM whitelisted_usernames")
for record in whitelist:
db['whitelisted_usernames'].upsert(
dict(username=record['username']),
['username']
)
return bot.get_message('patreon', 'whitelist_updated', language=language)
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, u.username username, "
"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:
if not bot.db['whitelisted_usernames'].find_one(username=record['username']):
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))
asyncio.ensure_future(invite_new_patrons(bot=bot))
return bot.get_message('patreon', 'list_updated', language=language)
async def handle_left_chat_event(bot, update):
if update['chat']['id'] != bot.shared_data['bic_chat_id']:
return
user_record = bot.db['users'].find_one(telegram_id=update['left_chat_member']['id'])
bot.db['patrons'].upsert(
dict(
user_id=user_record['id'],
is_in_chat=False
),
['user_id']
)
async def handle_new_members(bot, update):
if update['chat']['id'] != bot.shared_data['bic_chat_id']:
return
for member in update['new_chat_members']:
user_record = bot.db['users'].find_one(telegram_id=member['id'])
patron_record = bot.db['patrons'].find_one(user_id=user_record['id'],
order_by=['-id'])
# If user is not white-listed, kick them
if patron_record is None or not patron_record['tier']:
await bot.kickChatMember(chat_id=bot.shared_data['bic_chat_id'],
user_id=user_record['telegram_id'])
else: # Otherwise, take note of their joining
bot.db['patrons'].upsert(
dict(
user_id=user_record['id'],
is_in_chat=True
),
['user_id']
)
def init(telegram_bot: davtelepot.bot.Bot):
telegram_bot.messages['patreon'] = patreon_messages
# === whitelisted_usernames table
if 'whitelisted_usernames' not in telegram_bot.db.tables:
table = telegram_bot.db.create_table('whitelisted_usernames')
else:
table = telegram_bot.db.get_table('whitelisted_usernames')
if 'username' not in table.columns:
table.create_column('username', telegram_bot.db.types.string(50))
# === patrons table
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)
@telegram_bot.document_handler(condition=is_whitelist_file,
authorization_level='administrator')
async def _handle_whitelist_file(bot: davtelepot.bot.Bot,
update: dict,
language: str):
return await handle_whitelist_file(bot=bot, update=update,
language=language)
telegram_bot.set_message_handler('new_chat_members', handle_new_members)
telegram_bot.set_message_handler('left_chat_member', handle_left_chat_event)

2
requirements.txt
View File

@ -1,2 +0,0 @@
aiosmtplib
davtelepot

29
run_me.sh
View File

@ -1,29 +0,0 @@
#!/bin/bash
# This file must be executable, otherwise the service cannot run it.
# Run `$python_script` while exit code is 65.
# At each iteration, pull news from repository and update dependencies.
# Get current directory
this_script_directory=$(cd `dirname $0` && pwd);
cd "$this_script_directory";
if [ ! -d "$this_script_directory/env" ]; then
python3 -m venv env;
env/bin/pip install -r requirements.txt;
fi
python_virtual_environment="$this_script_directory/env/bin"
echo "Python script will be run while it exits with value===65.";
i=65;
while [ $i -eq 65 ]
do
echo "Pulling from repository..."
git pull;
echo "Updating dependencies";
"$python_virtual_environment/pip" install --upgrade --no-cache-dir \
--no-deps davtelepot;
echo "Running python script";
"$python_virtual_environment/python" -m bic_bot;
i=$?;
done