import logging from typing import Any, Dict, Optional import requests logger = logging.getLogger(__name__) class LeadSquaredService: """Python port of LeadSquared integration service used in tatsat2dev/leadsquared.""" def __init__( self, access_key: str, secret_key: str, api_host: Optional[str] = None, timeout: int = 30, ): self.access_key = (access_key or "").strip() self.secret_key = (secret_key or "").strip() self.api_host = (api_host or "https://api-in21.leadsquared.com/v2/").rstrip("/") self.timeout = int(timeout) def _auth_params(self) -> Dict[str, str]: return { "accessKey": self.access_key, "secretKey": self.secret_key, } def _request( self, method: str, endpoint: str, query_params: Optional[Dict[str, Any]] = None, payload: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: if not self.access_key or not self.secret_key: return { "success": False, "message": "LeadSquared credentials are missing", "data": None, } url = f"{self.api_host}/{endpoint.lstrip('/')}" params = self._auth_params() if query_params: params.update(query_params) try: response = requests.request( method=method.upper(), url=url, params=params, json=payload, timeout=self.timeout, ) if response.ok: parsed: Any try: parsed = response.json() except Exception: parsed = {"raw": response.text} return { "success": True, "message": "Request successful", "data": parsed, } logger.error( "LeadSquared API request failed: %s %s status=%s body=%s", method, endpoint, response.status_code, response.text, ) return { "success": False, "message": f"LeadSquared API error ({response.status_code}): {response.text}", "data": None, } except Exception as exc: logger.error("LeadSquared request exception: %s", exc, exc_info=True) return { "success": False, "message": f"Exception: {str(exc)}", "data": None, } def test_connection(self) -> Dict[str, Any]: result = self.search_leads( { "Paging": { "Offset": 0, "RowCount": 1, } } ) if result.get("success"): return { "success": True, "message": "LeadSquared connection successful", "credentials": { "access_key": f"{self.access_key[:10]}..." if self.access_key else "", "api_host": self.api_host, }, } return { "success": False, "message": f"LeadSquared connection failed: {result.get('message')}", } def search_leads(self, search_params: Dict[str, Any]) -> Dict[str, Any]: return self._request( method="POST", endpoint="LeadManagement.svc/Leads.Get", payload=search_params or {}, ) def get_lead(self, lead_id: str) -> Dict[str, Any]: return self._request( method="GET", endpoint="LeadManagement.svc/Leads.GetById", query_params={"id": str(lead_id)}, ) def create_lead(self, lead_data: Dict[str, Any]) -> Dict[str, Any]: return self._request( method="POST", endpoint="LeadManagement.svc/Lead.Create", payload=lead_data or {}, ) def update_lead(self, lead_id: str, lead_data: Dict[str, Any]) -> Dict[str, Any]: return self._request( method="POST", endpoint="LeadManagement.svc/Lead.Update", query_params={"leadId": str(lead_id)}, payload=lead_data or {}, ) def delete_lead(self, lead_id: str) -> Dict[str, Any]: # LeadSquared supports delete endpoint in v2 for lead records. return self._request( method="POST", endpoint="LeadManagement.svc/Lead.Delete", query_params={"leadId": str(lead_id)}, payload={}, ) def get_leads_in_date_range( self, from_date_str: str, to_date_str: str, offset: int = 0, row_count: int = 200, date_field: str = "ModifiedOn", ) -> Dict[str, Any]: """Fetch leads whose `date_field` falls within [from_date_str, to_date_str]. Uses Leads.Get with AdvancedSearch so it works on all LSQ account tiers. Pagination is offset-based (Offset / RowCount) matching Leads.Get convention. `date_field` can be 'ModifiedOn' (default) or 'CreatedOn'. On a fresh / first-run call, pass a very old from_date to retrieve all historical leads. """ payload: Dict[str, Any] = { "AdvancedSearch": { "Conditions": [ { "Field": date_field, "Condition": "GreaterOrEqualTo", "Value": from_date_str, }, { "Field": date_field, "Condition": "LessOrEqualTo", "Value": to_date_str, }, ], "Operator": "AND", }, "Paging": { "Offset": offset, "RowCount": row_count, }, } return self._request( method="POST", endpoint="LeadManagement.svc/Leads.Get", payload=payload, ) @staticmethod def _phone_digits(value: Any) -> str: """Return only digit characters from a phone string.""" return "".join(ch for ch in str(value or "") if ch.isdigit()) def _lead_matches_phone(self, lead: Dict[str, Any], query_digits: str) -> bool: """Return True if any phone field on the lead contains `query_digits` as a suffix. LSQ stores phones with country-code prefixes (+91-XXXXXXXXXX). We normalise both sides to digits and check that either the query is a suffix of the stored number or vice-versa (handles 10-digit vs 12-digit variants). """ core = query_digits[-10:] if len(query_digits) >= 10 else query_digits for field in ("Phone", "Mobile", "PhoneNumber"): stored_raw = lead.get(field) if not stored_raw: continue stored = self._phone_digits(stored_raw) stored_core = stored[-10:] if len(stored) >= 10 else stored if stored_core == core or stored == query_digits: return True return False def search_lead_by_phone(self, phone: str) -> Dict[str, Any]: normalized = self._phone_digits(phone) if not normalized: return {"success": False, "message": "Phone is required", "data": None} # Build search payloads: AdvancedSearch with exact conditions is most # reliable; plain SearchBy is kept as a fallback. core = normalized[-10:] if len(normalized) >= 10 else normalized search_values = list(dict.fromkeys([normalized, core])) # dedup, preserve order payloads = [] for val in search_values: payloads += [ { "AdvancedSearch": { "Conditions": [{"Field": "Phone", "Condition": "Equal", "Value": val}], }, "Paging": {"Offset": 0, "RowCount": 5}, }, { "AdvancedSearch": { "Conditions": [{"Field": "Mobile", "Condition": "Equal", "Value": val}], }, "Paging": {"Offset": 0, "RowCount": 5}, }, ] # Fallback: SearchBy Parameter (less reliable but covers edge cases) payloads += [ {"Parameter": {"SearchBy": "Phone", "Value": core}, "Paging": {"Offset": 0, "RowCount": 5}}, {"Parameter": {"SearchBy": "Mobile", "Value": core}, "Paging": {"Offset": 0, "RowCount": 5}}, ] for payload in payloads: result = self.search_leads(payload) if not result.get("success"): continue data = result.get("data") rows: list = [] if isinstance(data, list): rows = data elif isinstance(data, dict): rows = ( data.get("List") or data.get("Data") or data.get("list") or data.get("data") or [] ) # Verify each candidate actually has the queried phone for lead in rows: if isinstance(lead, dict) and self._lead_matches_phone(lead, normalized): return {"success": True, "message": "Lead found", "data": [lead]} return {"success": False, "message": "No LeadSquared lead found for phone", "data": None}