Skip to content

Instantly share code, notes, and snippets.

@ergoz
Forked from fskuratov/Msg2Trello.py
Created March 26, 2023 12:28
Show Gist options
  • Save ergoz/9924e61cddc001fad50c9327de8590b0 to your computer and use it in GitHub Desktop.
Save ergoz/9924e61cddc001fad50c9327de8590b0 to your computer and use it in GitHub Desktop.
Telegram bot, developed using Python, that seamlessly translates incoming messages or forwarded texts into tasks and subtasks with the help of the GPT-3.5 API. The tasks, along with their associated subtasks as checklists, are efficiently organized into specified boards and lists on Trello by utilizing the Trello API.
import logging
import openai
import re
from datetime import datetime, timedelta
from trello import TrelloClient
from telegram import Update
from telegram.ext import ApplicationBuilder, ContextTypes, CommandHandler, MessageHandler, filters
# Set up logging for easier debugging
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
logger = logging.getLogger(__name__)
# Trello API credentials
TRELLO_API_KEY = "INSERT YOUR TRELLO API KEY"
TRELLO_API_SECRET = "INSERT YOUR TRELLO API SECRET"
TRELLO_TOKEN = "INSERT YOU TRELLO TOKEN"
# Initialize the Trello client with the API credentials
trello_client = TrelloClient(
api_key=TRELLO_API_KEY,
api_secret=TRELLO_API_SECRET,
token=TRELLO_TOKEN
)
# Set OpenAI API key
openai.api_key = "INSERT YOUR OPENAPI KEY"
# Add log messages to relevant parts of the code
logger.info("Trello client initialized")
logger.info(f"OpenAI API key: {openai.api_key}")
# Function to create Trello task
# This function takes board_name, list_name, task_name, and optional parameters due_date,
# summary, and sub_tasks as input.
# It searches for the specified Trello board and list,
# creates a card with the provided information, sets the card's due date if provided, and returns the created card.
def create_trello_task(board_name, list_name, task_name, due_date=None, summary=None, sub_tasks=None):
# Initialize target_board and target_list as None
target_board = None
target_list = None
# Iterate through all boards and find the target board
for board in trello_client.list_boards():
logger.info(f"Found board: {board.name}")
if board.name.lower() == board_name.lower():
target_board = board
break
# If target board is not found, return early
if not target_board:
logger.warning(f"No board found with the name {board_name}")
return
# Iterate through all lists in the target board and find the target list
for lst in target_board.list_lists():
logger.info(f"Found list: {lst.name}")
if lst.name.lower() == list_name.lower():
target_list = lst
break
# If target list is not found, return early
if not target_list:
logger.warning(f"No list found with the name {list_name}")
return
# Create a card with the task name and summary
card_description = f"Summary: {summary}\n"
card = target_list.add_card(task_name, desc=card_description)
logger.info(f"Created Trello card with ID: {card.id}")
# Create a checklist for the subtasks (if any)
if sub_tasks:
card.add_checklist("Sub-tasks", sub_tasks)
logger.info(f"Created a checklist with the subtasks:")
for sub_task in sub_tasks:
logger.info(f"- {sub_task}")
logger.info(f"Created card with name '{task_name}' and description:\n{card_description}")
# Set the card's due date (if provided)
if due_date:
try:
due_date_dt = datetime.strptime(due_date, "%Y-%m-%d")
current_date = datetime.now()
# Check if the due date is earlier than the current date
if due_date_dt < current_date:
# Set the due date to the current date + 1 day
due_date_dt = current_date + timedelta(days=1)
due_date = due_date_dt.strftime("%Y-%m-%d")
logger.warning(f"Due date was in the past. Updated due date to {due_date}")
# Set the card's due date
card.set_due(due_date_dt)
logger.info(f"Set due date to {due_date}")
except ValueError:
# Handle invalid date format for due date
logger.error(f"Invalid date format for due date: {due_date}")
# If the card is archived, unarchive it
if card.closed:
card.set_closed(False)
logger.info(f"Unarchived card with ID: {card.id}")
return card
# Function to generate a response using GPT-3.5-turbo
# This function, generate_response, takes a user prompt and generates a response using GPT-3.5-turbo.
# It creates a list of messages as input, with a system message that provides context and a user message
# containing the prompt.
# It then calls the OpenAI API with this list of messages, and extracts the content of the response.
# The response content is then returned.
async def generate_response(prompt):
# Create a list of messages to send to GPT-3.5-turbo as input
messages = [{"role": "system", "content": "You are a helpful assistant that converts user messages into tasks."},
{"role": "user", "content": prompt}]
logger.info(f"Generated prompt: {prompt}")
# Call the OpenAI API with the list of messages
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages,
max_tokens=1000,
n=1,
stop=None,
temperature=0.2,
)
logger.info(f"Raw GPT-3.5-turbo response: {response}")
# Extract the content of the response and return it
extracted_content = response.choices[0].message['content'].strip()
logger.info(f"Extracted content: {extracted_content}")
return extracted_content
# Pass the message to GPT-3 and get the response
# This function, process_message, takes a user message as input and processes it by passing it to GPT-3.
# It calls the generate_response function to get the GPT-3 response, which is expected to include task name, due date
# (if any), a brief summary, and sub-tasks (if applicable).
# The GPT-3 response is then printed for debugging purposes and returned.
async def process_message(message):
logger.info(f"Received message: {message}")
# Pass the message to GPT-3 and get the response
gpt3_response = await generate_response(
f"Create a task from the following text: \"{message}\". Include task name, due date (if any, in YYYY-MM-DD format), a brief summary (no more than 30 words), and sub-tasks (if applicable).")
# Debugging: Print the GPT-3 response
logger.info(f"GPT-3 response: {gpt3_response}")
return gpt3_response
# Extract task details from GPT-3 response
# This function, extract_task_details, takes the GPT-3 response as input and extracts the task details from it.
# The extracted details include task name, due date, summary, and sub-tasks.
# The function iterates through the lines of the response, looking for specific patterns to identify and extract
# the relevant information. The extracted details are stored in a dictionary and returned.
def extract_task_details(gpt3_response):
logger.info(f"Extracting task details from GPT-3 response: {gpt3_response}")
# Initialize a dictionary to store the extracted task details
task_details = {}
# Split the GPT-3 response into lines
lines = gpt3_response.split('\n')
# Initialize a flag for parsing subtasks and an empty list to store subtasks
parsing_subtasks = False
subtasks = []
# Iterate through the lines of the response
for line in lines:
# Extract and store the task name
if line.startswith("Task Name:"):
task_details['task_name'] = line[len("Task Name:"):].strip()
# Extract and store the due date
elif line.startswith("Due Date:"):
task_details['due_date'] = line[len("Due Date:"):].strip()
# Extract and store the summary
elif line.startswith("Summary:"):
task_details['summary'] = line[len("Summary:"):].strip()
# Start parsing subtasks when the "Sub-tasks:" line is encountered
elif line.startswith("Sub-tasks:"):
parsing_subtasks = True
# Parse and store subtasks
elif parsing_subtasks:
# This regex pattern matches a numbered or bulleted list item (e.g., "1. ", "2. ", "- ", etc.)
pattern = r'^\s*[\d\-\*\+]+\.\s+'
match = re.match(pattern, line)
# If the line matches the pattern, store the subtask
if match:
subtasks.append(re.sub(pattern, '', line).strip())
# If the line doesn't match the pattern, stop parsing subtasks
else:
parsing_subtasks = False
# Store the subtasks in the task_details dictionary
task_details['sub_tasks'] = subtasks
# Log the extracted task details
logger.info(f"Extracted task details: {task_details}")
# Explicitly return the task_details dictionary or an empty dictionary
return task_details if any(task_details.values()) else {}
# Function to handle the /start command
# This function, start, is an asynchronous function that handles the /start command in a chatbot context.
# It takes two arguments: update, which contains information about the incoming message, and context,
# which provides access to the bot instance and other utilities.
# The function uses the generate_response function to create a welcome message from GPT-3,
# indicating that the bot is capable of converting text messages into tasks.
# It then sends this generated message to the user by calling context.bot.send_message,
# using the update.effective_chat.id to identify the chat where the message should be sent.
# The function uses the generate_response function to create a welcome message from GPT-3,
# indicating that the bot is capable of converting text messages into tasks.
# It then sends this generated message to the user by calling context.bot.send_message,
# using the update.effective_chat.id to identify the chat where the message should be sent.
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
logger.info(f"Received /start command")
# Generate a welcome message using GPT-3, stating that the bot can convert text messages into tasks
gpt3_response = await generate_response("Make a welcome message that says that the bot is able to convert text messages sent to it into tasks")
logger.info(f"Generated welcome message: {gpt3_response}")
# Send the generated welcome message to the user
await context.bot.send_message(chat_id=update.effective_chat.id, text=gpt3_response)
logger.info(f"Sent welcome message to the user")
# Function to handle incoming text messages (excluding commands)
# This function, message_handler, is an asynchronous function that handles incoming text messages (excluding commands)
# in a chatbot context.
# It takes two arguments: update, which contains information about the incoming message, and context,
# which provides access to the bot instance and other utilities.
# The function processes the received message using the process_message function,
# which generates a response using GPT-3.
# If a response is generated, it is sent back to the chat, and the task details are extracted from the response
# using the extract_task_details function.
# If task details are successfully extracted, a Trello task is created with the extracted details.
# If the task details could not be extracted or the GPT-3 response could not be generated,
# the function sends an error message to the chat.
async def message_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
logger.info(f"Received message: {update.message.text}")
# Process the received message using the process_message function
processed_message = await process_message(update.message.text)
if processed_message:
# Send the GPT response back to the chat
await context.bot.send_message(chat_id=update.effective_chat.id, text=processed_message)
logger.info(f"Sent GPT response to the chat")
# Extract task details from the GPT response
task_details = extract_task_details(processed_message)
logger.info(f"Extracted task details: {task_details}")
if task_details:
# Replace 'Your Board Name' and 'Your List Name' with the actual names of your Trello board and list.
# Create a Trello task with the extracted task details
create_trello_task('YOUR BOARD NAME', 'YOUR LIST NAME', task_details['task_name'], task_details['due_date'],
task_details['summary'], task_details['sub_tasks'])
logger.info("Created Trello task")
else:
# If the task details could not be extracted, send an error message to the chat
await context.bot.send_message(chat_id=update.effective_chat.id,
text="Sorry, I couldn't extract the task details from the message.")
logger.info("Failed to extract task details")
else:
# If the GPT-3 response could not be generated, send an error message to the chat
await context.bot.send_message(chat_id=update.effective_chat.id,
text="Sorry, I couldn't generate a response. Please try again.")
logger.info("Failed to generate GPT-3 response")
# Main function to set up and run the bot
# This script is the main function to set up and run the bot.
# It is executed when the script is run directly (not imported as a module).
# 1. The ApplicationBuilder is used to create an application object with the bot token.
# 2. The /start command handler is created using the CommandHandler class with the
# command name 'start' and the start function. It is then added to the application object using the add_handler method.
# 3. A message handler is created for non-command text messages using the MessageHandler class.
# It filters text messages that are not commands and uses the message_handler function.
# The message handler is then added to the application object using the add_handler method.
# 4. Finally, the run_polling method is called on the application object to start
# polling for updates from the Telegram API.
if __name__ == '__main__':
logger.info("Starting the bot")
# Create the application object with the bot token
application = ApplicationBuilder().token('INSERT YOUR TELEGRAM BOT TOKEN').build()
# Create and add the /start command handler
start_handler = CommandHandler('start', start)
application.add_handler(start_handler)
logger.info("Added /start command handler")
# Create and add the message handler for non-command text messages
message_handler_obj = MessageHandler(filters.TEXT & (~filters.COMMAND), message_handler)
application.add_handler(message_handler_obj)
logger.info("Added message handler")
# Start polling for updates
application.run_polling()
logger.info("Started polling for updates")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment