"""Field-mapping engine for outbound identity sync. A mapping is a list of *rules*. Each rule renders one downstream field from a source template, optional transform pipeline, optional value lookup table, and optional default. The full input domain is a single :class:`PersonContext` dictionary built from one ``IdentityPerson`` plus its primary org membership and (if needed) its manager — never the live ORM row, so the engine has zero database side effects. Rule shape (JSON, validated by :class:`MappingRule.validate`): { "source_template": "{first_name}.{last_name}", "transform": "lower" | ["lower", "trim"] | null, "value_map": {"CN": "APAC"} | null, "default": "n/a" | null, "target_field": "email" | "name.familyName" | "customSchemas.X.Y" } Why a small templating language instead of Jinja2 / Python eval: * The set of attributes a downstream system needs is small (a dozen per target). A regex-substitution template handles 100% of real cases without dragging Jinja2 into the dependency tree. * No code execution path means there's no sandbox to harden. The entire engine is a pure data transformation — easy to audit, easy to write unit tests for, can never read arbitrary attributes off a SQLAlchemy row. * Production-grade requirement: refuse unknown source fields LOUDLY rather than silently emit empty strings. ``getattr(model, "x", None)`` was the cause of three prior outages — see CLAUDE.md feedback memory ``feedback_no_getattr_default_for_model_columns``. The whitelist below enforces this at render time. Supported source field names — anything outside this set raises ``MappingError`` at template parse: work_email, employee_id, mobile_phone, first_name, last_name, full_name, name_pinyin, job_title, country, manager_email, manager_name, manager_employee_no, employment_status, hire_date, termination_date, org_path, org_path_level_0..org_path_level_9, department_l1..department_l4 (alias for the levels), cost_center, business_unit, location The runner adds ``org_path_level_*`` after applying the per-target ``runner_options.drop_root_segments`` if the runner is the ``google_schema`` runner — see ``runners.google_schema``. Other runners pass org_path raw and only the explicit indices are valid. """ from __future__ import annotations import hashlib import json import re from dataclasses import dataclass, field from datetime import date, datetime from typing import Any from app.core.name_sanitize import strip_sentinel_name_tokens # ── Whitelist of source field names ─────────────────────────────────── # Anything outside this set raises ``MappingError`` at parse time so # admins find typos in the editor instead of in production logs. _BASE_SOURCE_FIELDS: frozenset[str] = frozenset({ "work_email", "employee_id", "first_name", "last_name", "full_name", "name_pinyin", "display_name", # ``display_name_cjk`` = the CJK run of display_name (robust extraction # via the ``cjk_name`` transform), BLANK for foreign hires. NetSuite # maps firstName from it to keep Chinese names without romanising and # without disturbing foreign hires' English names (user decision # 2026-06-24). Derived in build_person_context. "display_name_cjk", # Resolved VENDOR-internal ids of a person's role-holders, used to set # NetSuite reference fields (supervisor / 部门负责人 / 预算负责人 / HRBP). # The orchestrator resolves them per-person ONLY for the netsuite kind # (工号 → NetSuite internal id, via the adopted ``eid:`` keys) and # injects them into ctx; every other vendor leaves them empty so the # mapping rule drops → the field is preserved. They are NetSuite # internal ids (numeric strings), set as ``{"id": }`` reference # objects via the ``…​.id`` target_field convention. "ns_supervisor_id", # ``ns_department_code`` = the person's dept org_node_source_id, injected # ONLY when that dept is adopted in NetSuite (→ department.externalId). "ns_department_code", "job_title", "country", "city", "manager_email", "manager_name", # ``manager_employee_no`` = the direct manager's HRIS 工号. Injected by # the orchestrator ONLY for the ``huilianyi`` kind (汇联易 keys people by # 工号, so its 主岗直属领导 field takes the manager's employeeID # directly — no vendor-id lookup, unlike NetSuite's ns_supervisor_id). # Absent for every other kind → the rule renders empty → the field drops # → the vendor keeps whatever it has. Gated on the manager being part of # the pushed population (active / on_leave, P-shaped 工号, not purged) # so we never point 汇联易 at a person it doesn't have. "manager_employee_no", # HR Business Partner — resolved from ``IdentityPerson.hris_hrbp_person_id`` # (SF ``EmpJobRelationships`` 2448). ``hrbp_email`` is what the Feishu # contact runner resolves to a Feishu user_id for the HRBP GENERIC_USER # custom attribute; ``hrbp_name`` is a human-readable fallback. Empty # string when the person has no HRBP declared in SF. "hrbp_email", "hrbp_name", "employment_status", "hire_date", "termination_date", # 手机号 — Authing's downstream feishu/汇联易 pushes carried it # (飞书#0 task 9 maps 手机号→phone; the live 汇联易 roster stores # ``mobile``). Needed for vendor parity after the Authing # decommission. Source: ``identity_persons.mobile_phone``. "mobile_phone", # ISO-2 of the mobile's own country (SF PerPhone.countryCode → # externalCode). Source: ``identity_persons.mobile_country_code``. "mobile_country_code", # ``mobile_country_iso2`` = the phone's country as a clean ISO-2, with a # FALLBACK to the person's work country. Derived, not a column: the raw # ``mobile_country_code`` is mixed ISO-2 / ISO-3 / free text and is NULL # for 21 of prod's 411 phone-holders — all of them foreign staff whose # numbers would otherwise be pushed to 汇联易 with no country and thus be # stored as China-mainland numbers (the vendor's documented default). "mobile_country_iso2", "org_path", # ``cost_center`` is intentionally listed even though our HRIS # sync does not yet populate it — admins can map to it now and # the field renders empty until a follow-up PR adds the SF # ``EmpJob.costCenter`` pull + an IdentityPerson column. "cost_center", "business_unit", # ``location`` is the WORK location name (e.g. "Beijing Office", # "San Mateo HQ"), NOT the country. Sourced from # ``IdentityPerson.work_location_name``. "location", "work_location_code", # ``company_name`` / ``company_code`` are the legal entity (SF # FOCompany) — what Navan calls "Subsidiary" / "Division". Each # employee belongs to one legal entity (PingCAP USA Inc., PingCAP # KK, 平凯星辰, etc.). "company_name", "company_code", # ``department_code`` is the LEAF dept code on the person's # primary org membership (= ``IdentityOrgMembership.org_node_source_id``). # Used by department-aware downstream targets (Feishu Contact) to # populate the user's ``department_ids`` slot. Stable across HRIS # renames; same string the dept runner uses as ``open_department_id``. "department_code", # SF-sourced raw picklist codes (migration 344), projected to Feishu by # the feishu_contact target's value-maps. ``gender`` → Feishu gender enum # (M/F), ``employee_type_code`` → Feishu employee_type enum # (1070/1072/…), ``office_code`` → Feishu "Location" custom-attr label # (2659→"SH Office"). Sourced from the IdentityPerson columns of the # same name. "gender", "employee_type_code", "office_code", # GitHub handle — SF cust_Other.cust_id (migration 369), keyed by # employee_no. → Feishu ``GitHub ID`` custom attr (no value_map). # Sourced from IdentityPerson.github_id. "github_id", # SF-verbatim labels (resolved in the SF sync), pushed to Google # Employee_Details VERBATIM — no value_map. ``office_mail_group`` = # EmailGroup label ('sh-office@pingcap.com'), ``job_function`` = # FOJobFunction name ('Engineer'), ``job_family`` = # cust_JobFamily name, ``residence_country`` = ISO-2 work-location # country ('CN'). Sourced from the IdentityPerson columns. "office_mail_group", "job_function", "job_family", "residence_country", }) # org_path_level_0 .. org_path_level_9 + department_l1 .. department_l4 # are derived; runners populate them. They're listed here so the parser # accepts them; an empty derived field renders as "" (then default fills # in). 10 levels covers the deepest org chart we've seen + headroom. # # ``department_combined`` is L2 / L3 / L4 joined by " / " with empty # segments dropped. Navan's enterprise.department slot wants a single # string showing the full department path below the legal entity — # admins previously had to pick one level which hid mid-tree placement. _DERIVED_SOURCE_FIELDS: frozenset[str] = frozenset( [f"org_path_level_{i}" for i in range(10)] + [f"department_l{i}" for i in range(1, 5)] + ["department_combined"] ) SUPPORTED_SOURCE_FIELDS: frozenset[str] = ( _BASE_SOURCE_FIELDS | _DERIVED_SOURCE_FIELDS ) # Note: dept-keyed targets (Feishu Contact Phase A) do NOT go through # the mapping engine — their wire body is built directly from raw # ``IdentityOrgNode`` fields in ``FeishuDepartmentRunner._build_body``. # So there's no dept-source-field whitelist here. If a future kind # needs admin-customisable dept rules, re-introduce # ``_DEPT_SOURCE_FIELDS`` + a ``build_dept_context`` function and # union into ``SUPPORTED_SOURCE_FIELDS``. # ── Built-in transforms ─────────────────────────────────────────────── # ── Country codes ──────────────────────────────────────────────────── # HRIS carries ISO-3 (``CHN``); most vendor APIs want ISO-2 (``CN``) — # Feishu's ``country`` field and 汇联易's phone ``countryCode`` both do. # Canonical home for the translation (presets re-exports it) so the two # don't drift: a country present in one copy and missing from the other # renders an empty field on exactly one vendor, which is the silent # half-migration class of bug this file already warns about elsewhere. COUNTRY_ISO3_TO_ISO2: dict[str, str] = { "CHN": "CN", "JPN": "JP", "USA": "US", "CAN": "CA", "SGP": "SG", "MYS": "MY", "HKG": "HK", "TWN": "TW", "KOR": "KR", "AUS": "AU", "IND": "IN", "IDN": "ID", "PHL": "PH", "THA": "TH", "VNM": "VN", "NZL": "NZ", "GBR": "GB", "DEU": "DE", "FRA": "FR", "NLD": "NL", "ESP": "ES", "ITA": "IT", "IRL": "IE", "CHE": "CH", "SWE": "SE", "DNK": "DK", "NOR": "NO", "FIN": "FI", "POL": "PL", "ROU": "RO", "ARE": "AE", "ISR": "IL", "ZAF": "ZA", "MEX": "MX", } def resolve_country_iso2(value: str | None) -> str: """Best-effort ISO-2 from a country column that carries MIXED shapes. ``identity_persons.mobile_country_code`` is ISO-2 for most people (``CN``/``JP``/``US``) but ISO-3 for others (``IDN``/``POL``/``NLD``) and occasionally free text (a real prod value is ``"GB / UK"``) — SF's picklist resolution differs per instance/field, so a single-shape assumption silently drops whole cohorts. Returns "" when the value can't be resolved, which callers must treat as "send nothing" rather than as a default. """ v = (value or "").strip().upper() if len(v) == 2 and v.isalpha(): return v return COUNTRY_ISO3_TO_ISO2.get(v, "") _PHONE_SEPARATOR_RE = re.compile(r"[\s\-()\.]") SUPPORTED_TRANSFORMS: frozenset[str] = frozenset({ "lower", "upper", "trim", # ``phone_digits`` strips the human separators HRIS phone numbers carry # (JP staff are stored as "90-1234-5678", US as "(415) 555-1234") so a # vendor that wants a bare number gets one. It strips ONLY separators — # a leading "+" / "00" international prefix is deliberately LEFT IN # PLACE so the consuming runner can refuse the value instead of silently # welding the dialing code onto the national number ("+8613…" must never # become "8613…", which is a different, possibly someone else's, number # on a vendor whose phone field is tenant-unique). "phone_digits", "title_case", "email_local", "email_domain", "iso_date", # ``strip_company_prefix`` drops the leading numeric ordering prefix # that PingCAP's HRIS bakes into FOCompany.name — e.g. # ``"01_PingCAP(US),Inc."`` → ``"PingCAP(US),Inc."``. Navan's # subsidiary picker has no use for the sort key, and the prefix # leaks an internal HRIS ordering into a vendor surface. "strip_company_prefix", # ``cjk_name`` extracts the Chinese (CJK) run from a bilingual # ``display_name`` ("冯帅 Shuai Feng" → "冯帅") for vendors that want the # native name (e.g. 汇联易, a Chinese expense system). No CJK present # (foreign staff) → the trimmed input passes through unchanged so an # English name still lands. Sourced from ``display_name`` because that # is the only context field carrying the CJK form (``full_name`` is # email-derived English; ``name_pinyin`` is romanised). "cjk_name", # ``cjk_name_leading`` = the GUARDED variant of ``cjk_name`` with a # passthrough fallback: only treat the input as a CJK name when it # LEADS with CJK ("冯帅 Shuai Feng" → "冯帅"); anything else passes # through unchanged ("Yukun Sun" → "Yukun Sun", and crucially # "John Smith 顾问" → "John Smith 顾问" — never the extracted trailing # remark "顾问" that raw ``cjk_name`` would pull out). Same leading-CJK # guard as the ``display_name_cjk`` context token, but where that token # BLANKS non-leading-CJK input (rule drops → vendor value preserved), # this transform keeps it — for fields that must render non-empty for # everyone (NetSuite rejects a nameless employee CREATE, #1949). "cjk_name_leading", # ``cjk_pinyin`` = pinyin of ONLY the leading CJK run of a bilingual # ``display_name`` ("冯帅 Shuai Feng" → "fengshuai"). SF has no pinyin # field, so Feishu's PinYin attr is derived here — from the CJK name # only, so it stays the clean surname+given pinyin instead of the # whole-string ``name_pinyin`` ("fengshuai shuai feng") that leaks the # romanised English tail. No CJK (foreign staff) → "" → rule drops. "cjk_pinyin", # ``cn_city`` strips the administrative "市" suffix that PingCAP's HRIS # bakes into the work-city ("上海市" → "上海"). Feishu's standard ``city`` # field only renders values that match its municipality dictionary — the # bare name "上海" displays, "上海市" is stored but silently NOT shown on # the profile card (same class of bug as country ISO-3 vs ISO-2). # Romanised overseas cities ("Tokyo") carry no "市" → pass through. "cn_city", # ``latin_name`` extracts the LATIN (English) part of a bilingual # ``display_name`` ("孟凡生 Jason Meng" → "Jason Meng"; "John Smith" → # "John Smith"; "张三" → ""). Feishu's en_name must carry the person's # PREFERRED English name, which SF stores in displayName after the CJK # name — NOT the email-derived romanisation (email fansheng.meng → # "Fansheng Meng", which is wrong: his preferred English is "Jason"). # Removes all CJK runs, collapses whitespace, trims. Pure-CJK name → # "" so the rule drops (Feishu keeps its existing en_name). "latin_name", }) # Matches one or more digits followed by an underscore / hyphen / dot # separator at the very start of the string, tolerant of leading # whitespace. Anchored — never trims a numeric prefix mid-string. _COMPANY_PREFIX_RE = re.compile(r"^\s*\d+\s*[_\-\.]\s*") # Contiguous CJK run (+ the · used in some transliterated names) for the # ``cjk_name`` transform. Covers CJK Unified Ideographs + Ext-A. _CJK_NAME_RE = re.compile(r"[㐀-鿿·]+") # Leading CJK block, bridging whitespace-separated runs (a compound surname # or a surname/given-name split like "欧阳 娜娜"). Anchored at the start so a # trailing / parenthetical CJK remark after the English name (the convention # is " ") is NOT pulled into the extracted name. _CJK_LEAD_RE = re.compile(r"[㐀-鿿·]+(?:\s+[㐀-鿿·]+)*") # Maximum render depth: we don't allow nested {} expansion. _TEMPLATE_TOKEN_RE = re.compile(r"\{([a-z][a-z0-9_]*)\}") class MappingError(ValueError): """Raised on invalid mapping configuration or render failure.""" # ── target_field path parser ───────────────────────────────────────── # SCIM enterprise extensions live at URN-prefixed attribute paths like # urn:ietf:params:scim:schemas:extension:enterprise:2.0:User.department # The URN itself contains a dot (the schema-version "2.0") which would # collide with our normal "." path separator. The parser below detects # the SCIM URN prefix, treats it as one atomic segment, and splits the # remainder on "." as usual. # # Recognised URN suffixes: ``:User`` and ``:Group`` — the only two # resource types in core SCIM 2.0. Custom extensions following the # same shape (e.g. third-party ``…:Account.*``) need to be added here. _SCIM_URN_BOUNDARIES = (":User", ":Group") def split_target_path(path: str) -> list[str]: """Split a target_field into ordered path segments. Handles SCIM URN-prefixed attribute paths specially: ``urn:ietf:params:scim:schemas:extension:enterprise:2.0:User.x.y`` → ``["urn:ietf:params:scim:schemas:extension:enterprise:2.0:User", "x", "y"]`` Anything else just splits on ``.``: ``name.familyName`` → ``["name", "familyName"]`` ``emails.0.value`` → ``["emails", "0", "value"]`` """ if path.startswith("urn:"): for boundary in _SCIM_URN_BOUNDARIES: idx = path.find(boundary + ".") if idx != -1: urn_end = idx + len(boundary) urn = path[:urn_end] rest = path[urn_end + 1:] return [urn] + (rest.split(".") if rest else []) # URN-only target (no extension attr after :User) — single seg. if path.endswith(boundary): return [path] return path.split(".") @dataclass(slots=True) class MappingRule: source_template: str target_field: str transform: list[str] = field(default_factory=list) value_map: dict[str, str] | None = None # When ``value_map`` is set, an unmapped source value renders empty # by default (STRICT translation — see :func:`render_value`). Set # ``value_map_passthrough`` to instead pass an unmapped value # through unchanged. Use this for *override* maps that rewrite a # handful of values and leave the rest alone — e.g. the Navan # manager rule remaps one manager's email to a delegate while every # other manager flows through verbatim. Ignored when ``value_map`` # is null. value_map_passthrough: bool = False default: str | None = None # Rule renders ONLY into the payload of a person Helix has not linked # yet (``IdentitySyncState.external_id`` empty) — i.e. the CREATE # shape. On merge-semantics vendors (NetSuite PATCH/PUT, Feishu # PATCH), omitting the field on UPDATE preserves the vendor's # existing value. Use for fields that are REQUIRED non-empty at # create time but must never clobber a vendor-curated value on # update (the NetSuite firstName case: create needs a name, update # must not overwrite a foreign hire's split first/last name). # Enforced by ``render_payload(..., exclude_create_only=True)``, # which the sweep / single-person push pass for already-linked rows. create_only: bool = False # When True, a render that produces nothing for this field (empty # after transforms + no default, or a STRICT value_map miss) is a # HARD BLOCK: the orchestrator marks the person ``blocked`` with an # actionable reason BEFORE any vendor call, instead of silently # dropping the field and letting the vendor 400 the whole push every # sweep (NetSuite「请输入值:子公司」/「非法 ID。请输入名称」) or — # worse — accepting a partial record. The block auto-clears when the # source data is fixed (the payload hash changes). Set on the vendor's # genuinely mandatory create fields (e.g. NetSuite subsidiary / name). required: bool = False @classmethod def from_dict(cls, raw: dict[str, Any]) -> MappingRule: if not isinstance(raw, dict): raise MappingError(f"mapping rule must be an object, got {type(raw)!r}") src = raw.get("source_template") tgt = raw.get("target_field") if not isinstance(src, str) or not src: raise MappingError("mapping rule needs non-empty source_template") if not isinstance(tgt, str) or not tgt: raise MappingError("mapping rule needs non-empty target_field") # target_field segments — see :func:`split_target_path` for the # special-case URN handling. Each non-URN segment must be # either a pure-digit array index (``emails.0.value``) or a # letter-led identifier. URN segments (``urn:…:User``) are # accepted verbatim because the SCIM URN syntax is fixed. for seg in split_target_path(tgt): if seg.startswith("urn:"): continue if not seg or not re.match( r"^([0-9]+|[A-Za-z][A-Za-z0-9_-]*)$", seg, ): raise MappingError( f"target_field segment {seg!r} not allowed in " f"{tgt!r}; non-URN segments must be either pure " "digits (array index) or letter-led identifiers " "(letters/digits/_/-)." ) # Validate the source template tokens against the whitelist. # _TEMPLATE_TOKEN_RE matches lowercase ascii identifiers which # is what we expect from admins; if they typed `{Foo}` it just # won't match (and the literal stays in output). That's fine. for token in _TEMPLATE_TOKEN_RE.findall(src): if token not in SUPPORTED_SOURCE_FIELDS: raise MappingError( f"unknown source field {{{token}}} in template " f"{src!r}. Supported: " f"{sorted(SUPPORTED_SOURCE_FIELDS)}" ) # Transform: accept str OR list[str]. Each entry must be in # SUPPORTED_TRANSFORMS. raw_xform = raw.get("transform") xform_list: list[str] = [] if raw_xform is None or raw_xform == []: xform_list = [] elif isinstance(raw_xform, str): xform_list = [raw_xform] elif isinstance(raw_xform, list): xform_list = [str(x) for x in raw_xform] else: raise MappingError( f"transform must be a string or list of strings, " f"got {type(raw_xform)!r}" ) for x in xform_list: if x not in SUPPORTED_TRANSFORMS: raise MappingError( f"unknown transform {x!r}. Supported: " f"{sorted(SUPPORTED_TRANSFORMS)}" ) # value_map: optional dict[str, str]. Keys/values must be # strings — JSON-safe and unambiguous when re-rendered to admin. raw_vmap = raw.get("value_map") vmap: dict[str, str] | None = None if raw_vmap is not None: if not isinstance(raw_vmap, dict): raise MappingError("value_map must be an object or null") vmap = {} for k, v in raw_vmap.items(): if not isinstance(k, str): raise MappingError( f"value_map key must be string, got {type(k)!r}" ) vmap[k] = "" if v is None else str(v) # ``None`` (the to_dict emission for rules without a value_map) # collapses to the default ``False``. raw_passthrough = raw.get("value_map_passthrough") or False if not isinstance(raw_passthrough, bool): raise MappingError( f"value_map_passthrough must be a boolean, got " f"{type(raw_passthrough)!r}" ) if raw_passthrough and vmap is None: raise MappingError( "value_map_passthrough is set but value_map is null; " "passthrough only applies alongside a value_map" ) # Reachability guard (silent-blank-population bug class): the # value_map lookup runs on the value AFTER the transform chain, so # a key the transforms can never emit is dead config. A STRICT map # whose keys are all unreachable blanks the field for the WHOLE # population with a green run (e.g. ``transform=lower`` + # ``value_map={"USA":"NA"}`` — every rendered value is lowercase, # so the uppercase key never hits). Catch the unambiguous casing # transforms at save time (zero false positives — a key survives # ``lower``/``upper`` iff it is already in that case). if vmap: if "lower" in xform_list: dead = sorted(k for k in vmap if k and k != k.lower()) if dead: raise MappingError( f"value_map keys {dead} can never match: the " f"'lower' transform lowercases every value before " f"lookup — use lowercase keys." ) if "upper" in xform_list: dead = sorted(k for k in vmap if k and k != k.upper()) if dead: raise MappingError( f"value_map keys {dead} can never match: the " f"'upper' transform uppercases every value before " f"lookup — use uppercase keys." ) raw_default = raw.get("default") default_str: str | None = None if raw_default is not None: if not isinstance(raw_default, (str, int, float, bool)): raise MappingError( f"default must be a scalar, got {type(raw_default)!r}" ) default_str = str(raw_default) # ``None`` (the to_dict emission for non-create-only rules) # collapses to the default ``False``. raw_create_only = raw.get("create_only") or False if not isinstance(raw_create_only, bool): raise MappingError( f"create_only must be a boolean, got " f"{type(raw_create_only)!r}" ) raw_required = raw.get("required") or False if not isinstance(raw_required, bool): raise MappingError( f"required must be a boolean, got {type(raw_required)!r}" ) return cls( source_template=src, target_field=tgt, transform=xform_list, value_map=vmap, value_map_passthrough=raw_passthrough, default=default_str, create_only=raw_create_only, required=raw_required, ) def to_dict(self) -> dict[str, Any]: return { "source_template": self.source_template, "target_field": self.target_field, "transform": self.transform or None, "value_map": self.value_map, # Only emit the flag when a value_map exists — keeps the # persisted JSON tidy for the overwhelming majority of # rules that don't use a value_map at all. "value_map_passthrough": ( self.value_map_passthrough if self.value_map else None ), "default": self.default, # Same tidy-JSON convention: emitted only when set. "create_only": self.create_only or None, # Emit only when set — keeps persisted JSON tidy for the # overwhelming majority of non-required rules. "required": self.required or None, } @dataclass(slots=True) class MappingConfig: rules: list[MappingRule] @classmethod def from_dict(cls, raw: dict[str, Any] | None) -> MappingConfig: if raw is None: return cls(rules=[]) if not isinstance(raw, dict): raise MappingError("mapping_config must be an object") raw_rules = raw.get("rules") or [] if not isinstance(raw_rules, list): raise MappingError("mapping_config.rules must be a list") rules = [MappingRule.from_dict(r) for r in raw_rules] # No duplicate target_field entries — last-write-wins surprises # are exactly the kind of footgun "production grade" forbids. seen: set[str] = set() for r in rules: if r.target_field in seen: raise MappingError( f"duplicate target_field {r.target_field!r} in mapping" ) seen.add(r.target_field) return cls(rules=rules) def to_dict(self) -> dict[str, Any]: return {"rules": [r.to_dict() for r in self.rules]} # ── Person context (input to the engine) ───────────────────────────── def build_person_context( person: Any, *, org_path: str | None, manager_email: str | None, manager_name: str | None, hrbp_email: str | None = None, hrbp_name: str | None = None, org_path_drop_root: list[str] | None = None, org_node_source_id: str | None = None, ) -> dict[str, str]: """Build the flat context dict the renderer reads from. Every value is normalised to a string (empty string when missing). Date columns render as ISO 8601 (``YYYY-MM-DD``) which is what every downstream system we target accepts; transforms can override if needed (currently ``iso_date`` is a no-op for already-ISO input, kept for forward compatibility). ``org_path_drop_root`` mirrors the runner option of the same name — pass the list of top-level segments (case-insensitive) to strip before populating ``org_path_level_*`` and ``department_l*``. This matches the legacy behavior that excluded the legal entity ("PingCAP") from the L1-L4 dept hierarchy. Whitelisted access only — we explicitly read each known column instead of ``getattr(person, field)`` to fail loudly on schema drift (see CLAUDE.md memory). """ def _str(v: Any) -> str: if v is None: return "" if isinstance(v, (date, datetime)): # Render dates as ISO calendar date; full datetimes also # collapse to date because per-day granularity is what # downstream HR/finance fields use (hire_date, # termination_date). if isinstance(v, datetime): return v.date().isoformat() return v.isoformat() return str(v).strip() raw_email = _str(person.work_email) # Direct attribute access (NOT getattr(..., None)) so a renamed/dropped # column surfaces as AttributeError at first use instead of silently # rendering "" and degrading every vendor's givenName/familyName — the # same policy this function states 100 lines below (memory # ``feedback_no_getattr_default_for_model_columns``). pinyin = _str(person.name_pinyin) display = _str(person.display_name) first = "" last = "" full_name = "" # Naming derivation — downstream systems (Navan, SCIM in # general) require both givenName + familyName non-empty. Order # of preference, most reliable first: # # 1. **Email local-part with a dot** ("liyang.zhang") — at PingCAP # every work_email follows ``.@pingcap.com`` # by HRIS convention. Splitting on the FIRST dot gives a # correct given/family pair regardless of how the HRIS rendered # name_pinyin or display_name. This bypasses the messy cases # below (CJK-prefixed display_name, concatenated pinyin). # 2. **name_pinyin with a space** ("Liyang Zhang") — clean HRIS # sync output for English-named hires. # 3. **display_name with a space, English suffix** — # "张利扬 Lian Zhang" pattern (CJK + romanised). Drop the # leading CJK token if present, then split the rest. # 4. **name_pinyin concatenated** ("zhangliyang") — legacy data # with no separator. Use as familyName, set givenName= # pinyin too (the SCIM server will accept it; admin can fix # the HRIS source data later). # 5. **Nothing usable** — Title-case the email local-part. def _is_cjk(s: str) -> bool: # Match the SAME character class as the ``cjk_name`` transform's # ``_CJK_NAME_RE`` / ``_CJK_LEAD_RE`` (CJK Ext-A U+3400 onward + the # ``·`` separator) — not just CJK Unified (U+4E00). A narrower guard # here let a display name leading with an Ext-A ideograph slip past # and blank ``display_name_cjk``, silently dropping the NetSuite # firstName. Keeping the two in sync closes that gap. return bool(_CJK_NAME_RE.search(s)) if "@" in raw_email: local = raw_email.split("@", 1)[0] if "." in local: given, family = local.split(".", 1) given = given.strip() family = family.strip() # Skip empty halves (e.g. ".foo@x" or "foo.@x"); fall # through to the next derivation strategy. if given and family: first = given.title() last = family.title() full_name = f"{first} {last}" if (not first or not last) and pinyin and " " in pinyin: parts = pinyin.split(maxsplit=1) if not first: first = parts[0] if not last and len(parts) > 1: last = parts[1] if not full_name: full_name = pinyin if (not first or not last) and display and " " in display: # Drop leading CJK tokens so "张利扬 Lian Zhang" yields # ["Lian", "Zhang"] not ["张利扬", "Lian Zhang"]. tokens = [t for t in display.split() if not _is_cjk(t)] if len(tokens) >= 2: if not first: first = tokens[0] if not last: last = " ".join(tokens[1:]) if not full_name: full_name = " ".join(tokens) if (not first or not last) and pinyin: # Concatenated pinyin fallback. Better than empty — Navan and # most SCIM servers accept ``Zhangliyang Zhangliyang`` rather # than rejecting the whole record. Admins fix the HRIS source. if not first: first = pinyin.title() if not last: last = pinyin.title() if not full_name: full_name = pinyin if not full_name: local = raw_email.split("@", 1)[0] if "@" in raw_email else raw_email full_name = local.replace(".", " ").replace("_", " ").title() # Final guard — never emit an empty givenName/familyName. Navan (and # most SCIM servers) hard-reject a create when either half is blank # ("User first name and last name are mandatory"). Single-token # records (e.g. a person whose only name is "Hice": no dot in the # email local-part, no name_pinyin, one-word display_name) reach # here with first/last still empty. Fill both from full_name; a # single token is duplicated into both halves, matching the # concatenated-pinyin fallback above — admins fix the HRIS source. if not first or not last: tokens = full_name.split() if len(tokens) >= 2: if not first: first = tokens[0] if not last: last = " ".join(tokens[1:]) else: token = tokens[0] if tokens else (full_name or "Unknown") if not first: first = token if not last: last = token # Direct attribute access (NOT ``getattr(person, name, None)``) # so a wrong column name surfaces as AttributeError at first use # rather than silently rendering as empty — see CLAUDE.md memory # ``feedback_no_getattr_default_for_model_columns``. # CJK name as written, extracted from display_name via the robust # ``cjk_name`` transform ("朱晓鹏 Xiaopeng Zhu" → "朱晓鹏"; bridges # "冯 帅"; ignores trailing remarks). ``cjk_name`` passes a pure-English # name through unchanged, so we BLANK the result when it carries no CJK # — that way a foreign hire renders empty → the consuming rule drops → # the vendor's existing (correct, English) name is preserved untouched. # Used by NetSuite (user decision 2026-06-24: keep Chinese names, don't # romanise). This is a NEW token; the romanised ``first_name`` / # ``last_name`` above are unchanged, so Navan / SCIM / Feishu / Google # mappings are unaffected. # Only treat the display_name as a CJK NAME when it LEADS with CJK # (e.g. "朱晓鹏 Xiaopeng Zhu"). ``cjk_name`` otherwise falls back to the # first CJK run anywhere in the string, which would wrongly pull a # trailing remark out of a foreign hire's name ("John Smith (财务部)" # → "财务部"). A foreign name never leads with CJK, so this blanks it # → firstName drops → the vendor's English name is preserved. _stripped_display = display.strip() display_cjk = ( _apply_transform(display, "cjk_name") if _stripped_display and _is_cjk(_stripped_display[0]) else "" ) raw_org_path = _str(org_path) bu = "" if raw_org_path: # Business Unit = first non-tenant-root segment of org_path. # Mirrors ``app.identity.bu_helpers.bu_from_org_path``; we # don't import that helper to avoid an identity-domain # dependency on this pure-data module. for part in [p.strip() for p in raw_org_path.split("/") if p.strip()]: if part.lower() not in ("pingcap", "pingcap-ati"): bu = part break ctx: dict[str, str] = { "work_email": raw_email, # ``employee_no`` on the model; ``employee_id`` is a property # alias kept for backward compatibility with older domain code. "employee_id": _str(person.employee_no), "first_name": first, "last_name": last, "full_name": full_name, # CJK name as written (see derivation above); empty for foreign # hires so the consuming rule drops → preserves the vendor name. "display_name_cjk": display_cjk, "name_pinyin": _str(person.name_pinyin), "display_name": _str(person.display_name), "job_title": _str(person.job_title), "country": _str(person.country), "city": _str(person.city), "manager_email": _str(manager_email), "manager_name": _str(manager_name), "hrbp_email": _str(hrbp_email), "hrbp_name": _str(hrbp_name), "employment_status": _str(person.employment_status), "hire_date": _str(person.hire_date), "termination_date": _str(person.termination_date), "mobile_phone": _str(person.mobile_phone), # ISO-2 of the mobile's OWN country (SF PerPhone.countryCode → # externalCode). Threaded to downstream phone formatters so a # host-country phone on an international assignee gets the right # dialing code instead of the home-country default. Empty when the # phone has no synced country → formatter falls back to ``country``. "mobile_country_code": _str(person.mobile_country_code), # Phone country, normalised, phone-first then work-country fallback. # "" when neither resolves — consumers must send nothing rather than # assume a default (汇联易 defaults an absent countryCode to 中国大陆, # so a wrong guess mislabels a foreign number). "mobile_country_iso2": ( resolve_country_iso2(person.mobile_country_code) or resolve_country_iso2(person.country) ), "org_path": raw_org_path, # ``cost_center`` is populated by the SF sync from # EmpJob.costCenter — backfills on next HRIS sync. Persons # whose SF record has no cost-center set (rare — usually # contractors / system accounts) render an empty string. "cost_center": _str(person.cost_center), # Business unit is the top non-tenant-root segment of org_path. "business_unit": bu, # Work location (e.g. "Beijing Office") — distinct from country. "location": _str(person.work_location_name), "work_location_code": _str(person.work_location_code), # SF picklist codes (migration 344) → Feishu enums/labels via the # feishu_contact value-maps. Rendered RAW here. "gender": _str(person.gender), "employee_type_code": _str(person.employee_type_code), "office_code": _str(person.office_code), # GitHub handle (migration 369) → Feishu GitHub ID custom attr. "github_id": _str(person.github_id), # SF-verbatim labels → Google Employee_Details (no value_map). "office_mail_group": _str(person.office_mail_group), "job_function": _str(person.job_function), "job_family": _str(person.job_family), "residence_country": _str(person.residence_country), # Legal entity / subsidiary (Navan calls this "division"). "company_name": _str(person.company_name), "company_code": _str(person.company_code), # Leaf dept code from the person's primary org membership — # passed in by the caller (the orchestrator already has the # membership row from ``_load_candidates``). Empty string when # the person has no primary membership, in which case Feishu's # ``department_ids`` slot falls back to the Feishu root # sentinel ``"0"`` on POST create (see # ``FeishuContactRunner.upsert``). "department_code": _str(org_node_source_id), } # org_path → levels (with optional drop_root stripping). parts = [p.strip() for p in (org_path or "").split("/") if p.strip()] drop = {s.strip().lower() for s in (org_path_drop_root or []) if s and s.strip()} while drop and parts and parts[0].lower() in drop: parts = parts[1:] # Allocate first 10 levels; index out of range → "". for i in range(10): ctx[f"org_path_level_{i}"] = parts[i] if i < len(parts) else "" # department_l1..l4 are friendly aliases for the first 4 levels. for i in range(1, 5): ctx[f"department_l{i}"] = parts[i - 1] if i - 1 < len(parts) else "" # ``department_combined`` = L2 / L3 / L4 with empty segments dropped. # L1 is the BU / division and lives in its own SCIM slot # (``business_unit``); the combined string is the path BELOW the # BU, which is the granularity Navan's department picker wants. ctx["department_combined"] = " / ".join( seg for seg in (ctx["department_l2"], ctx["department_l3"], ctx["department_l4"]) if seg ) return ctx # ── Render ─────────────────────────────────────────────────────────── def _apply_transform(value: str, transform: str) -> str: if transform == "lower": return value.lower() if transform == "phone_digits": # Separators only — see the SUPPORTED_TRANSFORMS note on why a # leading "+"/"00" is preserved rather than stripped. return _PHONE_SEPARATOR_RE.sub("", value or "") if transform == "upper": return value.upper() if transform == "trim": return value.strip() if transform == "title_case": return value.title() if transform == "email_local": return value.split("@", 1)[0] if "@" in value else value if transform == "email_domain": return value.split("@", 1)[1] if "@" in value else "" if transform == "iso_date": # Best-effort: accepts already-ISO date or YYYY-MM-DDTHH:MM:SS; # returns the date portion. Anything else passes through. if not value: return "" try: if "T" in value: return value.split("T", 1)[0] # Validate it parses as a date. date.fromisoformat(value[:10]) return value[:10] except ValueError: return value if transform == "strip_company_prefix": # ``01_PingCAP(US),Inc.`` → ``PingCAP(US),Inc.`` (single pass — # not stripping repeatedly so ``01_2_Foo`` stays ``2_Foo``, # which is the loud-failure behavior we prefer over silently # consuming everything that looks like a prefix). return _COMPANY_PREFIX_RE.sub("", value, count=1).strip() if transform == "cjk_name": # "冯帅 Shuai Feng" → "冯帅"; "欧阳 娜娜 Nana" → "欧阳娜娜" (a space # between surname and given name must NOT truncate to the surname); # "Shuai Feng" (no CJK) → "Shuai Feng". Take the LEADING CJK block # (bridging internal whitespace) so a trailing/parenthetical CJK # remark after the English name isn't concatenated in. If the string # doesn't start with CJK (atypical "English 中文" ordering) fall back # to the first contiguous CJK run — the original conservative pick. stripped = (value or "").strip() lead = _CJK_LEAD_RE.match(stripped) if lead: return re.sub(r"\s+", "", lead.group(0)) m = _CJK_NAME_RE.search(stripped) return m.group(0) if m else stripped if transform == "cjk_name_leading": # Guarded ``cjk_name``: leading CJK → the CJK name; anything else # (foreign name, or a foreign name with a TRAILING CJK remark like # "John Smith 顾问") passes through unchanged. See the # SUPPORTED_TRANSFORMS comment for the ``display_name_cjk``-token # comparison. stripped = (value or "").strip() lead = _CJK_LEAD_RE.match(stripped) if lead: return re.sub(r"\s+", "", lead.group(0)) return stripped if transform == "cjk_pinyin": # Leading CJK run → full pinyin ("冯帅 Shuai Feng" → "fengshuai"). # No CJK (foreign staff) → "" so the rule drops (Feishu keeps its # value). Reuses the ``cjk_name`` extraction, then pinyin — lazy # import so the pure whitelist/parse paths don't pull pypinyin. cjk = _apply_transform(value, "cjk_name") if not _CJK_NAME_RE.search(cjk or ""): return "" from app.core.pinyin_utils import to_pinyin return to_pinyin(cjk) if transform == "cn_city": # Normalise the HRIS city so Feishu renders it: # "上海市" → "上海" (strip the administrative 市 suffix — Feishu's # city dictionary keys on the bare municipality name, # same class as the country ISO-3→ISO-2 fix) # "北京" → "北京" (already clean, idempotent) # "Austin" / "Tokyo" / "Singapore" → passed through UNCHANGED. # Romanised overseas cities are NOT dropped: Feishu stores and shows # them fine (verified live 2026-07-16 — PATCH city="Austin" persisted # as "Austin"). The old "drop non-CJK" behaviour left EVERY overseas # employee's city blank — real production data loss, not "won't # render". Empty in → empty out (rule drops → the runner's # write-when-present preserves whatever Feishu already holds, so a # missing HRIS city never blanks an existing vendor value). s = (value or "").strip() if not s: return "" if len(s) > 1 and s.endswith("市") and _CJK_NAME_RE.search(s): s = s[:-1] return s if transform == "latin_name": # "孟凡生 Jason Meng" → "Jason Meng"; "John Smith" → "John Smith"; # "张三" → "". Strip every CJK run (+ interpunct), collapse the # whitespace left behind, trim. Order-agnostic (works whether the # CJK name leads or trails the English part). s = _CJK_NAME_RE.sub(" ", value or "") s = re.sub(r"\s+", " ", s).strip() # Defence in depth against the same "张东晖 null" sentinel the HRIS # ingest sanitiser catches: if a stray literal "null"/"None" reached # here (e.g. via a display_name row that predates that fix), it would # otherwise be pushed to Feishu as the person's en_name. Drop it — # whole-token only, so "Knull" survives — and return "" (rule drops, # write-when-present preserves the vendor value). return strip_sentinel_name_tokens(s) or "" # Should never happen — validation above rejects unknowns. raise MappingError(f"unknown transform {transform!r}") def render_value(rule: MappingRule, ctx: dict[str, str]) -> str | None: """Render one rule against a context. Returns: The rendered string when a value is produced, ``None`` when the rule produces nothing (empty after transforms AND no default). Returning None — vs. empty string — lets the runner decide whether to include the field in the payload. SCIM treats present-empty differently from absent: present-empty clears the value on the downstream side, absent leaves it alone. """ # Substitute tokens. ``re.sub`` with a callback handles missing # tokens gracefully (already validated at parse time). rendered = _TEMPLATE_TOKEN_RE.sub( lambda m: ctx.get(m.group(1), ""), rule.source_template, ) for x in rule.transform: rendered = _apply_transform(rendered, x) if rule.value_map is not None: # ``value_map`` is treated as a STRICT translation: unknown # source values render empty rather than passing through # untouched. This matches the way SCIM systems like Navan # validate enum-shaped fields (region / companyOffice / ...) — # passing an unknown value through would 400 the whole push. # The empty render then either falls back to ``default`` (if # set) or drops the field entirely (the runner sees absence). # # ``value_map_passthrough`` flips this to OVERRIDE semantics: # an unmapped value flows through unchanged. Used for rules # that rewrite only a few values (e.g. remap one manager's # email to a delegate) and must leave every other value alone. if rendered in rule.value_map: # Explicit mapping hit — honour it verbatim, INCLUDING a # deliberately blank mapping ("clear this field for source # value X"). A blank hit must drop the field (present-absent), # NOT be rescued by ``default`` below — distinguish a hit from # a STRICT miss so the two no longer collapse together. mapped = rule.value_map[rendered] return mapped if mapped else None # Miss: passthrough keeps the original value untouched; STRICT # blanks it (then ``default`` / drop handling below applies). rendered = rendered if rule.value_map_passthrough else "" if not rendered: if rule.default is not None: return rule.default return None return rendered def collect_render_diagnostics( config: MappingConfig, ctx: dict[str, str], *, exclude_create_only: bool = False, ) -> tuple[dict[str, str], list[tuple[str, str]]]: """Post-render diagnostics for the outcome-taxonomy layer. ``exclude_create_only`` MUST mirror the flag passed to :func:`render_payload` for the same person: on the UPDATE shape, ``create_only`` rules are excluded from both the payload and these verdicts (see the loop guard below). Returns ``(value_map_misses, dropped_required)``: * ``value_map_misses`` — ``{target_field: source_value}`` for every STRICT ``value_map`` rule whose (post-transform, non-empty) source value had no entry in the map. Surfaced as a "new downstream code with no value_map row" signal so a brand-new HRIS code (new legal entity, new employee-type) is alerted ONCE as a config task rather than silently blanking a field for the whole population. * ``dropped_required`` — ``[(target_field, reason)]`` for every rule marked ``required=True`` whose render produced nothing (empty source with no default, or a STRICT value_map miss with no default). The orchestrator turns a non-empty list into a per-person ``blocked`` outcome BEFORE the vendor call, with an actionable reason. Cheap: one extra pass over the rule list (no vendor I/O). Mirrors the exact staging of :func:`render_value` so verdicts can't drift. """ misses: dict[str, str] = {} dropped_required: list[tuple[str, str]] = [] for rule in config.rules: if exclude_create_only and rule.create_only: # UPDATE shape: ``create_only`` rules are not rendered at all # (see :func:`render_payload`), so neither a value_map miss # nor a required-drop verdict applies — flagging one would # falsely BLOCK a linked person whose create-only field # (e.g. NetSuite firstName) can't render, even though the # update payload deliberately omits it. continue after = _TEMPLATE_TOKEN_RE.sub( lambda m: ctx.get(m.group(1), ""), rule.source_template, ) for x in rule.transform: after = _apply_transform(after, x) vm_miss = False if ( rule.value_map is not None and not rule.value_map_passthrough and after # only a real (non-empty) source value counts as a miss and after not in rule.value_map ): vm_miss = True misses[rule.target_field] = after if rule.required and render_value(rule, ctx) is None: dropped_required.append(( rule.target_field, ( f"unmapped value {after!r} (no value_map entry, no default)" if vm_miss else "empty source value (no default)" ), )) return misses, dropped_required def render_payload( config: MappingConfig, ctx: dict[str, str], *, exclude_create_only: bool = False, ) -> dict[str, Any]: """Render every rule into a (possibly nested) payload dict. ``exclude_create_only=True`` renders the UPDATE shape: rules marked ``create_only`` are skipped entirely (field absent → merge-semantics vendors preserve their existing value). The sweep / single-person push pass it for persons Helix has already linked (``IdentitySyncState.external_id`` set); previews and unlinked persons render the full CREATE shape. Path segment semantics: * ``name.familyName`` → ``out["name"]["familyName"]`` (nested dict) * ``emails.0.value`` → ``out["emails"][0]["value"]`` (numeric segments materialise as list indices, which is what SCIM multi-valued attributes expect — a JSON object with key "0" would silently be ignored by every SCIM server) * ``urn:ietf:...:User.employeeNumber`` → the URN segment is a regular dict key (colons are part of the SCIM extension namespace identifier, not nesting) Numeric values for boolean and numeric SCIM fields: A rendered string ``"true"`` / ``"false"`` is emitted as a JSON boolean for paths that appear to be SCIM boolean attributes (``emails.N.primary``, ``active``, ``phoneNumbers.N.primary``). Without this coercion the SCIM server would reject the type. We keep this conservative — only the well-known boolean slots; for everything else the string rendering wins so admins can put arbitrary text through the mapping pipeline. Rules whose render result is ``None`` are dropped entirely. """ def _coerce(target_field: str, value: str) -> Any: last = target_field.rsplit(".", 1)[-1] # SCIM boolean slots — coerce "true"/"false"/"1"/"0". if last in ("primary", "active"): v = value.strip().lower() if v in ("true", "1", "yes"): return True if v in ("false", "0", "no", ""): return False return value out: dict[str, Any] = {} for rule in config.rules: if exclude_create_only and rule.create_only: continue value = render_value(rule, ctx) if value is None: continue coerced = _coerce(rule.target_field, value) path = split_target_path(rule.target_field) # Walk the path materialising containers as needed. ``cur`` # alternates between dict and list as we encounter alphanumeric # vs. numeric path segments. cur: Any = out for i, seg in enumerate(path): is_last = i == len(path) - 1 seg_is_index = seg.isdigit() next_seg_is_index = ( False if is_last else path[i + 1].isdigit() ) if seg_is_index: idx = int(seg) if not isinstance(cur, list): # Programming error — we'd only hit this if the # parent step set up a dict where this loop # expected a list. Skip the rule rather than # crash the whole render. break # Pad list with placeholders if the admin skipped a # gap (rare; SCIM emails[0] / [1] usually stays dense). while len(cur) <= idx: cur.append(None) if is_last: cur[idx] = coerced else: if cur[idx] is None or not isinstance( cur[idx], (dict, list), ): cur[idx] = [] if next_seg_is_index else {} cur = cur[idx] else: if not isinstance(cur, dict): break if is_last: cur[seg] = coerced else: existing = cur.get(seg) if not isinstance(existing, (dict, list)): existing = [] if next_seg_is_index else {} cur[seg] = existing cur = existing # Strip leftover None placeholders from sparse list rules. SCIM # servers reject ``emails: [null, {...}]``. The renderer above # only inserts None on numeric-index gaps, so this clean-up is # the matching guard for that pad path. def _clean(node: Any) -> Any: if isinstance(node, list): return [_clean(x) for x in node if x is not None] if isinstance(node, dict): return {k: _clean(v) for k, v in node.items()} return node cleaned = _clean(out) return cleaned if isinstance(cleaned, dict) else {} def trace_render( config: MappingConfig, ctx: dict[str, str], ) -> list[dict[str, Any]]: """Render each rule and return a per-rule trace. Drives the admin UI "Mapping Preview" drawer — for each rule it exposes the source template, the raw token-substituted value, the intermediate transform output, the final rendered value (after value_map / default), and whether the rule was dropped (rendered None → field absent from payload). Cheap to compute (one extra pass over the rules in addition to ``render_payload``), but kept in its own function so the hot path pays nothing. """ out: list[dict[str, Any]] = [] for rule in config.rules: # Mirror ``render_value`` step-by-step so the UI can show # exactly where empty inputs / missing maps lost data. The # intermediate strings here are debugging output — they don't # feed back into the final payload (which still comes from # ``render_payload``). raw = _TEMPLATE_TOKEN_RE.sub( lambda m: ctx.get(m.group(1), ""), rule.source_template, ) after_transforms = raw for x in rule.transform: after_transforms = _apply_transform(after_transforms, x) after_value_map: str | None = after_transforms value_map_hit: bool | None = None if rule.value_map is not None: # {} = strict map (drop on miss) value_map_hit = after_transforms in rule.value_map after_value_map = rule.value_map.get( after_transforms, after_transforms if rule.value_map_passthrough else "", ) final = after_value_map dropped = False used_default = False if not final: if value_map_hit: # Explicit blank mapping hit → deliberate drop; mirror # render_value and do NOT rescue with ``default``. final = None dropped = True elif rule.default is not None: final = rule.default used_default = True else: final = None dropped = True out.append({ "source_template": rule.source_template, "target_field": rule.target_field, "transform": list(rule.transform), "value_map_hit": value_map_hit, "default": rule.default, "raw": raw, "after_transforms": after_transforms, "after_value_map": after_value_map, "rendered": final, "dropped": dropped, "used_default": used_default, }) return out # ── Mapping metadata catalog (admin config builder) ─────────────────── # Source-field categories drive the grouped picker in the admin mapping # editor. Every name here MUST also be in ``SUPPORTED_SOURCE_FIELDS`` — # the catalog is the authoritative list the frontend renders, so the FE # can never drift from what the renderer actually accepts (the old # hard-coded FE transform list silently dropped ``strip_company_prefix``). _SOURCE_FIELD_CATEGORY: dict[str, str] = { "work_email": "identity", "employee_id": "identity", "mobile_phone": "identity", "mobile_country_code": "identity", "mobile_country_iso2": "identity", "github_id": "identity", "display_name": "name", "first_name": "name", "last_name": "name", "full_name": "name", "name_pinyin": "name", "job_title": "employment", "employment_status": "employment", "hire_date": "employment", "termination_date": "employment", "cost_center": "employment", # SF picklist codes (migration 344) projected to Feishu enums/labels. "gender": "employment", "employee_type_code": "employment", # SF-verbatim Job Function / Job Family labels (Google Employee_Details). "job_function": "employment", "job_family": "employment", "manager_email": "manager", "manager_name": "manager", "manager_employee_no": "manager", "country": "location", "city": "location", "location": "location", "work_location_code": "location", "office_code": "location", # SF-verbatim office mail-group + residence country (Google Employee_Details). "office_mail_group": "location", "residence_country": "location", "company_name": "org", "company_code": "org", "business_unit": "org", "department_code": "org", "org_path": "org", } def mapping_metadata() -> dict[str, Any]: """Authoritative catalog for the admin mapping-config builder. Returns the source fields (grouped by category) and the transform pipeline steps the renderer accepts. The frontend renders dropdowns from this so admins pick valid tokens/transforms instead of typing free text that only fails at save (``MappingError``) or — worse — silently renders empty. Derived ``org_path_level_*`` / ``department_l*`` are folded under the ``derived`` category. """ source_fields: list[dict[str, str]] = [] for name in sorted(_BASE_SOURCE_FIELDS): source_fields.append({ "name": name, "category": _SOURCE_FIELD_CATEGORY.get(name, "other"), }) for name in sorted(_DERIVED_SOURCE_FIELDS): source_fields.append({"name": name, "category": "derived"}) return { "source_fields": source_fields, "transforms": [{"name": x} for x in sorted(SUPPORTED_TRANSFORMS)], } def hash_payload(payload: dict[str, Any]) -> str: """sha256 of the payload's sorted-key JSON. Safe to compare across runs because Python dict iteration is insertion-ordered, and JSON rendering with ``sort_keys=True`` normalises that. """ encoded = json.dumps(payload, sort_keys=True, default=str) return hashlib.sha256(encoded.encode("utf-8")).hexdigest()