tools.py

import random
from abc import ABC, abstractmethod
from typing import get_origin, get_args
import os
# from langchain.tools import tool
import json
from pydantic import BaseModel, Field
from typing import Dict,  Union
import random
import copy
from types import UnionType

from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings

class VectorStore:
    def __init__(self, embeddings_model, vectorstore):
        embeddings = HuggingFaceEmbeddings(model_name=embeddings_model)
        self.vectore_store = FAISS.load_local(vectorstore, embeddings, allow_dangerous_deserialization=True)
    
    def get_context(self, instruction, number_of_contexts=2):
        documentos = self.vectore_store.similarity_search_with_score(instruction, k=number_of_contexts)
        return self._beautiful_context(documentos)
    
    def _beautiful_context(self, docs):
        context = ""
        for doc in docs:
            context += doc[0].page_content + "\n"
        return context
    
def read_json(data_path: str) -> tuple[list, dict]:
    try:
        with open(data_path, 'r', encoding="utf-8") as f:
            data = [json.loads(line) for line in f.readlines()]
    except:
        with open(data_path, 'r', encoding="utf-8") as f:
            data = json.loads(f.read())
    return data

json_data = read_json("data/val_de_nuria.json")
reservations = {}

class ToolBase(BaseModel, ABC):
    @abstractmethod
    def invoke(cls, input: Dict):
        pass

    @classmethod
    def to_openai_tool(cls):
        """
        Extracts function metadata from a Pydantic class, including function name, parameters, and descriptions.
        Formats it into a structure similar to OpenAI's function metadata.
        """
        function_metadata = {
            "type": "function",
            "function": {
                "name": cls.__name__,  # Function name is same as the class name, in lowercase
                "description": cls.__doc__.strip(),
                "parameters": {
                    "type": "object",
                    "properties": {},
                    "required": [],
                },
            },
        }

        # Iterate over the fields to add them to the parameters
        for field_name, field_info in cls.model_fields.items():
            # Field properties
            field_type = "string"  # Default to string, will adjust if it's a different type
            annotation = field_info.annotation.__args__[0] if getattr(field_info.annotation, "__origin__", None) is Union else field_info.annotation
            
            has_none = False
            if get_origin(annotation) is UnionType:  # Check if it's a Union type
                args = get_args(annotation)
                if type(None) in args:
                    has_none = True
                args = [arg for arg in args if type(None) != arg]
                if len(args) > 1:
                    raise TypeError("It can be union of only a valid type (str, int, bool, etc) and None")
                elif len(args) == 0:
                    raise TypeError("There must be a valid type (str, int, bool, etc) not only None")
                else:
                    annotation = args[0]
            
            if annotation == int:
                field_type = "integer"
            elif annotation == bool:
                field_type = "boolean"
            
            # Add the field's description and type to the properties
            function_metadata["function"]["parameters"]["properties"][field_name] = {
                "type": field_type,
                "description": field_info.description,
            }

            # Determine if the field is required (not Optional or None)
            if field_info.is_required():
                function_metadata["function"]["parameters"]["required"].append(field_name)
                has_none = True

            # If there's an enum (like for `unit`), add it to the properties
            if hasattr(field_info, 'default') and field_info.default is not None and isinstance(field_info.default, list):
                function_metadata["function"]["parameters"]["properties"][field_name]["enum"] = field_info.default
                if not has_none:
                    function_metadata["function"]["parameters"]["required"].append(field_name)

        return function_metadata

tools: Dict[str, ToolBase] = {}
oitools = []
vector_store = VectorStore(embeddings_model="BAAI/bge-m3", vectorstore="data/vs")

def tool_register(cls: BaseModel):
    oaitool = cls.to_openai_tool()
    
    oitools.append(oaitool)
    tools[oaitool["function"]["name"]] = cls

