""" Module: lead_enricher.py Description: Enriches lead data with company information. """ import logging import os import json from typing import Dict, Optional # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) class LeadEnricher: """ Enriches lead data with company information. """ def __init__(self, company_data_file: str): """ Initializes the LeadEnricher with a path to a JSON file containing company data. Args: company_data_file (str): The path to the JSON file containing company data. """ self.company_data_file = company_data_file self.company_data: Dict[str, Dict] = self._load_company_data() def _load_company_data(self) -> Dict[str, Dict]: """ Loads company data from the JSON file. Returns: Dict[str, Dict]: A dictionary where the keys are company names and the values are dictionaries containing company information. Returns an empty dictionary if there is an error loading the file. """ try: with open(self.company_data_file, "r") as f: data = json.load(f) logging.info(f"Successfully loaded company data from {self.company_data_file}") return data except FileNotFoundError: logging.error(f"Company data file not found: {self.company_data_file}") return {} except json.JSONDecodeError: logging.error(f"Error decoding JSON from {self.company_data_file}") return {} except Exception as e: logging.error(f"An unexpected error occurred: {e}") return {} def enrich_lead(self, lead: Dict) -> Dict: """ Enriches a lead dictionary with company information based on the company name. Args: lead (Dict): A dictionary representing the lead. Must contain a "company" key. Returns: Dict: The enriched lead dictionary. Returns the original lead if the company is not found or if the lead dictionary does not contain a 'company' key. """ if not isinstance(lead, dict): logging.warning(f"Invalid lead data: {lead}. Lead must be a dictionary.") return lead if "company" not in lead: logging.warning("Lead does not contain a 'company' key. Cannot enrich.") return lead company_name = lead["company"] if company_name in self.company_data: company_info = self.company_data[company_name] enriched_lead = lead.copy() # Avoid modifying the original lead enriched_lead.update(company_info) logging.info(f"Successfully enriched lead for company: {company_name}") return enriched_lead else: logging.warning(f"Company not found in data: {company_name}") return lead def main(): """ Main function to demonstrate the LeadEnricher. """ # Create a sample company data file company_data = { "Acme Corp": {"industry": "Technology", "employees": 1000}, "Beta Inc": {"industry": "Finance", "employees": 500}, } company_data_file = "/tmp/company_data.json" # Using /tmp for example try: with open(company_data_file, "w") as f: json.dump(company_data, f, indent=4) logging.info(f"Created sample company data file: {company_data_file}") except Exception as e: logging.error(f"Failed to create company data file: {e}") return # Initialize the LeadEnricher enricher = LeadEnricher(company_data_file) # Sample lead data lead = {"name": "John Doe", "email": "john.doe@example.com", "company": "Acme Corp"} # Enrich the lead enriched_lead = enricher.enrich_lead(lead) # Print the enriched lead print("Original Lead:", lead) print("Enriched Lead:", enriched_lead) # Clean up the sample file try: os.remove(company_data_file) logging.info(f"Deleted sample company data file: {company_data_file}") except Exception as e: logging.warning(f"Failed to delete company data file: {e}") if __name__ == "__main__": # Create directory if it doesn't exist to prevent errors output_dir = "/mnt/e/genesis-system/core/revenue/" if not os.path.exists(output_dir): os.makedirs(output_dir) # Write this file to disk file_path = os.path.join(output_dir, "lead_enricher.py") with open(file_path, "w") as f: f.write("""\"\"\" Module: lead_enricher.py Description: Enriches lead data with company information. \"\"\" import logging import os import json from typing import Dict, Optional # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) class LeadEnricher: \"\"\" Enriches lead data with company information. \"\"\" def __init__(self, company_data_file: str): \"\"\" Initializes the LeadEnricher with a path to a JSON file containing company data. Args: company_data_file (str): The path to the JSON file containing company data. \"\"\" self.company_data_file = company_data_file self.company_data: Dict[str, Dict] = self._load_company_data() def _load_company_data(self) -> Dict[str, Dict]: \"\"\" Loads company data from the JSON file. Returns: Dict[str, Dict]: A dictionary where the keys are company names and the values are dictionaries containing company information. Returns an empty dictionary if there is an error loading the file. \"\"\" try: with open(self.company_data_file, "r") as f: data = json.load(f) logging.info(f"Successfully loaded company data from {self.company_data_file}") return data except FileNotFoundError: logging.error(f"Company data file not found: {self.company_data_file}") return {} except json.JSONDecodeError: logging.error(f"Error decoding JSON from {self.company_data_file}") return {} except Exception as e: logging.error(f"An unexpected error occurred: {e}") return {} def enrich_lead(self, lead: Dict) -> Dict: \"\"\" Enriches a lead dictionary with company information based on the company name. Args: lead (Dict): A dictionary representing the lead. Must contain a "company" key. Returns: Dict: The enriched lead dictionary. Returns the original lead if the company is not found or if the lead dictionary does not contain a 'company' key. \"\"\" if not isinstance(lead, dict): logging.warning(f"Invalid lead data: {lead}. Lead must be a dictionary.") return lead if "company" not in lead: logging.warning("Lead does not contain a 'company' key. Cannot enrich.") return lead company_name = lead["company"] if company_name in self.company_data: company_info = self.company_data[company_name] enriched_lead = lead.copy() # Avoid modifying the original lead enriched_lead.update(company_info) logging.info(f"Successfully enriched lead for company: {company_name}") return enriched_lead else: logging.warning(f"Company not found in data: {company_name}") return lead def main(): \"\"\" Main function to demonstrate the LeadEnricher. \"\"\" # Create a sample company data file company_data = { "Acme Corp": {"industry": "Technology", "employees": 1000}, "Beta Inc": {"industry": "Finance", "employees": 500}, } company_data_file = "/tmp/company_data.json" # Using /tmp for example try: with open(company_data_file, "w") as f: json.dump(company_data, f, indent=4) logging.info(f"Created sample company data file: {company_data_file}") except Exception as e: logging.error(f"Failed to create company data file: {e}") return # Initialize the LeadEnricher enricher = LeadEnricher(company_data_file) # Sample lead data lead = {"name": "John Doe", "email": "john.doe@example.com", "company": "Acme Corp"} # Enrich the lead enriched_lead = enricher.enrich_lead(lead) # Print the enriched lead print("Original Lead:", lead) print("Enriched Lead:", enriched_lead) # Clean up the sample file try: os.remove(company_data_file) logging.info(f"Deleted sample company data file: {company_data_file}") except Exception as e: logging.warning(f"Failed to delete company data file: {e}") if __name__ == "__main__": main() """) logging.info(f"Successfully wrote lead_enricher.py to {file_path}") #Example usage. # You can uncomment the main() call to see the example run # main()