DonConfia_Dev/agents/app/seller/catalog_tools.py
2025-03-24 20:33:04 -05:00

129 lines
3.4 KiB
Python

from abc import ABC, abstractmethod
import requests
import json
class CatalogAbstract(ABC):
@abstractmethod
def list_products(self):
"""
Lista todos los productos disponibles en el catálogo con su disponibilidad
"""
pass
@abstractmethod
def search_products(self):
"""
Busca productos en el catálogo que coincidan con la consulta
"""
raise NotImplementedError
@abstractmethod
def check_price(self):
"""
Verifica el precio de un producto específico
"""
raise NotImplementedError
@abstractmethod
def check_availability(self):
"""
Verifica la disponibilidad de un producto espécifico
"""
raise NotImplementedError
@abstractmethod
def get_product_details(self):
"""
Obtiene detalles completos de un producto
"""
raise NotImplementedError
@abstractmethod
def list_tools(self):
"""Retorna los metodos de esta clase en una lista"""
raise NotImplementedError
class CatalogTrytonTools(CatalogAbstract):
def __init__(self, url, application_name, key, db):
self.tryton_url = url
self.application_name = application_name
self.tryton_db = db
self.tryton_key = key
self.base_url = "{}/{}/{}".format(
self.tryton_url,
self.tryton_db,
self.application_name,
)
def list_products(self) -> list:
"""
Lista todos los productos disponibles en el catálogo con su disponibilidad
"""
endpoint = f"{self.base_url}/products"
products = requests.get(
endpoint,
headers={
"Authorization": f"bearer {self.tryton_key}",
},
)
return products.json()
def search_products(self, product_name) -> list:
"""
Busca productos en el catálogo que coincidan con la consulta
"""
endpoint = f"{self.base_url}/search_products/{product_name}"
products = requests.get(
endpoint,
headers={
"Authorization": f"bearer {self.tryton_key}",
},
)
return products.json()
def check_price(self, product_name):
"""
Verifica el precio de un producto específico
"""
products = self.search_products(product_name)
return products.json()
def check_availability(self, product_name):
"""
Verifica la disponibilidad de un producto espécifico
"""
keys = ["id", "name", "template"]
products = self.clean_keys_from_records(
self.search_products(product_name), keys
)
return products.json()
def get_product_details(self, product_name):
"""
Obtiene detalles completos de un producto
"""
keys = ["id", "name", "description"]
products = self.search_products(product_name)
return products.json()
def list_tools(self):
"""Retorna los metodos de esta clase en una lista"""
raise NotImplementedError
def clean_keys_from_records(self, keys, records):
# Iterar sobre cada registro y eliminar las claves no deseadas
for record in records:
for clave in keys:
record.pop(clave, None) # Elimina la clave si existe
# Mostrar el resultado
return json.dumps(records, indent=4, ensure_ascii=False)