# @tool_register
class hotel_description(ToolBase):
    """Retrieves basic information about the hotel, such as its name, address, contact details, and overall description."""

    @classmethod
    def invoke(cls, input: Dict) -> str: 
        return """### **Nou Vall de Núria – Brief Description**  
Nestled in the stunning **Vall de Núria** in the Pyrenees, **Nou Vall de Núria** offers a perfect blend of comfort and adventure. Guests can enjoy breathtaking mountain views, premium accommodations, and excellent facilities, including an outdoor pool, gym, and sauna.  
The hotel features **two dining options**, serving traditional Catalan cuisine and refreshing drinks. Accommodations range from **cozy standard rooms to luxurious suites and fully equipped apartments**, catering to couples, families, and groups.  
For an unforgettable stay, guests can choose from **special packages**, including family-friendly stays, romantic getaways, ski adventures, and relaxation retreats. Outdoor enthusiasts can explore **hiking trails, ski slopes, and fishing spots** in the surrounding natural beauty.  
Whether for relaxation or adventure, **Nou Vall de Núria** promises a unique and memorable experience.""" 


@tool_register
class get_hotal_general_information(ToolBase):
    """    Retrieves general information about a hotel, including its background, amenities, location, and nearby attractions based on the query.    """
    query: str = Field(description="User's question or query for retrieving the information.")

    @classmethod
    def invoke(cls, input: Dict) -> str:
        query = input.get("query", None)
        if not query:
            return "Missing required argument: query."
        
        # return "We are currently working on it. You can't use this tool right now—please try again later. Thank you for your patience!"
        return vector_store.get_context(query)


@tool_register
class packs(ToolBase):
    """Provides a list of available activity pack at the hotel."""

    @classmethod
    def invoke(cls, input: Dict) -> str: 
        return json_data["packs"]

@tool_register
class hotel_facilities(ToolBase):
    """Provides a list of available general facilities at the hotel, which could include amenities like a spa, pool, gym, conference rooms, etc."""

    @classmethod
    def invoke(cls, input: Dict) -> str: 
        return json_data["general_facilities"]

@tool_register
class restaurants_details(ToolBase):
    """Provides a list of available restaurants with their details."""

    @classmethod
    def invoke(cls, input: Dict) -> str: 
        """
        Play a playlist by its name, starting with the first or a random song.
        """

        return json_data["restaurants"]
    

@tool_register
class restaurant_details(ToolBase):
    """Retrieves detailed information about a specific restaurant in the hotel, including its menu, ambiance, operating hours, and special features."""

    name: str = Field(default=[res["name"] for res in json_data["restaurants"]], description="Name of the resaturant")
    
    @classmethod
    def invoke(cls, input: Dict) -> str: 
        """
        Play a playlist by its name, starting with the first or a random song.
        """

        instance = cls(**input)
        name = instance.name

        restaurante = [res for res in json_data["restaurants"] if res["name"] == name]
        if restaurante:
            return restaurante
        else:
            return f"We don't have any restaurante with the name: {name}"


@tool_register
class rooms_information(ToolBase):
    """
    Provides a list of available hotel rooms, including brief descriptions of each room type.
    """
    @classmethod
    def invoke(cls, input: Dict) -> str: 
        return json_data["room_types"]


@tool_register
class check_room_availability(ToolBase):
    """
    Checks if a specified room type is available between the provided check-in and check-out dates for a given number of guests.
    """
    room_type: str = Field(default=list(json_data["room_types"].keys()), description="The type of room to check for availability.")
    reservation_start_date: str = Field(description="The check-in date for the reservation, formatted as 'YYYY-MM-DD' (e.g., '2025-04-01')")
    reservation_end_date: str = Field(description="The check-out date for the reservation, formatted as 'YYYY-MM-DD' (e.g., '2025-04-05')")
    guests: int = Field(description="The number of guests that will occupy the room.")
    
    @classmethod
    def invoke(cls, input: Dict) -> str: 
        input["room_type"] = input.get("room_type", None)
        room_type = input.get("room_type", None)
        reservation_start_date = input.get("reservation_start_date", None)
        reservation_end_date = input.get("reservation_end_date", None)
        guests = input.get("guests", None)

        missing = []
        if not room_type:
            missing.append("room_type")
        if not reservation_start_date:
            missing.append("reservation_start_date")
        if not reservation_end_date:
            missing.append("reservation_end_date")
        if not guests:
            missing.append("guests")

        if len(missing):
            value = ", ".join(missing)
            return f"Unable to check the room availability. The following required arguments are missing:{value}." 

        instance = cls(**input)
        room_type = instance.room_type
        reservation_start_date = instance.reservation_start_date
        reservation_end_date = instance.reservation_end_date
        guests = instance.guests
        
        rooms = [room for room in json_data["accomodations"]["rooms"] if room_type in room["type"]]
        if len(rooms) == 0:
            return f"There is no room exists with room type {room_type}"
        
        rooms2 = [room for room in rooms if guests <= room["number_of_guests"]]
        if len(rooms2) == 0:
            max_guests = json_data["room_types"][room_type]["number_of_guests"]
            return f"The number of guest is superior then the availibilty, maximum is {max_guests}"

        return rooms2




