#!/usr/bin/env python3 """Offline request preparation for a deliberately narrow HubSpot CRM v3 profile. No HTTP client, credentials, file/database writes or native CRM operations. The caller supplies invented, already qualified contact observations. This does not establish their freshness, permissions, portal authenticity or real identity. """ from copy import deepcopy from dataclasses import dataclass, replace import json import re from urllib.parse import quote @dataclass(frozen=True) class Contact: portal: str source_contact: str record_id: str archived: bool = False @dataclass(frozen=True) class UniqueProperty: portal: str name: str has_unique_value: bool value_type: str = 'string' ALLOWED = frozenset({'email', 'firstname', 'lastname', 'phone', 'jobtitle'}) CLEARABLE = frozenset({'phone', 'jobtitle'}) def hold(reason): return {'decision': 'hold', 'reason': reason, 'request': None} def plan(portal, source_contact, properties, contacts=(), *, operation='update_only', clear_fields=frozenset(), unique_property=None, allow_creation=False): """Prepare one descriptor; never send or apply it. update_only needs exactly one supplied contact identity. create_or_update needs explicit creation permission and a qualified custom unique property. Empty strings clear only explicitly authorized local-profile fields; null is deliberately retained for review, without predicting HubSpot's response. """ if not all(type(x) is str and x and x == x.strip() for x in (portal, source_contact)): return hold('invalid_declared_scope_or_source_contact') if type(properties) is not dict or not properties: return hold('missing_changes') if not all(type(k) is str and k in ALLOWED for k in properties): return hold('property_outside_local_profile') if type(clear_fields) is not frozenset or not clear_fields <= CLEARABLE: return hold('invalid_clear_authorization') if not clear_fields <= properties.keys(): return hold('clear_authorization_without_property') for key, value in properties.items(): if value is None: return hold('unknown_value_not_clear_instruction') if type(value) is not str: return hold('property_value_not_string') if value == '': if key not in clear_fields: return hold('clear_not_authorized') elif not value.strip() or value != value.strip(): return hold('unqualified_whitespace') elif key in clear_fields: return hold('clear_authorization_with_nonempty_value') if type(contacts) is not tuple: return hold('invalid_contact_inventory') for c in contacts: if not isinstance(c, Contact) or type(c.archived) is not bool: return hold('invalid_contact_inventory') if not all(type(x) is str and x and x == x.strip() for x in (c.portal, c.source_contact, c.record_id)): return hold('invalid_contact_inventory') matching = [c for c in contacts if c.portal == portal and c.source_contact == source_contact] if len(matching) > 1: return hold('several_contact_correspondences') if matching and matching[0].archived: return hold('archived_contact_requires_review') body = {'properties': deepcopy(properties)} if operation == 'update_only': if not matching: return hold('existing_contact_not_qualified') request = {'method': 'PATCH', 'path': '/crm/v3/objects/contacts/' + quote(matching[0].record_id, safe=''), 'body': body} elif operation == 'create_or_update': if type(allow_creation) is not bool or not allow_creation: return hold('creation_not_authorized') k = unique_property if not isinstance(k, UniqueProperty) or k.portal != portal: return hold('unique_property_not_qualified') if (type(k.name) is not str or not re.fullmatch(r'[a-z][a-z0-9_]*', k.name) or k.name in {'email', 'hs_object_id'} or k.name in ALLOWED or type(k.has_unique_value) is not bool or not k.has_unique_value or k.value_type != 'string'): return hold('custom_unique_property_required') # Emptying during a possible creation has a different purpose; outside this profile. if clear_fields: return hold('clearing_not_supported_in_create_or_update_profile') request = {'method': 'POST', 'path': '/crm/v3/objects/contacts/batch/upsert', 'body': {'inputs': [{'idProperty': k.name, 'id': source_contact, 'properties': deepcopy(properties)}]}} else: return hold('operation_outside_local_profile') return {'decision': 'prepared', 'reason': operation, 'declared_portal': portal, 'source_contact': source_contact, 'may_create': operation == 'create_or_update', 'request': request} def run_checks(): checks = [] def check(name, condition): if name in checks or not condition: raise AssertionError(name) checks.append(name) def held(name, result, reason): check(name, result == hold(reason)) c = Contact('portal-test', 'ERP-PERSON-007', '901') k = UniqueProperty('portal-test', 'erp_contact_ref', True) props = {'email': 'nouvelle@example.test', 'phone': '+33100000000'} inventory = (c,) snapshot = deepcopy((props, inventory, k)) first = plan(c.portal, c.source_contact, props, inventory) check('known_contact_uses_patch', first['request']['method'] == 'PATCH') check('record_id_identifies_target', first['request']['path'] == '/crm/v3/objects/contacts/901') check('new_email_is_value_not_lookup_key', first['request']['body']['properties']['email'] == props['email']) check('update_only_cannot_create', first['may_create'] is False) check('patch_has_no_id_property', 'idProperty' not in first['request']['body']) after_email = plan(c.portal, c.source_contact, {'phone': '+33200000000'}, inventory) check('later_update_keeps_same_target', after_email['request']['path'] == first['request']['path']) check('absent_email_is_not_emitted', 'email' not in after_email['request']['body']['properties']) held('unknown_contact_does_not_fall_back_to_upsert', plan(c.portal, c.source_contact, props), 'existing_contact_not_qualified') held('other_portal_does_not_identify_contact', plan('other', c.source_contact, props, inventory), 'existing_contact_not_qualified') held('other_person_does_not_identify_contact', plan(c.portal, 'ERP-PERSON-008', props, inventory), 'existing_contact_not_qualified') held('duplicate_correspondence_is_not_first_match', plan(c.portal, c.source_contact, props, (c, c)), 'several_contact_correspondences') held('conflicting_correspondences_are_held', plan(c.portal, c.source_contact, props, (c, replace(c, record_id='902'))), 'several_contact_correspondences') held('archived_contact_is_not_recreated', plan(c.portal, c.source_contact, props, (replace(c, archived=True),)), 'archived_contact_requires_review') held('null_does_not_become_empty', plan(c.portal, c.source_contact, {'phone': None}, inventory), 'unknown_value_not_clear_instruction') held('empty_does_not_silently_clear', plan(c.portal, c.source_contact, {'phone': ''}, inventory), 'clear_not_authorized') cleared = plan(c.portal, c.source_contact, {'phone': ''}, inventory, clear_fields=frozenset({'phone'})) check('explicit_clear_prepares_empty_string', cleared['request']['body'] == {'properties': {'phone': ''}}) check('explicit_clear_remains_update_only', cleared['may_create'] is False) held('email_clear_is_outside_profile', plan(c.portal, c.source_contact, {'email': ''}, inventory, clear_fields=frozenset({'email'})), 'invalid_clear_authorization') held('missing_clear_field_is_not_added', plan(c.portal, c.source_contact, {'phone': '1'}, inventory, clear_fields=frozenset({'jobtitle'})), 'clear_authorization_without_property') held('nonempty_clear_instruction_is_inconsistent', plan(c.portal, c.source_contact, {'phone': '1'}, inventory, clear_fields=frozenset({'phone'})), 'clear_authorization_with_nonempty_value') held('boolean_is_not_string_property', plan(c.portal, c.source_contact, {'phone': False}, inventory), 'property_value_not_string') held('readonly_property_is_not_forwarded', plan(c.portal, c.source_contact, {'hs_object_id': '902'}, inventory), 'property_outside_local_profile') held('unapproved_field_is_not_forwarded', plan(c.portal, c.source_contact, {'lifecyclestage': 'customer'}, inventory), 'property_outside_local_profile') held('whitespace_is_not_normalized_into_clear', plan(c.portal, c.source_contact, {'phone': ' '}, inventory), 'unqualified_whitespace') held('empty_changes_are_held', plan(c.portal, c.source_contact, {}, inventory), 'missing_changes') held('invalid_inventory_is_held', plan(c.portal, c.source_contact, props, [c]), 'invalid_contact_inventory') held('invalid_record_identity_is_held', plan(c.portal, c.source_contact, props, (replace(c, record_id=''),)), 'invalid_contact_inventory') held('unsupported_operation_is_held', plan(c.portal, c.source_contact, props, inventory, operation='replace'), 'operation_outside_local_profile') held('upsert_requires_creation_permission', plan(c.portal, c.source_contact, props, operation='create_or_update', unique_property=k), 'creation_not_authorized') held('integer_does_not_grant_permission', plan(c.portal, c.source_contact, props, operation='create_or_update', unique_property=k, allow_creation=1), 'creation_not_authorized') held('property_metadata_is_required', plan(c.portal, c.source_contact, props, operation='create_or_update', allow_creation=True), 'unique_property_not_qualified') held('email_is_not_custom_partial_upsert_key', plan(c.portal, c.source_contact, props, operation='create_or_update', unique_property=replace(k, name='email'), allow_creation=True), 'custom_unique_property_required') held('object_id_is_not_custom_upsert_key', plan(c.portal, c.source_contact, props, operation='create_or_update', unique_property=replace(k, name='hs_object_id'), allow_creation=True), 'custom_unique_property_required') held('ordinary_property_does_not_prove_uniqueness', plan(c.portal, c.source_contact, props, operation='create_or_update', unique_property=replace(k, has_unique_value=False), allow_creation=True), 'custom_unique_property_required') held('integer_does_not_prove_uniqueness', plan(c.portal, c.source_contact, props, operation='create_or_update', unique_property=replace(k, has_unique_value=1), allow_creation=True), 'custom_unique_property_required') held('other_portal_property_is_not_used', plan(c.portal, c.source_contact, props, operation='create_or_update', unique_property=replace(k, portal='other'), allow_creation=True), 'unique_property_not_qualified') held('numeric_key_profile_is_not_assumed', plan(c.portal, c.source_contact, props, operation='create_or_update', unique_property=replace(k, value_type='number'), allow_creation=True), 'custom_unique_property_required') upsert = plan(c.portal, c.source_contact, props, operation='create_or_update', unique_property=k, allow_creation=True) check('authorized_upsert_marks_creation_risk', upsert['may_create'] is True) check('upsert_is_post_not_patch', upsert['request']['method'] == 'POST') check('upsert_uses_documented_v3_path', upsert['request']['path'] == '/crm/v3/objects/contacts/batch/upsert') item = upsert['request']['body']['inputs'][0] check('id_property_is_per_input', item['idProperty'] == 'erp_contact_ref') check('external_person_reference_is_lookup_value', item['id'] == c.source_contact) check('upsert_properties_remain_selective', item['properties'] == props) held('upsert_does_not_reuse_clear_profile', plan(c.portal, c.source_contact, {'phone': ''}, operation='create_or_update', unique_property=k, allow_creation=True, clear_fields=frozenset({'phone'})), 'clearing_not_supported_in_create_or_update_profile') check('caller_inputs_unchanged', (props, inventory, k) == snapshot) first['request']['body']['properties']['phone'] = 'changed only in returned descriptor' check('returned_payload_does_not_alias_input', props == snapshot[0]) check('identical_input_produces_identical_plan', after_email == plan(c.portal, c.source_contact, {'phone': '+33200000000'}, inventory)) escaped = plan(c.portal, c.source_contact, {'phone': '1'}, (replace(c, record_id='invented/a?b'),)) check('invented_record_id_stays_one_path_segment', escaped['request']['path'].endswith('invented%2Fa%3Fb')) # Demonstration identities can be non-native; escaping is not CRM ID certification. return {'checks_passed': len(checks), 'check_names': checks, 'examples': {'update': plan(c.portal, c.source_contact, props, inventory), 'next_update': after_email, 'explicit_clear': cleared, 'custom_key_upsert': upsert}, 'scope': {'http_calls': 0, 'crm_operations': 0, 'storage_writes': 0, 'native_response_certification': False}} if __name__ == '__main__': print(json.dumps(run_checks(), ensure_ascii=False, indent=2))