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

Allow users to link an email to their Telegram account. If their email is listed in the patrons list, send them an invite link

Browse Source
This commit is contained in:
Davte 2021-06-20 00:42:20 +02:00
parent 66745a15ed
commit fc0b29709a
Signed by: Davte
GPG Key ID: 209AE674A0007425
6 changed files with 297 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
bic_bot/authorization.py
View File

@ -40,7 +40,11 @@ async def set_chat(bot: davtelepot.bot.Bot, update: dict, language: str):
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')

4
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, patreon
from . import authorization, email_verification, patreon
from .messages import language_messages, supported_languages
current_path = os.path.dirname(
@ -88,9 +88,11 @@ def run():
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)

218
bic_bot/email_verification.py Normal file
View File

@ -0,0 +1,218 @@
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)

66
bic_bot/messages.py
View File

@ -55,6 +55,72 @@ language_messages = {
}
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': "",

7
bic_bot/patreon.py
View File

@ -4,6 +4,7 @@ import os
import davtelepot
from .email_verification import invite_new_patrons
from .messages import patreon_messages
@ -42,6 +43,7 @@ async def handle_patrons_list_file(bot: davtelepot.bot.Bot, update: dict, langua
['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)
@ -63,12 +65,13 @@ async def handle_new_members(bot, update):
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'])
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 not of their joining
else: # Otherwise, take note of their joining
bot.db['patrons'].upsert(
dict(
user_id=user_record['id'],

1
requirements.txt
View File

@ -1 +1,2 @@
aiosmtplib
davtelepot