"""Run-failure alert dispatch for the Identity Sync Center. Why this exists --------------- The 数据同步 (outbound identity sync) center had exactly one failure- notification path: the cron dispatcher (``identity_sync.dispatch``) re-raises when any *due* target fails a tick, and the scheduled-jobs framework's :func:`maybe_notify_failure` then pages IT admins. But manual and **event-triggered** runs never go through that dispatcher — an ``event:hire_confirm`` Google Workspace dept push that 403s calls :meth:`IdentitySyncService.run_target` directly, finalises the run row as ``failed``, and … nothing. No admin signal. That was the reported gap (a 403 on a hire-confirm push with zero alert). This module is the trigger-source-agnostic fix. It is invoked from :meth:`IdentitySyncService._finalise_run` (the single chokepoint every run passes through, regardless of cron / manual / event origin) as a fire-and-forget task whenever a run lands ``failed`` / ``partial``. It dispatches to the target's configured ``alert_recipient_ids`` and only when the target's ``alert_enabled`` switch is on. There is deliberately NO IT-admin fallback: a target alerts exactly the people the admin picked, or — switch off / empty list — nobody. The whole-org admin broadcast was removed on 2026-06-19 ("不要默认通知管理员", see DECISIONS.md). The cron dispatcher keeps its own scheduled-job failure backstop that still pages admins on a cron-tick failure; that path is independent of this module. Dedup ----- A target failing every 60s cron tick must not spam the recipients. We reuse :func:`compute_failure_signature` (the scheduled-job notifier's pure signature function) keyed on the target slug + first error line. We notify only when the *immediately preceding* run for this target was NOT a same-signature failure — i.e. on the first failure, on a recovery- then-refailure, and when the *kind* of failure changes. Identical back- to-back failures are suppressed. No extra columns: the runs table is the dedup state (the just-finalised run is row[0], its predecessor row[1]). Best-effort ----------- Never raises. The run outcome is the source of truth; a notification glitch must not obscure it or block the run / HTTP path. """ from __future__ import annotations import asyncio import hashlib import logging from uuid import UUID from sqlalchemy import desc, select from app.domains.identity_sync.models import IdentitySyncRun, IdentitySyncTarget from app.identity.person_models import IdentityOrgNode, IdentityPerson from app.notifications.models import NotificationOutbox from app.notifications.service import ( LocalizedText, NotificationEnvelope, NotificationService, ) from app.scheduled_jobs.failure_notifier import compute_failure_signature logger = logging.getLogger(__name__) # Per-recipient dispatch ceiling — bounds how long a wedged Feishu can # tie up a single recipient's send. Mirrors the scheduled-job notifier. _DISPATCH_TIMEOUT_S = 10.0 # Statuses that warrant an alert. ``partial`` carries per-person or # dept-phase failures the admin should still see. # # Must stay identical to ``service._RUN_STATUSES_WITH_FAILURES`` (asserted # in tests): the per-target alert here and the dispatcher's admin page must # not disagree about what counts as a failure. Imported lazily rather than # aliased at module scope because ``service`` imports this module. _ALERTING_STATUSES = ("failed", "partial") # Dedup look-back window. Sized to comfortably bracket one nightly sweep # cycle (24h + margin) so a standing same-signature failure is deduped # against the previous cycle regardless of how many interleaved event runs # a high-cadence target fires — while a failure whose last occurrence rolled # out of this window re-alerts as a fresh regression. See _should_notify. _DEDUP_WINDOW_HOURS = 26 async def _resolve_recipients( session, target: IdentitySyncTarget, ) -> list: """Resolve the target's alert audience — the explicit list only. The target's ``alert_recipient_ids`` filtered to active people with a usable work_email (a NULL/blank work_email recipient otherwise spawns a retry storm on the Feishu / email channels). When the list is empty — or none of the ids resolve to an eligible person — the result is an empty list and the caller sends nothing. There is NO IT-admin fallback by design (see module docstring / DECISIONS.md 2026-06-19). """ explicit_ids = target.alert_recipient_ids or [] parsed: list[UUID] = [] for raw in explicit_ids: try: parsed.append(UUID(str(raw))) except (ValueError, TypeError): continue if not parsed: return [] rows = ( await session.execute( select(IdentityPerson).where( IdentityPerson.id.in_(parsed), IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", IdentityPerson.work_email.isnot(None), IdentityPerson.work_email != "", ) ) ).scalars().all() return list(rows) async def _should_notify( session, *, target_id: UUID, current_run_id: UUID, signature: str, ) -> bool: """Decide whether this failure deserves a fresh push. True on: the first failure, a changed failure kind, or a recovery→refailure. Suppresses an identical *standing* failure so a target failing every tick doesn't spam the recipients. Dedup is over a TIME window, not a fixed run count. A fixed ``LIMIT N`` conflates cadences and breaks both ways (why this was rewritten, 2026-07-17 review): * high-cadence targets (feishu_contact fires ≥24 hourly single-person ``event`` pushes/day, plus onboarding/offboarding bursts) can interleave *more than N* clean ``ok`` runs between two nightly ``partial`` sweeps — so a fixed ``LIMIT 25`` never reaches the prior alerting sweep, reports "no prior alert", and re-pages the SAME standing partial every night (the exact storm dedup exists to kill); * low-cadence targets (google daily cron, no event traffic) put ~N days inside ``LIMIT N`` — so a failure that genuinely recovered and recurred weeks later still matches the stale same-signature run and is wrongly suppressed. A ``_DEDUP_WINDOW_HOURS`` (≈ one nightly cycle) time window is cadence-independent: it always brackets ~1 sweep cycle regardless of how many event runs interleave. A real recovery pushes the last same-signature failure out of the window by the next cycle, so recovery→refailure re-fires. As a belt-and-braces for sub-window cron cadences, an intervening successful FULL sweep (a ``ok`` run that is NOT a per-person ``event`` push — those ``ok`` event runs are noise, they do not clear a standing nightly ``partial``) is also treated as a recovery that re-arms the alert. ``cancelled`` / ``running`` are neither alerting nor recovery and are skipped. """ from datetime import UTC, datetime, timedelta window_start = datetime.now(UTC) - timedelta(hours=_DEDUP_WINDOW_HOURS) recent = ( await session.execute( select(IdentitySyncRun) .where(IdentitySyncRun.target_id == target_id) .where(IdentitySyncRun.id != current_run_id) # Only FINISHED runs; an in-flight ``running`` row (finished_at # NULL) on a fast-cadence target is not a predecessor. .where(IdentitySyncRun.finished_at.is_not(None)) # Time-bounded, not count-bounded — see docstring. .where(IdentitySyncRun.started_at >= window_start) .order_by( desc(IdentitySyncRun.started_at), desc(IdentitySyncRun.id), ) # Safety cap only (the time window already bounds the set on any # sane cadence); keeps this a cheap indexed scan on a runaway. .limit(200) ) ).scalars().all() # Walk newest→oldest to the most recent alerting run, noting whether a # genuine recovery (successful FULL sweep) happened since — everything # visited before we hit that alerting run is strictly more recent than it. last_alerting = None saw_full_recovery = False for r in recent: if r.status in _ALERTING_STATUSES: last_alerting = r break if r.status == "ok" and not (r.trigger_source or "").startswith("event"): saw_full_recovery = True if last_alerting is None: # No alerting run in the window — first failure, or the standing # failure recovered long enough ago that this is a fresh regression. return True if saw_full_recovery: # Recovered (a full sweep went green) then re-failed → new incident. return True prior_signature = compute_failure_signature( f"identity_sync:{target_id}", last_alerting.error_message, ) return prior_signature != signature async def _dispatch_one(*, recipient, envelope_kwargs: dict, trace_id: str) -> None: """Per-recipient dispatch in its OWN session (blast-radius bound). Mirrors the scheduled-job notifier: a ``wait_for`` timeout poisons only this recipient's session, the outbox row persists for the retry worker, and the cron / HTTP path is never blocked. """ from app.core.database import async_session_factory # savepoint-exempt: dedicated per-recipient session (below) try: async with async_session_factory() as r_session: r_svc = NotificationService(r_session) # savepoint-exempt: dedicated per-recipient session try: await asyncio.wait_for( r_svc.dispatch( NotificationEnvelope(recipient=recipient, **envelope_kwargs), trace_id=trace_id, actor_user_id=None, ), timeout=_DISPATCH_TIMEOUT_S, ) await r_session.commit() # savepoint-exempt: dedicated per-recipient session except TimeoutError: logger.warning( "identity_sync alert to %s timed out after %.1fs — " "outbox row persists; retry worker re-delivers", getattr(recipient, "work_email", "?"), _DISPATCH_TIMEOUT_S, ) # savepoint-exempt: dedicated per-recipient session try: await r_session.commit() except Exception: # noqa: BLE001 — defence pass except Exception as exc: # noqa: BLE001 — best-effort logger.warning( "identity_sync alert to %s failed: %s", getattr(recipient, "work_email", "?"), exc, ) # savepoint-exempt: dedicated per-recipient session try: await r_session.commit() except Exception: # noqa: BLE001 — defence pass except Exception: # noqa: BLE001 — per-recipient must never raise logger.exception( "identity_sync alert per-recipient session crashed for %s", getattr(recipient, "work_email", "?"), ) async def maybe_notify_run_failure(*, run_id: UUID) -> None: """Top-level entry point — called from ``_finalise_run`` post-commit. Best-effort, never raises. Opens its own session, loads the run + target, applies the dedup gate, resolves the audience, and fans out one in-app + Feishu + email card per recipient (each in its own session, concurrently). """ try: from app.core.database import async_session_factory from app.domains.identity_sync.service import _safe_error from app.notifications.service import ( _SEE_TEMPLATE_SENTINEL, CardSection, CardStat, ) async with async_session_factory() as session: run = await session.get(IdentitySyncRun, run_id) if run is None or run.status not in _ALERTING_STATUSES: return target = await session.get(IdentitySyncTarget, run.target_id) if target is None or target.deleted_at is not None: return if not target.alert_enabled: logger.info( "identity_sync alert: target %s failed but failure " "alerting is switched off — skipping", target.slug, ) return signature = compute_failure_signature( f"identity_sync:{target.id}", run.error_message, ) if not await _should_notify( session, target_id=target.id, current_run_id=run.id, signature=signature, ): logger.info( "identity_sync alert dedup hit for target %s " "(signature=%s) — skipping", target.slug, signature, ) return recipients = await _resolve_recipients(session, target) if not recipients: logger.info( "identity_sync alert: target %s failed but no eligible " "recipients configured (alerting on, recipient list empty " "or unresolved) — skipping (no admin fallback by design)", target.slug, ) return # Scrub credential-shaped fragments before the error lands in # the most widely-distributed surface (Feishu card body). raw_err = run.error_message or "" error_brief = (_safe_error(raw_err) or "").strip().split("\n", 1)[0][:240] failed_count = int(run.failed_count or 0) in_scope = int(run.persons_in_scope or 0) # Locale-fragment triple. The value used to be Chinese-only, # which forced the English body to drop the distinction between # "failed" and "partially failed" entirely — an EN reader could # not tell a total outage from a partial one. partial = run.status != "failed" status_label_zh = "部分失败" if partial else "失败" status_label_en = "partially failed" if partial else "failed" status_label_ja = "一部失敗" if partial else "失敗" trigger = run.trigger_source or "—" started_label = ( run.started_at.strftime("%Y-%m-%d %H:%M") if run.started_at else "—" ) title_zh = f"⚠️ 数据同步失败 — {target.name}" card_subtitle = LocalizedText( zh=( "该同步目标的一次运行失败,请点击下方按钮查看运行历史并处理。" ), en=( "A run of this sync target failed. Use the button below to " "open the run history and handle it." ), ja=( "この同期ターゲットの実行が失敗しました。下のボタンから" "実行履歴を開いてご対応ください。" ), ) card_stats = [ CardStat( label=LocalizedText( zh="同步目标", en="Sync target", ja="同期ターゲット", ), value=target.name, hint=f"`{target.slug}`", ), CardStat( label=LocalizedText(zh="状态", en="Status", ja="ステータス"), value=LocalizedText( zh=status_label_zh, en=status_label_en, ja=status_label_ja, ), hint=LocalizedText( zh=f"类型 {target.kind}", en=f"Kind {target.kind}", ja=f"種別 {target.kind}", ), ), CardStat( label=LocalizedText( zh="失败 / 范围", en="Failed / in scope", ja="失敗 / 対象", ), value=f"{failed_count} / {in_scope}", ), CardStat( label=LocalizedText( zh="触发源", en="Trigger source", ja="トリガー元", ), value=trigger, hint=started_label, ), ] card_sections = [ CardSection( title=LocalizedText( zh="错误摘要", en="Error summary", ja="エラー概要", ), body=( error_brief or LocalizedText( zh="(无错误信息)", en="(no error message)", ja="(エラーメッセージなし)", ) ), tone="danger", ), ] url_path = ( f"/admin/identity-sync?tab=runs&target={target.id}" ) envelope_kwargs = { "title": title_zh, "body_text": _SEE_TEMPLATE_SENTINEL, "channels": ["feishu", "email"], "template_key": "identity_sync.run_failure", "context": { "target_name": target.name, "target_slug": target.slug, "kind": target.kind, "status_label_zh": status_label_zh, "status_label_en": status_label_en, "status_label_ja": status_label_ja, "error_brief": error_brief or "(无错误信息)", "failed_count": failed_count, "in_scope": in_scope, "trigger_source": trigger, }, "url": url_path, "card_subtitle": card_subtitle, "card_stats": card_stats, "card_sections": card_sections, "related_resource_type": "identity_sync_run", "related_resource_id": UUID(str(run.id)), } trace_id = f"identity_sync.run_failure:{target.slug}:{signature}" frozen = list(recipients) await asyncio.gather( *[ _dispatch_one( recipient=r, envelope_kwargs=envelope_kwargs, trace_id=trace_id, ) for r in frozen ], return_exceptions=True, ) logger.info( "identity_sync alert dispatched for target run %s to %d " "recipient(s) (signature=%s)", run_id, len(frozen), signature, ) except Exception: # noqa: BLE001 logger.exception( "identity_sync alert path crashed for run %s — suppressing " "so the run outcome is unaffected", run_id, ) # ── Unmapped-department audit (non-blocking heads-up) ──────────────── # Template key doubles as the outbox dedup label: the "last alert for # this target" lookup filters on it. _UNMAPPED_TEMPLATE_KEY = "identity_sync.unmapped_depts" def _unmapped_signature(codes: list[str]) -> str: """Stable short signature of the unmapped-code SET (order-independent).""" joined = ",".join(sorted(codes)) return hashlib.sha256(joined.encode("utf-8")).hexdigest()[:16] async def _last_alert_signature( session, *, target_id: UUID, template_key: str, sig_key: str, ) -> str | None: """Set-signature carried by the most recent alert of ``template_key`` for this target (``None`` if never alerted). The outbox IS the dedup state — no extra columns, mirroring the run-failure notifier's use of the runs table. Shared by the huilianyi (unmapped) and netsuite (unadopted) dept audits.""" row = ( await session.execute( select(NotificationOutbox) .where(NotificationOutbox.template_key == template_key) .where( NotificationOutbox.related_resource_type == "identity_sync_target", ) # CHAR(36) column — compare as string (the codebase convention # for related_resource_id; a bare UUID object won't match). .where(NotificationOutbox.related_resource_id == str(target_id)) .order_by(desc(NotificationOutbox.created_at)) .limit(1) ) ).scalars().first() if row is None or not row.context: return None sig = row.context.get(sig_key) return str(sig) if sig else None async def maybe_notify_unmapped_depts(*, target_id: UUID) -> None: """Alert IT (NON-blocking) when a 汇联易 recode target has active org nodes with NO ``huilianyi_dept_code_map`` entry. Such a node's dept code passes through UNTRANSLATED (raw SF code) and can spawn a parallel vendor department that doesn't match the NS/Authing-era code — the recurring "split" bug (seen 3× in prod: Business Operations, IT Services & Security, AI Lab; each discovered weeks later by finance when cost attribution diverged). Design intent (per user 2026-07-07): NOT a failure. The sync stays green — this is an actionable heads-up so a human maps the new dept (or maps it to itself to confirm it is genuinely new) *the day it appears*, not weeks later. Deliberately no auto-mapping: matching by name across systems is unreliable (Helix "AI Lab" vs 汇联易 "AI") and a wrong guess is silent corruption — the human decision is the point. Dedup: fires only when the unmapped SET changes (signature stored in the prior alert's outbox ``context``). Reuses the run-failure notifier's ``alert_enabled`` + ``alert_recipient_ids`` audience (no IT-admin fallback, per DECISIONS 2026-06-19). Best-effort, never raises — the sync outcome is the source of truth. """ try: from app.core.database import async_session_factory from app.notifications.service import ( _SEE_TEMPLATE_SENTINEL, CardSection, CardStat, ) async with async_session_factory() as session: target = await session.get(IdentitySyncTarget, target_id) if target is None or target.deleted_at is not None: return if target.kind != "huilianyi" or not target.alert_enabled: return code_map = (target.runner_options or {}).get( "huilianyi_dept_code_map", ) or {} # Empty map = passthrough-by-design (not a recode/cutover # target) — "unmapped" is expected there, nothing to flag. if not code_map: return map_keys = {str(k) for k in code_map} nodes = ( await session.execute( select(IdentityOrgNode).where( IdentityOrgNode.deleted_at.is_(None), IdentityOrgNode.node_status == "active", ) ) ).scalars().all() unmapped = sorted( ( (str(n.org_node_source_id), n.org_node_name or "") for n in nodes if str(n.org_node_source_id) not in map_keys ), key=lambda x: x[0], ) if not unmapped: return signature = _unmapped_signature([c for c, _ in unmapped]) if await _last_alert_signature( session, target_id=target.id, template_key=_UNMAPPED_TEMPLATE_KEY, sig_key="unmapped_signature", ) == signature: logger.info( "identity_sync unmapped-dept audit: target %s set " "unchanged (signature=%s) — skipping", target.slug, signature, ) return recipients = await _resolve_recipients(session, target) if not recipients: logger.info( "identity_sync unmapped-dept audit: target %s has %d " "unmapped dept(s) but no eligible recipients " "(alert_recipient_ids empty/unresolved) — skipping " "(no admin fallback by design)", target.slug, len(unmapped), ) return count = len(unmapped) listed = unmapped[:20] body_lines = "\n".join(f"• {code} {name}" for code, name in listed) overflow = count - len(listed) title_zh = f"🗂️ 数据同步:{count} 个部门缺少映射 — {target.name}" card_subtitle = LocalizedText( zh=( "下列部门在 Helix 有组织节点,但汇联易映射表里没有对应的老" "部门编码。同步会用原始编码推送,可能在汇联易新建一个与 " "NS/老系统不一致的平行部门,导致成本归集错乱。请为每个部门补" "一行映射(Helix 编码 → 汇联易老编码);若确属全新部门,映射到" "其自身编码以显式放行。" ), en=( "The departments below have an org node in Helix but no " "matching legacy department code in the Huilianyi mapping " "table. The sync pushes the raw code, which may create a " "parallel department in Huilianyi that disagrees with " "NS / the legacy system and scrambles cost attribution. " "Please add one mapping row per department (Helix code → " "Huilianyi legacy code); if a department really is new, " "map it to its own code to allow it explicitly." ), ja=( "以下の部門は Helix に組織ノードがありますが、Huilianyi の" "マッピング表に対応する旧部門コードがありません。同期は" "元のコードのまま連携するため、NS / 旧システムと一致しない" "並行部門が Huilianyi 側に新規作成され、コストの集計が" "崩れるおそれがあります。部門ごとにマッピングを 1 行ずつ" "追加してください(Helix コード → Huilianyi の旧コード)。" "本当に新設部門である場合は、自身のコードへマッピングして" "明示的に許可してください。" ), ) card_stats = [ CardStat( label=LocalizedText( zh="同步目标", en="Sync target", ja="同期ターゲット", ), value=target.name, hint=f"`{target.slug}`", ), CardStat( label=LocalizedText( zh="待映射部门数", en="Departments to map", ja="マッピング待ちの部門数", ), value=str(count), ), ] card_sections = [ CardSection( title=LocalizedText( zh="待映射部门", en="Departments to map", ja="マッピング待ちの部門", ), body=( LocalizedText( zh=f"{body_lines}\n… 其余 {overflow} 个", en=f"{body_lines}\n… and {overflow} more", ja=f"{body_lines}\n… ほか {overflow} 件", ) if overflow > 0 else body_lines ), tone="warn", ), ] envelope_kwargs = { "title": title_zh, # Sentinel, not the Chinese subtitle: a non-sentinel # ``body_text`` short-circuits ``render_inbox_body``, so the # bell would keep showing this literal Chinese while Feishu / # email render the localised template. "body_text": _SEE_TEMPLATE_SENTINEL, "channels": ["feishu", "email"], "template_key": _UNMAPPED_TEMPLATE_KEY, "context": { "target_slug": target.slug, "target_name": target.name, "unmapped_count": count, "unmapped_codes": [c for c, _ in listed], "unmapped_signature": signature, }, "url": f"/admin/identity-sync?tab=targets&target={target.id}", "card_subtitle": card_subtitle, "card_stats": card_stats, "card_sections": card_sections, "related_resource_type": "identity_sync_target", "related_resource_id": UUID(str(target.id)), } trace_id = f"{_UNMAPPED_TEMPLATE_KEY}:{target.slug}:{signature}" frozen = list(recipients) await asyncio.gather( *[ _dispatch_one( recipient=r, envelope_kwargs=envelope_kwargs, trace_id=trace_id, ) for r in frozen ], return_exceptions=True, ) logger.info( "identity_sync unmapped-dept audit dispatched for target %s: " "%d dept(s) to %d recipient(s) (signature=%s)", target_id, count, len(frozen), signature, ) except Exception: # noqa: BLE001 logger.exception( "identity_sync unmapped-dept audit crashed for target %s — " "suppressed (sync outcome unaffected)", target_id, ) # ── STRICT value_map miss audit (non-blocking heads-up) ───────────── _VALUE_MAP_MISS_TEMPLATE_KEY = "identity_sync.value_map_misses" async def maybe_notify_value_map_misses(*, run_id: UUID) -> None: """Alert IT (NON-blocking) when a sweep hit STRICT ``value_map`` misses. A STRICT ``value_map`` rule (region / company_code / employee_type / …) blanks any source value it has no entry for — so a brand-new HRIS code (new legal entity, new employee-type, a re-coded office) silently drops the field for everyone carrying it, freezing that vendor attribute with a GREEN run. (This is the class behind the 1074/1075 employee-type incident and the recurring "new company_code" NetSuite/汇联易 failures.) This turns that silent drop into a ONE-TIME, deduplicated heads-up: map the new code (or map it to itself to confirm it's intentional) the day it appears. Reads the just-finalised run's ``result_summary`` (populated by the sweep) — no re-render. Dedup by the miss SET signature via the outbox (same mechanism as the unmapped-dept audit). Fires for ANY run outcome (misses happen on green runs too). Best-effort, never raises. """ try: from app.core.database import async_session_factory from app.notifications.service import ( _SEE_TEMPLATE_SENTINEL, CardSection, CardStat, ) async with async_session_factory() as session: run = await session.get(IdentitySyncRun, run_id) if run is None: return misses = (run.result_summary or {}).get("value_map_misses") or {} # Shape: {target_field: [source_value, ...]}. Flatten to a # stable "field=value" set for the signature + display. pairs: list[str] = [] for field_name in sorted(misses): vals = misses[field_name] if isinstance(vals, (list, tuple, set)): for v in sorted(str(x) for x in vals): pairs.append(f"{field_name}={v}") elif vals: pairs.append(f"{field_name}={vals}") if not pairs: return target = await session.get(IdentitySyncTarget, run.target_id) if target is None or target.deleted_at is not None: return if not target.alert_enabled: return signature = _unmapped_signature(pairs) if await _last_alert_signature( session, target_id=target.id, template_key=_VALUE_MAP_MISS_TEMPLATE_KEY, sig_key="value_map_miss_signature", ) == signature: logger.info( "identity_sync value_map-miss audit: target %s set " "unchanged (signature=%s) — skipping", target.slug, signature, ) return recipients = await _resolve_recipients(session, target) if not recipients: logger.info( "identity_sync value_map-miss audit: target %s has %d " "unmapped code(s) but no eligible recipients — skipping " "(no admin fallback by design)", target.slug, len(pairs), ) return count = len(pairs) listed = pairs[:20] body_lines = "\n".join(f"• {p}" for p in listed) overflow = count - len(listed) title_zh = f"🔤 数据同步:{count} 个字段值缺少映射 — {target.name}" card_subtitle = LocalizedText( zh=( "下列「字段=源值」在同步目标的 value_map 里没有对应条目。" "STRICT 映射会把未命中的值渲染为空 → 该字段被丢弃、下游对应" "属性被冻结(且运行仍显示绿色)。请为每个新值补一行映射;若确属" "预期,映射到其自身以显式放行。补齐后相关人员会自动重新同步。" ), en=( "The \"field=source value\" pairs below have no entry in " "this sync target's value_map. A STRICT mapping renders an " "unmatched value as empty → the field is dropped and the " "matching downstream attribute freezes (while the run " "still shows green). Please add one mapping row per new " "value; if the value is expected, map it to itself to " "allow it explicitly. Once mapped, the affected people are " "re-synced automatically." ), ja=( "以下の「フィールド=ソース値」は、この同期ターゲットの " "value_map に対応するエントリがありません。STRICT " "マッピングでは未一致の値は空として出力されるため、" "そのフィールドは破棄され、連携先の該当属性は" "更新されないまま固定されます(実行はグリーンのままです)。" "新しい値ごとにマッピングを 1 行ずつ追加してください。" "想定どおりの値であれば、自身の値へマッピングして明示的に" "許可してください。補完後、対象者は次回自動的に" "再同期されます。" ), ) card_stats = [ CardStat( label=LocalizedText( zh="同步目标", en="Sync target", ja="同期ターゲット", ), value=target.name, hint=f"`{target.slug}`", ), CardStat( label=LocalizedText( zh="待映射值数", en="Values to map", ja="マッピング待ちの値の数", ), value=str(count), ), ] card_sections = [ CardSection( title=LocalizedText( zh="待映射字段值", en="Field values to map", ja="マッピング待ちのフィールド値", ), body=( LocalizedText( zh=f"{body_lines}\n… 其余 {overflow} 个", en=f"{body_lines}\n… and {overflow} more", ja=f"{body_lines}\n… ほか {overflow} 件", ) if overflow > 0 else body_lines ), tone="warn", ), ] envelope_kwargs = { "title": title_zh, # See the unmapped-dept audit: the bell renders the stored # body verbatim unless it is the sentinel. "body_text": _SEE_TEMPLATE_SENTINEL, "channels": ["feishu", "email"], "template_key": _VALUE_MAP_MISS_TEMPLATE_KEY, "context": { "target_slug": target.slug, "target_name": target.name, "miss_count": count, "misses": listed, "value_map_miss_signature": signature, }, "url": f"/admin/identity-sync?tab=targets&target={target.id}", "card_subtitle": card_subtitle, "card_stats": card_stats, "card_sections": card_sections, "related_resource_type": "identity_sync_target", "related_resource_id": UUID(str(target.id)), } trace_id = f"{_VALUE_MAP_MISS_TEMPLATE_KEY}:{target.slug}:{signature}" frozen = list(recipients) await asyncio.gather( *[ _dispatch_one( recipient=r, envelope_kwargs=envelope_kwargs, trace_id=trace_id, ) for r in frozen ], return_exceptions=True, ) logger.info( "identity_sync value_map-miss audit dispatched for target %s: " "%d value(s) to %d recipient(s) (signature=%s)", target.slug, count, len(frozen), signature, ) except Exception: # noqa: BLE001 logger.exception( "identity_sync value_map-miss audit crashed for run %s — " "suppressed (sync outcome unaffected)", run_id, ) # ── NetSuite unadopted-department audit (non-blocking heads-up) ────── _UNADOPTED_NS_TEMPLATE_KEY = "identity_sync.unadopted_ns_depts" # Locale-fragment triples for the one paragraph this alert can't express # with a shared sentence: with ``netsuite_create_missing`` OFF the unadopted # depts are safely skipped (a pre-flight warning), with it ON they are about # to be CREATED as parallel departments (an active hazard). Two different # things to say, so the call site picks the pair and every language reads its # own — the ``{create_missing_note_*}`` slots in the template. _NS_NOTE_CREATE_MISSING_ON: dict[str, str] = { "zh": ( "⚠️ 已开启 create_missing,但下列部门在 NetSuite 里还没有对应 " "externalId(未采纳)。本次同步会把它们**新建**为平行部门,可能与既有 " "Authing 老部门重复。请先给既有 NS 部门回填 externalId(就地采纳)再让" "其创建,或确认确属全新部门。" ), "en": ( "⚠️ create_missing is enabled, but the departments listed below still " "have no matching externalId in NetSuite (not yet adopted). This sync " "will **create** them as parallel departments, which may duplicate " "the existing Authing-era ones. Please backfill the externalId onto " "the existing NS department first (adopt it in place) before letting " "them be created, or confirm that they really are new." ), "ja": ( "⚠️ create_missing が有効になっていますが、以下の部門には NetSuite 側で" "対応する externalId がありません(未取り込み)。今回の同期でこれらは" "並行部門として**新規作成**され、既存の Authing 由来の部門と重複する" "おそれがあります。作成させる前に、既存の NS 部門へ externalId を" "登録して取り込む(その場で採用する)か、本当に新設部門であることを" "ご確認ください。" ), } _NS_NOTE_CREATE_MISSING_OFF: dict[str, str] = { "zh": ( "下列部门在 NetSuite 里还没有对应 externalId(未采纳),当前 " "update-only 模式下会被安全跳过、不会同步。开启 create_missing 之前," "请先给既有 NS 部门回填 externalId(就地采纳),否则开启后会被新建为" "平行部门、与老部门重复。" ), "en": ( "The departments listed below still have no matching externalId in " "NetSuite (not yet adopted). In the current update-only mode they are " "safely skipped and never synced. Before enabling create_missing, " "please backfill the externalId onto the existing NS department " "(adopt it in place) — otherwise, once it is enabled, they are " "created as parallel departments that duplicate the legacy ones." ), "ja": ( "以下の部門には NetSuite 側で対応する externalId がありません" "(未取り込み)。現在の update-only モードでは安全にスキップされ、" "同期されません。create_missing を有効にする前に、既存の NS 部門へ " "externalId を登録して取り込んで(その場で採用して)ください。" "そのまま有効にすると、並行部門として新規作成され、既存の部門と" "重複します。" ), } async def maybe_notify_unadopted_ns_depts(*, target_id: UUID) -> None: """Alert IT (NON-blocking) when a NetSuite dept-sync target has active org nodes NOT yet adopted in NetSuite (no NS department carries their HRIS code as ``externalId``). NS analog of :func:`maybe_notify_unmapped_depts`, same recurring bug in a different system: NS departments are Authing-era reference data keyed by INTERNAL id with an EMPTY externalId; Helix addresses them by ``eid:{HRIS code}``. An **unadopted** dept (no externalId match) is the split-risk seed — while the dept runner is update-only it is safely SKIPPED, but the moment ``netsuite_create_missing`` is flipped on, every unadopted node is CREATED as a PARALLEL SF-coded dept beside the real Authing-era one (exactly the 汇联易 "AI Lab split", on NS). So this fires as a **pre-flight before create_missing**: while running update-only, it lists the depts that must be adopted in place (backfill externalId onto the existing NS dept) FIRST. The sync stays green — unadopted depts are already skipped, this just surfaces the adoption gap. Fires from the NS dept phase (already gated behind ``netsuite_sync_departments``), so silent until dept sync is enabled. Reliability: if the NS ``externalId`` set can't be read (client/SuiteQL error) we ABORT rather than alert — a failed read would make every node look unadopted (false storm). Dedup by the unadopted SET signature, same audience/rules as the huilianyi audit. Best-effort, never raises. """ try: from app.core.database import async_session_factory from app.domains.identity_sync.service import IdentitySyncService from app.identity.person_models import IdentityOrgNode from app.notifications.service import ( _SEE_TEMPLATE_SENTINEL, CardSection, CardStat, ) async with async_session_factory() as session: target = await session.get(IdentitySyncTarget, target_id) if target is None or target.deleted_at is not None: return if target.kind != "netsuite" or not target.alert_enabled: return # Adopted set = NS department externalIds (HRIS codes already # backfilled). Build the client + query in a guarded block; a # failure aborts (never alert on an unreliable set). svc = IdentitySyncService(session) adopted: set[str] = set() client = None try: client = await svc._build_netsuite_client(target) rows = await client.suiteql( "SELECT externalid AS eid FROM department " "WHERE externalid IS NOT NULL ORDER BY id" ) adopted = { str(r.get("eid")).strip() for r in rows if (r.get("eid") or "").strip() } except Exception: # noqa: BLE001 logger.warning( "netsuite unadopted-dept audit: could not read NS " "externalId set for target %s — aborting (no alert on " "an unreliable adopted set)", target.slug, exc_info=True, ) return finally: if client is not None: try: await client.aclose() except Exception: # noqa: BLE001 pass nodes = ( await session.execute( select(IdentityOrgNode).where( IdentityOrgNode.deleted_at.is_(None), IdentityOrgNode.node_status == "active", ) ) ).scalars().all() unadopted = sorted( ( (str(n.org_node_source_id), n.org_node_name or "") for n in nodes if str(n.org_node_source_id) not in adopted ), key=lambda x: x[0], ) if not unadopted: return signature = _unmapped_signature([c for c, _ in unadopted]) if await _last_alert_signature( session, target_id=target.id, template_key=_UNADOPTED_NS_TEMPLATE_KEY, sig_key="unadopted_signature", ) == signature: logger.info( "netsuite unadopted-dept audit: target %s set unchanged " "(signature=%s) — skipping", target.slug, signature, ) return recipients = await _resolve_recipients(session, target) if not recipients: logger.info( "netsuite unadopted-dept audit: target %s has %d " "unadopted dept(s) but no eligible recipients — skipping", target.slug, len(unadopted), ) return create_missing = bool( (target.runner_options or {}).get("netsuite_create_missing"), ) count = len(unadopted) listed = unadopted[:20] body_lines = "\n".join(f"• {code} {name}" for code, name in listed) overflow = count - len(listed) title_zh = f"🗂️ NetSuite:{count} 个部门未采纳 — {target.name}" note = ( _NS_NOTE_CREATE_MISSING_ON if create_missing else _NS_NOTE_CREATE_MISSING_OFF ) card_subtitle = LocalizedText( zh=note["zh"], en=note["en"], ja=note["ja"], ) tone = "danger" if create_missing else "warn" card_stats = [ CardStat( label=LocalizedText( zh="同步目标", en="Sync target", ja="同期ターゲット", ), value=target.name, hint=f"`{target.slug}`", ), CardStat( label=LocalizedText( zh="未采纳部门数", en="Departments not yet adopted", ja="未取り込みの部門数", ), value=str(count), ), CardStat( label="create_missing", value=( LocalizedText(zh="已开启", en="Enabled", ja="有効") if create_missing else LocalizedText(zh="未开启", en="Disabled", ja="無効") ), ), ] card_sections = [ CardSection( title=LocalizedText( zh="未采纳部门", en="Departments not yet adopted", ja="未取り込みの部門", ), body=( LocalizedText( zh=f"{body_lines}\n… 其余 {overflow} 个", en=f"{body_lines}\n… and {overflow} more", ja=f"{body_lines}\n… ほか {overflow} 件", ) if overflow > 0 else body_lines ), tone=tone, ), ] envelope_kwargs = { "title": title_zh, # See the unmapped-dept audit: the bell renders the stored # body verbatim unless it is the sentinel. "body_text": _SEE_TEMPLATE_SENTINEL, "channels": ["feishu", "email"], "template_key": _UNADOPTED_NS_TEMPLATE_KEY, "context": { "target_slug": target.slug, "target_name": target.name, "unadopted_count": count, "unadopted_codes": [c for c, _ in listed], "create_missing": create_missing, # Locale-fragment triple — the template's # ``{create_missing_note_*}`` slot. Stored per language so # the bell still reads correctly if the recipient later # switches theirs. "create_missing_note_zh": note["zh"], "create_missing_note_en": note["en"], "create_missing_note_ja": note["ja"], "unadopted_signature": signature, }, "url": f"/admin/identity-sync?tab=targets&target={target.id}", "card_subtitle": card_subtitle, "card_stats": card_stats, "card_sections": card_sections, "related_resource_type": "identity_sync_target", "related_resource_id": UUID(str(target.id)), } trace_id = ( f"{_UNADOPTED_NS_TEMPLATE_KEY}:{target.slug}:{signature}" ) frozen = list(recipients) await asyncio.gather( *[ _dispatch_one( recipient=r, envelope_kwargs=envelope_kwargs, trace_id=trace_id, ) for r in frozen ], return_exceptions=True, ) logger.info( "netsuite unadopted-dept audit dispatched for target %s: %d " "dept(s) to %d recipient(s) (create_missing=%s, signature=%s)", target_id, count, len(frozen), create_missing, signature, ) except Exception: # noqa: BLE001 logger.exception( "netsuite unadopted-dept audit crashed for target %s — " "suppressed (sync outcome unaffected)", target_id, ) # ── Blocked-backlog audit (green run,人工待处理) ───────────────────── _BLOCKED_BACKLOG_TEMPLATE_KEY = "identity_sync.blocked_backlog" async def maybe_notify_blocked_backlog(*, run_id: UUID) -> None: """Alert IT (NON-blocking) when a sweep left people ``blocked``. ``blocked`` means the vendor refused a person for a NON-transient, data/config-dependent reason, so — by design — the run stays GREEN and the row is not retried until the underlying data changes. That design note has always claimed the backlog "is surfaced and alerted once"; it wasn't. There was no alert of any kind, on any target, so a blocked person was invisible unless somebody opened the sync ledger and looked. That is precisely how the Zoom gap was found by a person, not by the platform: bin.ye@pingcap.com had no Zoom account for a day, the runs were all green, and nothing anywhere said "17 people need an admin". Dedup by the blocked SET signature via the outbox (same mechanism as the unmapped-dept / value-map audits) — the first occurrence of a given backlog alerts, identical backlogs on later sweeps stay quiet, and a changed set (someone new blocked, or someone cleared) re-alerts. Reasons listed in ``SELF_DRAINING_BLOCKED_REASONS`` are excluded: they are expected, self-clearing states (e.g. a per-run create budget that deliberately spreads a backlog over several sweeps) and paging a human for them would train everyone to ignore the alert. Best-effort, never raises. """ try: from app.core.database import async_session_factory from app.domains.identity_sync.models import IdentitySyncRunItem from app.domains.identity_sync.runners.base import ( SELF_DRAINING_BLOCKED_REASONS, ) from app.notifications.service import ( _SEE_TEMPLATE_SENTINEL, CardSection, CardStat, ) async with async_session_factory() as session: run = await session.get(IdentitySyncRun, run_id) if run is None or not (run.blocked_count or 0): return target = await session.get(IdentitySyncTarget, run.target_id) if target is None or target.deleted_at is not None: return if not target.alert_enabled: return items = ( await session.execute( select(IdentitySyncRunItem) .where(IdentitySyncRunItem.run_id == run_id) .where(IdentitySyncRunItem.outcome == "blocked") .order_by(IdentitySyncRunItem.subject_display) ) ).scalars().all() actionable: list[tuple[str, str]] = [] for it in items: detail = it.response_detail if isinstance(it.response_detail, dict) else {} reason = str(detail.get("blocked_reason") or "").strip() if reason in SELF_DRAINING_BLOCKED_REASONS: continue # Prefer the machine reason; fall back to the vendor's # error tag, then to the first line of the error text. label = ( reason or str(detail.get("error_tag") or "").strip() or (str(it.error or "").strip().splitlines() or [""])[0][:120] or "unknown" ) actionable.append((it.subject_display, label)) if not actionable: return signature = _unmapped_signature( [f"{email}|{label}" for email, label in actionable], ) if await _last_alert_signature( session, target_id=target.id, template_key=_BLOCKED_BACKLOG_TEMPLATE_KEY, sig_key="blocked_backlog_signature", ) == signature: logger.info( "identity_sync blocked-backlog audit: target %s set " "unchanged (signature=%s) — skipping", target.slug, signature, ) return recipients = await _resolve_recipients(session, target) if not recipients: logger.info( "identity_sync blocked-backlog audit: target %s has %d " "blocked person(s) but no eligible recipients — skipping " "(no admin fallback by design)", target.slug, len(actionable), ) return count = len(actionable) listed = actionable[:20] body_lines = "\n".join(f"• {email} — {label}" for email, label in listed) overflow = count - len(listed) title_zh = f"🚧 数据同步:{count} 人被下游拒绝待处理 — {target.name}" card_subtitle = LocalizedText( zh=( "下列人员的同步被下游系统以「非临时性」原因拒绝 —— 属于数据" "或配置问题,重试不会自愈,所以运行状态仍是绿色、也不会反复" "重试。请按每行的原因处理(补数据 / 改配置 / 在下游腾出容量);" "处理完这些人会在下一次同步自动重新推送。" ), en=( "The downstream system rejected the sync for the people " "below for a NON-transient reason — a data or " "configuration problem that retrying will not heal, which " "is why the run still shows green and nothing is retried " "in a loop. Please act on the reason given on each line " "(fill in the data / fix the configuration / free up " "capacity downstream); once handled, these people are " "pushed again automatically on the next sync." ), ja=( "以下の対象者の同期は、連携先システムから「一時的ではない」" "理由で拒否されました。データまたは設定の問題であり、" "リトライしても解消しないため、実行ステータスは" "グリーンのままで、繰り返しリトライも行われません。" "各行の理由に応じてご対応ください(データの補完 / 設定の" "修正 / 連携先での空き枠の確保)。対応が完了すると、" "次回の同期で自動的に再連携されます。" ), ) card_stats = [ CardStat( label=LocalizedText( zh="同步目标", en="Sync target", ja="同期ターゲット", ), value=target.name, hint=f"`{target.slug}`", ), CardStat( label=LocalizedText( zh="待处理人数", en="People needing action", ja="対応が必要な人数", ), value=str(count), ), ] card_sections = [ CardSection( title=LocalizedText( zh="被拒绝人员", en="Rejected people", ja="拒否された対象者", ), body=( LocalizedText( zh=f"{body_lines}\n… 其余 {overflow} 人", en=( f"{body_lines}\n… and {overflow} more " + ("person" if overflow == 1 else "people") ), ja=f"{body_lines}\n… ほか {overflow} 名", ) if overflow > 0 else body_lines ), tone="warn", ), ] envelope_kwargs = { "title": title_zh, # See the unmapped-dept audit: the bell renders the stored # body verbatim unless it is the sentinel. "body_text": _SEE_TEMPLATE_SENTINEL, "channels": ["feishu", "email"], "template_key": _BLOCKED_BACKLOG_TEMPLATE_KEY, "context": { "target_slug": target.slug, "target_name": target.name, "blocked_count": count, "blocked": [ {"subject": email, "reason": label} for email, label in listed ], "blocked_backlog_signature": signature, }, "url": f"/admin/identity-sync?tab=targets&target={target.id}", "card_subtitle": card_subtitle, "card_stats": card_stats, "card_sections": card_sections, "related_resource_type": "identity_sync_target", "related_resource_id": UUID(str(target.id)), } trace_id = ( f"{_BLOCKED_BACKLOG_TEMPLATE_KEY}:{target.slug}:{signature}" ) frozen = list(recipients) slug = target.slug await asyncio.gather( *[ _dispatch_one( recipient=r, envelope_kwargs=envelope_kwargs, trace_id=trace_id, ) for r in frozen ], return_exceptions=True, ) logger.info( "identity_sync blocked-backlog audit dispatched for target %s: " "%d person(s) to %d recipient(s) (signature=%s)", slug, count, len(frozen), signature, ) except Exception: # noqa: BLE001 logger.exception( "identity_sync blocked-backlog audit crashed for run %s — " "suppressed (sync outcome unaffected)", run_id, ) __all__ = [ "maybe_notify_blocked_backlog", "maybe_notify_run_failure", "maybe_notify_unmapped_depts", "maybe_notify_unadopted_ns_depts", ]