@tool_register
class make_reservation(ToolBase):
    """
    Creates a new reservation for the hotel by booking a room of the specified type for the desired dates, and associating the booking with a user.
    """

    user_name: str = Field(description="The name of user who is doing the reservation.")
    room_type: str = Field(default=list(json_data["room_types"].keys()), description="The type of room being reserved.")
    reservation_start_date: str = Field(description="The check-in date for the reservation, formatted as 'YYYY-MM-DD' (e.g., '2025-04-01')")
    reservation_end_date: str = Field(description="The check-out date for the reservation, formatted as 'YYYY-MM-DD' (e.g., '2025-04-05')")
    guests: int = Field(description="The total number of guests for the reservation. Must be a positive integer.")
    
    @classmethod
    def invoke(cls, input: Dict) -> str:
        input["room_type"] = input.get("room_type", None)
        room_type = input.get("room_type", None)
        reservation_start_date = input.get("reservation_start_date", None)
        reservation_end_date = input.get("reservation_end_date", None)
        guests = input.get("guests", None)
        user_name = input.get("user_name", None)
        
        missing = []
        if not room_type:
            missing.append("room_type")
        if not reservation_start_date:
            missing.append("reservation_start_date")
        if not reservation_end_date:
            missing.append("reservation_end_date")
        if not guests:
            missing.append("guests")
        if not user_name:
            missing.append("user_name")

        if len(missing):
            value = ", ".join(missing)
            return f"Unable to complete the reservation. The following required arguments are missing:{value}."   
    

        instance = cls(**input)
        room_type = instance.room_type
        reservation_start_date = instance.reservation_start_date
        reservation_end_date = instance.reservation_end_date
        guests = instance.guests
        user_name = instance.user_name.lower()

        rooms = [room for room in json_data["accomodations"]["rooms"] if room_type in room["type"]]
        if len(rooms) == 0:
            return f"There is no room exists with room type {room_type}"
        
        rooms2 = [room for room in rooms if guests <= room["number_of_guests"]]
        if len(rooms2) == 0:
            max_guests = json_data["room_types"][room_type]["number_of_guests"]
            return f"The number of guest is superior then the availibilty, maximum is {max_guests}"

        room = rooms2[random.randint(0, len(rooms2) - 1)]

        rand = int(random.randint(0,10000000))
        while rand in reservations:
            rand = int(random.randint(0,10000000))
        
        tmp_data = {
            "status": "Reserved",
            "room_number": room["room_number"],
            "room_type": room_type,
            "reservation_start_date": reservation_start_date,
            "reservation_end_date": reservation_end_date,
            "guests": guests,
            "reservation_id": rand,
            "user_name": user_name,
        }

        reservations[rand] = tmp_data

        return json.dumps(tmp_data)

@tool_register
class cancel_reservation(ToolBase):
    """Playing a specific playlist by its name."""

    reservation_id: int = Field(description="The unique identifier of the reservation to be canceled.")
    
    @classmethod
    def invoke(cls, input: Dict) -> str: 

        reservation_id = input.get("reservation_id", None)

        missing = []
        if not reservation_id:
            missing.append("reservation_id")

        if len(missing):
            value = ", ".join(missing)
            return f"Unable to cancel the reservation. The following required arguments are missing:{value}."     
        
        instance = cls(**input)
        reservation_id = instance.reservation_id


        
        if reservation_id not in reservations:
            return f"There is no reservations with the id: {reservation_id}"
        
        reservations.pop(reservation_id)
        return f"The reservation {reservation_id} is cancled correctly"

