Untitled

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)