#!/usr/bin/env python3 """Offline, read-only Lead-to-existing-ERP-customer reconciliation laboratory. Invented already-decoded SOAP-shaped values, not an XML/HTTP/Salesforce client. No storage mutation, conversion, customer creation, merge or remote API call. Declared org labels and local business rules do not prove authenticated identity. Python 3.9+ standard-library constructs; run this file to execute the examples. """ from collections import defaultdict from copy import deepcopy from dataclasses import dataclass, replace import json from typing import Optional class ProfileError(ValueError): pass @dataclass(frozen=True) class Request: lead_id: str source_ref: str requested_account: Optional[str] = None @dataclass(frozen=True) class AccountLink: org: str account_id: str customer_ref: str @dataclass(frozen=True) class PriorCustomer: org: str lead_id: str customer_ref: str @dataclass(frozen=True) class Customer: org: str customer_ref: str active: bool def text(value): return isinstance(value, str) and bool(value.strip()) def optional_text(value): return value is None or text(value) def row(request, status, reason, account=None, customer=None): return dict(lead_id=request.lead_id, source_ref=request.source_ref, status=status, reason=reason, account_id=account, proposed_customer_ref=customer) def reconcile(org, requests, results, links, prior, customers, complete=True): """Propose only links to a known active ERP customer; never apply them. Every unqualified response holds the submitted cohort. Qualified native failures remain per Lead. Mapping/customer conflicts hold all successful Leads reaching the same Account, while unrelated Accounts stay independent. These are deliberately conservative LOCAL policies, not SOAP atomicity. """ if not text(org) or type(complete) is not bool: raise ProfileError('invalid scope or completion marker') if not isinstance(requests, tuple) or not requests: raise ProfileError('a retained nonempty request tuple is required') if any(not isinstance(r, Request) or not text(r.lead_id) or not text(r.source_ref) or not optional_text(r.requested_account) for r in requests): raise ProfileError('invalid retained request') if len({r.lead_id for r in requests}) != len(requests): raise ProfileError('repeated Lead in retained request') if len({r.source_ref for r in requests}) != len(requests): raise ProfileError('repeated local source reference') for values, cls, fields in [(links, AccountLink, ('org', 'account_id', 'customer_ref')), (prior, PriorCustomer, ('org', 'lead_id', 'customer_ref')), (customers, Customer, ('org', 'customer_ref'))]: if not isinstance(values, tuple) or any( not isinstance(v, cls) or any(not text(getattr(v, f)) for f in fields) for v in values ): raise ProfileError('invalid typed local inventory') if any(type(c.active) is not bool for c in customers): raise ProfileError('customer activity must be boolean') customer_keys = [(c.org, c.customer_ref) for c in customers] if len(set(customer_keys)) != len(customer_keys): raise ProfileError('repeated scoped customer identity') def hold_all(): return [row(r, 'hold', 'unqualified_conversion_evidence') for r in requests] if not complete or not isinstance(results, list) or len(results) != len(requests): return hold_all() parsed = [] for req, result in zip(requests, results): if not isinstance(result, dict) or type(result.get('success')) is not bool: return hold_all() if result.get('leadId') is not None and result['leadId'] != req.lead_id: return hold_all() if not isinstance(result.get('errors'), list): return hold_all() if result.get('relatedPersonAccountId') is not None: return hold_all() # Deliberately outside the business-Account profile. if result['success']: if (result.get('leadId') != req.lead_id or result['errors'] or not text(result.get('accountId')) or not text(result.get('contactId')) or not optional_text(result.get('opportunityId'))): return hold_all() account = result['accountId'] if req.requested_account is not None and req.requested_account != account: return hold_all() parsed.append((req, account)) else: if not result['errors'] or any( not isinstance(e, dict) or not text(e.get('statusCode')) for e in result['errors'] ): return hold_all() if any(result.get(k) is not None for k in ['accountId', 'contactId', 'opportunityId']): return hold_all() parsed.append((req, None)) accounts = defaultdict(list) previous = defaultdict(list) active = {c.customer_ref: c.active for c in customers if c.org == org} for link in links: if link.org == org: accounts[link.account_id].append(link.customer_ref) for item in prior: if item.org == org: previous[item.lead_id].append(item.customer_ref) grouped = defaultdict(list) decisions = {} for req, account in parsed: if account is None: decisions[req.lead_id] = row(req, 'conversion_failed', 'native_failure') else: grouped[account].append(req) for account, members in grouped.items(): targets = accounts[account] target = targets[0] if len(targets) == 1 else None reason = None if not targets: reason = 'missing_account_customer_mapping' elif len(targets) != 1: reason = 'ambiguous_account_customer_mapping' elif target not in active: reason = 'unknown_erp_customer' elif not active[target]: reason = 'inactive_erp_customer' elif any(len(previous[r.lead_id]) > 1 for r in members): reason = 'ambiguous_prior_lead_customer' elif any(previous[r.lead_id] and previous[r.lead_id][0] != target for r in members): reason = 'prior_customer_conflict_for_account_group' for req in members: decisions[req.lead_id] = ( row(req, 'hold', reason, account) if reason else row(req, 'propose_existing_link', 'known_active_customer', account, target) ) return [decisions[r.lead_id] for r in requests] def success(lead='lead-A', account='account-A', contact='contact-A', opportunity=None): return dict(success=True, leadId=lead, accountId=account, contactId=contact, opportunityId=opportunity, errors=[]) def main(): checks = [] def check(name, condition): assert name not in checks if not condition: raise AssertionError(name) checks.append(name) req = (Request('lead-A', 'PROSPECT-01', 'account-A'), Request('lead-B', 'PROSPECT-02', 'account-A')) values = [success(), success('lead-B', contact='contact-B')] links = (AccountLink('org-A', 'account-A', 'CLI-100'),) customers = (Customer('org-A', 'CLI-100', True), Customer('org-A', 'CLI-200', True)) original = deepcopy((req, values, links, customers)) call = lambda rs=values, rq=req, ls=links, pr=(), cs=customers, **kw: reconcile( 'org-A', rq, rs, ls, pr, cs, **kw) nominal = call() check('two_leads_have_two_proposals', len(nominal) == 2 and all(x['status'] == 'propose_existing_link' for x in nominal)) check('two_leads_reuse_one_existing_customer', {x['proposed_customer_ref'] for x in nominal} == {'CLI-100'}) check('different_contacts_do_not_split_customer', [x['source_ref'] for x in nominal] == ['PROSPECT-01', 'PROSPECT-02']) check('optional_opportunity_is_not_required', all(x['status'] == 'propose_existing_link' for x in call(rs=[success(opportunity='opportunity-A'), values[1]]))) check('existing_consistent_lead_alias_is_accepted', all(x['status'] == 'propose_existing_link' for x in call(pr=(PriorCustomer('org-A', 'lead-A', 'CLI-100'),)))) conflict = call(pr=(PriorCustomer('org-A', 'lead-B', 'CLI-200'),)) check('late_prior_conflict_holds_both_leads', all(x['status'] == 'hold' and x['reason'] == 'prior_customer_conflict_for_account_group' for x in conflict)) check('held_group_exposes_no_customer_proposal', all(x['proposed_customer_ref'] is None for x in conflict)) rq3 = req + (Request('lead-C', 'PROSPECT-03', 'account-B'),) mixed = call(rq=rq3, rs=values + [success('lead-C', 'account-B', 'contact-C')], ls=links + (AccountLink('org-A', 'account-B', 'CLI-200'),), pr=(PriorCustomer('org-A', 'lead-B', 'CLI-200'),)) check('unrelated_account_keeps_its_proposal', [x['status'] for x in mixed] == ['hold', 'hold', 'propose_existing_link'] and mixed[2]['proposed_customer_ref'] == 'CLI-200') check('request_order_is_retained', [x['lead_id'] for x in mixed] == ['lead-A', 'lead-B', 'lead-C']) check('missing_mapping_never_proposes_a_new_customer', all( x['reason'] == 'missing_account_customer_mapping' and x['proposed_customer_ref'] is None for x in call(ls=()))) check('contradictory_account_mappings_hold_group', all( x['reason'] == 'ambiguous_account_customer_mapping' for x in call( ls=links + (AccountLink('org-A', 'account-A', 'CLI-200'),)))) check('duplicate_mapping_is_not_silently_deduplicated', all( x['status'] == 'hold' for x in call(ls=links + links))) check('unknown_customer_holds_group', all(x['reason'] == 'unknown_erp_customer' for x in call(cs=()))) check('inactive_customer_holds_group', all(x['reason'] == 'inactive_erp_customer' for x in call(cs=(Customer('org-A', 'CLI-100', False),)))) check('duplicate_prior_alias_holds_group', all(x['reason'] == 'ambiguous_prior_lead_customer' for x in call(pr=(PriorCustomer('org-A', 'lead-A', 'CLI-100'),) * 2))) check('business_reference_case_is_not_normalized', all(x['status'] == 'hold' for x in call(pr=(PriorCustomer('org-A', 'lead-A', 'cli-100'),)))) other_links = (AccountLink('org-B', 'account-A', 'CLI-200'),) check('other_declared_org_mapping_does_not_replace_target', call(ls=links+other_links) == nominal) check('other_declared_org_alias_does_not_create_conflict', call(pr=(PriorCustomer('org-B', 'lead-A', 'CLI-200'),)) == nominal) check('other_declared_org_customer_does_not_validate_local_target', all(x['status'] == 'hold' for x in call(cs=(Customer('org-B', 'CLI-100', True),)))) check('unrequested_account_requires_known_local_customer_mapping', call(rq=tuple(replace(r, requested_account=None) for r in req)) == nominal) check('reordered_plan_and_results_keep_same_projection', sorted(call(rq=req[::-1], rs=values[::-1]), key=lambda x:x['lead_id']) == nominal) failed = dict(success=False, leadId='lead-B', errors=[dict(statusCode='FIELD_CUSTOM_VALIDATION_EXCEPTION')]) partial = call(rs=[values[0], failed]) check('qualified_native_failure_is_per_lead', [x['status'] for x in partial] == ['propose_existing_link', 'conversion_failed']) check('failed_conversion_proposes_no_customer', partial[1]['proposed_customer_ref'] is None) check('known_failure_accepts_null_target_ids', call(rs=[values[0], dict(failed, accountId=None, contactId=None, opportunityId=None)]) == partial) no_failed_id = dict(failed); del no_failed_id['leadId'] check('missing_failed_lead_id_uses_documented_result_position', call(rs=[values[0],no_failed_id]) == partial) shared_target = call(rq=rq3, rs=values+[success('lead-C','account-B','contact-C')], ls=links+(AccountLink('org-A','account-B','CLI-100'),)) check('distinct_accounts_can_share_a_prequalified_erp_customer', all(x['status']=='propose_existing_link' and x['proposed_customer_ref']=='CLI-100' for x in shared_target)) def held(name, rs=values, **kwargs): d = call(rs=rs, **kwargs) check(name, all(x['status'] == 'hold' and x['reason'] == 'unqualified_conversion_evidence' and x['proposed_customer_ref'] is None for x in d)) held('missing_result_holds_cohort', rs=values[:1]) held('extra_result_holds_cohort', rs=values+[values[0]]) held('absent_response_holds_cohort', rs=None) held('incomplete_transport_holds_cohort', complete=False) held('swapped_results_without_plan_are_not_recorrelated_by_guess', rs=values[::-1]) held('unknown_lead_identity_holds_cohort', rs=[dict(values[0],leadId='unknown'),values[1]]) missing_success_id = dict(values[0]); del missing_success_id['leadId'] held('successful_conversion_requires_returned_lead_identity', rs=[missing_success_id,values[1]]) held('contradictory_failed_lead_identity_is_not_ignored', rs=[values[0],dict(failed,leadId='unknown')]) held('integer_success_is_not_boolean', rs=[dict(values[0],success=1),values[1]]) held('success_with_error_is_unqualified', rs=[dict(values[0],errors=[dict(statusCode='X')]),values[1]]) held('success_missing_account_is_unqualified', rs=[dict(values[0],accountId=None),values[1]]) held('success_missing_contact_is_unqualified', rs=[dict(values[0],contactId=''),values[1]]) held('wrong_requested_account_holds_cohort', rs=[dict(values[0],accountId='account-Z'),values[1]]) held('explicit_related_person_account_is_outside_profile', rs=[dict(values[0],relatedPersonAccountId='person-A'),values[1]]) held('invalid_optional_opportunity_is_not_discarded', rs=[dict(values[0],opportunityId=7),values[1]]) held('failure_with_success_target_is_unqualified', rs=[values[0],dict(failed,accountId='account-A')]) held('failure_without_errors_is_unqualified', rs=[values[0],dict(failed,errors=[])]) held('errorcode_does_not_replace_soap_statuscode', rs=[values[0],dict(failed,errors=[dict(errorCode='X')])]) def invalid(name, **kwargs): try: call(**kwargs) except ProfileError: check(name, True) else: raise AssertionError(name) invalid('duplicate_lead_plan_is_refused', rq=(req[0],req[0])) invalid('duplicate_source_ref_is_refused', rq=(req[0],replace(req[1],source_ref=req[0].source_ref))) invalid('nontext_retained_lead_is_refused', rq=(replace(req[0],lead_id=7),req[1])) invalid('nonboolean_activity_is_refused', cs=(Customer('org-A','CLI-100',1),)) invalid('duplicate_customer_identity_is_refused', cs=(customers[0],customers[0])) invalid('blank_account_mapping_is_refused', ls=(AccountLink('org-A',' ', 'CLI-100'),)) check('all_original_inputs_are_unchanged', (req,values,links,customers) == original) check('repeated_read_only_analysis_is_deterministic', call() == nominal) print(json.dumps(dict(checks=len(checks),check_names=checks,nominal=nominal, shared_account_conflict=conflict,independent_account=mixed, native_partial_failure=partial,http_calls=0,native_conversions=0, erp_customer_creations=0,erp_customer_merges=0,storage_writes=0),indent=2)) if __name__ == '__main__': main()