Buyable ships: Switched from starcitizen.tools to UEX for more reliable information

llm_tools/get_buyable_ships.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+
+import requests
+import sqlite3
+
+# --- Configuration ---
+API_URL = "https://api.uexcorp.space/2.0/vehicles_purchases_prices_all"
+with open("uex_api_key", "r") as f:
+    BEARER_TOKEN = f.read().strip()
+
+DB_NAME = "buyable_ships.db"
+TABLE_NAME = "buyable_ships"
+
+def setup_database():
+    """
+    Sets up the SQLite database and creates the table if it doesn't exist.
+    The table uses a composite primary key (id_vehicle, id_terminal)
+    to ensure each vehicle at each terminal has only one latest entry.
+    """
+    conn = sqlite3.connect(DB_NAME)
+    cursor = conn.cursor()
+
+    # Using "IF NOT EXISTS" prevents errors on subsequent runs.
+    # The schema is derived from your provided image.
+    # We use INSERT OR REPLACE later, so a primary key is important.
+    # (id_vehicle, id_terminal) is a good candidate for a unique key.
+    cursor.execute(f"""
+        CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
+            id_vehicle TEXT,
+            id_terminal TEXT,
+            price_buy REAL,
+            vehicle_name TEXT,
+            terminal_name TEXT,
+            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
+            PRIMARY KEY (id_vehicle, id_terminal)
+        )
+    """)
+    conn.commit()
+    conn.close()
+    print("Database setup complete.")
+
+def fetch_data_from_api():
+    """
+    Fetches the latest vehicle purchase price data from the UAX Corp API.
+    Returns the data as a list of dictionaries or None if an error occurs.
+    """
+    headers = {"Authorization": f"Bearer {BEARER_TOKEN}"}
+    try:
+        response = requests.get(API_URL, headers=headers)
+        response.raise_for_status()  # Raise an error for HTTP errors
+        data = response.json()
+        return data.get("data", [])
+    except requests.RequestException as e:
+        print(f"Error fetching API data: {e}")
+        return None
+
+def save_data_to_db(data):
+    """
+    Saves the fetched vehicle purchase price data to the SQLite database.
+    """
+    conn = sqlite3.connect(DB_NAME)
+    cursor = conn.cursor()
+
+    for item in data:
+        cursor.execute(f"""
+            INSERT OR REPLACE INTO {TABLE_NAME} (id_vehicle, id_terminal, price_buy, vehicle_name, terminal_name)
+            VALUES (?, ?, ?, ?, ?)
+        """, (
+            item.get("id_vehicle"),
+            item.get("id_terminal"),
+            item.get("price_buy"),
+            item.get("vehicle_name"),
+            item.get("terminal_name")
+        ))
+
+    conn.commit()
+    conn.close()
+    print("Data saved to database.")
+
+if __name__ == "__main__":
+    setup_database()
+    data = fetch_data_from_api()
+    if data:
+        save_data_to_db(data)

llm_tools/star_citizen_info_retrieval.py

@@ -597,74 +597,33 @@ class Tools:
         self, __event_emitter__: Callable[[dict], Any] = None
     ):
         """
-        Fetches all buyable ships, their prices, and locations from the Star Citizen Tools wiki.
+        Fetches all buyable ships, their prices, and locations from the buyable_ships.db database.
         """
         emitter = EventEmitter(__event_emitter__)
-        api_url = "https://starcitizen.tools/api.php"
+        await emitter.progress_update("Fetching purchasable ships from the database...")
         ship_data = {}
-        page_title = "Purchasing_ships"
+        final_output = "No purchasable ships found in the database."
 
-        await emitter.progress_update(f"Fetching data from {page_title}...")
-        params = {
-            "action": "parse",
-            "page": page_title,
-            "format": "json",
-            "prop": "text",
-        }
         try:
-            response = await asyncio.to_thread(requests.get, api_url, params=params)
+            conn = sqlite3.connect(self.db_path + "/buyable_ships.db")
-            response.raise_for_status()
+            cursor = conn.cursor()
-            data = response.json()
+            cursor.execute(
-
+                "SELECT vehicle_name, price_buy, terminal_name FROM buyable_ships ORDER BY vehicle_name"
-            if "error" in data:
-                await emitter.error_update(
-                    f"API Error for {page_title}: {data['error']['info']}"
             )
-                return
+            rows = cursor.fetchall()
-            html_content = data.get("parse", {}).get("text", {}).get("*", "")
+            conn.close()
-            if not html_content:
-                await emitter.error_update(f"No content found for {page_title}.")
-                return
 
-            await emitter.progress_update(f"Parsing data from {page_title}...")
+            if not rows:
-            soup = BeautifulSoup(html_content, "html.parser")
+                await emitter.error_update("No purchasable ships found in the database.")
-            tables = soup.find_all("table", class_="wikitable")
+                print(final_output)
+                return final_output
 
-            for table in tables:
+            await emitter.progress_update("Processing ship data...")
-                header_row = table.find("tr")
-                if not header_row:
-                    continue
-                headers = [th.get_text(strip=True) for th in header_row.find_all("th")]
-
-                rows = table.find_all("tr")[1:]
             for row in rows:
-                    cells = row.find_all("td")
+                ship_name, price, location = row
-                    if not cells or len(cells) < 3:
-                        continue
-
-                    ship_name_tag = cells[1].find("a")
-                    if not ship_name_tag or not ship_name_tag.get("title"):
-                        continue
-                    ship_name = ship_name_tag.get("title").strip()
-                    price = cells[2].get_text(strip=True)
-
                 if ship_name not in ship_data:
                     ship_data[ship_name] = []
-
+                ship_data[ship_name].append({"price": price, "location": location})
-                    location_headers = headers[3:]
-                    for i, cell in enumerate(cells[3:]):
-                        if "✔" in cell.get_text():
-                            location = location_headers[i]
-                            ship_data[ship_name].append(
-                                {"price": price + " aUEC (alpha United Earth Credits)", "location": location}
-                            )
-
-            await emitter.success_update(f"Successfully processed {page_title}.")
-
-        except requests.exceptions.RequestException as e:
-            await emitter.error_update(f"Error fetching data for {page_title}: {e}")
-        except json.JSONDecodeError:
-            await emitter.error_update(f"Error decoding JSON for {page_title}.")
 
             output_lines = []
             for ship_name, locations in sorted(ship_data.items()):
@@ -672,11 +631,23 @@ class Tools:
                 output_lines.append("Buyable at:")
                 for item in locations:
                     output_lines.append(
-                    f"  - Location: {item['location']}, Price: {item['price']}"
+                        f"  - Location: {item['location']}, Price: {int(item['price'])} aUEC"
                     )
 
             final_output = "\n".join(output_lines)
-        await emitter.success_update(f"Found {len(ship_data)} unique buyable ships.")
+            await emitter.success_update(
+                f"Found {len(ship_data)} unique buyable ships."
+            )
+
+        except sqlite3.Error as e:
+            error_message = f"Database error while fetching purchasable ships: {e}"
+            await emitter.error_update(error_message)
+            final_output = error_message
+        except Exception as e:
+            error_message = f"An unexpected error occurred: {e}"
+            await emitter.error_update(error_message)
+            final_output = error_message
+
         print(final_output)
         return final_output
 
@@ -789,4 +760,4 @@ class Tools:
 
 if __name__ == "__main__":
     info_printer = Tools()
-    asyncio.run(info_printer.get_ship_owners("Perseus"))
+    asyncio.run(info_printer.list_rentable_ships())