Refactor scripts with better docs

This commit is contained in:
Roger Gonzalez 2024-11-28 15:21:33 -03:00
parent 5fef9939fd
commit f5f6871b64
Signed by: rogs
GPG Key ID: C7ECE9C6C36EC2E6
2 changed files with 44 additions and 74 deletions

13
bofa.py
View File

@ -1,12 +1,19 @@
#!/usr/bin/env python3 """Process Bank of America expense data from spreadsheets."""
from pyexcel_ods3 import get_data from pyexcel_ods3 import get_data
from settings import EXPENSES_FILENAME from settings import EXPENSES_FILENAME
class BofaData(): class BofaData:
def get(self) -> None: """Handle Bank of America expense data operations."""
def get(self) -> dict:
"""Retrieve and format expense data from spreadsheet.
Returns:
dict: Formatted data with headlines as keys and amounts as values
"""
with open(EXPENSES_FILENAME, "rb") as expenses: with open(EXPENSES_FILENAME, "rb") as expenses:
data = get_data(expenses) data = get_data(expenses)
main_page = data.get("Main") main_page = data.get("Main")

105
bot.py
View File

@ -1,4 +1,7 @@
"""A Matrix bot that manages TODOs, expenses, and AI interactions."""
import os import os
import openai import openai
import simplematrixbotlib as botlib import simplematrixbotlib as botlib
import validators import validators
@ -7,11 +10,6 @@ import wget
from bofa import BofaData from bofa import BofaData
from org import OrgData from org import OrgData
from settings import ( from settings import (
BANK_ACCOUNT_NUMBERS,
BROU_PASSWORD,
BROU_USERNAME,
ITAU_PASSWORD,
ITAU_USERNAME,
MATRIX_PASSWORD, MATRIX_PASSWORD,
MATRIX_URL, MATRIX_URL,
MATRIX_USER, MATRIX_USER,
@ -38,8 +36,8 @@ for username in MATRIX_USERNAMES:
@bot.listener.on_message_event @bot.listener.on_message_event
async def todo(room, message): async def todo(room, message):
""" """Add new TODOs with specified details.
Function that adds new TODOs
Usage: Usage:
user: !todo title - objective - extra (optional) user: !todo title - objective - extra (optional)
bot: TODO added! bot: TODO added!
@ -70,13 +68,14 @@ async def todo(room, message):
print(f"Room: {room_id}, User: {user}, Message: {message}") print(f"Room: {room_id}, User: {user}, Message: {message}")
OrgData().add_new_todo(keyword, todo_title, todo_objective, todo_extra) OrgData().add_new_todo(keyword, todo_title, todo_objective, todo_extra)
await bot.api.send_text_message(room_id, "TODO added!") return await bot.api.send_text_message(room_id, "TODO added!")
return None
@bot.listener.on_message_event @bot.listener.on_message_event
async def list_todos(room, message): async def list_todos(room, message):
""" """List today's plan from org files.
Function that lists today's plan
Usage: Usage:
user: !list [free,work] user: !list [free,work]
bot: [prints a list with today's todos] bot: [prints a list with today's todos]
@ -97,13 +96,14 @@ async def list_todos(room, message):
print(f"Room: {room_id}, User: {user}, Message: {message}") print(f"Room: {room_id}, User: {user}, Message: {message}")
plan = OrgData().list_plan(message) plan = OrgData().list_plan(message)
await bot.api.send_text_message(room_id, plan) return await bot.api.send_text_message(room_id, plan)
return None
@bot.listener.on_message_event @bot.listener.on_message_event
async def list_bofa_status(room, message): async def list_bofa_status(room, message):
""" """Show Bank of America account status.
Function that lists bofa status
Usage: Usage:
user: !bofa user: !bofa
bot: [prints bofa current status] bot: [prints bofa current status]
@ -114,7 +114,6 @@ async def list_bofa_status(room, message):
if user == MATRIX_USERNAME: if user == MATRIX_USERNAME:
room_id = room.room_id room_id = room.room_id
print(f"Room: {room_id}, User: {user}, Message: bofa") print(f"Room: {room_id}, User: {user}, Message: bofa")
bofa_data = BofaData().get() bofa_data = BofaData().get()
@ -124,57 +123,17 @@ async def list_bofa_status(room, message):
if amount != "0 USD": if amount != "0 USD":
return_data += f"{person}: {amount}\n" return_data += f"{person}: {amount}\n"
await bot.api.send_text_message(room_id, return_data) return await bot.api.send_text_message(room_id, return_data)
return None
def generate_return_data_from_account(bank: str, accounts: list) -> str:
bank_message = f"{bank.upper()}\nNot available. Try again later!\n"
for account in accounts:
if account.get("number", "") in BANK_ACCOUNT_NUMBERS:
account_number = account.get("number", "")
balance = account.get("balance", "")
bank_message = f"{bank.upper()}\nAccount Number: {account_number}\nBalance: {balance} USD\n"
return bank_message
@bot.listener.on_message_event
async def list_bank_information(room, message):
"""
Function that lists banks information
Usage:
user: !banks
bot: [prints current status of banks]
"""
match = botlib.MessageMatch(room, message, bot, PREFIX)
if match.is_not_from_this_bot() and match.prefix() and match.command("banks"):
user = message.sender
if user == MATRIX_USERNAME:
room_id = room.room_id
print(f"Room: {room_id}, User: {user}, Message: banks")
await bot.api.send_text_message(room_id, "Looking for bank data, just a sec...")
brou_accounts = get_bank_information("brou", BROU_USERNAME, BROU_PASSWORD)
itau_accounts = get_bank_information("itau", ITAU_USERNAME, ITAU_PASSWORD)
return_data = ""
return_data += generate_return_data_from_account("brou", brou_accounts)
return_data += "\n"
return_data += generate_return_data_from_account("itau", itau_accounts)
await bot.api.send_text_message(room_id, return_data)
@bot.listener.on_message_event @bot.listener.on_message_event
async def save_link(room, message): async def save_link(room, message):
""" """Save a URL to the links file.
Function that lists banks information
Usage: Usage:
user: !banks user: [any valid URL]
bot: [prints current status of banks] bot: Link added!
""" """
match = botlib.MessageMatch(room, message, bot) match = botlib.MessageMatch(room, message, bot)
message_content = message.body message_content = message.body
@ -183,17 +142,17 @@ async def save_link(room, message):
if user == MATRIX_USERNAME: if user == MATRIX_USERNAME:
room_id = room.room_id room_id = room.room_id
print(f"Room: {room_id}, User: {user}, Message: {message_content}") print(f"Room: {room_id}, User: {user}, Message: {message_content}")
OrgData().add_new_link(f"- {message_content}\n") OrgData().add_new_link(f"- {message_content}\n")
await bot.api.send_text_message(room_id, "Link added!") return await bot.api.send_text_message(room_id, "Link added!")
return None
@bot.listener.on_message_event @bot.listener.on_message_event
async def chatgpt(room, message): async def chatgpt(room, message):
""" """Start a conversation with ChatGPT.
Function that starts a conversation with chatgpt
Usage: Usage:
user: !chatgpt Hello! user: !chatgpt Hello!
bot: [prints chatgpt response] bot: [prints chatgpt response]
@ -222,13 +181,14 @@ async def chatgpt(room, message):
print(f"Error: {e}") print(f"Error: {e}")
response = "There was a problem with your prompt" response = "There was a problem with your prompt"
await bot.api.send_text_message(room_id, response) return await bot.api.send_text_message(room_id, response)
return None
@bot.listener.on_message_event @bot.listener.on_message_event
async def reset_chatgpt(room, message): async def reset_chatgpt(room, message):
""" """Reset the ChatGPT conversation history.
Function that resets a conversation with chatgpt
Usage: Usage:
user: !reset user: !reset
bot: Conversation reset! bot: Conversation reset!
@ -241,13 +201,14 @@ async def reset_chatgpt(room, message):
CONVERSATION[user] = [starting_prompt] CONVERSATION[user] = [starting_prompt]
room_id = room.room_id room_id = room.room_id
await bot.api.send_text_message(room_id, "Conversation reset!") return await bot.api.send_text_message(room_id, "Conversation reset!")
return None
@bot.listener.on_message_event @bot.listener.on_message_event
async def dall_e(room, message): async def dall_e(room, message):
""" """Generate an image using DALL-E.
Function that generates a Dall-E image
Usage: Usage:
user: !dalle A sunny caribbean beach user: !dalle A sunny caribbean beach
bot: returns an image bot: returns an image
@ -270,9 +231,11 @@ async def dall_e(room, message):
await bot.api.send_image_message(room_id, image_filename) await bot.api.send_image_message(room_id, image_filename)
os.remove(image_filename) os.remove(image_filename)
return None
except Exception as e: except Exception as e:
print(f"Error sending image: {e}") print(f"Error sending image: {e}")
await bot.api.send_text_message(room_id, f"Error sending image: {e}") return await bot.api.send_text_message(room_id, f"Error sending image: {e}")
return None
bot.run() bot.run()