import logging import json import math from typing import Dict, Optional, List, Any from urllib.parse import urlparse from sqlalchemy.orm import Session from src.database import engine, SessionLocal from src.config import gemini_settings from .tech_detector import LeadEnricher from .analyzer import MarketingLeadAnalyzer from .models import EnrichedLead, Base logger = logging.getLogger(__name__) class EnrichmentService: def __init__(self): self.tech_detector = LeadEnricher() self.ai_analyzer = MarketingLeadAnalyzer(api_key=gemini_settings.api_key) Base.metadata.create_all(bind=engine) def _get_base_url(self, url: str) -> str: try: parsed = urlparse(url) if parsed.scheme and parsed.netloc: return f"{parsed.scheme}://{parsed.netloc}" return url except Exception: return url def process_lead(self, lead_data: Dict) -> Optional[EnrichedLead]: # 0. Check for existing lead place_id = lead_data.get('place_id') if place_id: with SessionLocal() as db: existing_lead = db.query(EnrichedLead).filter(EnrichedLead.place_id == place_id).first() if existing_lead: return {"existing_lead":True, "data":existing_lead} raw_website = lead_data.get('website') website = self._get_base_url(raw_website) if raw_website else None extracted_info = {"emails": [], "phones": [], "socials": [], "confidence_score": 0} ai_info = {} # Initial data collection initial_emails = lead_data.get('emails', []) extracted_info["emails"] = initial_emails if isinstance(initial_emails, list) else [initial_emails] if lead_data.get('phone'): extracted_info["phones"].append(lead_data.get('phone')) # 1. Enrichment if website and isinstance(website, str) and website.startswith('http'): try: wa_info = self.tech_detector.process_target(website) deep_data = wa_info.get("extracted_data", {}) extracted_info["emails"].extend(deep_data.get("emails", [])) extracted_info["phones"].extend(deep_data.get("phones", [])) extracted_info["socials"] = deep_data.get("socials", []) extracted_info["confidence_score"] = wa_info.get("confidence_score", 0) ai_info = self.ai_analyzer.perform_audit(website) except Exception as e: logger.error(f"Enrichment error for {website}: {e}") # 2. Helper Functions for Cleaning (Inside scope for lead-specific logic) def is_val_nan(val): if val is None: return True if isinstance(val, float) and math.isnan(val): return True if isinstance(val, str) and val.lower() in ['nan', 'none', 'null', '']: return True return False def clean_value(val, max_str_len=1000): """Deeply cleans and truncates strings to prevent DB overflow.""" if is_val_nan(val): return None if isinstance(val, list): return [clean_value(v, max_str_len) for v in val[:20]] # Limit lists to 20 items if isinstance(val, dict): return {str(k): clean_value(v, max_str_len) for k, v in val.items()} if isinstance(val, str): return val[:max_str_len] # Hard cap on string length return val def safe_json_prepare(val, max_items=10): """Ensures JSON fields are valid, non-null, and sized reasonably.""" if is_val_nan(val): return None data = val if isinstance(val, str): try: data = json.loads(val) except: data = val return clean_value(data) # 3. Data Deduplication unique_emails = list(set([e for e in extracted_info["emails"] if e and not is_val_nan(e)])) unique_phones = list(set([p for p in extracted_info["phones"] if p and not is_val_nan(p)])) social_map = clean_value(self._map_social_links(extracted_info["socials"])) # 4. Map to Model res_val = safe_json_prepare(lead_data.get('reservations')) order_val = safe_json_prepare(lead_data.get('order_online')) try: enriched_lead = EnrichedLead( campaign_name=str(lead_data.get('campaign_name', ''))[:255], run_id=str(lead_data.get('run_id', ''))[:255], user_id=str(lead_data.get('user_id', ''))[:255], job_id=str(lead_data.get('job_id', ''))[:255], website=website, email=unique_emails[0] if unique_emails else None, whatsapp_number=unique_phones[0] if unique_phones else None, additional_phone_numbers=unique_phones[1:10], # Keep first 10 social_links=social_map, enrichment_confidence_score=float(extracted_info.get("confidence_score", 0)) / 100.0, seo_score=int(ai_info.get("seo_score", 0) or 0), conversion_score=int(ai_info.get("conversion_score", 0) or 0), identified_problems=clean_value(ai_info.get("weaknesses", [])), recommended_actions=clean_value(ai_info.get("recommended_actions", [])), tech_stack=clean_value(ai_info.get("tech_stack_detected", [])), business_type=str(ai_info.get("business_type", ""))[:255], input_id=str(lead_data.get('input_id', '')), title=str(lead_data.get('title', lead_data.get('Title', '')))[:500], category=str(lead_data.get('category', ''))[:255], address=str(lead_data.get('address', '')), review_count=int(lead_data.get('review_count') or 0), review_rating=float(lead_data.get('review_rating') or 0), place_id=str(lead_data.get('place_id', '')), # These are now cleaned/truncated instead of stripped to keys open_hours=safe_json_prepare(lead_data.get('open_hours')), popular_times=safe_json_prepare(lead_data.get('popular_times')), descriptions=safe_json_prepare(lead_data.get('descriptions')), reservations=res_val, order_online=order_val, booking_system=bool(res_val or order_val), menu=safe_json_prepare(lead_data.get('menu')), owner=safe_json_prepare(lead_data.get('owner')), about=safe_json_prepare(lead_data.get('about')), user_reviews=safe_json_prepare(lead_data.get('user_reviews')) ) # 5. Database Save with SessionLocal() as db: try: db.add(enriched_lead) db.commit() db.refresh(enriched_lead) return {"existing_lead":False, "data":enriched_lead} except Exception as db_err: db.rollback() if "unique" in str(db_err).lower(): return None logger.error(f"DB Insert Failure: {db_err}") return None except Exception as mapping_err: logger.error(f"Mapping Error: {mapping_err}") return None def _map_social_links(self, links: List[str]) -> Dict[str, str]: mapping = {} platforms = ['facebook', 'instagram', 'linkedin', 'twitter', 'youtube'] for link in links: if not isinstance(link, str): continue for platform in platforms: if platform in link.lower(): mapping[platform] = link return mapping