#!/usr/bin/env python3 import argparse import hashlib import logging import time from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from dotenv import load_dotenv from config import Config from db_handler import DatabaseHandler from leadsquared_service import LeadSquaredService def _build_config_dict() -> Dict[str, Any]: cfg: Dict[str, Any] = {} for key in dir(Config): if key.isupper(): cfg[key] = getattr(Config, key) return cfg def _normalize_phone_variants(phone: Any) -> List[str]: digits = "".join(ch for ch in str(phone or "") if ch.isdigit()) if not digits: return [] core10 = digits[-10:] if len(digits) > 10 else digits variants = {digits, f"+{digits}", core10} if len(core10) == 10: variants.update({f"91{core10}", f"+91{core10}", f"0{core10}"}) return [v for v in variants if v] def _extract_lsq_lead_list(payload: Any) -> List[Dict[str, Any]]: if payload is None: return [] if isinstance(payload, list): return payload if isinstance(payload, dict): for key in ("List", "Data", "list", "data"): value = payload.get(key) if isinstance(value, list): return value return [] def _lsq_field(record: Optional[Dict[str, Any]], *keys: str) -> Optional[Any]: if not isinstance(record, dict): return None for key in keys: if key in record and record.get(key) not in (None, ""): return record.get(key) for key in keys: for record_key, value in record.items(): if str(record_key).lower() == str(key).lower() and value not in (None, ""): return value return None def _first_present_str(*values: Any) -> Optional[str]: for value in values: if value is None: continue text = str(value).strip() if text: return text return None def _lsq_display_name(record: Optional[Dict[str, Any]]) -> Optional[str]: if not isinstance(record, dict): return None first = _first_present_str(_lsq_field(record, "FirstName", "first_name")) last = _first_present_str(_lsq_field(record, "LastName", "last_name")) full_name = " ".join([part for part in [first, last] if part]).strip() or None return _first_present_str( _lsq_field(record, "ProspectName", "prospect_name", "ContactName", "contact_name"), full_name, _lsq_field(record, "Name", "name"), ) def _lsq_external_id(record: Optional[Dict[str, Any]], fallback_phone: Optional[str] = None) -> Optional[str]: if not isinstance(record, dict): return None lead_id = _first_present_str(_lsq_field(record, "ProspectID", "LeadId", "Id", "lead_id", "ProspectId")) if lead_id: return lead_id phone = _first_present_str(_lsq_field(record, "Phone", "Mobile", "PhoneNumber"), fallback_phone) or "" email = _first_present_str(_lsq_field(record, "EmailAddress", "email")) or "" name = _lsq_display_name(record) or "" digest_source = f"{phone}|{email}|{name}" return hashlib.sha256(digest_source.encode("utf-8")).hexdigest() def _upsert_lead(db: DatabaseHandler, bid: str, lead: Dict[str, Any]) -> None: """Extract fields from a raw LSQ lead record and upsert into crm_leads_cache.""" phone = _first_present_str(_lsq_field(lead, "Phone", "Mobile", "PhoneNumber")) external_id = _lsq_external_id(lead, phone) db.upsert_crm_lead_cache( bid=bid, provider="leadsquared", external_lead_id=external_id, lead_name=_lsq_display_name(lead), owner_name=_first_present_str(_lsq_field(lead, "OwnerName", "Owner", "owner_name")), email=_first_present_str(_lsq_field(lead, "EmailAddress", "email")), phone_primary=phone, phone_variants=_normalize_phone_variants(phone), lead_status=_first_present_str(_lsq_field(lead, "LeadStatus", "Status", "lead_status")), next_task_due_date=_first_present_str(_lsq_field(lead, "NextTaskDueDate", "next_task_due_date")), lead_payload=lead, last_synced_at=datetime.utcnow(), ) def _sync_bid_leadsquared( db: DatabaseHandler, bid: str, access_key: str, secret_key: str, api_host: Optional[str], page_size: int, max_pages: int, lookback_days_first_run: int, ) -> Dict[str, Any]: """Sync LSQ leads for a bid using GetRecentlyModified with a stored watermark. On the first run it looks back `lookback_days_first_run` days so we catch all existing leads. On subsequent runs it re-uses the last watermark with a 5-minute overlap to avoid missed records near the boundary. PageIndex/PageSize pagination is used (1-based) as required by the GetRecentlyModified endpoint. """ service = LeadSquaredService( access_key=access_key, secret_key=secret_key, api_host=api_host, timeout=60, ) watermark_key = "lsq_leads" last_watermark = db.get_sync_watermark(bid, watermark_key) if last_watermark is None: from_dt = datetime.utcnow() - timedelta(days=lookback_days_first_run) else: from_dt = last_watermark - timedelta(minutes=5) to_dt = datetime.utcnow() from_str = from_dt.strftime("%Y-%m-%d %H:%M:%S") to_str = to_dt.strftime("%Y-%m-%d %H:%M:%S") synced = 0 pages = 0 page_index = 1 new_watermark = from_dt offset = 0 while pages < max_pages: pages += 1 result = service.get_leads_in_date_range( from_date_str=from_str, to_date_str=to_str, offset=offset, row_count=page_size, date_field="ModifiedOn", ) if not result.get("success"): # If AdvancedSearch date query failed, log and stop this bid — don't # silently leave the cache stale. return { "bid": bid, "success": False, "message": result.get("message") or "LeadSquared date-range fetch failed", "pages": pages, "synced": synced, } leads = _extract_lsq_lead_list(result.get("data")) if not leads: break for lead in leads: # Advance watermark using the lead's ModifiedOn / CreatedOn modified_on = _parse_lsq_datetime( _lsq_field(lead, "ModifiedOn", "modified_on", "UpdatedOn", "updated_on") ) created_on = _parse_lsq_datetime( _lsq_field(lead, "CreatedOn", "created_on") ) ts = modified_on or created_on if ts and ts > new_watermark: new_watermark = ts _upsert_lead(db, bid, lead) synced += 1 if len(leads) < page_size: break offset += len(leads) # Advance watermark: use latest lead timestamp if we got one, else advance to to_dt if new_watermark > from_dt: db.set_sync_watermark(bid, watermark_key, new_watermark) else: db.set_sync_watermark(bid, watermark_key, to_dt) return {"bid": bid, "success": True, "pages": pages, "synced": synced} def _parse_lsq_datetime(value: Any) -> Optional[datetime]: """Parse a LeadSquared datetime string to a Python datetime (UTC).""" if not value: return None for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%SZ"): try: return datetime.strptime(str(value)[:19], fmt) except ValueError: continue return None def _sync_bid_lsq_activities( db: DatabaseHandler, bid: str, access_key: str, secret_key: str, api_host: Optional[str], page_size: int, lookback_days_first_run: int, ) -> Dict[str, Any]: """Sync activities for a bid using the RetrieveRecentlyModified endpoint. Uses a watermark (stored in sync_watermarks) so each run only fetches activities modified since the last successful sync. On the first run it goes back `lookback_days_first_run` days. """ service = LeadSquaredService( access_key=access_key, secret_key=secret_key, api_host=api_host, timeout=60, ) watermark_key = "lsq_activities" last_watermark = db.get_sync_watermark(bid, watermark_key) if last_watermark is None: from_dt = datetime.utcnow() - timedelta(days=lookback_days_first_run) else: # Small 5-minute overlap to catch any activities we may have missed from_dt = last_watermark - timedelta(minutes=5) to_dt = datetime.utcnow() from_str = from_dt.strftime("%Y-%m-%d %H:%M:%S") to_str = to_dt.strftime("%Y-%m-%d %H:%M:%S") synced = 0 pages = 0 page_index = 1 new_watermark = from_dt while True: pages += 1 payload = { "Parameter": { "FromDate": from_str, "ToDate": to_str, "IncludeCustomFields": 1, }, "Paging": { "PageIndex": page_index, "PageSize": page_size, }, "Sorting": { "ColumnName": "CreatedOn", "Direction": 0, # ascending so watermark advances correctly }, } result = service._request( method="POST", endpoint="ProspectActivity.svc/RetrieveRecentlyModified", payload=payload, ) if not result.get("success"): return { "bid": bid, "success": False, "message": result.get("message") or "Activity fetch failed", "pages": pages, "synced": synced, } data = result.get("data") or {} activities = [] if isinstance(data, list): activities = data elif isinstance(data, dict): activities = ( data.get("ProspectActivities") or data.get("List") or data.get("Data") or [] ) if not activities: break for act in activities: if not isinstance(act, dict): continue activity_id = _first_present_str(act.get("Id"), act.get("id")) if not activity_id: continue created_on = _parse_lsq_datetime(act.get("CreatedOn") or act.get("created_on")) modified_on = _parse_lsq_datetime(act.get("ModifiedOn") or act.get("modified_on")) if modified_on and modified_on > new_watermark: new_watermark = modified_on elif created_on and created_on > new_watermark: new_watermark = created_on db.upsert_crm_lead_activity( bid=bid, provider="leadsquared", activity_id=activity_id, lead_id=_first_present_str( act.get("RelatedProspectId"), act.get("ProspectId"), act.get("LeadId"), ), event_code=_first_present_str(act.get("EventCode"), act.get("event_code")), event_name=_first_present_str(act.get("EventName"), act.get("event_name")), activity_created_on=created_on, activity_modified_on=modified_on, activity_data=act.get("Data") or act.get("data"), activity_fields=act.get("Fields") or act.get("fields") or act.get("ActivityFields"), ) synced += 1 if len(activities) < page_size: break page_index += 1 # Advance the watermark only if we made progress if new_watermark > from_dt: db.set_sync_watermark(bid, watermark_key, new_watermark) else: # Nothing came back but the run succeeded — move watermark to to_dt db.set_sync_watermark(bid, watermark_key, to_dt) return {"bid": bid, "success": True, "pages": pages, "synced": synced} def run_once( db: DatabaseHandler, leads_page_size: int, max_pages: int, leads_lookback_days: int = 365, activity_page_size: int = 200, activity_lookback_days: int = 90, ) -> Dict[str, Any]: integrations = db.get_active_crm_integrations(provider="leadsquared") if not integrations: return {"success": True, "message": "No active LeadSquared integrations found", "results": []} results = [] for item in integrations: bid = str(item.get("bid")) access_key = item.get("access_key") secret_key = item.get("secret_key") api_host = item.get("api_host") if not access_key or not secret_key: results.append({"bid": bid, "success": False, "message": "Missing decrypted credentials", "lead_pages": 0, "leads_synced": 0, "activity_pages": 0, "activities_synced": 0}) continue try: lead_res = _sync_bid_leadsquared( db=db, bid=bid, access_key=access_key, secret_key=secret_key, api_host=api_host, page_size=leads_page_size, max_pages=max_pages, lookback_days_first_run=leads_lookback_days, ) act_res = _sync_bid_lsq_activities( db=db, bid=bid, access_key=access_key, secret_key=secret_key, api_host=api_host, page_size=activity_page_size, lookback_days_first_run=activity_lookback_days, ) results.append({ "bid": bid, "success": lead_res.get("success") and act_res.get("success"), "lead_pages": lead_res.get("pages", 0), "leads_synced": lead_res.get("synced", 0), "activity_pages": act_res.get("pages", 0), "activities_synced": act_res.get("synced", 0), "lead_message": lead_res.get("message"), "activity_message": act_res.get("message"), }) except Exception as exc: results.append({"bid": bid, "success": False, "message": str(exc), "lead_pages": 0, "leads_synced": 0, "activity_pages": 0, "activities_synced": 0}) return {"success": True, "message": "CRM sync run complete", "results": results} def main(): parser = argparse.ArgumentParser(description="Sync LeadSquared CRM leads + activities into local DB cache") parser.add_argument("--once", action="store_true", help="Run one sync iteration and exit") parser.add_argument("--interval-seconds", type=int, default=300, help="Loop interval in seconds for continuous mode") parser.add_argument("--leads-page-size", type=int, default=200, help="Leads per page from LSQ GetRecentlyModified") parser.add_argument("--max-pages", type=int, default=50, help="Maximum lead pages per bid per cycle") parser.add_argument("--leads-lookback-days", type=int, default=365, help="Days to look back on first leads sync run (default: 365)") parser.add_argument("--activity-page-size", type=int, default=200, help="Records per activity page fetch") parser.add_argument("--activity-lookback-days", type=int, default=90, help="Days to look back on first activity sync run (default: 90)") args = parser.parse_args() load_dotenv() logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger("sync_crm_leads") config = _build_config_dict() db = DatabaseHandler(config) db.ensure_crm_integrations_table() db.ensure_crm_leads_cache_table() db.ensure_crm_lead_activities_table() db.ensure_sync_watermarks_table() safe_leads_page_size = max(10, min(int(args.leads_page_size), 500)) safe_max_pages = max(1, int(args.max_pages)) safe_interval = max(30, int(args.interval_seconds)) safe_leads_lookback_days = max(1, int(args.leads_lookback_days)) safe_activity_page_size = max(10, min(int(args.activity_page_size), 500)) safe_activity_lookback_days = max(1, int(args.activity_lookback_days)) if args.once: outcome = run_once( db=db, leads_page_size=safe_leads_page_size, max_pages=safe_max_pages, leads_lookback_days=safe_leads_lookback_days, activity_page_size=safe_activity_page_size, activity_lookback_days=safe_activity_lookback_days, ) logger.info("CRM sync (once): %s", outcome) return logger.info( "Starting continuous CRM sync worker interval=%ss leads_page_size=%s max_pages=%s " "leads_lookback_days=%s activity_page_size=%s activity_lookback_days=%s", safe_interval, safe_leads_page_size, safe_max_pages, safe_leads_lookback_days, safe_activity_page_size, safe_activity_lookback_days, ) while True: started = time.time() outcome = run_once( db=db, leads_page_size=safe_leads_page_size, max_pages=safe_max_pages, leads_lookback_days=safe_leads_lookback_days, activity_page_size=safe_activity_page_size, activity_lookback_days=safe_activity_lookback_days, ) logger.info("CRM sync cycle complete: %s", outcome) elapsed = time.time() - started sleep_for = max(1, safe_interval - int(elapsed)) time.sleep(sleep_for) if __name__ == "__main__": main()