trying to create a bot at replit;
for managing interviews that can:
-
On the first start, Create a dictionary of all members in the group to prevent cheating by users who change their usernames, set a field as previous interviews with default value 0. On next start, refer to this dictionary to tally poll responses. Connected to google sheets
-
On start, ask the interview creator the following options:
Text for the poll, Time limit of the poll, time limit to send the profile, minimum previous interviews.
-
Create a Poll, based on the input, provide the text and time limit of the poll, with these options:
Yes, no, mentor poll check. -
After time limit, From yes responses, select a random responder who’s number of previous interviews are equal or less than the input provided, compare from the already created dictionary. If no responder matches, select at random.
-
Tag the selected user to send a "profile brief" within the time limit to send the profile, to the poll creator and replying yes to the bot after sending the file, add 1 to the previous interviews for that user once yes is replied.
-
If the selected user fails to reply yes, go back to step 4. If the 2nd selected user also fails, write: selected respondents did not send a profile, the interview is cancelled.
`
import logging
import random
try:
from telegram.ext import Update, Updater, CommandHandler, CallbackContext , PollHandler
except ImportError:
import os
os.system("pip install python-telegram-bot --upgrade")
import gspread
from oauth2client.service_account import ServiceAccountCredentials
# Enable logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)
# Load Google Sheets credentials
scope = ["https://spreadsheets.google.com/feeds", "https://www.googleapis.com/auth/drive"]
creds = ServiceAccountCredentials.from_json_keyfile_name("*************.json", scope)
client = gspread.authorize(creds)
# Open the Google Sheet by title
sheet_title = "*****"
sheet = client.open(sheet_title).sheet1
# Dictionary to store members and their previous interviews count
members_dict = {}
# Load data from Google Sheet into members_dict at the beginning of your script
def load_from_sheet():
global members_dict
members_dict = {}
data = sheet.get_all_records()
for row in data:
user_id = row["User ID"]
prev_interviews = row["Previous Interviews"]
members_dict[user_id] = prev_interviews
# Function to update the Google Sheet with the members_dict
def update_sheet():
# Get existing data from the sheet
existing_data = sheet.get_all_records()
# Update data in the existing_data list
for user_id, prev_interviews in members_dict.items():
# Check if the user_id is already in the sheet
user_row = next((row for row in existing_data if row["User ID"] == user_id), None)
if user_row:
# Update the existing row
user_row["Previous Interviews"] = prev_interviews
else:
# Add a new row
existing_data.append({"User ID": user_id, "Previous Interviews": prev_interviews})
# Clear existing data in the sheet
sheet.clear()
# Write header
header = ["User ID", "Previous Interviews"]
sheet.append_row(header)
# Write modified data
for row in existing_data:
sheet.append_row([row["User ID"], row["Previous Interviews"]])
# Callback function for handling the /start command
def start(update: Update, context: CallbackContext) -> None:
update.message.reply_text("Welcome to the interview bot! Please provide the following options:n"
"/setoptions <poll_text> <poll_time_limit> <profile_time_limit> <min_previous_interviews>")
# Callback function for handling the /setoptions command
def set_options(update: Update, context: CallbackContext) -> None:
args = context.args
if len(args) != 4:
update.message.reply_text("Usage: /setoptions <poll_text> <poll_time_limit> <profile_time_limit> <min_previous_interviews>")
return
poll_text, poll_time_limit, profile_time_limit, min_previous_interviews = args
context.user_data['poll_text'] = poll_text
context.user_data['poll_time_limit'] = int(poll_time_limit) * 60 # Convert minutes to seconds
context.user_data['profile_time_limit'] = int(profile_time_limit) * 60 # Convert minutes to seconds
context.user_data['min_previous_interviews'] = int(min_previous_interviews)
update.message.reply_text("Options set successfully!")
# Callback function for handling the /interview command
def interview(update: Update, context: CallbackContext) -> None:
if 'poll_text' not in context.user_data:
update.message.reply_text("Please set options using /setoptions command first.")
return
# Create a poll
poll_message = update.message.reply_poll(
context.user_data['poll_text'],
options=["Yes", "No", "Mentor Poll Check"],
type=PollHandler.ANSWER_OPTIONS,
is_anonymous=False,
allows_multiple_answers=False,
open_period=context.user_data['poll_time_limit']
)
# Get the poll ID
poll_id = poll_message.poll.id
# Set a timer to check the poll results
context.job_queue.run_once(check_poll_results, context.user_data['poll_time_limit'], context=poll_id)
# Display time limit
update.message.reply_text(f"Poll created with time limit: {context.user_data['poll_time_limit'] / 60} minutes.")
# Callback function to check poll results
def check_poll_results(context: CallbackContext) -> None:
poll_id = context.job.context
poll_results = context.bot.stop_poll(chat_id=context.job.context, message_id=poll_id)
# Process poll results
process_poll_results(context.bot, context.user_data, poll_results)
# Function to process poll results
def process_poll_results(bot, user_data, poll_results):
yes_responses = poll_results['options'][0]['voter_count']
if yes_responses == 0:
update.message.reply_text("No one responded 'Yes' to the poll. The interview is cancelled.")
return
# Filter members with previous interviews less than or equal to the specified limit
eligible_members = [user_id for user_id, prev_interviews in members_dict.items() if prev_interviews <= user_data['min_previous_interviews']]
if not eligible_members:
selected_user_id = random.choice(list(members_dict.keys()))
else:
selected_user_id = random.choice(eligible_members)
selected_user = bot.get_chat_member(chat_id=context.chat_data['chat_id'], user_id=selected_user_id).user
# Notify the selected user to send a profile
bot.send_message(selected_user_id, f"You've been selected for an interview! Please send your profile brief within {user_data['profile_time_limit'] / 60} minutes. Reply with 'yes' once done.")
# Set a timer to check if the user replies 'yes' in time
context.job_queue.run_once(check_profile_submission, user_data['profile_time_limit'], context=(selected_user_id, user_data))
# Callback function to check profile submission
def check_profile_submission(context: CallbackContext) -> None:
selected_user_id, user_data = context.job.context
if members_dict[selected_user_id] == 0:
# Notify the poll creator about the selected user's failure to reply 'yes'
context.bot.send_message(context.chat_data['chat_id'], f"Selected user @{selected_user_id} did not reply 'yes'.")
# Retry the process with another user
process_poll_results(context.bot, user_data, context.chat_data)
else:
# Notify the poll creator about the successful profile submission
context.bot.send_message(context.chat_data['chat_id'], f"Profile received from @{selected_user_id}.")
# Increment the previous interviews count for the selected user
members_dict[selected_user_id] += 1
# Update Google Sheet
update_sheet()
# Callback function for handling the /stop command
def stop(update: Update, context: CallbackContext) -> None:
update.message.reply_text("Bot is stopping...")
context.job_queue.stop()
context.bot.stop()
# Main function to start the bot
def main() -> None:
# Set your Telegram Bot Token
updater = Updater("**********")
# Get the dispatcher to register handlers
dp = updater.dispatcher
# Register command handlers
dp.add_handler(CommandHandler("start", start))
dp.add_handler(CommandHandler("setoptions", set_options))
dp.add_handler(CommandHandler("interview", interview))
dp.add_handler(CommandHandler("stop", stop)) # Added /stop command handler
# Start the Bot
updater.start_polling()
# Run the bot until you send a signal to stop it
updater.idle()
if __name__ == '__main__':
main()
getting error,
File "/home/runner/CalmHeartyObject/main.py", line 73, in <module> def start(update: Update, context: CallbackContext) -> None: NameError: name 'Update' is not defined
// had written wrong error earlier
tried changing import statements, but unable to understand the error..
2
Answers
you have a parameter named Update in the start function, and this conflicts with the Update class from the telegram.ext module.
Change this:
To this:
You need to update the parameter name in the function definition and also in any references within the function. Repeat this for other function definitions where you encounter a similar issue.
Look at your block:
In case of an ImportError these variables will not be defined
Update, Updater, CommandHandler, CallbackContext , PollHandler
.It looks like this is happening, some ImportError occurred but you do not handle it adequately leaving the variables undefined.
In my opinion you should not try to handle a ImportError at all as you need the
Updater
class.Otherwise you could go with Michael Butscher’s comment by not evaluating the type hints.
Further, installing packages from inside a script is bad practice. You could add
from telegram.ext import Update, Updater, CommandHandler, CallbackContext , PollHandler
at the end of the except block to try a second import.TL;DR: Provide a requirements file with the appropriate package version and assume that everything is provided, do not try to install on the fly.