You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

195 lines
6.5 KiB

This is a telegram bot to fetch random word definition.
Currently supported commands are:
word - get a random word with definition
import datetime
import logging
import os
import psycopg2
import requests
import sqlite3
import telegram
import telegram.ext
import time
from bs4 import BeautifulSoup
from dotenv import load_dotenv
from telegram import ParseMode, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, CallbackQueryHandler
log_file_name = f"logs/wordbot_log_{'%Y_%m_%d')}.log"
logging.basicConfig(filename=log_file_name, filemode='a', format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
logger = logging.getLogger()
def connect_db():
database = os.getenv('DATABASE')
if database == 'postgres':
con = psycopg2.connect(
print("connected to postgres")
elif database == 'sqlite':
con = sqlite3.connect(os.getenv('DB_FILE'))
print("connected to sqlite")
return con
def start(
update: telegram.update.Update,
context: telegram.ext.callbackcontext.CallbackContext
) -> None:
""" Handles the start command to the bot. Responds with help content """
start_text = "Welcome to Random Word Bot. To fetch a word, just send /word", text=start_text)
def slugify(message: str) -> str:
""" Adds relevant escape characters as per Telegram's markdown parsing rules """
message = message.replace(".", "\\.").replace("*", "\\*").replace("(", "\\(").replace(")", "\\)").replace("-", "\\-")
return message
def scrape_def(word: str) -> dict:
""" Scrapes the definition and audio file link of a word from"""
ox_url = f'{word}?locale=en'
def_page = requests.get(ox_url)
if def_page.status_code != 200:
print(f'Word not found: {selected_word}')
return None
my_dict = {}
def_soup = BeautifulSoup(def_page.content, 'html.parser')
word = def_soup.find_all("span", {"class": "hw"})
my_dict['word'] = slugify(word[0].text.capitalize())
grammatical_type = def_soup.find_all("span", {"class": "pos"})
my_dict['type'] = u'\u2022 ' + grammatical_type[0].text.capitalize()
definition = def_soup.find_all("span", {"class": "ind"})
audio = def_soup.find_all("audio")
my_dict['audio'] = audio[0]['src'].replace('', '')
my_dict['definition'] = []
for i, j in enumerate(definition):
return my_dict
def word_def(
update: telegram.update.Update,
context: telegram.ext.callbackcontext.CallbackContext
) -> None:
""" Responds to the /word command on the bot
Fetches the definition from
Constructs the message and callback keyboard
Sends the message
con = connect_db()
if con is None:
print("Db connection failed")
word_query = 'select * from gre_words order by random() limit 1'
while True:
cur = con.cursor()
row = cur.fetchall()
selected_word = row[0][1].lower()
my_dict = scrape_def(selected_word)
if my_dict is not None:
msg = "*{word}*\n_{type}_\n{defn}".format(
word=my_dict['word'], type=slugify(my_dict['type']),
defn='\n'.join([slugify(i) for i in my_dict['definition']]))
button_callback_data_audio = f"p~{my_dict['word']}~{my_dict['audio']}"
button_callback_data_synonyms = f"s~{my_dict['word']}"
keyboard = [
InlineKeyboardButton("Pronunciation", callback_data=button_callback_data_audio),
InlineKeyboardButton("Synonyms", callback_data=button_callback_data_synonyms)
reply_markup = InlineKeyboardMarkup(keyboard)
msg_response =, text=msg, parse_mode=ParseMode.MARKDOWN_V2, reply_markup=reply_markup)
def get_synonyms(word: str) -> str:
""" Fetches the synonyms of a word from"""
syn_url = '{}?locale=en'
response = requests.get(syn_url.format(word.lower()))
soup = BeautifulSoup(response.content, 'html.parser')
synonyms = soup.find_all(['strong', 'span'], {'class': 'syn'})
synonym_list = [f'\n*{slugify(i.string)}*' if == 'strong' else slugify(i.string) for i in synonyms]
return (''.join(synonym_list))
def button(
update: telegram.update.Update,
context: telegram.ext.callbackcontext.CallbackContext
) -> None:
""" Responds to callback buttons in the original message.
If the callback type is audio, then responds with already available audio URL
If the callback type is synonyms, then fetchs the synonyms and responds
query = update.callback_query
data =
request_type = data.split('~')[0]
word = data.split('~')[1]
if request_type == 'p':
audio = f"{data.split('~')[2]}", audio=audio, reply_to_message_id=query.message.message_id)
elif request_type == 's':
synonyms = get_synonyms(word)
print(synonyms), text=synonyms, reply_to_message_id=query.message.message_id, parse_mode=ParseMode.MARKDOWN_V2)
def main() -> None:
""" Runs the bot and keeps it running """
wordbot_token = os.getenv('WORDBOT_TOKEN')
updater = Updater(token=wordbot_token, use_context=True)
dispatcher = updater.dispatcher
dispatcher.add_handler(CommandHandler('start', start))
dispatcher.add_handler(CommandHandler('word', word_def))
if __name__ == '__main__':
print('word_bot is running. Press ctrl+c to stop.')