@tool_register
class modify_reservation(ToolBase):
    """
    Allows a user to modify an existing reservation by updating the check-in/check-out dates or changing the room type, subject to availability.
    One of these parameter is mandatory: new_room_type, new_reservation_start_date, new_reservation_end_date, guests.
    """

    reservation_id: int = Field(description="The unique identifier of the reservation to be modified.")
    new_room_type: str | None = Field(default=list(json_data["room_types"].keys()), description=f"The type of new room to be modified, if {None} same room will be modified.")
    new_reservation_start_date: str = Field(default=None, description="New check out date in format DD/MM/YYYY")
    new_reservation_end_date: str = Field(default=None, description="New check out date in format DD/MM/YYYY")
    new_guests: int = Field(default=None, description="New number of guests for the reservation.")

    @classmethod
    def invoke(cls, input: Dict) -> str: 
        input["new_room_type"] = input.get("new_room_type", None)
        reservation_id = input.get("reservation_id", None)

        missing = []
        if not reservation_id:
            missing.append("reservation_id")

        instance = cls(**input)
        new_room_type = instance.new_room_type
        new_reservation_start_date = instance.new_reservation_start_date
        new_reservation_end_date = instance.new_reservation_end_date
        new_guests = instance.new_guests
        reservation_id = instance.reservation_id

        if len(missing):
            value = ", ".join(missing)
            return f"Unable to modify the reservation. The following required arguments are missing:{value}."     
    
        if not (new_room_type or new_reservation_start_date or new_reservation_end_date or new_guests):
            return "Unable to modify the reservation. One of the following arguments must be passed: new_room_type, new_reservation_start_date, new_reservation_end_date, new_guests."     


        if reservation_id not in reservations:
            return f"There is no reservations with the id: {reservation_id}"
        
        if new_room_type or new_guests:
            rooms = [room for room in json_data["accomodations"]["rooms"] if new_room_type in room["type"]]
            if len(rooms) == 0:
                return f"There is no room exists with room type {new_room_type}"
            
            rooms = [room for room in rooms if new_guests <= room["number_of_guests"]]
            if len(rooms) == 0:
                max_guests = json_data["room_types"][new_room_type]["number_of_guests"]
                return f"The number of guest is superior then the availibilty, maximum is {max_guests}"
            
            room = rooms[random.randint(0, len(rooms) - 1)]
            room_number = room["room_number"]
        else:
            room_number = reservations[reservation_id]["room_number"]

        
        reservations[reservation_id]["guests"] = new_guests if new_guests else reservations[reservation_id]["guests"]
        reservations[reservation_id]["reservation_start_date"] = new_reservation_start_date if new_reservation_start_date else reservations[reservation_id]["reservation_start_date"]
        reservations[reservation_id]["reservation_end_date"] = new_reservation_end_date if new_reservation_end_date else reservations[reservation_id]["reservation_end_date"]
        reservations[reservation_id]["room_type"] = new_room_type if new_room_type else reservations[reservation_id]["room_type"]
        reservations[reservation_id]["room_number"] = room_number
        tmp_data = reservations[reservation_id]
        return f"The reservation {reservation_id} is modified correctly: {json.dumps(tmp_data)}"

@tool_register
class get_reservation_detail(ToolBase):
    """Retrieve comprehensive details of a specific room reservation by providing its unique reservation ID."""

    reservation_id: int = Field(description="Id of the reservation")

    @classmethod
    def invoke(cls, input: Dict) -> str: 
        reservation_id = input.get("reservation_id", None)
    
        missing = []
        if not reservation_id:
            missing.append("reservation_id")

        if len(missing):
            value = ", ".join(missing)
            return f"Unable to get the details. The following required arguments are missing:{value}."     

        instance = cls(**input)
        reservation_id = instance.reservation_id

        if reservation_id not in reservations:
            return f"There is no reservations with the id: {reservation_id}"
        
        
        tmp_data = copy.deepcopy(reservations[reservation_id])
        return json.dumps(tmp_data)