"""Identity data-quality scanner + read/decision API. The scanner runs five reconciliation checks over the identity graph and upserts findings (deduped across runs); the read methods power the admin panel; the decision method lets an admin acknowledge / accept-risk / resolve a finding. See :mod:`app.identity.data_quality_models` for the finding-type catalog and lifecycle. """ from __future__ import annotations import logging import uuid from datetime import UTC, datetime from typing import Any from zoneinfo import ZoneInfo from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased from app.core.exceptions import NotFoundError, ValidationError from app.core.txn import best_effort from app.domains.identity_sync.models import ( IdentitySyncState, IdentitySyncTarget, ) from app.identity.data_quality_models import ( FINDING_SEVERITY, FINDING_TYPES, IdentityDataQualityFinding, IdentityDataQualityRun, ) from app.identity.person_models import ( IdentityExternalIdentity, IdentityPerson, ) logger = logging.getLogger(__name__) # Statuses that represent a still-relevant finding (anything but resolved). _LIVE_STATUSES = ("open", "acknowledged", "accepted_risk") _DECISION_STATUSES = ("acknowledged", "accepted_risk", "resolved") _NOTIFY_TEMPLATE = "identity.data_quality_new_high" class _Detected: """One detected problem before it's reconciled against stored rows.""" __slots__ = ("finding_type", "dedup_key", "person_id", "subject_display", "title", "detail") def __init__(self, finding_type, dedup_key, person_id, subject_display, title, detail): self.finding_type = finding_type self.dedup_key = dedup_key self.person_id = person_id self.subject_display = subject_display self.title = title self.detail = detail class IdentityDataQualityService: def __init__(self, db: AsyncSession): self.db = db # ── Detection (pure reads) ──────────────────────────────────────── async def _detect_all( self, ) -> tuple[list[_Detected], set[str], set[tuple[str, str]]]: """Run every detector. Returns ``(detected, incomplete_types, protected_keys)``. ``incomplete_types`` names finding-types whose detection could NOT be fully performed this run (e.g. an SF outage blocked the divergence pull). ``run_scan`` skips auto-resolution for those types so an upstream outage can never silently clear real open findings. ``protected_keys`` are individual ``(finding_type, dedup_key)`` that must NOT auto-resolve even though their type is otherwise complete — e.g. a divergence finding for a person pinned to a DISABLED instance we had no ground truth for this run.""" out: list[_Detected] = [] incomplete: set[str] = set() protected: set[tuple[str, str]] = set() out += await self._detect_orphan_login_identities() out += await self._detect_stale_tombstones() out += await self._detect_manager_departed() out += await self._detect_missing_manager() out += await self._detect_sync_drift_failed() out += await self._detect_offboarding_session_not_cut() out += await self._detect_hris_email_name_malformed() hris_divergence, hris_complete, hris_protected = ( await self._detect_hris_status_divergence() ) out += hris_divergence if not hris_complete: incomplete.add("hris_status_divergence") protected |= hris_protected return out, incomplete, protected async def _detect_orphan_login_identities(self) -> list[_Detected]: """A live, login-enabled binding for a departed / deleted person — the person can still SSO in despite having left. Security gap.""" rows = (await self.db.execute( select(IdentityExternalIdentity, IdentityPerson) .join(IdentityPerson, IdentityPerson.id == IdentityExternalIdentity.person_id) .where( IdentityExternalIdentity.deleted_at.is_(None), IdentityExternalIdentity.is_login_identity.is_(True), IdentityExternalIdentity.identity_status == "active", or_( IdentityPerson.deleted_at.is_not(None), IdentityPerson.employment_status == "left", ), ) )).all() out: list[_Detected] = [] for ext, person in rows: out.append(_Detected( "orphan_login_identity", str(ext.id), person.id, person.display_name or person.work_email, f"{person.display_name or person.work_email} 已离职/删除,但 " f"{ext.provider} 登录绑定仍可用", { "provider": ext.provider, "external_identity_id": str(ext.id), "employment_status": person.employment_status, "person_deleted": person.deleted_at is not None, }, )) return out async def _detect_stale_tombstones(self) -> list[_Detected]: """A person purged from the HRIS that still carries a live, login-enabled binding — purged upstream yet can still log in. ``identity_status == 'active'`` is part of the predicate, matching the sibling :meth:`_detect_orphan_login_identities`. Without it the check reported any ``is_login_identity`` row as "can still log in", including the states that exist precisely BECAUSE the login surface is already closed: * ``pending_welcome`` — the mailbox was minted but the welcome mail carrying the first-login password failed, so the provisioner SUSPENDED it (see ``account_provisioner``). A suspended mailbox is the opposite of a live login. * ``inactive`` / ``suspended`` — offboarding-suspended, or an admin freeze. Prod 2026-08-18: the SF admin account ``CBadmin`` was paged to five IT admins as a high-severity "purged yet can still log in" finding while its Google mailbox had been ``suspended: ADMIN`` since creation with ``lastLoginTime`` never set — and Helix's own ``is_effectively_departed`` already refuses a session to any ``hris_record_state='purged'`` row. Two closed doors reported as an open one. A high-severity finding that overstates its risk teaches admins to discount the channel, so keep this predicate to bindings that are genuinely usable. """ rows = (await self.db.execute( select(IdentityPerson) .join(IdentityExternalIdentity, IdentityExternalIdentity.person_id == IdentityPerson.id) .where( IdentityPerson.hris_record_state == "purged", IdentityExternalIdentity.deleted_at.is_(None), IdentityExternalIdentity.is_login_identity.is_(True), IdentityExternalIdentity.identity_status == "active", ) .distinct() )).scalars().all() return [ _Detected( "stale_tombstone", str(p.id), p.id, p.display_name or p.work_email, f"{p.display_name or p.work_email} 已从 HRIS 清除(purged),但仍保留可登录绑定", {"hris_record_state": p.hris_record_state}, ) for p in rows ] async def _detect_manager_departed(self) -> list[_Detected]: mgr = aliased(IdentityPerson) rows = (await self.db.execute( select(IdentityPerson, mgr) .join(mgr, mgr.id == IdentityPerson.manager_person_id) .where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", or_( mgr.deleted_at.is_not(None), mgr.employment_status == "left", ), ) )).all() return [ _Detected( "manager_departed", str(person.id), person.id, person.display_name or person.work_email, f"{person.display_name or person.work_email} 的汇报对象 " f"{m.display_name or m.work_email} 已离职", { "manager_person_id": str(m.id), "manager_display": m.display_name or m.work_email, "manager_status": m.employment_status, }, ) for person, m in rows ] async def _detect_missing_manager(self) -> list[_Detected]: """Flag an active real employee with no usable manager — whatever the cause (HRIS declares none, or the declared manager is gone/absent and ``backfill_manager_links`` couldn't resolve it). Two exclusions strip the noise the unfiltered check produced (prod 2026-06-09: 13 findings, 10 of which were Helix link-failures that the new backfill resolves on the next sync — those simply stop having ``manager_person_id IS NULL``): * has ≥1 active direct report → org-apex (CEO / founder), who legitimately has no manager. Excluded. * ``job_title IS NULL`` → seeded SF service/admin rows (TimeAdmin / LifecyleAdmin / ECadmin …) that carry no job and no manager by nature. Excluded. Deliberately does NOT exclude ``manager_external_id IS NOT NULL``: because backfill runs at the tail of every sync, a row that still has a declared id but no resolved ``manager_person_id`` at scan time means the manager is genuinely unresolvable (left / deleted / not in Helis) — a real "manager departed, never reassigned" gap that would otherwise be invisible (``manager_departed`` can't fire on a NULL pointer either). ``manager_external_id`` is surfaced in the detail so an admin can see who HRIS *claims* the manager is. """ subordinate = aliased(IdentityPerson) has_active_report = ( select(subordinate.id) .where( subordinate.manager_person_id == IdentityPerson.id, subordinate.employment_status == "active", subordinate.deleted_at.is_(None), ) .exists() ) rows = (await self.db.execute( select(IdentityPerson).where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", IdentityPerson.manager_person_id.is_(None), IdentityPerson.job_title.is_not(None), ~has_active_report, ) )).scalars().all() return [ _Detected( "missing_manager", str(p.id), p.id, p.display_name or p.work_email, f"{p.display_name or p.work_email} 没有汇报对象(manager 为空)", { "job_title": p.job_title, "declared_manager_external_id": p.manager_external_id, }, ) for p in rows ] async def _detect_hris_email_name_malformed(self) -> list[_Detected]: """HR's 邮箱名 (SF ``PerPersonal.customString4``) isn't an email name. Predicate is a **round-trip**, not a character blocklist: run the stored value through the provisioner's own derivation and ask whether it survived unchanged. That way the check can never disagree with the thing it is checking (the failure mode the reconcile's own comment warns about), and it needs no list of "bad characters" to keep in sync with the sanitiser. * ``minted == ""`` → Helix cannot use the value at all and falls back to the firstName/lastName derivation. HR's intent is being ignored — e.g. a 3-token display name, a ``@qq.com`` personal address, or CJK text. * ``minted != raw`` → usable, but only after Helix reinterpreted it (``"Koji Betsuyaku"`` → ``koji.betsuyaku``). The address is right; the SF field is still dirty. Deliberately ``low`` severity and deliberately on the HR-facing data-quality panel rather than in an IT alert: nothing is broken for the employee, and the fix belongs to whoever owns the SF record. Leavers are excluded — nobody edits a departed person's HRIS fields to clear a finding. """ # Local imports: app.identity must not take a hard edge on the # google_workspace domain (same rule the email-name reconcile obeys). from app.core.corporate_domains import ( # noqa: PLC0415 get_corporate_email_domains, ) from app.domains.google_workspace.account_provisioner import ( # noqa: PLC0415 _PROVISIONING_SETTING, _local_part_from_hris_email, ) from app.integrations.service import get_platform_setting # noqa: PLC0415 settings = await get_platform_setting( _PROVISIONING_SETTING, default={}, session=self.db, ) or {} max_local = int( (settings if isinstance(settings, dict) else {}).get( "local_part_max_length", ) or 30 ) corporate_domains = await get_corporate_email_domains(self.db) rows = (await self.db.execute( select(IdentityPerson).where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", IdentityPerson.hris_email_name.isnot(None), IdentityPerson.hris_email_name != "", ) )).scalars().all() out: list[_Detected] = [] for person in rows: raw = (person.hris_email_name or "").strip() raw_local = raw.lower().partition("@")[0] minted = _local_part_from_hris_email( raw, max_length=max_local, corporate_domains=corporate_domains, ) if minted == raw_local: continue subject = person.display_name or person.work_email usable = bool(minted) title = ( f"{subject} 的 HR 邮箱名 {raw!r} 不是邮箱名格式,Helix 按 " f"{minted!r} 理解" if usable else f"{subject} 的 HR 邮箱名 {raw!r} 无法使用,Helix 已改用姓名派生" ) out.append(_Detected( "hris_email_name_malformed", str(person.id), person.id, subject, title, { "hris_email_name": raw, "interpreted_as": minted or None, "usable": usable, "work_email": person.work_email, "hris_person_id": person.hris_person_id, "sf_field": "PerPersonal.customString4", }, )) return out async def _detect_sync_drift_failed(self) -> list[_Detected]: rows = (await self.db.execute( select(IdentitySyncState, IdentitySyncTarget, IdentityPerson) .join(IdentitySyncTarget, IdentitySyncTarget.id == IdentitySyncState.target_id) .join(IdentityPerson, IdentityPerson.id == IdentitySyncState.person_id) .where( IdentitySyncState.last_status == "failed", IdentitySyncTarget.enabled.is_(True), IdentitySyncTarget.deleted_at.is_(None), ) )).all() return [ _Detected( "sync_drift_failed", f"{target.id}:{state.person_id}", person.id, person.display_name or person.work_email, f"{person.display_name or person.work_email} 同步到 {target.name} 持续失败", { "target_id": str(target.id), "target_name": target.name, "last_error": state.last_error, "retry_count": state.retry_count, }, ) for state, target, person in rows ] async def _detect_offboarding_session_not_cut(self) -> list[_Detected]: """Leavers whose identity-plane session cutoff FAILED or NEVER RAN. Two classes, both meaning "the leaver may still hold a live session": 1. **Failed** — the Google ``users.signOut`` recorded ``session_enforcement.google_session.status='failed'`` on the process (migration 383 proof written by the force-logout path). 2. **Never ran (review 2026-07-08 #1739)** — a COMPLETED offboarding whose ``session_enforcement`` is still NULL. This is the deferred-future-leaver hole: post-commit dispatch deferred the cutoff for a future ``leaver_date`` and nothing replayed it, so the process finalised with the session/account cutoff never attempted. The catch-up in advance finalisation now prevents new cases, but this branch still surfaces any historical / stuck process (e.g. one that completed before the fix, or whose T+0 tick never re-ran) so IT can trigger the manual re-enforce. Cancelled processes (mis-flagged leavers) are excluded — they should never have had a cutoff attempted. Rehired employees (``employment_status='active'`` with a stale COMPLETED row) are excluded from the NULL branch — a deliberately un-cut active employee is correct, not a finding.""" from app.domains.offboarding.models import ( # noqa: PLC0415 PROC_CANCELLED, PROC_COMPLETED, OffboardingProcess, ) out: list[_Detected] = [] seen_process_ids: set[str] = set() # Class 1: a recorded FAILED cutoff channel. rows = (await self.db.execute( select(OffboardingProcess, IdentityPerson) .join(IdentityPerson, IdentityPerson.id == OffboardingProcess.person_id) .where( OffboardingProcess.session_enforcement.is_not(None), OffboardingProcess.status != PROC_CANCELLED, IdentityPerson.deleted_at.is_(None), ) )).all() for proc, person in rows: gs = (proc.session_enforcement or {}).get("google_session") or {} if gs.get("status") != "failed": continue subject = person.display_name or person.work_email seen_process_ids.add(str(proc.id)) out.append(_Detected( "offboarding_session_not_cut", str(proc.id), person.id, subject, f"{subject} 的 Google 会话未成功切断", { "process_id": str(proc.id), "work_email": person.work_email, "reason": "cutoff_failed", "error": gs.get("error"), "http_status": gs.get("http_status"), "at": gs.get("at"), }, )) # Class 2 (#1739): a COMPLETED process with a NULL ledger — the cutoff # was deferred and never replayed. ``employment_status='active'`` # rehires are excluded (their previous cutoff was legitimately never # run / undone by cancel). null_rows = (await self.db.execute( select(OffboardingProcess, IdentityPerson) .join(IdentityPerson, IdentityPerson.id == OffboardingProcess.person_id) .where( # A deferred cutoff leaves the column as SQL NULL (never # assigned by start_process/dispatch); the reopen-clear path # (#1749) assigns Python None which the JSON type persists as # a JSON ``null`` literal. Match BOTH so neither shape hides a # never-attempted cutoff. or_( OffboardingProcess.session_enforcement.is_(None), func.json_type(OffboardingProcess.session_enforcement) == "NULL", ), OffboardingProcess.status == PROC_COMPLETED, IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status != "active", ) )).all() for proc, person in null_rows: # Dedup key is the process id; a process is either failed-ledger # (class 1) or NULL-ledger (class 2), never both, so this guard is # belt-and-braces against a race where a ledger appears mid-scan. if str(proc.id) in seen_process_ids: continue subject = person.display_name or person.work_email out.append(_Detected( "offboarding_session_not_cut", str(proc.id), person.id, subject, f"{subject} 的 Google 会话切断从未执行（离职流程已完成）", { "process_id": str(proc.id), "work_email": person.work_email, "reason": "cutoff_never_attempted", "leaver_date": ( proc.leaver_date.isoformat() if proc.leaver_date else None ), }, )) return out async def _detect_hris_status_divergence( self, ) -> tuple[list[_Detected], bool, set[tuple[str, str]]]: """Helix ``employment_status`` disagrees with SF ground truth: Helix says ``active`` but the live SF roster reports the user ``status='f'`` (forfeited). ``leaver_pending`` is deliberately NOT flagged — offboarding is already in flight there and its SLA escalation owns stuck cases. The P0051 class — a real leaver still counted active in Helix. Pulls the full SF roster (``list_all_users`` → both ``t`` + ``f``) per enabled HRIS instance. Returns ``(findings, complete, protected_keys)``. ``complete=False`` when any instance's SF pull failed — the caller then MUST NOT auto-resolve existing ``hris_status_divergence`` findings this run, or an SF outage would falsely clear every divergence (a silent "all-clear"). ``protected_keys`` are the ``(finding_type, dedup_key)`` of active persons pinned to an instance we did NOT pull this run (it's DISABLED — ``list_hris_instances`` returns enabled only — so we have no ground truth for them). ``complete`` stays True so the rest of the fleet still auto-resolves, but these specific findings must be preserved: a real un-offboarded leaver whose instance was later disabled (e.g. an entity wind-down) must not be silently cleared.""" # SF-active statuses — shared with the lifecycle classifier so # "non-active" means the same thing here as it does on the sync # paths (``t`` + ``active_external_suite`` are employed; anything # else — ``f``, blank, or a status SF invents tomorrow — is not). from app.domains.hr.service import ( # noqa: PLC0415 normalize_reason_codes, sf_status_is_active, ) from app.identity.successfactors_auth import ( # noqa: PLC0415 build_auth_from_config, ) from app.identity.successfactors_client import ( # noqa: PLC0415 SuccessFactorsClient, ) from app.integrations.compat import list_hris_instances # noqa: PLC0415 complete = True out: list[_Detected] = [] protected: set[tuple[str, str]] = set() try: instances = await list_hris_instances(session=self.db) except Exception: # noqa: BLE001 logger.warning( "hris_status_divergence: list_hris_instances failed — " "skipping divergence detection this run", exc_info=True, ) return [], False, protected # Phase 1 — pull EVERY instance's roster first. Review round-1 fix # (2026-07-04): the previous per-instance loop evaluated persons # with ``hris_instance_id=NULL`` once PER instance — duplicate # findings with flip-flopping instance attribution, and worse, a # cross-tenant employee reported 'f' by one instance but 't' by # the other (mobility) was wrongly flagged as a leaver. A single # Phase-2 pass fixes the duplication; for a person's own tenant we # flag only when that tenant reports 'f' and not 't'. # SF ``userId`` is NOT globally unique across the CN (api30) and EU # (api55) tenants — two DIFFERENT people can share one userId # (documented hazard, PLATFORM.md; guarded everywhere else via # ``hris_instance_id``). Keying a single global map on bare userId # (round-1 shape) lets an unrelated EU 't' suppress a real CN leaver's # 'f' (false negative — the exact P0051 blind spot this detector # exists to close), or vice-versa (false positive → a live employee # paged as gone). So aggregate PER (instance, userId) and, in Phase 2, # consult only each person's OWN tenant. A merged fallback is kept for # legacy ``hris_instance_id IS NULL`` rows, where we genuinely don't # know the authoritative tenant and "any live 't' anywhere = still # employed" is the conservative (false-positive-avoiding) choice. status_by_instance_user: dict[Any, dict[str, set[str]]] = {} status_by_user_any: dict[str, set[str]] = {} instance_slug_by_id: dict[Any, str] = {} instance_by_user_any: dict[str, str] = {} pulled_instance_ids: set[Any] = set() # Kept for the Phase-3 termination-reason enrichment (client is # already authenticated; the flagged set is tiny). client_by_instance_id: dict[Any, Any] = {} excluded_by_instance: dict[Any, frozenset[str]] = {} for row, cfg in instances: base_url = (cfg.get("base_url") or "").strip() if not base_url: # An ENABLED instance we cannot verify (blank base_url — # config edit / secret-rotation glitch). Detection is # incomplete for its persons; suppress auto-resolve so a # real leaver pinned to it is never falsely cleared — same # semantics as the SF-pull-failure branch below. logger.warning( "hris_status_divergence: instance %s enabled but has no " "base_url — treating divergence detection as incomplete", getattr(row, "slug", row.id), ) complete = False continue try: client = SuccessFactorsClient( base_url=base_url, auth=build_auth_from_config(cfg), ) sf_users = await client.list_all_users() except Exception: # noqa: BLE001 logger.warning( "hris_status_divergence: SF roster pull failed for " "instance %s — not resolving existing divergence " "findings this run", getattr(row, "slug", row.id), exc_info=True, ) complete = False continue pulled_instance_ids.add(row.id) slug = getattr(row, "slug", None) or str(row.id) instance_slug_by_id[row.id] = slug client_by_instance_id[row.id] = client excluded_by_instance[row.id] = normalize_reason_codes( (cfg or {}).get("excluded_termination_reason_codes") ) per_user = status_by_instance_user.setdefault(row.id, {}) for u in sf_users: uid = u.get("userId") if not uid: continue st = str(u.get("status") or "").strip().lower() per_user.setdefault(uid, set()).add(st) status_by_user_any.setdefault(uid, set()).add(st) # Attribution for the NULL-instance fallback path only. # Any non-active status can now produce a finding (see the # Phase-2 widening below), so attribute those too. if not sf_status_is_active(st) and uid not in instance_by_user_any: instance_by_user_any[uid] = slug # Any pull failure ⇒ a person could look forfeited only because the # instance that knows them as active didn't answer. Detect nothing # this run (complete=False already suppresses auto-resolve). if not complete: return [], False, protected # Phase 2 — evaluate each Helix-active person exactly once. # Logic round-1 (2026-07-04): ``active`` ONLY. ``leaver_pending`` # means offboarding is already in flight — flagging them here is # semantic noise (the finding would sit open for days until the # advance cron flips them to ``left``, inviting duplicate tickets). # A STUCK leaver_pending is the offboarding domain's problem and is # already escalated by its SLA tiers (warn/urgent/breach); this # detector owns exactly one question: "does SF say gone while Helix # still says employed-with-access?" persons = (await self.db.execute( select(IdentityPerson).where( IdentityPerson.source_system == "successfactors", IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", ) )).scalars().all() # Same Asia/Shanghai ``>=`` convention as # ``classify_inactive_sf_user`` (see its docstring): a pre-start / # day-one hire is inactive in SF by design (status='f' until day 1 # — the P10013 shape the incremental sync now onboards as active), # so flagging them as "SF 已离职" would page admins daily until # their start date for a divergence that is entirely expected. today_sh = datetime.now(ZoneInfo("Asia/Shanghai")).date() # instance_id → [(finding, hris_person_id)] for Phase-3 reason # enrichment below. reason_targets: dict[Any, list[tuple[_Detected, str]]] = {} for p in persons: if p.hire_date is not None and p.hire_date >= today_sh: # Not yet started — expected to be inactive in SF; never a # divergence. (Any stale pre-fix finding auto-resolves.) continue # A person pinned to an instance we didn't pull (disabled / # no base_url) has no ground truth this run — skip, never guess. # Protect any existing finding for them from auto-resolve: a # DISABLED instance (absent from list_hris_instances) otherwise # leaves complete=True, silently clearing a real leaver's finding. if ( p.hris_instance_id is not None and p.hris_instance_id not in pulled_instance_ids ): protected.add(("hris_status_divergence", str(p.id))) continue if p.hris_instance_id is not None: # Scope to the person's OWN tenant — a different person's # identical userId in another tenant must never vote. statuses = ( status_by_instance_user.get(p.hris_instance_id, {}) .get(p.hris_person_id) or set() ) attributed = instance_slug_by_id.get(p.hris_instance_id) else: # Legacy unpinned row — no authoritative tenant; fall back to # the merged view so any live 't' anywhere suppresses the flag. statuses = status_by_user_any.get(p.hris_person_id) or set() attributed = instance_by_user_any.get(p.hris_person_id) # PR-audit widening (2026-07-16): flag ANY non-active SF status, # not just ``'f'`` — the lifecycle classifier deliberately skips # unknown/inactive statuses without a substantiated termination # (NO_TERMINATION, never fabricating a leaver_date), so this # detector is the designated surface where those shapes become # visible; matching only ``'f'`` left every other non-active # status invisible everywhere. An empty ``statuses`` set means # the user is ABSENT from the roster entirely — that is # ``hris.absent_person_check``'s jurisdiction, not a divergence. inactive_statuses = sorted( s for s in statuses if not sf_status_is_active(s) ) if not inactive_statuses or any( sf_status_is_active(s) for s in statuses ): continue hris_status = ",".join(s or "(blank)" for s in inactive_statuses) subject = p.display_name or p.work_email finding = _Detected( "hris_status_divergence", str(p.id), p.id, subject, ( f"{subject} 在 Helix 为 {p.employment_status},SF 已离职" if "f" in inactive_statuses else f"{subject} 在 Helix 为 {p.employment_status}," f"SF 状态非在职({hris_status})" ), { "helix_status": p.employment_status, "hris_status": hris_status, "hris_instance": attributed, "employee_no": p.employee_no, }, ) out.append(finding) if ( "f" in inactive_statuses and p.hris_instance_id in client_by_instance_id ): reason_targets.setdefault(p.hris_instance_id, []).append( (finding, p.hris_person_id) ) # Phase 3 — best-effort termination-reason enrichment (2026-07-30). # A divergence caused by an EXCLUDED reason (人员分类转换 / 跨DC调动 / # 兼岗结束 — configured "do not close accounts") is EXPECTED: the # person keeps working while their SF slice stays 'f' forever. # Without the reason on the finding, the operator re-derives it # from SF by hand every time. Flagged set is tiny (usually 0–3); # any lookup failure just leaves the finding un-annotated. for instance_id, targets in reason_targets.items(): try: reasons = await client_by_instance_id[ instance_id ].get_termination_reasons_for_users( [uid for _, uid in targets] ) except Exception: # noqa: BLE001 logger.warning( "hris_status_divergence: termination-reason enrichment " "failed for instance %s — findings stay un-annotated", instance_slug_by_id.get(instance_id, instance_id), exc_info=True, ) continue excluded = excluded_by_instance.get(instance_id) or frozenset() for finding, uid in targets: code = (reasons.get(uid) or "").strip().upper() if not code: continue finding.detail["termination_reason"] = code if code in excluded: finding.detail["termination_reason_excluded"] = True finding.title += ( f"(SF 离职原因 {code},已配置为不关号——预期差异," "人仍在职)" ) else: finding.title += f"(SF 离职原因 {code})" return out, complete, protected # ── Scan (mutating) ─────────────────────────────────────────────── async def run_scan(self, *, trigger_source: str = "cron") -> dict[str, Any]: started = datetime.now(UTC) detected, incomplete_types, protected_keys = await self._detect_all() detected_by_key: dict[tuple[str, str], _Detected] = { (d.finding_type, d.dedup_key): d for d in detected } # Load ALL findings — resolved ones included. The (finding_type, # dedup_key) UNIQUE spans every status, so a problem that resolved # and later recurred must REOPEN its row, never insert a sibling # (prod 2026-06-11: a recurred missing_manager hit 1062 on the # dedup key and aborted the whole scan). Bounded: one row per # (check, subject), mirroring auditor_findings' sweep. existing = (await self.db.execute( select(IdentityDataQualityFinding) )).scalars().all() existing_by_key = { (f.finding_type, f.dedup_key): f for f in existing } new_high: list[IdentityDataQualityFinding] = [] new_count = 0 reopened_count = 0 for key, d in detected_by_key.items(): sev = FINDING_SEVERITY.get(d.finding_type, "medium") row = existing_by_key.get(key) if row is None: row = IdentityDataQualityFinding( finding_type=d.finding_type, dedup_key=d.dedup_key, severity=sev, person_id=d.person_id, subject_display=d.subject_display, title=d.title, detail=d.detail, status="open", first_seen_at=started, last_seen_at=started, ) self.db.add(row) new_count += 1 if sev == "high": new_high.append(row) continue # Re-seen — refresh the snapshot. row.last_seen_at = started row.severity = sev row.title = d.title row.detail = d.detail row.person_id = d.person_id row.subject_display = d.subject_display if row.status == "resolved": # It came back — it was never actually fixed. Reopen and # clear the stale decision (same semantics as auditor # findings' _reopen). row.status = "open" row.auto_resolved = False row.resolved_at = None row.decided_by_person_id = None row.decided_at = None row.decision_note = None reopened_count += 1 if sev == "high": new_high.append(row) # open / acknowledged / accepted_risk → keep the human decision. # Auto-resolve live findings that disappeared from this scan — # EXCEPT finding-types whose detection was incomplete this run (e.g. # SF was unreachable for the divergence pull). Resolving those would # be a false "all-clear" on an upstream outage. resolved_count = 0 for key, row in existing_by_key.items(): if key[0] in incomplete_types: continue if key in protected_keys: # Detection was complete fleet-wide, but this specific person # is pinned to an instance we couldn't verify this run (its # HRIS integration is disabled) — preserve, never false-clear. continue if key not in detected_by_key and row.status != "resolved": row.status = "resolved" row.auto_resolved = True row.resolved_at = started resolved_count += 1 await self.db.flush() # Open totals (post-scan) for the run-history trend. by_type, open_total, open_high = await self._open_counts() finished = datetime.now(UTC) run = IdentityDataQualityRun( started_at=started, finished_at=finished, duration_ms=int((finished - started).total_seconds() * 1000), status="ok", trigger_source=trigger_source, open_total=open_total, open_high=open_high, # Reopened findings are newly actionable — the run trend has a # single "new" axis, so they count there. new_count=new_count + reopened_count, resolved_count=resolved_count, by_type=by_type, ) self.db.add(run) await self.db.commit() if new_high: await self._notify_new_high(new_high) return { "new_count": new_count, "reopened_count": reopened_count, "resolved_count": resolved_count, "open_total": open_total, "open_high": open_high, "by_type": by_type, } async def _open_counts(self) -> tuple[dict[str, int], int, int]: rows = (await self.db.execute( select( IdentityDataQualityFinding.finding_type, IdentityDataQualityFinding.severity, func.count().label("n"), ) .where(IdentityDataQualityFinding.status.in_(_LIVE_STATUSES)) .group_by( IdentityDataQualityFinding.finding_type, IdentityDataQualityFinding.severity, ) )).all() by_type: dict[str, int] = {ft: 0 for ft in FINDING_TYPES} total = 0 high = 0 for ft, sev, n in rows: by_type[ft] = by_type.get(ft, 0) + int(n) total += int(n) if sev == "high": high += int(n) return by_type, total, high async def _notify_new_high( self, findings: list[IdentityDataQualityFinding], ) -> None: """Dedup-aware in-app push to IT admins on NEW high-severity findings. Best-effort — never raises into the scan.""" async with best_effort( self.db, label="identity dq: new-high notification", log=logger, ) as dq_notify_step: from app.notifications.models import NotificationOutbox from app.notifications.service import ( NotificationEnvelope, NotificationService, ) from app.scheduled_jobs.failure_notifier import list_admin_recipients admins = await list_admin_recipients(self.db) if not admins: return day_start = datetime.now(UTC).replace( hour=0, minute=0, second=0, microsecond=0, ) notif = NotificationService(self.db) count = len(findings) sample = [f.subject_display or "?" for f in findings[:5]] for admin in admins: already = (await self.db.execute( select(NotificationOutbox.id).where( and_( NotificationOutbox.recipient_person_id == admin.id, NotificationOutbox.template_key == _NOTIFY_TEMPLATE, NotificationOutbox.created_at >= day_start, ) ).limit(1) )).scalar_one_or_none() if already is not None: continue body = ( f"身份数据质量扫描发现 {count} 个新的高危问题" f"(离职仍可登录 / HRIS 已清除仍可登录)。\n\n" + "\n".join(f" · {s}" for s in sample) + "\n\n请到「身份数据质量」面板处理。" ) try: await notif.dispatch( NotificationEnvelope( recipient=admin, title=f"[身份] {count} 个新的高危身份数据质量问题", body_text=body, template_key=_NOTIFY_TEMPLATE, url="/admin/identity/data-quality", related_resource_type="identity_data_quality", related_resource_id=None, context={ "count": count, "samples": sample, # ``samples`` is a list — interpolating it # renders a Python repr, so the template reads # this pre-joined bullet block instead. "sample_lines": "\n".join( f" · {s}" for s in sample ), }, ), ) except Exception as exc: # noqa: BLE001 logger.warning("identity dq notify failed for %s: %s", admin.id, exc) await self.db.commit() if not dq_notify_step.ok: logger.error("identity data-quality new-high notification failed") # ── Read (admin panel) ──────────────────────────────────────────── async def list_findings( self, *, status: str | None = None, finding_type: str | None = None, severity: str | None = None, page: int = 1, page_size: int = 50, ) -> dict[str, Any]: conds = [] if status: conds.append(IdentityDataQualityFinding.status == status) else: conds.append(IdentityDataQualityFinding.status.in_(_LIVE_STATUSES)) if finding_type: conds.append(IdentityDataQualityFinding.finding_type == finding_type) if severity: conds.append(IdentityDataQualityFinding.severity == severity) total = (await self.db.execute( select(func.count()).select_from(IdentityDataQualityFinding).where(*conds) )).scalar_one() # high → medium → low ordering, newest-seen first within a tier. sev_rank = func.field( IdentityDataQualityFinding.severity, "high", "medium", "low", ) rows = (await self.db.execute( select(IdentityDataQualityFinding) .where(*conds) .order_by(sev_rank, IdentityDataQualityFinding.last_seen_at.desc()) .limit(page_size) .offset((page - 1) * page_size) )).scalars().all() return { "items": [self._serialize(f) for f in rows], "total": total, "page": page, "page_size": page_size, } async def summary(self) -> dict[str, Any]: by_type, open_total, open_high = await self._open_counts() last_run = (await self.db.execute( select(IdentityDataQualityRun) .order_by(IdentityDataQualityRun.started_at.desc()) .limit(1) )).scalars().first() trend = (await self.db.execute( select(IdentityDataQualityRun) .order_by(IdentityDataQualityRun.started_at.desc()) .limit(30) )).scalars().all() return { "open_total": open_total, "open_high": open_high, "by_type": by_type, "last_run_at": last_run.started_at.isoformat() if last_run else None, "trend": [ { "started_at": r.started_at.isoformat(), "open_total": r.open_total, "open_high": r.open_high, "new_count": r.new_count, "resolved_count": r.resolved_count, } for r in reversed(trend) ], } def _serialize(self, f: IdentityDataQualityFinding) -> dict[str, Any]: return { "id": str(f.id), "finding_type": f.finding_type, "severity": f.severity, "status": f.status, "person_id": str(f.person_id) if f.person_id else None, "subject_display": f.subject_display, "title": f.title, "detail": f.detail, "auto_resolved": f.auto_resolved, "first_seen_at": f.first_seen_at.isoformat() if f.first_seen_at else None, "last_seen_at": f.last_seen_at.isoformat() if f.last_seen_at else None, "resolved_at": f.resolved_at.isoformat() if f.resolved_at else None, "decision_note": f.decision_note, "decided_at": f.decided_at.isoformat() if f.decided_at else None, } # ── Decision (mutating) ─────────────────────────────────────────── async def decide( self, *, finding_id: uuid.UUID, status: str, note: str | None, actor_user_id: uuid.UUID, ) -> dict[str, Any]: if status not in _DECISION_STATUSES: raise ValidationError( f"status must be one of {_DECISION_STATUSES}, got {status!r}", ) f = await self.db.get(IdentityDataQualityFinding, finding_id) if f is None: raise NotFoundError("IdentityDataQualityFinding", str(finding_id)) f.status = status f.decision_note = note f.decided_by_person_id = actor_user_id f.decided_at = datetime.now(UTC) if status == "resolved": f.auto_resolved = False f.resolved_at = datetime.now(UTC) await self.db.commit() await self.db.refresh(f) return self._serialize(f)