Untitled
unknown
plain_text
10 months ago
3.7 kB
6
Indexable
import boto3 from pydantic import BaseModel from typing import List import json class UserPrompt(BaseModel): query: str class CustomerSearchResultItem(BaseModel): score: str customer_type: str customer_key: str customer_name: str child_details: str customer_id: str class CustomerSearchResults(BaseModel): hits: List[CustomerSearchResultItem] found: int class CustomerCloudSearchClient: def __init__(self, domain_endpoint: str): self.client = boto3.client('cloudsearchdomain', endpoint_url=domain_endpoint, region_name="us-east-1", aws_access_key_id="AKIA5YW5EQVDQGM6NQAU",aws_secret_access_key="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") def search(self, fields:List[str], query: str) -> CustomerSearchResults: response = self.client.search( query=query, queryParser='simple', size=10, queryOptions = json.dumps({"fields":fields}), sort = '_score desc' ) search_results = [] for hit in response['hits']['hit']: search_results.append(CustomerSearchResultItem( score=hit['fields'].get('_score', [''])[0], customer_type=hit['fields'].get('customer_type', [''])[0], customer_key=hit['fields'].get('customer_key', [''])[0], customer_name=hit['fields'].get('customer_name', [''])[0], child_details=hit['fields'].get('child_details', [''])[0], customer_id=hit['fields'].get('customer_id', [''])[0] )) return CustomerSearchResults( hits=search_results, found=response['hits']['found'] ) def format_customer_search_results(customer_search_results: CustomerSearchResults, keyword: str) -> List[dict]: """ Formats the customer search results into a list of dictionaries representing each result item. Args: - customer_search_results (CustomerSearchResults): The customer search results object containing customer information. - keyword (str): The keyword used for the search. Returns: - list: List of dictionaries, each representing a formatted customer search result. """ formatted_results = [] for item in customer_search_results.hits: # Parse child_details as JSON child_details = json.loads(item.child_details) # Format matches list matches = [ { "query_by_value": detail["key"], "lookup_value": detail["name"], "user_friendly_value": "" } for detail in child_details ] # Construct formatted result dictionary formatted_result = { "type": "customer", "index_name": "fincopilot-dim-customer", "total_count": customer_search_results.found, "extracted_count": len(matches), "matched_on": keyword, "matches": matches } formatted_results.append(formatted_result) return formatted_results def handle_user_prompt_for_Customer(user_prompt: UserPrompt, field:str) -> CustomerSearchResults: with open('config.json', 'r') as f: config = json.load(f) client = CustomerCloudSearchClient(config['CUSTOMER_DOMAIN_ENDPOINT']) results = client.search(fields= [field], query=user_prompt.query) return results if __name__ == "__main__": # Replace with your CloudSearch domain endpoint query="Australia" user_prompt = UserPrompt(query="Australia") results = handle_user_prompt_for_Customer(user_prompt, field = 'customer_name') formatted_results = format_customer_search_results(results, query) print(formatted_results)
Editor is loading...
Leave a Comment