#!/usr/bin/env python3 import json,sys,argparse,re from datetime import datetime from zoneinfo import ZoneInfo, ZoneInfoNotFoundError UNKNOWN={None,"","Unknown","unknown"} REQ={"route":"REQ_ROUTE_SUPPORT","permissions":"REQ_ROUTE_SUPPORT","source":"REQ_ROUTE_SUPPORT","connector":"REQ_ROUTE_SUPPORT","field":"REQ_FIELD_VALIDATION","person":"REQ_PERSON_MATCH","member":"REQ_CAMPAIGN_MEMBER","owner":"REQ_OWNER_ASSIGNMENT","population":"REQ_POPULATION_RECONCILIATION"} def get(o,path,default=None): cur=o for k in path.split('.'): if not isinstance(cur,dict) or k not in cur:return default cur=cur[k] return cur def known(v): return not (v is None or v=="" or v=="Unknown" or v=="unknown") def evidence_value(v): if not isinstance(v,str) or not v.strip(): return False value=v.strip().lower() blocked=(r'\bunknown\b',r'\bpending\b',r'\bawait(?:ing)?\b',r'\bfuture\b',r'\bwill be\b',r'\bplanned\b',r'\brequested\b',r'\bunverified\b',r'\bnot observed\b',r'\bdenied\b',r'\bmissing\b',r'\binsufficient\b',r'\bno access\b',r'\btbd\b') return not any(re.search(pattern,value) for pattern in blocked) def data_value(v): return isinstance(v,str) and bool(v.strip()) and v.strip().lower()!='unknown' ISO_TIME=re.compile(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})$') def instant(v): if not isinstance(v,str) or not ISO_TIME.fullmatch(v): return None try: result=datetime.fromisoformat(v.replace('Z','+00:00')) return result if result.utcoffset() is not None else None except ValueError: return None def valid_cutoff(c): if not isinstance(c,dict): return None start,end,deadline=instant(c.get('start')),instant(c.get('end')),instant(c.get('receipt_deadline')) if not start or not end or not deadline or start>=end or deadlineend and not evidence_value(c.get('lag_rule_reference')): return None return start,end,deadline def receipt_time(v,cutoff,prior,cohort_only=False): timestamp=instant(v) latest=cutoff[1] if cohort_only else cutoff[2] return timestamp if timestamp and cutoff[0]<=timestamp<=latest and timestamp>=prior else None def diagnostic(v): if v is None: kind='null' elif isinstance(v,list): kind='array' elif isinstance(v,dict): kind='object' elif isinstance(v,bool): kind='boolean' elif isinstance(v,str): kind='string' elif isinstance(v,(int,float)): kind='number' else: kind=type(v).__name__.lower() try: value=json.dumps(v,ensure_ascii=False,separators=(',',':')) except TypeError: value=json.dumps(str(v),ensure_ascii=False,separators=(',',':')) return {'kind':kind,'value':value} def result(disposition,state,request_id,detail,findings=None): return {"disposition":disposition,"first_broken_state":state,"request_id":request_id,"detail":detail,"findings":findings or []} def evaluate(o): if not isinstance(o,dict): finding={'path':'$',**diagnostic(o)} return result("REPAIR_REQUIRED","record","REQ_ROUTE_SUPPORT","Record must be a JSON object.",[finding]) if get(o,'route.support_state')!='documented' or not evidence_value(get(o,'route.support_reference')): return result("REPAIR_REQUIRED","route_support",REQ['route'],"Document the chosen paid-form route and support reference.") for p in ['scope.linkedin_ad_account_id','scope.linkedin_page_id','scope.linkedin_form_id','scope.linkedin_response_id','scope.salesforce_environment','scope.salesforce_org_reference','scope.campaign_id']: if not evidence_value(get(o,p)): return result("REPAIR_REQUIRED","scope",REQ['route'],f"Supply current observed {p}; pending, denied, future, missing, or unverified values do not prove identity.") cutoff=valid_cutoff(get(o,'scope.cutoff')) if not cutoff: return result("REPAIR_REQUIRED","scope",REQ['route'],"Require a parseable ordered source cutoff and receipt deadline, with a valid IANA timezone, matching offsets and evidence for any added processing lag.") if not evidence_value(get(o,'route.observed_authorization')): return result("REPAIR_REQUIRED","permissions",REQ['permissions'],"Require observed authorization for the chosen connector route and Salesforce org.") for p in ['permissions.linkedin_ad_account_role','permissions.linkedin_page_role','permissions.business_manager_alignment','permissions.salesforce_access','permissions.observed_at']: if not evidence_value(get(o,p)): return result("REPAIR_REQUIRED","permissions",REQ['permissions'],f"Supply current observed {p}; pending, denied, future, missing, or unverified values do not prove access.") permission_time=instant(get(o,'permissions.observed_at')) if not permission_time or permission_time>cutoff[1]: return result("REPAIR_REQUIRED","permissions",REQ['permissions'],"Permission observation needs a parseable timestamp no later than the cutoff end.") rid=get(o,'scope.linkedin_response_id'); env=get(o,'scope.salesforce_environment') src=get(o,'receipts.source_response'); con=get(o,'receipts.connector') if not isinstance(src,dict): return result("REPAIR_REQUIRED","source_response",REQ['source'],"receipts.source_response must be an object.",[{'path':'receipts.source_response',**diagnostic(src)}]) if src.get('state')!='observed' or not evidence_value(src.get('linkedin_response_id')) or src.get('linkedin_response_id')!=rid or not evidence_value(src.get('observed_at')) or not evidence_value(src.get('evidence_reference')): return result("REPAIR_REQUIRED","source_response",REQ['source'],"Observe the exact LinkedIn response ID with a current timestamp and evidence reference before checking downstream states.") prior=permission_time prior=receipt_time(src.get('observed_at'),cutoff,prior,cohort_only=True) if not prior: return result("REPAIR_REQUIRED","source_response",REQ['source'],"receipts.source_response.observed_at must be parseable, within cutoff, and no earlier than the preceding observation.") if not isinstance(con,dict): return result("REPAIR_REQUIRED","connector_acceptance",REQ['connector'],"receipts.connector must be an object.",[{'path':'receipts.connector',**diagnostic(con)}]) if con.get('state')!='accepted' or not evidence_value(con.get('linkedin_response_id')) or con.get('linkedin_response_id')!=rid or not evidence_value(con.get('salesforce_environment')) or con.get('salesforce_environment')!=env or not evidence_value(con.get('run_id')) or not evidence_value(con.get('observed_at')): return result("REPAIR_REQUIRED","connector_acceptance",REQ['connector'],"Require one accepted connector run with a current timestamp for the same response and Salesforce environment.") prior=receipt_time(con.get('observed_at'),cutoff,prior) if not prior: return result("REPAIR_REQUIRED","connector_acceptance",REQ['connector'],"receipts.connector.observed_at must be parseable, within cutoff, and no earlier than the preceding observation.") maps=o.get('field_map') if not isinstance(maps,list) or not maps or any(not isinstance(m,dict) or not evidence_value(m.get('source')) or not evidence_value(m.get('destination')) or not isinstance(m.get('required'),bool) or m.get('transformation') not in {'identity','trim','trim_lowercase'} for m in maps): return result("REPAIR_REQUIRED","field_validation",REQ['field'],"Document at least one concrete source-to-destination field mapping.") if len({m['source'] for m in maps})!=len(maps) or len({m['destination'] for m in maps})!=len(maps): return result("REPAIR_REQUIRED","field_validation",REQ['field'],"Each source and Salesforce destination may appear only once in this scalar field map.") field=get(o,'receipts.field_validation') if not isinstance(field,dict): return result("REPAIR_REQUIRED","field_validation",REQ['field'],"receipts.field_validation must be an object.",[{'path':'receipts.field_validation',**diagnostic(field)}]) if field.get('state')!='accepted' or not evidence_value(field.get('field')) or not evidence_value(field.get('detail')): return result("REPAIR_REQUIRED","field_validation",REQ['field'],"Resolve the rejected, pending or unknown Salesforce field before person verification.") readback=field.get('read_back') def transformed(source,mode): if not isinstance(source,str): return None if mode=='identity': return source if mode=='trim': return source.strip() if mode=='trim_lowercase': return source.strip().lower() return None if not isinstance(readback,list) or len(readback)!=len(maps) or any(not isinstance(r,dict) or r.get('source')!=m['source'] or r.get('destination')!=m['destination'] or not data_value(r.get('source_value')) or r.get('expected_value')!=transformed(r.get('source_value'),m.get('transformation')) or r.get('observed_value')!=r.get('expected_value') or not evidence_value(r.get('person_id')) or r.get('linkedin_response_id')!=rid or r.get('salesforce_environment')!=env or not evidence_value(r.get('evidence_reference')) for m,r in zip(maps,readback)): return result("REPAIR_REQUIRED","field_validation",REQ['field'],"Require each read-back to preserve the raw source, apply the approved transformation, and match the observed destination value.") response_rows=[r for r in readback if r.get('source')=='linkedin_response_id'] if len(response_rows)!=1 or response_rows[0].get('source_value')!=rid or response_rows[0].get('expected_value')!=rid: return result("REPAIR_REQUIRED","field_validation",REQ['field'],"The mapped LinkedIn response ID must match the declared response identity byte for byte.") prior=receipt_time(field.get('observed_at'),cutoff,prior) if not prior: return result("REPAIR_REQUIRED","field_validation",REQ['field'],"receipts.field_validation.observed_at must be parseable, within cutoff, and no earlier than the preceding observation.") policy=get(o,'person_policy'); keys=policy.get('match_keys') if isinstance(policy,dict) else None if not isinstance(keys,list) or not keys or any(not evidence_value(k) for k in keys) or any(not evidence_value(policy.get(k)) for k in ['no_match_action','single_contact_action','single_lead_action','contact_and_lead_action','multiple_match_action','retry_key']) or policy.get('retry_key')!='linkedin_response_id': return result("REPAIR_REQUIRED","person_resolution",REQ['person'],"Document nonempty match keys, all match outcomes, and linkedin_response_id as the retry key.") if policy.get('same_response_action')!='deduplicate retry' or policy.get('new_response_action')!='count as new source response': return result("REPAIR_REQUIRED","person_resolution",REQ['person'],"Treat only the same LinkedIn response ID as a retry; a new response ID is a new source response.") if policy['no_match_action']!='create Lead' or policy['single_contact_action']!='update Contact' or policy['single_lead_action']!='update Lead' or policy['contact_and_lead_action']!='hold for review' or policy['multiple_match_action']!='hold for review': return result("REPAIR_REQUIRED","person_resolution",REQ['person'],"Match outcomes must cover new Lead, existing Contact, existing Lead, and ambiguous matches.") person=get(o,'receipts.person') if not isinstance(person,dict): return result("REPAIR_REQUIRED","person_resolution",REQ['person'],"receipts.person must be an object.",[{'path':'receipts.person',**diagnostic(person)}]) if person.get('state')!='observed' or person.get('object_type') not in {'Lead','Contact'} or not evidence_value(person.get('person_id')) or not evidence_value(person.get('linkedin_response_id')) or person.get('linkedin_response_id')!=rid or not evidence_value(person.get('observed_at')): return result("REPAIR_REQUIRED","person_resolution",REQ['person'],"Require one observed Lead or Contact with a current timestamp tied to the exact response ID.") if any(r['person_id']!=person['person_id'] for r in readback): return result("REPAIR_REQUIRED","person_resolution",REQ['person'],"Field read-back and person receipt must identify the same Salesforce person.") if any(not r.get('destination','').startswith(person['object_type']+'.') for r in readback): return result("REPAIR_REQUIRED","person_resolution",REQ['person'],"Mapped destinations must belong to the resolved Lead or Contact object.") branch=(person.get('object_type'),person.get('operation'),con.get('action')) branches={('Lead','created','Create Lead'):'accepted_new_lead',('Lead','updated','Update Lead'):'accepted_existing_lead',('Contact','updated','Update Contact'):'accepted_existing_contact'} if branch not in branches: return result("REPAIR_REQUIRED","person_resolution",REQ['person'],"Connector action and observed person operation must agree on the selected match branch.") prior=receipt_time(person.get('observed_at'),cutoff,prior) if not prior: return result("REPAIR_REQUIRED","person_resolution",REQ['person'],"receipts.person.observed_at must be parseable, within cutoff, and no earlier than the preceding observation.") expectations=get(o,'expectations'); expectations=expectations if isinstance(expectations,dict) else {} if expectations.get('approval_state')!='approved' or not evidence_value(expectations.get('approval_reference')): return result("REPAIR_REQUIRED","campaign_member",REQ['member'],"Member and owner expectations require recorded approval with evidence.") for k in ['member_status','member_evidence_reference']: if not evidence_value(expectations.get(k)): return result("REPAIR_REQUIRED","campaign_member",REQ['member'],f"Supply approved {k} expectation and evidence reference.") member=get(o,'receipts.campaign_member') if not isinstance(member,dict): return result("REPAIR_REQUIRED","campaign_member",REQ['member'],"receipts.campaign_member must be an object.",[{'path':'receipts.campaign_member',**diagnostic(member)}]) if member.get('state')!='observed' or not evidence_value(member.get('campaign_member_id')) or not evidence_value(member.get('campaign_id')) or member.get('campaign_id')!=get(o,'scope.campaign_id') or not evidence_value(member.get('person_id')) or member.get('person_id')!=person.get('person_id') or not evidence_value(member.get('observed_at')): return result("REPAIR_REQUIRED","campaign_member",REQ['member'],"Require the Campaign Member with a current timestamp for the same person and Campaign.") if member.get('member_status')!=expectations['member_status'] or not evidence_value(member.get('evidence_reference')): return result("REPAIR_REQUIRED","campaign_member",REQ['member'],"Observed Campaign Member status must match the approved expectation and have evidence.") prior=receipt_time(member.get('observed_at'),cutoff,prior) if not prior: return result("REPAIR_REQUIRED","campaign_member",REQ['member'],"receipts.campaign_member.observed_at must be parseable, within cutoff, and no earlier than the preceding observation.") for k in ['owner_id','assignment_rule','owner_evidence_reference']: if not evidence_value(expectations.get(k)): return result("REPAIR_REQUIRED","owner_assignment",REQ['owner'],f"Supply approved {k} expectation and evidence reference.") owner=get(o,'receipts.owner') if not isinstance(owner,dict): return result("REPAIR_REQUIRED","owner_assignment",REQ['owner'],"receipts.owner must be an object.",[{'path':'receipts.owner',**diagnostic(owner)}]) if owner.get('state')!='observed' or not evidence_value(owner.get('owner_id')) or not evidence_value(owner.get('person_id')) or owner.get('person_id')!=person.get('person_id') or not evidence_value(owner.get('observed_at')): return result("REPAIR_REQUIRED","owner_assignment",REQ['owner'],"Require the observed owner or queue with a current timestamp for the same person.") if owner.get('owner_id')!=expectations['owner_id'] or owner.get('assignment_rule')!=expectations['assignment_rule'] or not evidence_value(owner.get('evidence_reference')): return result("REPAIR_REQUIRED","owner_assignment",REQ['owner'],"Observed owner and assignment rule must match the approved expectation and have evidence.") prior=receipt_time(owner.get('observed_at'),cutoff,prior) if not prior: return result("REPAIR_REQUIRED","owner_assignment",REQ['owner'],"receipts.owner.observed_at must be parseable, within cutoff, and no earlier than the preceding observation.") pop=get(o,'populations') if not isinstance(pop,dict): return result("REPAIR_REQUIRED","population_reconciliation",REQ['population'],"populations must be an object.",[{'path':'populations',**diagnostic(pop)}]) keys=['source_responses','accepted_new_lead','accepted_existing_lead','accepted_existing_contact','rejected','duplicate_retry_attempts','unresolved','distinct_people','campaign_members','owned_people'] for k in keys: if not isinstance(pop.get(k),int) or isinstance(pop.get(k),bool) or pop[k]<0: return result("REPAIR_REQUIRED","population_reconciliation",REQ['population'],f"{k} must be a known non-negative integer; blank and Unknown are not zero.") accepted=pop['accepted_new_lead']+pop['accepted_existing_lead']+pop['accepted_existing_contact'] if pop['source_responses']!=accepted+pop['rejected']+pop['unresolved']: return result("REPAIR_REQUIRED","population_reconciliation",REQ['population'],"Source responses must equal accepted new Leads + accepted existing Leads + accepted existing Contacts + rejected + unresolved.") if pop['unresolved']>0: return result("REPAIR_REQUIRED","population_reconciliation",REQ['population'],"Positive unresolved blocks verification.") if pop['distinct_people']>accepted: return result("REPAIR_REQUIRED","population_reconciliation",REQ['population'],"Distinct people cannot exceed accepted source responses.") if pop['campaign_members']>pop['distinct_people'] or pop['owned_people']>pop['distinct_people']: return result("REPAIR_REQUIRED","population_reconciliation",REQ['population'],"Campaign Member and owner counts cannot exceed distinct people.") # observation-to-population invariants if person.get('state')=='observed' and accepted==0: return result("REPAIR_REQUIRED","population_reconciliation",REQ['population'],"Observed person contradicts an accepted-source-response population of zero.") if person.get('state')=='observed' and pop['distinct_people']==0: return result("REPAIR_REQUIRED","population_reconciliation",REQ['population'],"Observed person contradicts a distinct-person population of zero.") if pop[branches[branch]]==0: return result("REPAIR_REQUIRED","population_reconciliation",REQ['population'],"Observed person branch contradicts a branch population of zero.") if member.get('state')=='observed' and pop['campaign_members']==0: return result("REPAIR_REQUIRED","population_reconciliation",REQ['population'],"Observed Campaign Member contradicts a Campaign Member population of zero.") if owner.get('state')=='observed' and pop['owned_people']==0: return result("REPAIR_REQUIRED","population_reconciliation",REQ['population'],"Observed owner contradicts an owned-person population of zero.") if pop['campaign_members']