"""Notification category taxonomy — the single source of truth that maps a raw ``template_key`` onto (a) a human-meaningful *category* the employee can govern in their preference centre, and (b) an inbox *bucket* used to group the bell / notification-centre list. Why a curated registry instead of "family = first key segment" -------------------------------------------------------------- The ``template_key`` namespace grew organically and is genuinely mixed: some keys are dotted (``cloud.scope_request``, ``feed.like``) but many are flat (``welcome``, ``password_reset``, ``leaver_reminder``, ``transfer_received``). Splitting on the first ``.`` would scatter semantically-identical notifications across a dozen meaningless buckets and leave the flat keys in a junk "no-dot" group. So we classify by *meaning*: each category lists the prefixes and exact keys that belong to it, the classifier returns the first match, and anything unmatched falls through to ``other`` (mutable, action bucket — fail-towards-visible, never silently dropped from a preference toggle). This one registry backs THREE surfaces so they can never drift: * the preference matrix (which categories an employee may mute, per channel) * the inbox bucketing (action / info / social) * the MCP preference tools Hard safety floor (independent of this registry): templates matched by ``notifications.service.is_critical_template`` ALWAYS push, regardless of any mute or quiet-hours setting. A category being ``mutable`` only governs the employee-facing toggle; it never overrides the critical-delivery guarantee. """ from __future__ import annotations from dataclasses import dataclass, field # Inbox buckets. ``action`` = something the employee may need to do or # acknowledge; ``info`` = FYI / digest / lifecycle status; ``social`` = the # Moments / feed family (never drives the red badge — kept consistent with # ``inbox_service._SOCIAL_TEMPLATE_PREFIXES``). BUCKET_ACTION = "action" BUCKET_INFO = "info" BUCKET_SOCIAL = "social" # Stable category keys (used as JSON map keys in notification_preferences and # as i18n label keys ``notifications.category.`` on the frontend). Never # rename a key without a data migration of stored preference rows. CAT_REQUESTS = "requests_approvals" CAT_ACCOUNT_ACCESS = "account_access" CAT_ASSETS = "assets" CAT_CLOUD = "cloud" CAT_LIFECYCLE = "lifecycle" # 转岗复核。**刻意从 ``lifecycle`` 里拆出来**(2026-09-05): # ① 桶不对 —— 它落在 ``info``,而 ``role_change_review_open`` 是在要一个 # 人做决断,不是周知。红点该为它亮。 # ② 粒度不对 —— 一个想关掉「转岗复核」的经理,在合并的类别下会**连带** # 关掉入职、离职、HRIS 同步、identity_sync 和 procurement 的推送。一个 # 只能整片静音的开关,最后被按下的方式是整片静音。 # 这是唯一一档「同事的权限要你判断」的通知,量小、可关、进 action 桶。 CAT_ROLE_CHANGE = "role_change" CAT_LEAVE = "leave" CAT_SECURITY = "security" # 安全宣贯的**运营**档。刻意与 ``security`` 分开:必修宣贯是关不掉的 # action(公司要求他做一件事),运营宣贯是可关的 info(我们想让他知道 # 一件事)。合成一个类别会让每一份安全漫画都变成关不掉的推送 —— 而那 # 正是一个渠道被静音的方式,之后连必修的那些也没人读了。 CAT_AWARENESS = "awareness" CAT_SOCIAL = "social" CAT_SYSTEM = "system" CAT_OTHER = "other" @dataclass(frozen=True, slots=True) class Category: key: str bucket: str # ``mutable`` = the employee may turn off push channels for this category # in their preference centre. Lockable categories (False) still render in # the matrix but with the toggles disabled, so the employee can SEE that # these always reach them — they are never silently un-governable. mutable: bool # ``employee_facing`` = show in the self-service preference centre at all. # IT-ops-only categories (scheduled-job failures, reaper) are not shown to # plain employees — they would only ever reach IT recipients anyway. employee_facing: bool prefixes: tuple[str, ...] = field(default_factory=tuple) exact: tuple[str, ...] = field(default_factory=tuple) # Order matters: the classifier returns the FIRST category whose prefix/exact # set matches. Put the most specific (security / system) before the broad # domain buckets so e.g. ``cloud.grant.manual_revoke_escalated`` lands in # ``system`` (IT escalation) rather than the employee ``cloud`` category. _REGISTRY: tuple[Category, ...] = ( Category( key=CAT_SYSTEM, bucket=BUCKET_INFO, mutable=False, employee_facing=False, prefixes=( "scheduled_job.", "provisioning.revoke_failure", "cloud.grant.manual_revoke_escalated", "approval.reaper", "leaver_blocked_sla", # real keys: leaver_blocked_sla_warn/_urgent/_breach (underscore) "leaver_blocked.dual_sign", "leaver_escalation", "leaver_it_grace", ), ), Category( key=CAT_SECURITY, bucket=BUCKET_ACTION, mutable=False, employee_facing=True, prefixes=( "compliance_", "access_review", "jira.leaver_account_audit", "jira.recently_revoked_audit", # A named business owner being asked to confirm an externally # reported vulnerability. Immutable like the rest of this # category: an outside researcher is waiting on the answer, so # this is not a ping an employee should be able to mute. "bounty.", "system_registry.", ), # 必修宣贯 —— 用 exact 而不是前缀,这样同族的运营 key # (``security_awareness.campaign.notice``)不会被这一档吞掉: # 注册表是**首个命中即返回**,而 security 排在 awareness 前面。 # # 🔴 **催办也必须在这一档。** 它是必修那条推送的续集:如果原推送 # 关不掉、驱动红点,而「你还欠着」的提醒可以被关掉,那员工恰好能 # 静音掉唯一在告诉他还有必修没做的消息,红点也永远不会为这件未完成 # 的事亮起来。上线时这一条差点落进 awareness(可关 / info 桶),是 # 把分类打印出来逐条核对时抓到的。 # # 主管摘要**不**在这一档:那是给主管的知情通知,不是主管自己的活, # 它落 awareness(可关 / info)是对的。 exact=( "security_awareness.campaign.mandatory", "security_awareness.campaign.reminder", ), ), Category( # 运营宣贯。INFO 桶 —— 一份安全漫画不该驱动红点;红点是留给 # 「有人在等你做一件事」的。可关,而且它必须可关:一个员工关掉了 # 漫画推送,不影响他仍然收到必修的那一档。 key=CAT_AWARENESS, bucket=BUCKET_INFO, mutable=True, employee_facing=True, prefixes=("security_awareness.",), ), Category( key=CAT_SOCIAL, bucket=BUCKET_SOCIAL, mutable=True, employee_facing=True, prefixes=( "feed.celebration.", "feed.blessing.", "feed.comment.", "feed.like.", "feed.mention.", ), ), Category( key=CAT_CLOUD, bucket=BUCKET_ACTION, mutable=True, employee_facing=True, prefixes=("cloud.", "cmdb_iam_revival"), ), Category( key=CAT_ASSETS, bucket=BUCKET_ACTION, mutable=True, employee_facing=True, prefixes=( "asset.", "transfer_received", "transfer_accepted_it", "force_transfer_received", "borrow_auto_returned", ), # Same semantic family as ``transfer_received`` ("somebody's stuff is # now yours"), and something the recipient can ACT on (confirm or # reassign) — so it must land in the action bucket that drives the red # dot. Listed exactly here rather than left to the ``offboarding.`` # prefix below, which would classify it as a lifecycle FYI and hide it # from the badge the recipient actually watches. exact=("offboarding.handover_assigned",), ), Category( key=CAT_ACCOUNT_ACCESS, bucket=BUCKET_ACTION, mutable=True, employee_facing=True, prefixes=( "apikey.", "software_seat", "sso.", "mail_group_", "alias_", "password_reset", "provisioning.", "zoom.", "jira.", "vpn", ), ), Category( key=CAT_REQUESTS, bucket=BUCKET_ACTION, mutable=True, employee_facing=True, prefixes=("approval.", "support.", "github_id_change_"), exact=(), ), Category( # 假期提醒。Its own category rather than a member of ``lifecycle`` # for one load-bearing reason: lifecycle is an INFO bucket, and an # info-bucket row does not drive the red badge # (``inbox_service.unread_actionable``). A 年假到期提醒 carries a hard # deadline and asks the employee to go file a request — filing it under # an FYI bucket would hide the one notification whose whole value is # being noticed before a date. Mutable, so an employee who does not # want leave nudges can still turn the push channels off. key=CAT_LEAVE, bucket=BUCKET_ACTION, mutable=True, employee_facing=True, prefixes=("hr_leave.",), ), Category( # 转岗复核 —— 必须排在 ``lifecycle`` 之前:分类器返回第一个命中的 # 类别,而 ``lifecycle`` 曾经用 ``role_change`` 前缀把它吞掉。 key=CAT_ROLE_CHANGE, # ACTION,不是 info:卡片在要一个人对同事的权限做判断。 bucket=BUCKET_ACTION, # 可关。这一档确实是「有人在等你做一件事」,但它等的是一件 # 没有截止时间、不处理也不会自动回收任何东西的事(见 # ``mover_review_settings``),所以不该是关不掉的那一类。 mutable=True, employee_facing=True, prefixes=("role_change",), ), Category( key=CAT_LIFECYCLE, bucket=BUCKET_INFO, mutable=True, employee_facing=True, prefixes=( "offboarding.", "onboarding.", "leaver_reminder", "hire_day_it_info", "hris.", "identity_sync.", "lifecycle_daily_digest", "procurement.", # Automation-exception echoes. Registered HERE rather than # under a new category on purpose: the employee-facing half # is "the thing you were waiting for now works", which is # lifecycle news, and routing it through an existing # category means every mute / do-not-disturb preference an # employee already set governs it. A new category would # have been easier and would have silently handed everyone # a ping they had never agreed to. "automation_exception.", ), ), ) # Fallback — anything unmatched is a mutable, employee-facing action item. # Fail-towards-visible: an un-classified real notification still shows in the # inbox action bucket and still has a (per-channel) preference toggle. _OTHER = Category( key=CAT_OTHER, bucket=BUCKET_ACTION, mutable=True, employee_facing=True, ) def classify(template_key: str | None) -> Category: """Map a ``template_key`` to its :class:`Category` (first match wins).""" if template_key: for cat in _REGISTRY: for p in cat.prefixes: if template_key.startswith(p): return cat if template_key in cat.exact: return cat return _OTHER def category_key_for(template_key: str | None) -> str: return classify(template_key).key def bucket_for(template_key: str | None) -> str: return classify(template_key).bucket def registry_in_order() -> tuple[Category, ...]: """The classified categories in precedence order (first match wins). The ``other`` fallback is NOT included — it is everything that matches none of these, and callers building a SQL bucket clause handle it explicitly.""" return _REGISTRY def employee_categories() -> list[Category]: """Categories surfaced in the self-service preference centre, in display order (action items first, then info, then social).""" facing = [c for c in _REGISTRY if c.employee_facing] facing.append(_OTHER) order = {BUCKET_ACTION: 0, BUCKET_INFO: 1, BUCKET_SOCIAL: 2} return sorted(facing, key=lambda c: order.get(c.bucket, 9)) ALL_CATEGORY_KEYS: tuple[str, ...] = tuple(c.key for c in _REGISTRY) + (CAT_OTHER,) __all__ = [ "ALL_CATEGORY_KEYS", "BUCKET_ACTION", "BUCKET_INFO", "BUCKET_SOCIAL", "registry_in_order", "CAT_ACCOUNT_ACCESS", "CAT_ASSETS", "CAT_CLOUD", "CAT_LEAVE", "CAT_LIFECYCLE", "CAT_ROLE_CHANGE", "CAT_OTHER", "CAT_REQUESTS", "CAT_SECURITY", "CAT_SOCIAL", "CAT_SYSTEM", "Category", "bucket_for", "category_key_for", "classify", "employee_categories", ]