Spaces:
Running
Running
from typing import List, Dict, Optional, Any | |
import os | |
try: | |
from common.base.api_base_async import BaseAsyncAPI | |
from common.base.api_base import BaseAPI | |
except ImportError: | |
from ..common.base.api_base_async import BaseAsyncAPI | |
from ..common.base.api_base import BaseAPI | |
class RfpRecommend(BaseAPI): | |
def __init__(self): | |
super().__init__( | |
url=os.getenv("RFP_RECOMMENDATION_URL"), | |
headers={"x-api-key": os.getenv("RFP_RECOMMENDATION_API_KEY")}, | |
total_retries=5, | |
backoff_factor=10 | |
) | |
def __call__(self, candid_entity_id: int): | |
results = self.get(candid_entity_id=candid_entity_id) | |
return results | |
class RfpFeedback(BaseAPI): | |
def __init__(self): | |
super().__init__( | |
url=f'{os.getenv("RFP_RECOMMENDATION_URL")}/feedback', | |
headers={"x-api-key": os.getenv("RFP_RECOMMENDATION_API_KEY")}, | |
total_retries=2, | |
backoff_factor=5 | |
) | |
def __call__( | |
self, | |
recommendation_data: Dict[str, Any], | |
ratings: List[str], | |
info_is_correct: Optional[bool] = None, | |
info_is_sufficient: Optional[bool] = None, | |
comments: Optional[str] = None, | |
email: Optional[str] = None, | |
): | |
data = { | |
"recommendations": recommendation_data, | |
"ratings": ratings, | |
"correct_info": info_is_correct, | |
"sufficient_info": info_is_sufficient | |
} | |
if comments: | |
data["comments"] = comments | |
if email: | |
data["email"] = email | |
results = self.post(payload=data) | |
return results | |
class Inferencing(BaseAsyncAPI): | |
def __init__(self): | |
super().__init__( | |
url=os.getenv("AUTOCODING_API_URL"), | |
headers={"x-api-key": os.getenv("AUTOCODING_API_KEY")} | |
) | |
async def __call__(self, text: str, taxonomy: str = "pcs", | |
chunk: bool = False, high_precision: bool = False | |
): | |
result = self.post(payload={ | |
"text": text, | |
"taxonomy": taxonomy, | |
"chunk": chunk, | |
"high_precision": high_precision | |
}) | |
return result | |
class Entities(BaseAsyncAPI): | |
def __init__(self): | |
super().__init__( | |
url=os.getenv("DOCUMENT_API_URL").replace("/analyze", "/entities"), | |
headers={"x-api-key": os.getenv("DOCUMENT_API_KEY")} | |
) | |
async def __call__(self, text: str): | |
result = self.post(payload={"text": text}) | |
return result | |