o IEi@sUddlmZddlZddlZddlZddlmZddlmZm Z m Z hdZ hdZ e ZddddZded <d(d d Zd)ddZd*d+ddZd,ddZd-ddZdddd d.d&d'ZdS)/) annotationsN)deepcopy)AnyDictOptional>lenmodtyperegexlengthprefixmax_lenmin_lenweightsocr_kind validator check_rule max_length min_length must_parse check_index weight_list expected_len> id_number passport_nolicense_numberdriver_license_nopathmtimedataDict[str, Any]_CACHEreturnstrcCsvtjd}|r |Stjtjt}tj|dtj|dtj|dg}|D] }tj|r6|Sq*|dS)NKYC_IDENTITY_RULES_PATHzidentity_rules.jsonzidentity_registry.jsonzcountry_identity_rules.jsonr) osenvirongetrdirnameabspath__file__joinexists)envhere candidatespr2"/var/www/html/identity_registry.py_pick_default_path1s     r4rOptional[float]cCs$ztj|WStyYdSwN)r&rgetmtime Exception)rr2r2r3 _file_mtimeCs  r9 Optional[str]c Cs|pdpt}t|}ttddur-td|kr-td|kr-tdWdStj|sJdid}t|||d|WdSt |d d d  }| }Wdn1s`wYzt |}Wnt y}z td |d ||d}~wwt|tstd|d|vst|dtstdd|Drd|d}n t|dpdid}t|||d|WdS1swYdS)a1 Loads + caches the identity registry JSON. Auto-reloads on file mtime change. Schema (suggested): { "schema_version": 1, "countries": { "IR": { "kinds": { "national_id": { "type":"numeric", "length":10, "validator":"iran_national_code", "regex":"^[0-9]{10}$" } }, "doc_types": { "id_card": { "fields": { "id_number": { "use": "national_id" } } }, "driver_license": { "fields": { "id_number": { "use": "national_id" } } } } } } } r Nrr)schema_version countriesrrz utf-8-sig)encodingz*Identity registry JSON parse failed: path=z err=z.Identity registry must be a JSON object: path=r>css(|]\}}t|tot|tVqdSr6) isinstancer$dict).0kvr2r2r3 ws&z)load_identity_registry..r=)stripr4r9 _CACHE_LOCKr"r&rr-updateopenreadjsonloadsr8 RuntimeErrorrArBr(allitemsint)rr1rr frawer2r2r3load_identity_registryJs:$     $rUdcCs.i}|piD] \}}|tvr|||<q|Sr6)rP _ALLOWED_KEYS)rVoutrDrEr2r2r3_sanitize_specs rY country_entrydoc_type field_keyOptional[Dict[str, Any]]cCs:|pd}|p d}|d}t|tsi}||p%||p%i}t|ts-i}|d}|dur8|}t|ts?dS||}|durJdSt|trSd|i}t|tsZdS|d} | r|dpgi} t| tr| t| } t| tr~| | } t| trt| } |D] \} }| dkrq|| | <q| S|S)Nr; doc_typesfieldsusekinds)rGlowerr(rArBr$rP)rZr[r\ doc_type_l field_key_sr^dt_entryr_specr`rabasemergedrDrEr2r2r3_resolve_field_specsF               rir;F)rstrict strict_keyscountryrjboolrk Optional[set]c Cst|}t|tr|dnd}t|tsi}|pd}||p*||} t| ts@|r>|p5d|p8tvr>ddiSdSt| ||} t| ts[|rY|pPd|pStvrYddiSdSt t | S)a" Returns a dict of defaults to be merged into validation_logic for a given (country, doc_type, field_key). - If no defaults exist: returns None (unless strict=True and field_key is in strict_keys) - If defaults exist: returns a sanitized dict containing only allowed keys. r>Nr;rmissing_identity_rule) rUrArBr(rGupperrb_STRICT_DEFAULT_KEYSrirYr) rlr[r\rrjrkr r>ccentryrfr2r2r3get_identity_defaultss      rt)r#r$)rr$r#r5r6)rr:r#r!)rVr!r#r!)rZr!r[r$r\r$r#r])rlr$r[r$r\r$rr$rjrmrkrnr#r]) __future__rrLr& threadingcopyrtypingrrrrWrqLockrHr"__annotations__r4r9rUrYrirtr2r2r2r3s,      6 7