"""Managed-location resolver for ``asset_items.location``. The free-text ``asset_items.location`` column is the source of every dirty-data incident in the asset domain: * HRIS-sourced ``identity_persons.work_location_name`` lands in the column at claim time (e.g. ``Santa Clara``, ``北京市``, ``Würzburg``) — values the managed ``asset_locations`` master does NOT list, so the location dropdown then can't filter them. * Legacy 易盘点 xlsx imports wrote city / state strings (``California``, ``Sunnyvale``) into the column. * Bulk admin actions copy raw user-typed text through unchecked. The historical pattern of fixing this with one-off prod SQL backfills (Santa Clara → 美国; 北京市 → 北京; Würzburg → 德国) keeps the data clean for ~24h before the next HRIS sync re-dirties it. This resolver is the single canonical write-time gate. Every path that mutates ``AssetItem.location`` MUST route the value through ``resolve(raw)`` first. The contract: * Return ``None`` for None / empty / whitespace input. * If the trimmed value matches an active ``asset_locations.name`` row exactly, return that name (canonicalised by master). * Otherwise consult the bilingual normalization map (中国 城市名 + 城市市后缀 → 城市; 外国 城市 → 国家) and re-check against the master. If the normalized result is in master, return it. * Otherwise return ``None`` (the field is left blank rather than polluted; admins can still set the location via the dropdown). The resolver is constructed with an explicit master snapshot — either passed in or fetched on first use — so callers can amortize the master read across a bulk operation. The snapshot is per-request (not a process-wide cache), which keeps semantics simple: master edits (location create / update / delete from the settings page) take effect on the very next request. """ from __future__ import annotations import logging import unicodedata import uuid from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.asset.models import AssetLocation _logger = logging.getLogger(__name__) # ── normalization map ────────────────────────────────────────────── # # Population rule: only values that have been observed in prod # ``asset_items.location`` OR ``identity_persons.work_location_name`` # get an explicit entry. We do NOT build an aspirational world-cities # oracle — every entry is rooted in a real dirty value we want to # normalize. When a new dirty value shows up, add an explicit entry # (and the corresponding ``asset_locations`` master row in a # follow-up migration) — the resolver intentionally returns ``None`` # for unknowns so the field is left blank instead of polluted. # # Keys are matched after Unicode NFKC normalization + ``.strip()``; # the lookup is case-sensitive (HRIS sourced names are mixed-case # proper nouns, and English city names like ``LA`` vs ``La`` aren't # semantically equivalent to us). _NORMALIZATION_MAP: dict[str, str] = { # ── 中国 PingCAP 主城市 (HRIS uses 市 suffix; master drops it) ── "北京市": "北京", "上海市": "上海", "广州市": "广州", "深圳市": "深圳", "杭州市": "杭州", "成都市": "成都", "南京市": "南京", "长沙市": "长沙", "郑州市": "郑州", "沈阳市": "沈阳", "武汉市": "武汉", "大连市": "大连", "宁波市": "宁波", # ── 美国 (city/state-level → country) ── "Santa Clara": "美国", "Sunnyvale": "美国", "California": "美国", "San Jose": "美国", "San Francisco": "美国", "San Mateo": "美国", "San Diego": "美国", "Seattle": "美国", "King County": "美国", "Snohomish": "美国", "Florida City": "美国", "Chapel Hill": "美国", "Los Angeles": "美国", "Sacramento": "美国", "Castro Valley": "美国", "Riverside": "美国", "Denver": "美国", "Williamson County": "美国", "Norfolk": "美国", "Denton": "美国", "Alameda": "美国", "Jacksonville": "美国", "Chicago": "美国", "Cincinnati": "美国", "New Haven": "美国", "Austin": "美国", "Santa Cruz": "美国", # 2026-08-24 sweep of every active person's work location — these # nine values resolved to nothing, so their holders sat outside # every regional administrator's scope with no signal. "New York": "美国", "Brockton": "美国", "Highland": "美国", "Bothell": "美国", # ── 日本 ── "Tokyo": "日本", "Fukuoka": "日本", "Toyonaka": "日本", # ── 新加坡 ── "Singapore": "新加坡", # ``Other(SG)`` is a real SuccessFactors HRIS work-location value # (1 active person carried it in the 2026-05-30 snapshot) — SF # records the literal string when the upstream HR record fell into # the "Other" bucket for the Singapore region. Treat it as 新加坡 # for location purposes; the audit trail still captures the raw # value via ``operation.detail['location_raw']``. "Other(SG)": "新加坡", # ── 马来西亚 ── "Kuala Lumpur": "马来西亚", "Johor Bahru": "马来西亚", # ── 印度 ── "Bangalore": "印度", # ── 香港 ── "Hong Kong": "香港", # ── 荷兰 ── "Amsterdam": "荷兰", "Rotterdam": "荷兰", "The Hague": "荷兰", "Breukelen": "荷兰", # ── 德国 ── "Frankfurt": "德国", "Würzburg": "德国", # ── 澳大利亚 ── "Sydney": "澳大利亚", # ── 加拿大 ── "Toronto": "加拿大", "Cochrane": "加拿大", "Markham": "加拿大", "Vancouver": "加拿大", # ── 印度尼西亚 ── "Jakarta": "印度尼西亚", # ── 爱尔兰 ── "Dublin": "爱尔兰", # ── 台湾 ── "Taipei": "台湾", # ── 泰国 ── "Bangkok": "泰国", # ── 韩国 ── "Seoul": "韩国", # ── 阿联酋 ── "Dubai": "阿联酋", # ── 越南 ── "Hanoi": "越南", # ── 波兰 ── "Warsaw": "波兰", # ── 英国 ── "Leeds": "英国", "London": "英国", # ── 巴西 / 瑞典 (master rows seeded by migration 535) ── "São Paulo": "巴西", "Niterói": "巴西", "Stockholm": "瑞典", # ── 意大利 ── "Brindisi": "意大利", # ── 罗马尼亚 ── "Popeşti-Leordeni": "罗马尼亚", } # ``resolve()`` NFKC-normalizes its input before the lookup, so the keys # have to be NFKC too. Several are non-ASCII (Würzburg, São Paulo, # Niterói, Popeşti-Leordeni) and an editor that saves decomposed forms # would leave a key that can never match — a failure with no error, just # a value that quietly stops resolving. Normalize once at import. _NORMALIZATION_MAP = { unicodedata.normalize("NFKC", key): value for key, value in _NORMALIZATION_MAP.items() } class ManagedLocationResolver: """Resolve raw location strings to canonical ``asset_locations.name``. Construct one per request (or per bulk operation) so the master snapshot read happens once. The resolver is read-only — it never mutates the master or the asset row. """ def __init__(self, master_names: frozenset[str]) -> None: self._master = master_names @classmethod async def load( cls, db: AsyncSession, *, include_inactive: bool = False, ) -> ManagedLocationResolver: """Snapshot the active ``asset_locations.name`` set. Apply the same NFKC + ``.strip()`` normalization to the master names that ``resolve()`` applies to caller input. Without this the master frozenset would carry the raw DB form while the membership test below normalizes the input — a TiDB ``PAD SPACE`` collation, an off-by-one fullwidth-character master row, or a manual SQL insert with stray whitespace would all silently break resolution (Round-1 review TiDB lens finding). Empty/whitespace-only rows are filtered out so they can't poison the set. ``include_inactive`` widens the snapshot to deactivated locations. Write paths must NOT set it — a deactivated location is exactly one an operator has retired, and letting new writes land there re-pollutes the column. The regional permission gate does set it: an administrator keeps authority over the assets left behind at a location after it is deactivated, and dropping those rows to "unresolvable" would silently hand them to the global tier only. """ stmt = select(AssetLocation.name) if not include_inactive: stmt = stmt.where(AssetLocation.is_active.is_(True)) result = await db.execute(stmt) normalised: set[str] = set() for (raw,) in result.all(): if not raw: continue canon = unicodedata.normalize("NFKC", raw).strip() if canon: normalised.add(canon) return cls(frozenset(normalised)) def resolve(self, raw: str | None) -> str | None: """Return the canonical managed name, or ``None`` if no match. Pipeline: 1. ``None`` / empty / whitespace → ``None``. 2. NFKC-normalize + strip whitespace. 3. Exact-match against master → return. 4. Apply normalization map → re-check against master → return if the normalized result is in master. 5. Strip a single trailing ``市`` and re-check against master. 6. Otherwise log a debug line and return ``None``. Step 4 intentionally re-checks against the master rather than trusting the map blindly: if an admin disables / renames a managed row, the map entry that pointed at it stops resolving without code change. Step 5 exists because the map is explicit-only by design (entries are added when a dirty value is observed), which leaves a silent gap the moment an operator adds a new China city to the master: HRIS writes ``苏州市`` while the master row says ``苏州``, and resolution fails with nothing to alert on. The generic suffix strip closes that without weakening anything — it only ever resolves to a name the master already lists. This mirrors the ``_norm_city`` tolerance the office region gate carried before locations became the authority. """ if raw is None: return None if not isinstance(raw, str): # Defensive: a caller passing e.g. an int would have # crashed at DB write anyway; bail early with None so the # asset row keeps its previous location. return None cleaned = unicodedata.normalize("NFKC", raw).strip() if not cleaned: return None if cleaned in self._master: return cleaned normalized = _NORMALIZATION_MAP.get(cleaned) if normalized is not None and normalized in self._master: return normalized if cleaned.endswith("市"): stripped = cleaned[:-1].strip() if stripped and stripped in self._master: return stripped _logger.debug( "ManagedLocationResolver: dropping unrecognised location %r " "(no master / map hit) — asset row left blank", cleaned, ) return None # ── module-level convenience for callers that just need a one-shot # resolve and don't want to manage a resolver instance themselves. async def resolve_managed_location( db: AsyncSession, raw: str | None, ) -> str | None: """One-shot wrapper that loads the master + resolves ``raw``. Prefer ``ManagedLocationResolver.load(db)`` when the caller has several values to resolve back-to-back, so the master read is amortized. This convenience function pays the cost of one master SELECT every call — fine for one-off paths (single asset claim/return), but wasteful in a bulk loop. """ resolver = await ManagedLocationResolver.load(db) return resolver.resolve(raw) class LocationIndex: """Raw location string → ``asset_locations.id``. Wraps :class:`ManagedLocationResolver` (raw string → canonical managed *name*) with the name → id mapping that anything keyed on the location row needs — the asset region gate, the office→location lookup, the coverage rollups. Build one per request or per bulk operation so the master read is amortised across the rows. Loaded with ``include_inactive=True``: a deactivated location still has to resolve, or the assets parked there would silently fall out of every regional administrator's scope. """ def __init__( self, resolver: ManagedLocationResolver, ids_by_name: dict[str, uuid.UUID], ) -> None: self._resolver = resolver self._ids_by_name = ids_by_name @classmethod async def load(cls, db: AsyncSession) -> LocationIndex: resolver = await ManagedLocationResolver.load(db, include_inactive=True) rows = await db.execute(select(AssetLocation.id, AssetLocation.name)) ids_by_name: dict[str, uuid.UUID] = {} for location_id, name in rows.all(): if not name: continue canon = unicodedata.normalize("NFKC", name).strip() if canon: # Deterministic on a duplicate name: the resolver's # master set is name-deduplicated anyway, so either row # is equally (in)correct — stable beats arbitrary. ids_by_name.setdefault(canon, location_id) return cls(resolver, ids_by_name) def resolve_name(self, raw: str | None) -> str | None: """Canonical managed location *name* for a raw string, or None.""" return self._resolver.resolve(raw) def resolve_id(self, raw: str | None) -> uuid.UUID | None: """Canonical location id for a raw location string, or None.""" name = self._resolver.resolve(raw) if name is None: return None return self._ids_by_name.get(name) def resolve_first_id(self, *values: str | None) -> uuid.UUID | None: """Location id of the first of ``values`` that resolves, else None.""" for value in values: resolved = self.resolve_id(value) if resolved is not None: return resolved return None async def resolve_location_id_by_name( db: AsyncSession, raw: str | None, ) -> uuid.UUID | None: """One-shot ``raw location string -> asset_locations.id``. Prefer building a :class:`LocationIndex` when several values need resolving — this pays for a full master read per call. Fine for the one-asset-at-a-time paths (offboarding blockers, handover hints). """ index = await LocationIndex.load(db) return index.resolve_id(raw) # ── helpers for tests / data-fix scripts ─────────────────────────── def get_normalization_map_snapshot() -> dict[str, str]: """Return a shallow copy of the normalization map. Test code uses this to assert the migration backfill SQL is consistent with the runtime resolver. Production code should NOT consult this directly — call ``resolve()`` instead, which factors in the live master state. """ return dict(_NORMALIZATION_MAP)