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