""" Customer Data Consistency Service Compares customer data across BMC Hub, vTiger Cloud, and e-conomic """ import logging import asyncio from typing import Dict, List, Optional, Tuple, Any from app.core.database import execute_query, execute_query_single, execute_update, execute_insert from app.services.vtiger_service import VTigerService from app.services.economic_service import EconomicService from app.core.config import settings logger = logging.getLogger(__name__) class CustomerConsistencyService: """Service for checking and syncing customer data across systems""" # Field mapping: hub_field -> (vtiger_field, economic_field) FIELD_MAP = { 'name': ('accountname', 'name'), 'cvr_number': ('cf_856', 'corporateIdentificationNumber'), 'address': ('bill_street', 'address'), 'city': ('bill_city', 'city'), 'postal_code': ('bill_code', 'zip'), 'country': ('bill_country', 'country'), 'phone': ('phone', 'telephoneAndFaxNumber'), 'mobile_phone': ('mobile', 'mobilePhone'), 'email': ('email1', 'email'), 'website': ('website', 'website'), 'invoice_email': ('email2', 'email'), } def __init__(self): self.vtiger = VTigerService() self.economic = EconomicService() @staticmethod def _clean_contact_value(value: Any) -> Optional[str]: if value is None: return None text = str(value).strip() return text or None @classmethod def _normalize_contact(cls, source: str, row: Dict[str, Any]) -> Dict[str, Any]: first_name = cls._clean_contact_value( row.get('first_name') if source == 'hub' else row.get('firstname') ) last_name = cls._clean_contact_value( row.get('last_name') if source == 'hub' else row.get('lastname') ) email = cls._clean_contact_value(row.get('email')) phone = cls._clean_contact_value(row.get('phone')) mobile = cls._clean_contact_value(row.get('mobile')) title = cls._clean_contact_value(row.get('title')) department = cls._clean_contact_value(row.get('department')) vtiger_id = cls._clean_contact_value( row.get('vtiger_id') if source == 'hub' else row.get('id') ) display_name = " ".join(part for part in [first_name, last_name] if part).strip() if not display_name: display_name = email or phone or mobile or vtiger_id or 'Ukendt kontakt' return { 'id': row.get('id'), 'vtiger_id': vtiger_id, 'first_name': first_name, 'last_name': last_name, 'email': email, 'phone': phone, 'mobile': mobile, 'title': title, 'department': department, 'display_name': display_name, 'is_primary': bool(row.get('is_primary')) if source == 'hub' else False, 'role': cls._clean_contact_value(row.get('role')) if source == 'hub' else None, 'customer_count': row.get('customer_count', 0) or 0, } @staticmethod def _contact_match_key(contact: Dict[str, Any]) -> str: vtiger_id = str(contact.get('vtiger_id') or '').strip().lower() if vtiger_id: return f"vtiger:{vtiger_id}" email = str(contact.get('email') or '').strip().lower() if email: return f"email:{email}" first_name = str(contact.get('first_name') or '').strip().lower() last_name = str(contact.get('last_name') or '').strip().lower() phone = str(contact.get('phone') or '').strip().lower() mobile = str(contact.get('mobile') or '').strip().lower() return f"name:{first_name}|{last_name}|{phone}|{mobile}" async def fetch_hub_contacts(self, customer_id: int) -> List[Dict[str, Any]]: query = """ SELECT c.*, cc.is_primary, cc.role, ( SELECT COUNT(*) FROM contact_companies cc2 WHERE cc2.contact_id = c.id ) AS customer_count FROM contacts c JOIN contact_companies cc ON cc.contact_id = c.id WHERE cc.customer_id = %s AND c.is_active = TRUE ORDER BY cc.is_primary DESC, c.first_name, c.last_name, c.id """ rows = await asyncio.to_thread(execute_query, query, (customer_id,)) return [self._normalize_contact('hub', row) for row in (rows or [])] async def fetch_vtiger_contacts(self, vtiger_customer_id: Optional[str]) -> List[Dict[str, Any]]: if not vtiger_customer_id or not settings.VTIGER_URL: return [] safe_customer_id = self.vtiger._sanitize_vtiger_id(vtiger_customer_id) if not safe_customer_id: return [] rows: List[Dict[str, Any]] = [] seen_ids = set() # vTiger installations are inconsistent about which relation field is exposed on Contacts. relation_fields = ("account_id", "accountid", "parent_id", "account") for field_name in relation_fields: query = f"SELECT * FROM Contacts WHERE {field_name}='{safe_customer_id}';" result = await self.vtiger.query(query) for row in (result or []): row_id = str(row.get("id") or "").strip() dedupe_key = row_id or f"{row.get('email')}|{row.get('firstname')}|{row.get('lastname')}" if dedupe_key in seen_ids: continue seen_ids.add(dedupe_key) rows.append(row) return [self._normalize_contact('vtiger', row) for row in rows] @staticmethod def normalize_value(value: Any) -> Optional[str]: """ Normalize value for comparison - Convert to string - Strip whitespace - Lowercase - Convert empty strings to None """ if value is None: return None # Convert to string str_value = str(value).strip() # Empty string to None if not str_value: return None # Lowercase for case-insensitive comparison return str_value.lower() async def fetch_all_data(self, customer_id: int) -> Dict[str, Any]: """ Fetch customer data from all three systems in parallel Args: customer_id: Hub customer ID Returns: Dict with keys 'hub', 'vtiger', 'economic' containing raw data (or None) """ logger.info(f"šŸ” Fetching customer data from all systems for customer {customer_id}") # Fetch Hub data first to get mapping IDs hub_query = """ SELECT * FROM customers WHERE id = %s """ hub_data = await asyncio.to_thread(execute_query_single, hub_query, (customer_id,)) if not hub_data: raise ValueError(f"Customer {customer_id} not found in Hub") # Prepare async tasks for vTiger and e-conomic vtiger_task = None economic_task = None hub_contacts_task = self.fetch_hub_contacts(customer_id) vtiger_contacts_task = None # Fetch vTiger data if we have an ID and vTiger is configured if hub_data.get('vtiger_id') and settings.VTIGER_URL: vtiger_task = self.vtiger.get_account_by_id(hub_data['vtiger_id']) vtiger_contacts_task = self.fetch_vtiger_contacts(hub_data['vtiger_id']) # Fetch e-conomic data if we have a customer number and e-conomic is configured if hub_data.get('economic_customer_number') and settings.ECONOMIC_APP_SECRET_TOKEN: economic_task = self.economic.get_customer(hub_data['economic_customer_number']) # Parallel fetch with error handling tasks = {} if vtiger_task: tasks['vtiger'] = vtiger_task if economic_task: tasks['economic'] = economic_task tasks['hub_contacts'] = hub_contacts_task if vtiger_contacts_task: tasks['vtiger_contacts'] = vtiger_contacts_task results = {} if tasks: task_results = await asyncio.gather( *tasks.values(), return_exceptions=True ) # Map results back for key, result in zip(tasks.keys(), task_results): if isinstance(result, Exception): logger.error(f"āŒ Error fetching {key} data: {result}") results[key] = None else: results[key] = result return { 'hub': hub_data, 'vtiger': results.get('vtiger'), 'economic': results.get('economic'), 'hub_contacts': results.get('hub_contacts') or [], 'vtiger_contacts': results.get('vtiger_contacts') or [], } @classmethod def compare_data(cls, all_data: Dict[str, Optional[Dict[str, Any]]]) -> Dict[str, Dict[str, Any]]: """ Compare data across systems and identify discrepancies Args: all_data: Dict with 'hub', 'vtiger', 'economic' data (values may be None) Returns: Dict of discrepancies: { field_name: { 'hub': value, 'vtiger': value, 'economic': value, 'discrepancy': True/False } } """ discrepancies = {} hub_data = all_data.get('hub', {}) vtiger_data = all_data.get('vtiger', {}) economic_data = all_data.get('economic', {}) for hub_field, (vtiger_field, economic_field) in cls.FIELD_MAP.items(): # Get raw values hub_value = hub_data.get(hub_field) vtiger_value = vtiger_data.get(vtiger_field) if vtiger_data else None economic_value = economic_data.get(economic_field) if economic_data else None # Normalize for comparison hub_norm = cls.normalize_value(hub_value) vtiger_norm = cls.normalize_value(vtiger_value) economic_norm = cls.normalize_value(economic_value) # Check if all values are the same # Only compare systems that are available available_values = [] if hub_data: available_values.append(hub_norm) if vtiger_data: available_values.append(vtiger_norm) if economic_data: available_values.append(economic_norm) # Has discrepancy if there are different non-None values has_discrepancy = len(set(available_values)) > 1 if len(available_values) > 1 else False discrepancies[hub_field] = { 'hub': hub_value, 'vtiger': vtiger_value, 'economic': economic_value, 'discrepancy': has_discrepancy } return discrepancies @classmethod def compare_contacts(cls, all_data: Dict[str, Any]) -> List[Dict[str, Any]]: hub_contacts = all_data.get('hub_contacts') or [] vtiger_contacts = all_data.get('vtiger_contacts') or [] hub_by_key = {cls._contact_match_key(contact): contact for contact in hub_contacts} discrepancies: List[Dict[str, Any]] = [] for vtiger_contact in vtiger_contacts: match_key = cls._contact_match_key(vtiger_contact) hub_contact = hub_by_key.get(match_key) if not hub_contact: discrepancies.append({ 'match_key': match_key, 'action': 'create_or_link', 'reason': 'Kontakten findes i vTiger men ikke pĆ„ denne kunde i Hub', 'hub': None, 'vtiger': vtiger_contact, 'selectable': True, }) continue changed_fields = [] for field in ('first_name', 'last_name', 'email', 'phone', 'mobile', 'title', 'department'): if cls.normalize_value(hub_contact.get(field)) != cls.normalize_value(vtiger_contact.get(field)): changed_fields.append(field) if changed_fields: discrepancies.append({ 'match_key': match_key, 'action': 'update_hub', 'reason': 'Kontakt findes begge steder men har feltforskelle', 'hub': hub_contact, 'vtiger': vtiger_contact, 'changed_fields': changed_fields, 'selectable': True, }) else: discrepancies.append({ 'match_key': match_key, 'action': 'matched', 'reason': 'Kontakt findes allerede i Hub og matcher vTiger', 'hub': hub_contact, 'vtiger': vtiger_contact, 'changed_fields': [], 'selectable': False, }) return discrepancies async def sync_field( self, customer_id: int, field_name: str, source_system: str, source_value: Any ) -> Dict[str, bool]: """ Sync a field value to all enabled systems Args: customer_id: Hub customer ID field_name: Hub field name (from FIELD_MAP keys) source_system: 'hub', 'vtiger', or 'economic' source_value: The correct value to sync Returns: Dict with sync status: {'hub': True/False, 'vtiger': True/False, 'economic': True/False} """ logger.info(f"šŸ”„ Syncing {field_name} from {source_system} with value: {source_value}") if field_name not in self.FIELD_MAP: raise ValueError(f"Unknown field: {field_name}") _, economic_field = self.FIELD_MAP[field_name] # Fetch Hub data to get mapping IDs hub_query = "SELECT * FROM customers WHERE id = %s" hub_data = await asyncio.to_thread(execute_query_single, hub_query, (customer_id,)) if not hub_data: raise ValueError(f"Customer {customer_id} not found") results = {} # Update Hub if not the source if source_system != 'hub': try: update_query = f"UPDATE customers SET {field_name} = %s WHERE id = %s" await asyncio.to_thread(execute_update, update_query, (source_value, customer_id)) results['hub'] = True logger.info(f"āœ… Hub {field_name} updated") except Exception as e: logger.error(f"āŒ Failed to update Hub: {e}") results['hub'] = False else: results['hub'] = True # Already correct # vTiger is read-only for this workflow. results['vtiger'] = True # Update e-conomic if enabled and not the source if settings.ECONOMIC_SYNC_ENABLED and source_system != 'economic' and hub_data.get('economic_customer_number'): try: # e-conomic update requires different handling based on field update_data = {economic_field: source_value} # Check safety flags if settings.ECONOMIC_READ_ONLY or settings.ECONOMIC_DRY_RUN: logger.warning(f"āš ļø e-conomic update blocked by safety flags (READ_ONLY={settings.ECONOMIC_READ_ONLY}, DRY_RUN={settings.ECONOMIC_DRY_RUN})") results['economic'] = False else: await self.economic.update_customer(hub_data['economic_customer_number'], update_data) results['economic'] = True logger.info(f"āœ… e-conomic {economic_field} updated") except Exception as e: logger.error(f"āŒ Failed to update e-conomic: {e}") results['economic'] = False else: results['economic'] = True # Not applicable or already correct return results async def sync_vtiger_contacts_to_hub( self, customer_id: int, selected_match_keys: List[str], ) -> Dict[str, Any]: if not selected_match_keys: return {"selected": 0, "created": 0, "updated": 0, "linked": 0, "skipped": 0} all_data = await self.fetch_all_data(customer_id) discrepancies = self.compare_contacts(all_data) selected = {str(item or '').strip() for item in selected_match_keys if str(item or '').strip()} selected_rows = [row for row in discrepancies if row.get('match_key') in selected] stats = {"selected": len(selected_rows), "created": 0, "updated": 0, "linked": 0, "skipped": 0} for row in selected_rows: vtiger_contact = row.get('vtiger') or {} hub_contact = row.get('hub') if not vtiger_contact: stats["skipped"] += 1 continue contact_id = hub_contact.get('id') if hub_contact else None if not contact_id: vtiger_id = vtiger_contact.get("vtiger_id") email = vtiger_contact.get("email") if vtiger_id: existing_global = await asyncio.to_thread( execute_query_single, "SELECT id FROM contacts WHERE vtiger_id = %s LIMIT 1", (vtiger_id,), ) contact_id = existing_global.get("id") if existing_global else None if not contact_id and email: existing_global = await asyncio.to_thread( execute_query_single, "SELECT id FROM contacts WHERE LOWER(COALESCE(email, '')) = %s LIMIT 1", (str(email).strip().lower(),), ) contact_id = existing_global.get("id") if existing_global else None if contact_id: update_fields = { "first_name": vtiger_contact.get("first_name"), "last_name": vtiger_contact.get("last_name"), "email": vtiger_contact.get("email"), "phone": vtiger_contact.get("phone"), "mobile": vtiger_contact.get("mobile"), "title": vtiger_contact.get("title"), "department": vtiger_contact.get("department"), "vtiger_id": vtiger_contact.get("vtiger_id"), } await asyncio.to_thread( execute_update, """ UPDATE contacts SET first_name = %s, last_name = %s, email = %s, phone = %s, mobile = %s, title = %s, department = %s, vtiger_id = COALESCE(%s, vtiger_id) WHERE id = %s """, ( update_fields["first_name"], update_fields["last_name"], update_fields["email"], update_fields["phone"], update_fields["mobile"], update_fields["title"], update_fields["department"], update_fields["vtiger_id"], contact_id, ), ) stats["updated"] += 1 else: contact_id = await asyncio.to_thread( execute_insert, """ INSERT INTO contacts ( first_name, last_name, email, phone, mobile, title, department, vtiger_id ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) RETURNING id """, ( vtiger_contact.get("first_name"), vtiger_contact.get("last_name"), vtiger_contact.get("email"), vtiger_contact.get("phone"), vtiger_contact.get("mobile"), vtiger_contact.get("title"), vtiger_contact.get("department"), vtiger_contact.get("vtiger_id"), ), ) stats["created"] += 1 link_exists = await asyncio.to_thread( execute_query_single, """ SELECT id FROM contact_companies WHERE contact_id = %s AND customer_id = %s """, (contact_id, customer_id), ) if not link_exists: await asyncio.to_thread( execute_update, """ INSERT INTO contact_companies (contact_id, customer_id, is_primary, role) VALUES (%s, %s, %s, %s) ON CONFLICT (contact_id, customer_id) DO NOTHING """, (contact_id, customer_id, False, None), ) stats["linked"] += 1 return stats