"""Outbound notification for the one thing an employee can do here. The register is IT-maintained: an employee cannot re-point their own ownership row, and they should not be able to — ownership is an accountability record, not a preference. What they CAN do is say "this is not mine", which is a claim IT has to adjudicate. Recipient discipline matches the system registry: active ``global_admin`` holders only, because they are the people who can actually change the row. One human action produces one push. """ from __future__ import annotations import logging import uuid # noqa: F401 — used in a string annotation from collections.abc import Sequence from datetime import UTC, datetime from typing import TYPE_CHECKING from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.auth.models import AuthRole, AuthUserRole from app.core.rbac import GLOBAL_ADMIN_ROLE from app.domains.approval_registry import constants as const from app.domains.approval_registry.models import ApprovalFlow from app.domains.approval_registry.serializers import ( feishu_admins as feishu_admins_of, ) from app.domains.approval_registry.service import person_label as svc_person_label from app.identity.person_models import IdentityPerson from app.notifications.service import ( CardAction, LocalizedText, NotificationEnvelope, NotificationService, ) if TYPE_CHECKING: # pragma: no cover - typing only from app.domains.approval_registry.models import ( ApprovalFlowRequest, FlowConfirmationItem, FlowConfirmationRound, ) logger = logging.getLogger(__name__) #: Where the push lands: the registry page, filtered to the disputed #: row. A reach has to land somewhere the recipient can act, not on a #: list they then have to search. ADMIN_PAGE = "/admin/approval-flows?q={code}" #: Where the 「飞书后台待同步」 digest lands. A tab, not a filter, because #: the list is a worklist somebody drains rather than a slice of the #: register — and because a push whose landing page is 「the register, #: go find them」 makes the reader redo the comparison the card already #: did for them. DRIFT_PAGE = "/admin/approval-flows?tab=feishu-drift" async def _registry_maintainers(db: AsyncSession) -> list[IdentityPerson]: """Active ``global_admin`` holders — the only role that can re-point an owner (``can_manage_approval_registry``).""" rows = ( await db.execute( select(IdentityPerson) .join(AuthUserRole, AuthUserRole.user_id == IdentityPerson.id) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where( AuthRole.name == GLOBAL_ADMIN_ROLE, IdentityPerson.employment_status == "active", IdentityPerson.deleted_at.is_(None), ), ) ).scalars().all() seen: set = set() unique: list[IdentityPerson] = [] for person in rows: if person.id in seen: continue seen.add(person.id) unique.append(person) return unique async def notify_ownership_disputed( db: AsyncSession, flow: ApprovalFlow, *, disputer: IdentityPerson, note: str | None, ) -> int: """Tell IT that a named owner says the flow is not theirs. Best-effort: a Feishu hiccup must not roll back the event just written to the flow's trail. Returns how many operators were reached, so the caller can report that honestly instead of claiming 「已通知」. """ url = ADMIN_PAGE.format(code=flow.approval_code) who = disputer.display_name or disputer.work_email context = { "flow_name": flow.name, "approval_code": flow.approval_code, "disputer_name": who, "disputer_email": disputer.work_email, "owner_source": flow.owner_source, "note_block": (note or "").strip() or "(未填写说明)", } service = NotificationService(db) sent = 0 for operator in await _registry_maintainers(db): env = NotificationEnvelope( recipient=operator, title=f"{who} 表示不该由自己负责审批流程「{flow.name}」", body_text="", channels=["feishu"], template_key="approval_registry.ownership_disputed", context=context, url=url, related_resource_type="registry_approval_flow", related_resource_id=flow.id, card_actions=[ CardAction(label="打开流程台账", url=url, style="primary"), ], ) try: await service.push_inbox_and_feishu(env) sent += 1 except Exception: logger.warning( "approval_registry: dispute notification failed for %s → %s", flow.approval_code, operator.work_email, exc_info=True, ) return sent async def _person_by_id( db: AsyncSession, person_id: uuid.UUID | None, ) -> IdentityPerson | None: """Resolve a person by id instead of through an ORM relationship. Every relationship in this domain is ``lazy="raise"``, which is the right default — but it makes a notification helper depend on how its caller happened to load the row, and these helpers run inside try/except blocks where that failure is invisible. """ if person_id is None: return None return ( await db.execute( select(IdentityPerson).where(IdentityPerson.id == person_id), ) ).scalar_one_or_none() async def _portal_url(db: AsyncSession) -> str: from app.integrations.service import get_platform_setting return str( await get_platform_setting( "app.frontend_base_url", default="", session=db, ) or "", ) def _flow_list_text(names: list[str], cap: int = 10) -> LocalizedText: """The bulleted flow list for a confirmation card. Flow names themselves are never translated — they are the exact strings a person searches for in Feishu 审批, and a helpfully localized name is a name they cannot find. Only the overflow line is, because it is our sentence, not theirs. """ shown = "\n".join(f" · {n}" for n in names[:cap]) extra = len(names) - cap if extra <= 0: return LocalizedText(zh=shown, en=shown, ja=shown) return LocalizedText( zh=f"{shown}\n · …… 还有 {extra} 条", en=f"{shown}\n · … and {extra} more", ja=f"{shown}\n · …… ほか {extra} 件", ) def _drift_lines( rows: list[tuple[ApprovalFlow, str, list[str]]], ) -> LocalizedText: """One line per drifting flow: 台账负责人 vs 后台管理员. Both names, always. A digest that said only 「这条对不上」 would send the reader to the console to compare 49 rows by hand — the whole value of the push is that it already did the comparison. Flow names are never translated (they are the strings to search for in the console); only our own scaffolding is. """ def render(sep_from: str, sep_to: str, more: str) -> str: out = [] for flow, owner_name, admin_names in rows[: const.MAX_DRIFT_LISTED]: admins = "、".join(admin_names) if admin_names else "—" out.append( f" · {flow.name} ({flow.approval_code})\n" f" {sep_from}{owner_name} {sep_to}{admins}", ) extra = len(rows) - const.MAX_DRIFT_LISTED if extra > 0: out.append(f" · {more.format(count=extra)}") return "\n".join(out) return LocalizedText( zh=render("台账负责人：", "飞书流程管理员：", "…… 还有 {count} 条"), en=render("register owner: ", "Feishu admin: ", "… and {count} more"), ja=render("台帳担当者：", "Feishu 管理者：", "…… ほか {count} 件"), ) async def notify_transfer_console_gap( db: AsyncSession, flow: ApprovalFlow, *, actor: IdentityPerson, successor: IdentityPerson, previous_owner_name: str | None, ) -> int: """A person just pressed 「转交」 and the console did not follow. Fires at the moment of the transfer, in addition to the daily digest — user decision, 2026-08-28. The digest was built first and deliberately, to avoid one offboarding hand-over (which moves every flow a leaver owned inside a single transaction) delivering nine cards about nine rows of the same console screen. That reasoning still holds, and this push does not break it, because the seam it hangs off is narrow: ``set_owner`` calls it only for ``announce == HANDOVER_TRANSFER``, which is passed from exactly two places, both an employee pressing the button themselves (``confirmation.answer_item`` for a round item, ``confirmation.record_self_answer`` out of round). The offboarding coordinator passes ``announce=None`` and the request-registration paths pass ``announce=None`` too, so a batch hand-over still reaches IT as one digest line rather than a burst. The caller stamps ``feishu_admin_drift_notified_at`` when this returns non-zero, so the 09:00 digest does not raise the same flow again the next morning. A failed push leaves the stamp NULL, which is what makes the digest the safety net rather than a duplicate. Only called when the transfer actually left a gap — the successor is not among the console's 流程管理员. Telling IT 「有人转交了」 about a flow whose console entry already names the new owner would be an alert with no action behind it, and the fastest way to have this one muted. """ admins = feishu_admins_of(flow) portal = await _portal_url(db) page = f"{portal}{DRIFT_PAGE}" if portal else DRIFT_PAGE who = svc_person_label(actor) or (actor.work_email or "—") to_whom = svc_person_label(successor) or (successor.work_email or "—") context = { "flow_name": flow.name, "approval_code": flow.approval_code, "actor_name": who, "successor_name": to_whom, "successor_email": successor.work_email or "—", "previous_owner_name": previous_owner_name or who, # The full list, not just the mismatching entry: a definition # with three admins is a different console edit from one with a # single stale name, and the reader is about to make that edit. "console_admins": "、".join(a["name"] for a in admins) or "—", "portal_url": page, } service = NotificationService(db) sent = 0 for operator in await _registry_maintainers(db): env = NotificationEnvelope( recipient=operator, title=( f"{who} 把审批流程「{flow.name}」转交给了 {to_whom}" f"——飞书后台的流程管理员需要手动改" ), body_text="", channels=["feishu"], template_key="approval_registry.transfer_console_gap", context=context, url=DRIFT_PAGE, related_resource_type="registry_approval_flow", related_resource_id=flow.id, card_actions=[ CardAction( label=LocalizedText( zh="去飞书后台改管理员", en="Fix the admin in Feishu", ja="Feishu で管理者を修正", ), url=const.FEISHU_APPROVAL_ADMIN_URL, style="primary", ), CardAction( label=LocalizedText( zh="看待同步清单", en="See the full list", ja="未同期の一覧を見る", ), url=page, style="default", ), ], ) try: await service.push_inbox_and_feishu(env) sent += 1 except Exception: logger.warning( "approval_registry: transfer console-gap alert failed for " "%s → %s", flow.approval_code, operator.work_email, exc_info=True, ) return sent async def notify_feishu_admin_drift( db: AsyncSession, flows: Sequence[ApprovalFlow], ) -> int: """Tell IT which flows the Feishu console still points at the old owner — the one hand-off step the platform structurally cannot do. **One digest, not one push per flow.** The per-event shape was the obvious first design and it is wrong here for a reason worth writing down: the transfers that create drift arrive in clusters (an offboarding hand-over moves every flow a leaver owned, in one transaction), so per-event would deliver nine cards about nine rows of the same console screen. The same 「one card per person, never one per flow」 rule the confirmation round already follows. The cost is that a 14:00 transfer is raised the next morning. That is the right trade: the console list is edited by a human on a working day, and a same-minute push buys nothing while a storm costs the channel's credibility. Best-effort per recipient, and the return value is how many were actually reached — the caller reports that rather than 「已通知」. """ if not flows: return 0 people = { p.id: p for p in ( await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_( [f.owner_person_id for f in flows if f.owner_person_id], ), ), ) ).scalars().all() } rows: list[tuple[ApprovalFlow, str, list[str]]] = [] for flow in flows: owner = people.get(flow.owner_person_id) if flow.owner_person_id else None rows.append(( flow, (svc_person_label(owner) or "—") if owner is not None else "—", # Every admin the console names, not just the mismatching # one: 「后台写的是谁」 is the fact the reader needs in front # of them, and a definition with three admins is a different # edit from one with a single stale name. [a["name"] for a in feishu_admins_of(flow)], )) portal = await _portal_url(db) page = f"{portal}{DRIFT_PAGE}" if portal else DRIFT_PAGE context = { "drift_count": str(len(flows)), "drift_list": _drift_lines(rows), "portal_url": page, # Localized like the flow list, and for the same reason: it # lands INSIDE an otherwise translated body, where one Chinese # sentence reads as a bug rather than as a detail. "renag_line": LocalizedText( zh=f"仍然对不上的会在 {const.DRIFT_RENAG_DAYS} 天后再提醒一次。", en=( f"Anything still mismatched is raised again in " f"{const.DRIFT_RENAG_DAYS} days." ), ja=( f"不一致が残っている場合は {const.DRIFT_RENAG_DAYS} 日後に" f"再度お知らせします。" ), ), } service = NotificationService(db) sent = 0 for operator in await _registry_maintainers(db): env = NotificationEnvelope( recipient=operator, title=( f"飞书审批后台待同步：{len(flows)} 条流程的管理员和台账对不上" ), body_text="", channels=["feishu"], template_key="approval_registry.feishu_admin_drift", context=context, url=DRIFT_PAGE, related_resource_type="registry_approval_flow", # No single row owns a digest about many. Left unset rather # than pinned to the first flow, which would make the inbox # entry claim to be about one row and hide the other eight. card_actions=[ # Console first: it is where the work happens, and the # register page cannot do this edit at all. Not # role-gated — Feishu judges permission itself, and # hiding it behind Helix roles would hide it from the # 流程管理员 population this card is about. CardAction( label=LocalizedText( zh="去飞书后台改管理员", en="Fix the admin in Feishu", ja="Feishu で管理者を修正", ), url=const.FEISHU_APPROVAL_ADMIN_URL, style="primary", ), CardAction( label=LocalizedText( zh="看待同步清单", en="See the full list", ja="未同期の一覧を見る", ), url=page, style="default", ), ], ) try: await service.push_inbox_and_feishu(env) sent += 1 except Exception: logger.warning( "approval_registry: drift digest failed for %s", operator.work_email, exc_info=True, ) return sent async def notify_confirmation_round( db: AsyncSession, round_row: FlowConfirmationRound, items: list[FlowConfirmationItem], ) -> int: """One card per owner listing every flow recorded against them. **Never one card per flow.** An owner with nine flows gets one card listing nine; making it nine cards is how the whole round gets muted. The card carries a no-login 「全部确认」 button (an action token, purpose-scoped so it cannot be redirected at another action) and a deep link into the page where a per-flow answer — including handing one over — can be given. Best-effort per recipient: one unreachable person must not cost the other owners their card. Returns how many were actually reached, so the caller can report that instead of claiming 「已通知」. """ from app.core.security import create_action_token from app.domains.approval_registry import confirmation as conf from app.domains.approval_registry.models import ApprovalFlow grouped = conf.group_by_owner(items) if not grouped: return 0 people = { p.id: p for p in ( await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(list(grouped)), ), ) ).scalars().all() } flow_ids = {i.flow_id for i in items} flows = { f.id: f for f in ( await db.execute( select(ApprovalFlow).where(ApprovalFlow.id.in_(flow_ids)), ) ).scalars().all() } portal = await _portal_url(db) due = round_row.due_at.strftime("%Y-%m-%d") if round_row.due_at else "" service = NotificationService(db) sent = 0 now = datetime.now(UTC).replace(tzinfo=None) for person_id, own_items in grouped.items(): person = people.get(person_id) if person is None or person.employment_status != "active": continue names = [ flows[i.flow_id].name for i in own_items if i.flow_id in flows ] if not names: continue actions: list[CardAction] = [] confirm_url = "" if portal: token = create_action_token( purpose=const.ACTION_TOKEN_PURPOSE, subject_id=str(person.id), resource_id=str(round_row.id), ttl_days=const.ACTION_TOKEN_TTL_DAYS, ) confirm_url = ( f"{portal}/api/v1/approval-flows/confirm-by-token?token={token}" ) actions.append(CardAction( # The body of this card is already rendered per-recipient # from a zh/en/ja template; the BUTTONS were the one part # still hardcoded, so a Japanese owner read a Japanese # message and then had to press 「✅ 全部确认仍由我负责」. label=LocalizedText( zh="✅ 全部确认仍由我负责", en="✅ Confirm all are still mine", ja="✅ すべて自分の担当として確認", ), url=confirm_url, style="primary", )) actions.append(CardAction( label=LocalizedText( zh="逐条处理 / 转交他人", en="Review one by one / hand over", ja="個別に対応 / 引き継ぐ", ), url=f"{portal}/my/approval-flows" if portal else "/my/approval-flows", style="default", )) env = NotificationEnvelope( recipient=person, title=f"{round_row.label} 审批流程负责人确认：你名下有 {len(names)} 条", body_text="", channels=["feishu", "email"], template_key="approval_registry.confirmation_round", context={ "round_label": round_row.label, "flow_count": str(len(names)), # Bounded on purpose: a card listing forty flow names is # a wall nobody reads, and the page behind the button # has the full list anyway. The overflow line is # localized because it lands INSIDE an otherwise # translated body — one Chinese sentence in the middle of # an English message reads as a bug, not as a detail. "flow_list": _flow_list_text(names), "due_date": due, "confirm_url": confirm_url, "portal_url": portal, }, url="/my/approval-flows", related_resource_type="registry_approval_flow_round", related_resource_id=round_row.id, card_actions=actions, ) try: await service.push_inbox_and_feishu(env) for item in own_items: item.notified_at = now sent += 1 except Exception: logger.warning( "approval_registry: confirmation card failed for %s", person.work_email, exc_info=True, ) return sent async def notify_confirmation_overdue( db: AsyncSession, round_row: FlowConfirmationRound, items: list[FlowConfirmationItem], ) -> int: """Chase the owner, and copy their manager. The manager is copied rather than asked: they cannot answer for the flow, but an unanswered governance item is exactly the thing a lead can unblock in one conversation. Deduplicated per person — an owner with six unanswered flows is chased once. """ from app.domains.approval_registry import confirmation as conf from app.domains.approval_registry.models import ApprovalFlow grouped = conf.group_by_owner(items) if not grouped: return 0 people = { p.id: p for p in ( await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(list(grouped)), ), ) ).scalars().all() } flows = { f.id: f for f in ( await db.execute( select(ApprovalFlow).where( ApprovalFlow.id.in_({i.flow_id for i in items}), ), ) ).scalars().all() } portal = await _portal_url(db) service = NotificationService(db) now = datetime.now(UTC).replace(tzinfo=None) sent = 0 for person_id, own_items in grouped.items(): person = people.get(person_id) if person is None or person.employment_status != "active": continue names = [flows[i.flow_id].name for i in own_items if i.flow_id in flows] context = { "round_label": round_row.label, "flow_count": str(len(names)), "flow_list": "\n".join(f" · {n}" for n in names[:10]), "owner_name": person.display_name or person.work_email, "due_date": ( round_row.due_at.strftime("%Y-%m-%d") if round_row.due_at else "" ), "portal_url": portal, } recipients = [(person, "approval_registry.confirmation_overdue")] manager = None if person.manager_person_id: manager = ( await db.execute( select(IdentityPerson).where( IdentityPerson.id == person.manager_person_id, IdentityPerson.employment_status == "active", IdentityPerson.deleted_at.is_(None), ), ) ).scalar_one_or_none() if manager is not None: recipients.append( (manager, "approval_registry.confirmation_overdue_manager"), ) for recipient, template_key in recipients: env = NotificationEnvelope( recipient=recipient, title=( f"{round_row.label} 审批流程确认逾期：" f"{context['owner_name']} 还有 {len(names)} 条未回" ), body_text="", channels=["feishu"], template_key=template_key, context=context, url="/my/approval-flows", related_resource_type="registry_approval_flow_round", related_resource_id=round_row.id, card_actions=[CardAction( label=LocalizedText( zh="去确认", en="Confirm now", ja="確認する", ), url=( f"{portal}/my/approval-flows" if portal else "/my/approval-flows" ), style="primary", )], ) try: await service.push_inbox_and_feishu(env) except Exception: logger.warning( "approval_registry: overdue nudge failed for %s", recipient.work_email, exc_info=True, ) for item in own_items: item.escalated_at = now sent += 1 return sent async def notify_flow_request_submitted( db: AsyncSession, row: ApprovalFlowRequest, ) -> int: """Tell IT somebody needs a flow that does not exist. To ``global_admin`` only — they are the ones who can decide it and then build it in the Feishu console. Returns how many were reached. """ portal = await _portal_url(db) url = "/admin/approval-flows?tab=requests" requester = await _person_by_id(db, row.requester_person_id) who = requester.display_name or requester.work_email if requester else "员工" context = { "requester_name": who, "requester_email": requester.work_email if requester else "", "proposed_name": row.proposed_name, "purpose": row.purpose[:600], "category": row.proposed_category or "未指定", "portal_url": portal, } service = NotificationService(db) sent = 0 for operator in await _registry_maintainers(db): env = NotificationEnvelope( recipient=operator, title=f"{who} 申请新建审批流程「{row.proposed_name}」", body_text="", channels=["feishu"], template_key="approval_registry.flow_request_submitted", context=context, url=url, related_resource_type="registry_approval_flow_request", related_resource_id=row.id, card_actions=[CardAction( label="去处理", url=f"{portal}{url}" if portal else url, style="primary", )], ) try: await service.push_inbox_and_feishu(env) sent += 1 except Exception: logger.warning( "approval_registry: new-flow request push failed → %s", operator.work_email, exc_info=True, ) return sent async def notify_flow_request_decided( db: AsyncSession, row: ApprovalFlowRequest, ) -> int: """Tell the requester what IT decided. An approval says 「批了，正在建」 rather than 「好了」 — the flow does not exist until somebody builds it, and telling an employee it is done when it is not is how a request goes quiet for a fortnight. """ requester = await _person_by_id(db, row.requester_person_id) if requester is None or requester.employment_status != "active": return 0 portal = await _portal_url(db) approved = row.status == const.FLOW_REQ_APPROVED context = { "proposed_name": row.proposed_name, "note_block": (row.decision_note or "").strip() or "(未填写说明)", "portal_url": portal, } env = NotificationEnvelope( recipient=requester, title=( f"你申请的审批流程「{row.proposed_name}」已通过，IT 正在创建" if approved else f"你申请的审批流程「{row.proposed_name}」未通过" ), body_text="", channels=["feishu"], template_key=( "approval_registry.flow_request_approved" if approved else "approval_registry.flow_request_rejected" ), context=context, url="/my/approval-flows", related_resource_type="registry_approval_flow_request", related_resource_id=row.id, ) try: await NotificationService(db).push_inbox_and_feishu(env) return 1 except Exception: logger.warning( "approval_registry: decision push failed for request %s", row.id, exc_info=True, ) return 0 async def notify_flow_request_registered( db: AsyncSession, row: ApprovalFlowRequest, flow: ApprovalFlow, ) -> int: """The flow exists. Tell the requester its real name, and that they now own it — being handed a responsibility silently is how an owner first learns of it from a half-year card.""" requester = await _person_by_id(db, row.requester_person_id) if requester is None or requester.employment_status != "active": return 0 portal = await _portal_url(db) owner_id = row.proposed_owner_person_id or row.requester_person_id recipients = [requester] if owner_id != requester.id: nominee = ( await db.execute( select(IdentityPerson).where( IdentityPerson.id == owner_id, IdentityPerson.employment_status == "active", ), ) ).scalar_one_or_none() if nominee is not None: recipients.append(nominee) context = { "flow_name": flow.name, "approval_code": flow.approval_code, # NOT ``flow.owner``: the relationship is ``lazy="raise"`` and a # freshly created flow has never loaded it, so naming the owner # this way raised — inside a best-effort push, which swallowed it. # The requester simply never heard that their flow was ready. "owner_name": svc_person_label( await _person_by_id(db, flow.owner_person_id), ), "portal_url": portal, } service = NotificationService(db) sent = 0 for recipient in recipients: env = NotificationEnvelope( recipient=recipient, title=f"审批流程「{flow.name}」已经建好，可以用了", body_text="", channels=["feishu"], template_key="approval_registry.flow_request_registered", context=context, url="/my/approval-flows", related_resource_type="registry_approval_flow", related_resource_id=flow.id, card_actions=[CardAction( label=LocalizedText( zh="看看我负责的审批流程", en="View the flows I own", ja="担当している承認フローを見る", ), url=( f"{portal}/my/approval-flows" if portal else "/my/approval-flows" ), style="primary", )], ) try: await service.push_inbox_and_feishu(env) sent += 1 except Exception: logger.warning( "approval_registry: registration push failed → %s", recipient.work_email, exc_info=True, ) return sent # Re-exported from ``constants`` so a caller that already imports the # notifier does not need a second import to name a reason. HANDOVER_TRANSFER = const.HANDOVER_TRANSFER HANDOVER_OFFBOARDING = const.HANDOVER_OFFBOARDING HANDOVER_ASSIGNED = const.HANDOVER_ASSIGNED def _handover_line( reason: str, previous_name: str | None, actor_name: str | None, ) -> LocalizedText: """Opening line naming who handed the flow over, and why. A :class:`LocalizedText` rather than three suffixed context keys: the envelope fans out to one recipient here, but the language is resolved per recipient at send time, and ``_localized_context`` already knows how to do that. Hand-rolling ``{handover_line_en}`` into the English body is how a producer ends up shipping an English card with a Chinese sentence in the middle of it. """ who = previous_name or actor_name if reason == HANDOVER_OFFBOARDING: # NOT ``who``: on this path the actor is the IT admin or the cron # executing the transfer, and 「{actor} 离职」 about them is simply # a false sentence. Only the person we can actually name as the # previous owner is the one leaving. leaver = previous_name return LocalizedText( zh=f"{leaver or '前负责人'}离职,这条流程交接给了你。", en=( f"{leaver or 'The previous owner'} is leaving, and this " "flow has been handed over to you." ), ja=( f"{leaver or '前任担当者'}の退職に伴い、このフローを" "引き継ぎました。" ), ) if reason == HANDOVER_TRANSFER: return LocalizedText( zh=f"{who or '上一位负责人'}在半年确认里把这条流程转交给了你。", en=( f"{who or 'The previous owner'} passed this flow to you " "during the half-year ownership check." ), ja=( f"{who or '前任担当者'}が半年ごとの確認でこのフローをあなたに" "引き継ぎました。" ), ) # A cron or a script has no actor, and 「IT(IT)」 reads like a bug. by_zh = f"IT({actor_name})" if actor_name else "IT" by_en = f"IT ({actor_name})" if actor_name else "IT" return LocalizedText( zh=f"{by_zh}把这条流程的负责人登记成了你。", en=f"{by_en} recorded you as this flow's owner.", ja=f"{by_zh}がこのフローの担当者としてあなたを登録しました。", ) async def notify_owner_assigned( db: AsyncSession, flows: Sequence[ApprovalFlow], *, new_owner_id: uuid.UUID | None, previous_owner_id: uuid.UUID | None, reason: str, actor: IdentityPerson | None = None, ) -> int: """Tell somebody they now own these flows. Returns 1 if a push went out. Takes a LIST because this domain's rule is **one card per person, never one per flow** — the same rule the half-year round follows. An offboarding handover moves everything a leaver owned in one action, and five separate 「你现在负责…」 cards for one handover is the shape people learn to ignore. Reached mostly through ``service.set_owner`` / ``service.create_flow``, which announce **by default** so no path can forget; the offboarding coordinator calls it directly because its whole handover is one card. Nightly inference reaches neither function — ``apply_inferred_owner`` writes the columns itself — so the first night after deploy does not fire one card per inferred flow at every person in the company. New recipient faces stay narrow on purpose. Without this, a flow could change hands in silence and its new owner would first learn of the responsibility from a confirmation card six months later — or from a colleague asking about a flow they had never heard of. """ live = [f for f in flows if f is not None] if not live or new_owner_id is None or new_owner_id == previous_owner_id: return 0 if actor is not None and actor.id == new_owner_id: # You do not need a card telling you what you just did. IT # registering a flow against themselves is the common case. return 0 owner = await _person_by_id(db, new_owner_id) if owner is None or owner.employment_status != "active": return 0 previous = await _person_by_id(db, previous_owner_id) handover = _handover_line( reason, svc_person_label(previous), svc_person_label(actor), ) portal = await _portal_url(db) target = f"{portal}/my/approval-flows" if portal else "/my/approval-flows" context = { "flow_summary": _flow_summary(live), "flow_list": _flow_list(live), "handover_line": handover, "cta_line": LocalizedText( zh=f"在这里看你负责的全部流程:{target}", en=f"All the flows you own: {target}", ja=f"担当フローの一覧:{target}", ), "portal_url": portal, } env = NotificationEnvelope( recipient=owner, title=( f"审批流程「{live[0].name}」现在由你负责" if len(live) == 1 else f"你现在负责 {len(live)} 条审批流程" ), body_text="", channels=["feishu"], template_key="approval_registry.owner_assigned", context=context, url="/my/approval-flows", related_resource_type="registry_approval_flow", related_resource_id=live[0].id, card_actions=[CardAction( label=LocalizedText( zh="看看我负责的审批流程", en="View the flows I own", ja="担当している承認フローを見る", ), url=target, style="primary", )], ) try: # SAVEPOINT, because this one is called from inside ``set_owner`` # rather than after the router's commit: a failed outbox flush # that is merely swallowed leaves the session poisoned, and the # NEXT statement — the caller's — dies with an error about # something it never did. Rolling back to the savepoint keeps the # ownership change intact and loses only the card. async with db.begin_nested(): await NotificationService(db).push_inbox_and_feishu(env) except Exception: logger.warning( "approval_registry: owner-assigned push failed → %s", owner.work_email, exc_info=True, ) return 0 return 1 def _flow_summary(flows: Sequence[ApprovalFlow]) -> LocalizedText | str: """What the subject line calls the thing being handed over.""" if len(flows) == 1: name = flows[0].name return LocalizedText( zh=f"审批流程「{name}」", en=f"the approval flow \u201c{name}\u201d", ja=f"承認フロー「{name}」", ) return LocalizedText( zh=f"这 {len(flows)} 条审批流程", en=f"{len(flows)} approval flows", ja=f"{len(flows)} 件の承認フロー", ) def _flow_list(flows: Sequence[ApprovalFlow]) -> LocalizedText | str: """The flows themselves. One flow gets its code and purpose spelled out; several get a list, because a wall of descriptions is not what somebody reads on a phone.""" if len(flows) == 1: flow = flows[0] desc = (flow.description or "").strip() zh = f"{flow.name}\n编号:{flow.approval_code}" en = f"{flow.name}\nCode: {flow.approval_code}" ja = f"{flow.name}\nコード:{flow.approval_code}" if desc: zh += f"\n用途:{desc}" en += f"\nPurpose: {desc}" ja += f"\n用途:{desc}" return LocalizedText(zh=zh, en=en, ja=ja) # Bounded: a leaver with thirty flows produces a readable card, not a # scroll, and the page behind the button has the rest. shown = list(flows[:const.MAX_LISTED_FLOWS]) body = "\n".join(f" · {f.name}({f.approval_code})" for f in shown) body_en = "\n".join(f" · {f.name} ({f.approval_code})" for f in shown) body_ja = body rest = len(flows) - len(shown) if rest > 0: # The overflow line is prose, so it needs its own translation — # reusing the Chinese one put 「另有 N 条」 at the bottom of an # otherwise Japanese card. body += f"\n … 另有 {rest} 条" body_en += f"\n … and {rest} more" body_ja += f"\n … ほか {rest} 件" return LocalizedText(zh=body, en=body_en, ja=body_ja)