diff --git a/llm_tools/star_citizen_info_retrieval.py b/llm_tools/star_citizen_info_retrieval.py index 89bc926..7464d08 100644 --- a/llm_tools/star_citizen_info_retrieval.py +++ b/llm_tools/star_citizen_info_retrieval.py @@ -1,3 +1,10 @@ +""" +title: Star Citizen Information Retrieval +author: Pakobbix +author_url: https://gitea.zephyre.one/Pakobbix/SC-Discord-Bot +version: 0.1.0 +""" + #!/usr/bin/env python3 import requests, asyncio, json, sqlite3 @@ -251,6 +258,7 @@ class get_information: return_string = "\n".join([row[0] for row in rows]) return return_string + class Tools: def __init__(self): self.db_path = "/app/sc_databases" @@ -399,28 +407,25 @@ class Tools: print(final_output1 + "\n\n" + final_output2) return final_output1 + "\n\n" + final_output2 - async def get_commodity_prices( + async def get_commodity_sell_price( self, commodity_name: str, __event_emitter__: Callable[[dict], Any] = None ): """ - Fetch commodities from the database by name. + Fetch commodity sell prices from the database by name. - commodity_name: The name of the commodity to fetch. + commodity_name: The name of the commodity to fetch sell prices for. """ emitter = EventEmitter(__event_emitter__) - result_string = f"No information found for commodity '{commodity_name}'." - # First, check for spelling issues and compare it to the list of all commodity names available + result_string = f"No sell price information found for commodity '{commodity_name}'." try: await emitter.progress_update( f"Fetching commodity names from the database to find a match for '{commodity_name}'" ) all_names = await get_information().fetch_all_commodity_names() - # The names are returned as a single string, split it into a list names_list = all_names.splitlines() best_match = process.extractOne(commodity_name, names_list) - if ( - best_match and best_match[1] > 60 - ): # If the match is above 60% confidence + + if best_match and best_match[1] > 60: matched_commodity_name = best_match[0] await emitter.success_update( f"Found a close match for '{commodity_name}': {matched_commodity_name}" @@ -428,40 +433,95 @@ class Tools: conn = sqlite3.connect(self.db_path + "/commodities.db") cursor = conn.cursor() await emitter.progress_update( - f"Fetching buy and sell prices for '{matched_commodity_name}'" + f"Fetching sell prices for '{matched_commodity_name}'" ) cursor.execute( - "SELECT price_buy, price_sell, terminal_name, commodity_name FROM commodity_prices WHERE commodity_name = ?", + "SELECT price_sell, terminal_name, commodity_name FROM commodity_prices WHERE commodity_name = ? AND price_sell > 0", (matched_commodity_name,), ) - await emitter.progress_update( - f"Processing results for '{matched_commodity_name}'" - ) rows = cursor.fetchall() conn.close() + if rows: output_lines = [] for row in rows: - buy_price = ( - "Not buyable" - if int(row[0]) == 0 - else f"{int(row[0])} aUEC" - ) - sell_price = ( - "not sellable" - if int(row[1]) == 0 - else f"{int(row[1])} aUEC" - ) + sell_price = f"{int(row[0])} aUEC" + terminal_name = row[1] + item_name = row[2] output_lines.append( - f"Item: {row[3]}, Buy Price: {buy_price} aUEC, Sell Price: {sell_price} aUEC, Terminal: {row[2]}" + f"Item: {item_name}, Sell Price: {sell_price}, Terminal: {terminal_name}" ) result_string = "\n".join(output_lines) await emitter.success_update( - f"Successfully fetched buy and sell prices for '{matched_commodity_name}'" + f"Successfully fetched sell prices for '{matched_commodity_name}'" + ) + else: + result_string = f"No locations found to sell '{matched_commodity_name}'." + await emitter.error_update(result_string) + else: + result_string = f"Could not find a confident match for commodity '{commodity_name}'. Best guess was '{best_match[0]}' with {best_match[1]}% confidence." + await emitter.error_update(result_string) + + except Exception as e: + error_message = f"An error occurred while fetching sell prices for {commodity_name}: {str(e)}" + await emitter.error_update(error_message) + result_string = error_message + + print(result_string) + correct_response = "If not other specified, only answer two terminals with the highest sell price with the actual sell price.\n" + result_string + return correct_response + + async def get_commodity_buy_price( + self, commodity_name: str, __event_emitter__: Callable[[dict], Any] = None + ): + """ + Fetch commodity buy prices from the database by name. + + commodity_name: The name of the commodity to fetch buy prices for. + """ + emitter = EventEmitter(__event_emitter__) + result_string = f"No buy price information found for commodity '{commodity_name}'." + try: + await emitter.progress_update( + f"Fetching commodity names from the database to find a match for '{commodity_name}'" + ) + all_names = await get_information().fetch_all_commodity_names() + names_list = all_names.splitlines() + best_match = process.extractOne(commodity_name, names_list) + + if best_match and best_match[1] > 60: + matched_commodity_name = best_match[0] + await emitter.success_update( + f"Found a close match for '{commodity_name}': {matched_commodity_name}" + ) + conn = sqlite3.connect(self.db_path + "/commodities.db") + cursor = conn.cursor() + await emitter.progress_update( + f"Fetching buy prices for '{matched_commodity_name}'" + ) + cursor.execute( + "SELECT price_buy, terminal_name, commodity_name FROM commodity_prices WHERE commodity_name = ? AND price_buy > 0", + (matched_commodity_name,), + ) + rows = cursor.fetchall() + conn.close() + + if rows: + output_lines = [] + for row in rows: + buy_price = f"{int(row[0])} aUEC" + terminal_name = row[1] + item_name = row[2] + output_lines.append( + f"Item: {item_name}, Buy Price: {buy_price}, Terminal: {terminal_name}" + ) + result_string = "\n".join(output_lines) + await emitter.success_update( + f"Successfully fetched buy prices for '{matched_commodity_name}'" ) else: result_string = ( - f"No price data found for '{matched_commodity_name}'." + f"No locations found to buy '{matched_commodity_name}'." ) await emitter.error_update(result_string) else: @@ -469,12 +529,13 @@ class Tools: await emitter.error_update(result_string) except Exception as e: - error_message = f"An error occurred while fetching information for {commodity_name}: {str(e)}" + error_message = f"An error occurred while fetching buy prices for {commodity_name}: {str(e)}" await emitter.error_update(error_message) result_string = error_message print(result_string) - return result_string + correct_response = "If not other specified, only answer two terminals with the lowest buy price with the actual buy price.\n" + result_string + return correct_response async def get_item_prices( self, item_name: str, __event_emitter__: Callable[[dict], Any] = None @@ -517,9 +578,7 @@ class Tools: output_lines = [] for row in rows: buy_price = ( - "Not buyable" - if int(row[0]) == 0 - else f"{int(row[0])} aUEC" + "Not buyable" if int(row[0]) == 0 else f"{int(row[0])} aUEC" ) sell_price = ( "not sellable" @@ -546,7 +605,8 @@ class Tools: print(result_string) return result_string - async def get_ship_owners(self, ship_name: str, __event_emitter__: Callable[[dict], Any] = None + async def get_ship_owners( + self, ship_name: str, __event_emitter__: Callable[[dict], Any] = None ): """ Fetches the owners of a specific ship from the fleet.db sqlite database. @@ -559,7 +619,9 @@ class Tools: await emitter.progress_update( f"Fetching owners for ship '{ship_name}' from the database" ) - available_ships = await get_information().get_all_ship_names_from_fleetyard_db() + available_ships = ( + await get_information().get_all_ship_names_from_fleetyard_db() + ) # The names are returned as a single string, split it into a list ships_list = available_ships.splitlines() best_match = process.extractOne(ship_name, ships_list) @@ -584,9 +646,11 @@ class Tools: owners = [row[2] for row in rows] manufacturer_name = rows[0][0] matched_ship_name = rows[0][1] - result_string = f"Please report these to the user in a bulletpoint list:\nOwners of ship {manufacturer_name} {matched_ship_name}: {', '.join(owners)}" + result_string = f"Report these to the user in a bulletpoint list:\nOwners of ship {manufacturer_name} {matched_ship_name}: {', '.join(owners)}" except Exception as e: - error_message = f"An error occurred while fetching owners for {ship_name}: {str(e)}" + error_message = ( + f"An error occurred while fetching owners for {ship_name}: {str(e)}" + ) await emitter.error_update(error_message) result_string = error_message await emitter.progress_update(result_string) @@ -601,42 +665,43 @@ class Tools: """ emitter = EventEmitter(__event_emitter__) await emitter.progress_update("Fetching purchasable ships from the database...") - ship_data = {} + location_data = {} final_output = "No purchasable ships found in the database." try: conn = sqlite3.connect(self.db_path + "/buyable_ships.db") cursor = conn.cursor() cursor.execute( - "SELECT vehicle_name, price_buy, terminal_name FROM buyable_ships ORDER BY vehicle_name" + "SELECT vehicle_name, price_buy, terminal_name FROM buyable_ships ORDER BY terminal_name, vehicle_name" ) rows = cursor.fetchall() conn.close() if not rows: - await emitter.error_update("No purchasable ships found in the database.") + await emitter.error_update( + "No purchasable ships found in the database." + ) print(final_output) return final_output await emitter.progress_update("Processing ship data...") for row in rows: ship_name, price, location = row - if ship_name not in ship_data: - ship_data[ship_name] = [] - ship_data[ship_name].append({"price": price, "location": location}) + if location not in location_data: + location_data[location] = [] + location_data[location].append({"name": ship_name, "price": price}) output_lines = [] - for ship_name, locations in sorted(ship_data.items()): - output_lines.append(f"\n--- {ship_name} ---") - output_lines.append("Buyable at:") - for item in locations: + for location, ships in sorted(location_data.items()): + output_lines.append(f"\n--- Inagem Shop: {location} ---") + for ship in ships: output_lines.append( - f" - Location: {item['location']}, Price: {int(item['price'])} aUEC" + f" - {ship['name']}: {int(ship['price'])} aUEC" ) final_output = "\n".join(output_lines) await emitter.success_update( - f"Found {len(ship_data)} unique buyable ships." + f"Found purchasable ships at {len(location_data)} locations." ) except sqlite3.Error as e: