Wikelo ship missions are now a db to reduce the clutter of information via RAG and get accurate and non-hallucinated output.
This commit is contained in:
parent
c8fba0594c
commit
007aa63528
BIN
databases/wikelo_crafting.db
Normal file
BIN
databases/wikelo_crafting.db
Normal file
Binary file not shown.
@ -822,7 +822,113 @@ class Tools:
|
||||
print(final_output)
|
||||
return final_output
|
||||
|
||||
async def fetch_wikelo_information(
|
||||
self,
|
||||
mission_ship_name: str,
|
||||
__event_emitter__: Callable[[dict], Any] = None,
|
||||
):
|
||||
"""
|
||||
Retrieves all columns for the Wikelo entry whose *mission* or *ship_name*
|
||||
best matches ``mission_ship_name`` using fuzzy search.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
mission_ship_name : str
|
||||
The string to match against the `missions` and `ship_name`
|
||||
fields of the table.
|
||||
__event_emitter__ : Callable[[dict], Any] | None
|
||||
Optional async callback that receives status updates.
|
||||
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
A formatted summary of the matched entry, or an error message.
|
||||
"""
|
||||
emitter = EventEmitter(__event_emitter__)
|
||||
|
||||
try:
|
||||
await emitter.progress_update(
|
||||
f"Searching Wikelo database for '{mission_ship_name}'"
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 1. Pull all rows from the database
|
||||
# ------------------------------------------------------------------
|
||||
conn = sqlite3.connect(self.db_path + "/wikelo_crafting.db")
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT missions, ship_name, components, costs, rewards
|
||||
FROM Wikelo_information
|
||||
"""
|
||||
)
|
||||
rows = cursor.fetchall()
|
||||
conn.close()
|
||||
|
||||
if not rows:
|
||||
await emitter.error_update("No entries found in the Wikelo database.")
|
||||
return "No entries found."
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 2. Build a searchable list that keeps a reference to each row.
|
||||
# We create two separate search items per row – one for
|
||||
# the mission and one for the ship name.
|
||||
# ------------------------------------------------------------------
|
||||
search_items = [] # List of tuples: (search_string, label, full_row)
|
||||
for row in rows:
|
||||
mission, ship_name, components, costs, rewards = row
|
||||
search_items.append((mission, "mission", row))
|
||||
search_items.append((ship_name, "ship", row))
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 3. Find the best fuzzy match against all searchable strings.
|
||||
# ------------------------------------------------------------------
|
||||
best_match = process.extractOne(
|
||||
mission_ship_name,
|
||||
[item[0] for item in search_items],
|
||||
)
|
||||
|
||||
if not best_match:
|
||||
await emitter.error_update(f"No close match found for '{mission_ship_name}'.")
|
||||
return f"No close match found for '{mission_ship_name}'."
|
||||
|
||||
matched_string, score = best_match
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 4. Retrieve the full row that produced the matched string.
|
||||
# ------------------------------------------------------------------
|
||||
matched_row = None
|
||||
for text, label, row in search_items:
|
||||
if text == matched_string:
|
||||
matched_row = row
|
||||
break
|
||||
|
||||
if not matched_row:
|
||||
await emitter.error_update("Matched entry could not be found.")
|
||||
return "Error: matched entry could not be located."
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 5. Build the output string.
|
||||
# ------------------------------------------------------------------
|
||||
mission, ship_name, components, costs, rewards = matched_row
|
||||
result_text = (
|
||||
f"**Mission:** {mission}\n"
|
||||
f"**Ship Name:** {ship_name}\n"
|
||||
f"**Components:** {components}\n"
|
||||
f"**Costs:** {costs}\n"
|
||||
f"**Rewards:** {rewards}"
|
||||
)
|
||||
|
||||
await emitter.success_update(
|
||||
f"Found a close match ({matched_string}) with score {score}."
|
||||
)
|
||||
except Exception as exc:
|
||||
result_text = f"Error retrieving Wikelo information: {exc}"
|
||||
await emitter.error_update(result_text)
|
||||
|
||||
print(result_text)
|
||||
return "If possible, also mention where to find the materials needed.\n" + result_text
|
||||
|
||||
if __name__ == "__main__":
|
||||
info_printer = Tools()
|
||||
asyncio.run(info_printer.list_rentable_ships())
|
||||
asyncio.run(info_printer.fetch_wikelo_information("Nox"))
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user