"""HR 「年假到期提醒」的 HTTP 面。 **这是平台第一次把一个通知群发面交给非 global_admin 的角色。** 口子开得很窄 且是刻意的:``require_hr_admin()``(= global_admin + hr),而且只能发这**一个** 模板、只能发给命中假期口径的员工。整个 ``/notifications`` 后台仍然是 global_admin 的地盘 —— HR 拿到的是「一个 campaign 页」,不是通知系统的钥匙。 写侧(``PUT /config`` 与 ``POST /send``)额外要求真实浏览器会话 (``require_browser_session``):一次群发会打到几百名员工,是典型的 catastrophic-if-stolen 动作,不该由一个被偷走的 PAT 触发。这也意味着**没有** MCP 发送工具 —— MCP 侧只读(见 ``mcp/tools.py`` 的 ``list_leave_expiry_reminder_candidates``)。 """ from __future__ import annotations import logging import uuid from collections.abc import Sequence from datetime import datetime, timedelta from fastapi import APIRouter, Depends, Query, Request from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.background import spawn as _spawn_background from app.core.database import get_db from app.core.dependencies import ( get_real_ip, require_browser_session, require_hr_admin, require_hr_read, ) from app.core.distributed_lock import LockBusy, distributed_lock from app.core.exceptions import ConflictError, NotFoundError from app.core.locale import normalize_lang from app.core.rbac import GLOBAL_ADMIN_ROLE, HR_ROLE, has_role from app.core.response import success_response from app.domains.hr import leave_reminder_service as svc from app.domains.hr.leave_reminder_models import ( BATCH_RESOURCE_TYPE, BATCH_STATUS_COMPLETED, BATCH_STATUS_FAILED, BATCH_STATUS_PARTIAL, BATCH_STATUS_SENDING, HrLeaveReminderBatch, ) from app.domains.hr.leave_reminder_schemas import ( ALLOWED_CHANNELS, AccountSnapshot, BatchDetailResponse, BatchRecipientRow, BatchRow, ConfigResponse, ConfigUpdateRequest, ExclusionRequest, ExclusionRow, PreviewMessageRequest, PreviewRequest, SendRequest, TestRecipientRow, TestRecipientsRequest, ) from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/hr/leave/expiry-reminder", tags=["hr-leave-reminder"]) #: 发送路径的互斥锁 key。全局一把:这个功能一次只该有一个批次在准备/投递。 _SEND_LOCK_KEY = "hr_leave_expiry_reminder:send" async def _audit_send( db: AsyncSession, request: Request, current_user: AuthUser, batch: HrLeaveReminderBatch, body: SendRequest, trace_id: str, ) -> None: """群发的审计行 —— 和批次行在同一个事务里落盘。""" await AuditService(db).log( trace_id=trace_id, actor_user_id=current_user.id, actor_type="user", action="hr.leave_expiry_reminder.send", resource_type="hr_leave_reminder_batch", resource_id=str(batch.id), status="success", detail={ "test_mode": batch.test_mode, "channels": batch.channels, "recipient_count": batch.recipient_count, "account_count": batch.account_count, "at_risk_days": float(batch.at_risk_days or 0), "filters": body.filters.model_dump(mode="json"), "allow_resend": body.allow_resend, "explicit_selection": bool(body.person_ids), }, ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) # ── 配置(文案 / CTA / 默认口径) ──────────────────────────────────────── @router.get("/config", summary="年假到期提醒 · 页面初始化(文案+码表+覆盖面)") async def get_config( request: Request, current_user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): copy, overridden = await svc.effective_copy(db) options, years = await svc.list_account_type_options(db) active, covered = await svc.count_active_and_covered(db) payload = ConfigResponse( template_key=svc.TEMPLATE_KEY, template_copy=copy, builtin_template_copy=svc.builtin_copy(), copy_overridden=overridden, allowed_placeholders=list(svc.ALLOWED_PLACEHOLDERS), auto_placeholders=list(svc.AUTO_PLACEHOLDERS), hris_url=await svc.get_hris_url(db), resend_window_days=await svc.get_resend_window_days(db), signoff=await svc.get_signoff(db), builtin_signoff=svc.default_signoff(), defaults=await svc.get_default_filters(db), default_channels=await svc.get_default_channels(db), account_types=options, fiscal_years=years, as_of_date=svc.today_sh(), active_person_count=active, persons_with_leave_data=covered, ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.put("/config", summary="更新文案 / HRIS 入口 / 默认口径") async def update_config( request: Request, body: ConfigUpdateRequest, current_user: AuthUser = Depends(require_hr_admin()), _gate: None = Depends(require_browser_session), db: AsyncSession = Depends(get_db), ): """HR 自助改文案。 走 ``platform_settings`` 的 ``email_template.`` 覆盖层 —— 和 global_admin 的邮件模板编辑器**同一个存储**,所以运行时 ``_load_template`` 不需要认识这个功能。写入前逐字段校验占位符,详见 ``leave_reminder_service.validate_copy``。 """ from app.integrations.service import PlatformSettingService # 投递中不许改文案。outbox 行存的是 ``(see template)`` 哨兵,正文在投递时 # 按**当时**的覆盖层渲染 —— drain 跑到一半改一版,同一批人前后会收到两种 # 说法(而 ``copy_snapshot`` 只是给 HR 事后审计看的,收件人拿不到它)。 if body.template_copy is not None or body.reset_copy: in_flight = (await db.execute( select(HrLeaveReminderBatch.id).where( HrLeaveReminderBatch.status == BATCH_STATUS_SENDING, HrLeaveReminderBatch.finished_at.is_(None), ).limit(1) )).scalars().first() if in_flight is not None: raise ConflictError( f"A reminder batch ({in_flight}) is still sending. Editing the " "copy now would give the same batch's recipients two different " "versions — wait for it to finish.", ) settings = PlatformSettingService(db) trace_id = str(getattr(request.state, "trace_id", "") or "") changed: list[str] = [] if body.reset_copy: from app.integrations.models import PlatformSetting row = await db.get(PlatformSetting, svc.TEMPLATE_SETTING_KEY) if row is not None: await settings.delete( svc.TEMPLATE_SETTING_KEY, trace_id=trace_id, actor_user_id=current_user.id, ) changed.append("copy_reset") elif body.template_copy is not None: svc.validate_copy(body.template_copy) await settings.upsert( svc.TEMPLATE_SETTING_KEY, body.template_copy.model_dump(), "年假到期提醒文案(HR 自助维护)", trace_id=trace_id, actor_user_id=current_user.id, ) changed.append("copy") if body.hris_url is not None: await settings.upsert( svc.SETTING_HRIS_URL, svc.validate_hris_url(body.hris_url), "年假到期提醒卡片上的 HRIS 请假入口", trace_id=trace_id, actor_user_id=current_user.id, ) changed.append("hris_url") if body.resend_window_days is not None: await settings.upsert( svc.SETTING_RESEND_WINDOW, int(body.resend_window_days), "多少天内收到过提醒的人默认不再入选", trace_id=trace_id, actor_user_id=current_user.id, ) changed.append("resend_window_days") if body.signoff is not None: await settings.upsert( svc.SETTING_SIGNOFF, svc.validate_signoff(body.signoff), "年假到期提醒卡片的结尾寄语(随机 / 固定一句)", trace_id=trace_id, actor_user_id=current_user.id, ) changed.append("signoff") if body.defaults is not None: await settings.upsert( svc.SETTING_DEFAULTS, body.defaults.model_dump(mode="json"), "年假到期提醒的默认圈人口径", trace_id=trace_id, actor_user_id=current_user.id, ) changed.append("defaults") if body.default_channels is not None: picked = [ c for c in dict.fromkeys(body.default_channels) if c in ALLOWED_CHANNELS ] await settings.upsert( svc.SETTING_DEFAULT_CHANNELS, picked or list(ALLOWED_CHANNELS), "年假到期提醒的默认渠道", trace_id=trace_id, actor_user_id=current_user.id, ) changed.append("default_channels") await AuditService(db).log( trace_id=trace_id, actor_user_id=current_user.id, actor_type="user", action="hr.leave_expiry_reminder.config_update", resource_type="hr_leave_reminder_config", resource_id=svc.TEMPLATE_KEY, status="success", detail={"changed": changed}, ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) await db.commit() copy, overridden = await svc.effective_copy(db) return success_response( data={ "changed": changed, "template_copy": copy.model_dump(), "copy_overridden": overridden, }, request=request, ) # ── 圈人预览 ──────────────────────────────────────────────────────────── @router.post("/preview", summary="按口径圈人(不发送)") async def preview_cohort( request: Request, body: PreviewRequest, # 刻意比 ``require_hr_read`` 窄:这个响应是**全量**逐人 PII(姓名/邮箱/部门 # + 每个假期账户的授予·已用·余额·清零日),不分页。发送本来就只有 # global_admin + hr 能做,把整份名册也开给 auditor / management 没有业务 # 动机,而洞察板恰恰有一条「小组下钻要防止还原到个人」的隐私闸。 current_user: AuthUser = Depends(require_hr_admin()), db: AsyncSession = Depends(get_db), ): """无副作用的名单预演 —— 「这个口径命中多少人、各剩几天」。 返回**完整**名单(组织约 500 人,有假期账户的约 280 人),不分页:发送时 要用的正是这份 id 集合,分页会让「我勾掉了 3 个人」这件事变得不可靠。 """ payload = await svc.preview(db, body.filters) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post("/preview-message", summary="用真实收件人的数字渲染一份成品") async def preview_message( request: Request, body: PreviewMessageRequest, # 同 ``/preview``:渲染成品会带上一个真实收件人的姓名与余额。 current_user: AuthUser = Depends(require_hr_admin()), db: AsyncSession = Depends(get_db), ): """所见即所得:渲染走的是发送路径的同一套原语。 ``template_copy`` 传了就渲染这份草稿(HR 边改边看,不必先保存),没传就用 当前生效文案。 """ cohort = await svc.resolve_cohort(db, body.filters) target = None if body.person_id is not None: target = next( (c for c in cohort.candidates if str(c.person_id) == str(body.person_id)), None, ) if target is None: raise NotFoundError( "Person", f"{body.person_id} is not in the current cohort", ) elif cohort.candidates: target = cohort.candidates[0] copy = body.template_copy if copy is None: copy, _ = await svc.effective_copy(db) else: svc.validate_copy(copy) hris_url = await svc.get_hris_url(db) # 预览用的寄语。固定模式下就是真实会发的那句;随机模式下这是**预览专用**的 # 一次抽取(真实发送按批次 id 另抽),前端据 ``signoff.mode`` 如实说明。 signoff_config = await svc.get_signoff(db) signoff_line = svc.pick_signoff(signoff_config, seed="preview") today = svc.today_sh() # 与投递路径同一个来源(``NotificationService._resolve_absolute_url`` 读的 # 就是这个 platform setting),所以预览里的链接和员工真正收到的一致;没配 # 就退回相对路径,和投递路径的降级行为也一致。 from app.integrations.service import get_platform_setting base = await get_platform_setting( "app.frontend_base_url", default=None, session=db, ) portal_url = ( f"{str(base).rstrip('/')}{svc.MY_LEAVE_PATH}" if isinstance(base, str) and base else svc.MY_LEAVE_PATH ) if target is None: # 名单是空的(口径太窄 / 数据还没同步)。仍然给 HR 一份示例渲染, # 让他能继续打磨文案 —— 但 ``is_sample`` 如实标注这不是真人数据。 sample_deadline = today + timedelta( days=body.filters.expiring_within_days or 30, ) sample = [AccountSnapshot( account_id=uuid.uuid4(), account_type=body.filters.account_types[0], fiscal_year=today.year - 1, balance_days=5.0, entitled_days=5.0, used_days=0.0, deadline=sample_deadline, expires_in_days=(sample_deadline - today).days, forfeits=True, )] lang: str = normalize_lang(body.language) or "zh" sample_names = { "zh": "示例员工", "en": "Sample Employee", "ja": "サンプル社員", } context = svc.build_context( display_name=sample_names.get(lang, sample_names["zh"]), accounts=sample, lang=lang, hris_url=hris_url, today=today, ) rendered = svc.render_message( copy=copy, context=context, lang=lang, display_name=None, person_id=None, accounts=sample, hris_url=hris_url, portal_url=portal_url, signoff=svc.signoff_text(signoff_line, lang), ) return success_response( data={**rendered.model_dump(mode="json"), "is_sample": True}, request=request, ) # 不指定语言就按这个收件人自己的语言渲染 —— HR 想确认「日本同事会收到什么」 # 时再显式传 language。 lang = normalize_lang(body.language) or target.language context = svc.build_context( display_name=target.display_name, accounts=target.accounts, lang=lang, hris_url=hris_url, today=today, ) rendered = svc.render_message( copy=copy, context=context, lang=lang, display_name=target.display_name, person_id=target.person_id, accounts=target.accounts, hris_url=hris_url, portal_url=portal_url, signoff=svc.signoff_text(signoff_line, lang), ) return success_response( data={**rendered.model_dump(mode="json"), "is_sample": False}, request=request, ) @router.get("/exclusions", summary="默认不勾选名单") async def list_exclusions( request: Request, # 逐人 PII(姓名 / 邮箱 / 部门 + HR 写的排除理由),与圈人预览同一档。 current_user: AuthUser = Depends(require_hr_admin()), db: AsyncSession = Depends(get_db), ): """HR 维护的长期排除决定。命中的人**照常出现在名单里**,只是默认不勾选。""" rows = await svc.list_exclusions(db) actor_ids = [str(r.created_by) for r, _ in rows if r.created_by] # 空列表必须**跳过查询**,不能靠 ``.in_(x or [""])`` 兜底 —— ``GUID`` 类型的 # bind 会拿 ``""`` 去构造 UUID 并当场 ``ValueError``,整个接口 500。 # (名单为空是这个页面最常见的状态,恰好也是最容易漏测的那个。) actors: dict[str, IdentityPerson] = {} if actor_ids: actors = { str(p.id): p for p in (await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(actor_ids)) )).scalars().all() } out = [ ExclusionRow( person_id=r.person_id, display_name=( (person.display_name or person.work_email or str(r.person_id)) if person else str(r.person_id) ), work_email=person.work_email if person else None, employee_no=person.employee_no if person else None, department=None, note=r.note, created_by_name=( (actors[str(r.created_by)].display_name or None) if r.created_by and str(r.created_by) in actors else None ), created_at=r.created_at, ).model_dump(mode="json") for r, person in rows ] return success_response(data={"items": out, "total": len(out)}, request=request) @router.post("/exclusions", summary="加入默认不勾选名单") async def add_exclusion( request: Request, body: ExclusionRequest, current_user: AuthUser = Depends(require_hr_admin()), # 写侧与群发同档:被偷的 PAT 不该能悄悄把某个人从每一次提醒里拿掉。 _gate: None = Depends(require_browser_session), db: AsyncSession = Depends(get_db), ): trace_id = getattr(request.state, "trace_id", "") or "" row = await svc.set_exclusion( db, person_id=body.person_id, note=body.note, actor_person_id=current_user.id, ) await AuditService(db).log( trace_id=trace_id, actor_user_id=current_user.id, actor_type="user", action="hr.leave_expiry_reminder.exclusion_set", resource_type="identity_person", resource_id=str(body.person_id), status="success", detail={"note": row.note}, ) await db.commit() return success_response(data={"person_id": str(body.person_id)}, request=request) @router.delete("/exclusions/{person_id}", summary="移出默认不勾选名单") async def delete_exclusion( request: Request, person_id: uuid.UUID, current_user: AuthUser = Depends(require_hr_admin()), _gate: None = Depends(require_browser_session), db: AsyncSession = Depends(get_db), ): removed = await svc.remove_exclusion(db, person_id) if not removed: raise NotFoundError("HrLeaveReminderExclusion", str(person_id)) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "") or "", actor_user_id=current_user.id, actor_type="user", action="hr.leave_expiry_reminder.exclusion_cleared", resource_type="identity_person", resource_id=str(person_id), status="success", ) await db.commit() return success_response(data={"person_id": str(person_id)}, request=request) @router.post("/test-recipients", summary="试发预检:这几个人各自会收到什么") async def preflight_test_recipients( request: Request, body: TestRecipientsRequest, # 同 ``/preview``:响应里带真实姓名与余额。 current_user: AuthUser = Depends(require_hr_admin()), db: AsyncSession = Depends(get_db), ): """无副作用。**和发送路径共用** :func:`svc.resolve_test_recipients`。 存在的理由:试发的选择器搜的是全员通讯录(名单外的同事也该能被搜到、能收到 样信),所以「这个人能不能收到」只有服务端知道。把判定放在发送前的这一次 调用里,HR 在点「发送」之前就看到逐人结论,而不是点完弹一个 422。 """ resolved = await svc.resolve_test_recipients( db, filters=body.filters, person_ids=body.person_ids, actor_person_id=current_user.id, ) rows = [ TestRecipientRow( person_id=r.person_id, display_name=r.display_name, work_email=r.work_email, source=r.source, reason=r.reason, total_balance_days=r.row.total_balance_days if r.row else None, earliest_deadline=r.row.earliest_deadline if r.row else None, expires_in_days=r.row.expires_in_days if r.row else None, language=r.row.language if r.row else None, ) for r in resolved ] return success_response( data={ "recipients": [r.model_dump(mode="json") for r in rows], "max_recipients": svc.MAX_TEST_RECIPIENTS, "sendable_count": sum( 1 for r in rows if r.source != svc.TEST_SOURCE_BLOCKED ), }, request=request, ) # ── 发送 ──────────────────────────────────────────────────────────────── @router.post("/send", summary="群发年假到期提醒") async def send_reminders( request: Request, body: SendRequest, current_user: AuthUser = Depends(require_hr_admin()), _gate: None = Depends(require_browser_session), db: AsyncSession = Depends(get_db), ): """创建批次 → 立即返回 → 后台投递。 批次行(含收件人与数字快照)在**提交事务后**才把投递交给后台任务: ``enqueue_then_drain`` 会 commit,绝不能在请求持有的事务里跑,更不能在 N 次外部 HTTP 调用期间攥着 N 个 ``identity_persons`` 行锁。 """ trace_id = str(getattr(request.state, "trace_id", "") or "") # 「同一时刻只有一个批次」必须真的被强制。``assert_no_batch_in_flight`` 是 # 纯 SELECT,两个几百毫秒内到达的请求(= 双击的物理形状)会双双通过、各建 # 一个批次、各自投递,几百名员工每人两张卡。所以检查 + 建行整段包在一把 # Redis 锁里,并且这也是 CLAUDE.md 规则 5(状态变更必须幂等)在这条路径上 # 的落地方式。 try: async with distributed_lock(_SEND_LOCK_KEY, ttl=120): if not body.test_mode: await svc.assert_no_batch_in_flight(db) batch = await svc.create_batch( db, filters=body.filters, channels=body.channels, person_ids=body.person_ids, expected_recipient_count=body.expected_recipient_count, allow_resend=body.allow_resend, test_mode=body.test_mode, actor_person_id=current_user.id, trace_id=trace_id, ) await _audit_send(db, request, current_user, batch, body, trace_id) await db.commit() except LockBusy as exc: raise ConflictError( "Another send is being prepared right now. Wait a moment and " "check the send history before trying again.", ) from exc batch_id = batch.id _spawn_background( svc.run_batch_in_background(batch_id), name="hr_leave.expiry_reminder.send", ) return success_response( data={ "batch_id": str(batch_id), "status": batch.status, "recipient_count": batch.recipient_count, "test_mode": batch.test_mode, }, request=request, ) # ── 发送记录 ──────────────────────────────────────────────────────────── def _batch_row(batch: HrLeaveReminderBatch, creator_name: str | None) -> BatchRow: return BatchRow( id=batch.id, status=batch.status, test_mode=batch.test_mode, channels=list(batch.channels or []), # ``_hris_url`` 是内部快照字段,不进 HR 的口径展示。 filters={ k: v for k, v in (batch.filters or {}).items() if not k.startswith("_") }, recipient_count=batch.recipient_count, account_count=batch.account_count, at_risk_days=( float(batch.at_risk_days) if batch.at_risk_days is not None else None ), enqueued=batch.enqueued, delivered=batch.delivered, failed=batch.failed, error_summary=batch.error_summary, created_by_person_id=( uuid.UUID(batch.created_by_person_id) if batch.created_by_person_id else None ), created_by_name=creator_name, created_at=batch.created_at, started_at=batch.started_at, finished_at=batch.finished_at, ) @router.get("/batches", summary="发送记录") async def list_batches( request: Request, include_test: bool = Query(default=False), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), current_user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): # 顺手收尸:``hr.leave_reminder_reap`` 每 20 分钟跑一次,但 HR 刷这个页面的 # 时刻恰恰是「僵死批次」最该被说清楚的时刻,所以这里也跑一次(幂等,只动 # 心跳已停的行)。**只在 hr_admin 调用时做** —— auditor / management 按设计 # 只读,不该由一个 GET 触发状态迁移 + commit。 # best-effort:收尸失败绝不该让发送记录页打不开。 if has_role(current_user.role_names, GLOBAL_ADMIN_ROLE, HR_ROLE): # savepoint-exempt: 只读端点上的机会性维护,失败即 rollback 后照常返回; # 不 rollback 会让后面的列表查询以 PendingRollbackError 爆掉。 try: await svc.reap_stale_batches(db) except Exception: # noqa: BLE001 logger.warning( "leave_expiry_reminder: opportunistic reap failed", exc_info=True, ) await db.rollback() q = select(HrLeaveReminderBatch) count_q = select(func.count()).select_from(HrLeaveReminderBatch) if not include_test: q = q.where(HrLeaveReminderBatch.test_mode.is_(False)) count_q = count_q.where(HrLeaveReminderBatch.test_mode.is_(False)) total = int((await db.execute(count_q)).scalar() or 0) rows = (await db.execute( q.order_by(HrLeaveReminderBatch.created_at.desc()) .offset((page - 1) * page_size).limit(page_size) )).scalars().all() creators = await _creator_names(db, rows) return success_response( data=[ _batch_row(b, creators.get(str(b.created_by_person_id))).model_dump( mode="json", ) for b in rows ], meta={"page": page, "page_size": page_size, "total": total}, request=request, ) async def _creator_names( db: AsyncSession, batches: Sequence[HrLeaveReminderBatch], ) -> dict[str, str]: ids = [ b.created_by_person_id for b in batches if b.created_by_person_id ] if not ids: return {} rows = (await db.execute( select(IdentityPerson.id, IdentityPerson.display_name).where( IdentityPerson.id.in_(ids), ) )).all() return {str(pid): name or "" for pid, name in rows} @router.get("/batches/{batch_id}", summary="某次发送的明细 + 逐人投递结果") async def get_batch( batch_id: uuid.UUID, request: Request, # 与 ``/preview`` 同样刻意窄于 ``require_hr_read``:这个响应逐人回姓名、 # 工作邮箱、精确余额与清零日(一批最多 1000 人),正是 ``/preview`` 对 # auditor / management 关掉的那类数据。前门关了后门开着没有意义 —— # 他们要查「发过什么」有聚合的 ``GET /batches`` 和 send 的审计行。 current_user: AuthUser = Depends(require_hr_admin()), db: AsyncSession = Depends(get_db), ): """逐人投递结果直接读 outbox —— 状态、渠道、错误、已读都在那里。 刻意不另建 per-recipient 表:投递状态有两份就一定会有一天对不上,而对不上 的那一天,没人知道该信哪份。 """ from app.notifications.models import NotificationOutbox batch = await db.get(HrLeaveReminderBatch, batch_id) if batch is None: raise NotFoundError("HrLeaveReminderBatch", str(batch_id)) creators = await _creator_names(db, [batch]) rows = (await db.execute( select(NotificationOutbox, IdentityPerson.display_name) .outerjoin( IdentityPerson, IdentityPerson.id == NotificationOutbox.recipient_person_id, ) .where( NotificationOutbox.related_resource_type == BATCH_RESOURCE_TYPE, NotificationOutbox.related_resource_id == str(batch_id), ) .order_by(NotificationOutbox.created_at) )).all() # 发送时冻结的数字:从批次快照里按 person_id 取,用来回答「他当时被告知 # 剩几天」。outbox 的 context 里也有,但快照是 HR 点确认时看到的那一份。 snapshot_by_person = { str(r.get("person_id")): r for r in (batch.recipients_snapshot or []) } recipients: list[BatchRecipientRow] = [] for outbox, display_name in rows: snap = snapshot_by_person.get(str(outbox.recipient_person_id), {}) recipients.append(BatchRecipientRow( outbox_id=outbox.id, person_id=outbox.recipient_person_id, display_name=display_name or snap.get("display_name"), work_email=outbox.recipient_email or snap.get("work_email"), status=outbox.status, channels=list(outbox.channels or []), delivered_channels=list(outbox.delivered_channels or []), failed_channels=list(outbox.failed_channels or []), last_error=outbox.last_error, attempts=outbox.attempts, delivered_at=outbox.delivered_at, read_at=outbox.read_at, created_at=outbox.created_at, balance_days=( float(snap["total_balance_days"]) if snap.get("total_balance_days") is not None else None ), deadline=snap.get("earliest_deadline"), )) payload = BatchDetailResponse( batch=_batch_row(batch, creators.get(str(batch.created_by_person_id))), copy_snapshot=batch.copy_snapshot, recipients=recipients, ) return success_response(data=payload.model_dump(mode="json"), request=request) @router.post( "/batches/{batch_id}/resend-failed", summary="只对这批里投递失败的人再发一次", ) async def resend_failed( batch_id: uuid.UUID, request: Request, current_user: AuthUser = Depends(require_hr_admin()), _gate: None = Depends(require_browser_session), db: AsyncSession = Depends(get_db), ): """复用通知域的 ``resend_outbox_rows``,只挑这一批里**确定失败**的行。 判据是 ``status IN (failed, dead)``,**不是** ``!= delivered`` —— 后者会 连带克隆三类「其实还在路上」的行,每一类都等于给员工发第二遍: * ``pending`` + 静默时段:命中免打扰的收件人被 park 成 pending 等窗口结束, HR 看着像失败 → 点重发 → 当场击穿他自己设的静默时段,窗口到点后重试 worker 再投一次原行。 * ``pending``,只是这批还没 drain 到:``/send`` 立即返回、投递在后台跑几十 分钟,期间未处理的行全是 pending。 * ``delivering`` / ``delivery_uncertain``:provider 可能已经接受了。 ``resend_outbox_rows`` 的契约明确写着「clones every resolvable row **regardless of source status**」,所以过滤责任在调用方这一侧。 另外要求批次已收尾(``finished_at``),否则「重发」和还在跑的 drain 会互相 叠加。已送达的人永远不会被再打扰 —— 这也是这里刻意**不**提供「整批重发」 的原因:要再催一轮就重新走一次圈人流程,那样数字才是新的。 """ from app.notifications.models import ( STATUS_DEAD, STATUS_DELIVERED, STATUS_FAILED, STATUS_PARTIAL, NotificationOutbox, ) from app.notifications.service import NotificationService batch = await db.get(HrLeaveReminderBatch, batch_id) if batch is None: raise NotFoundError("HrLeaveReminderBatch", str(batch_id)) if batch.finished_at is None: raise ConflictError( "This batch is still sending. Wait for it to finish — rows that " "have not been attempted yet are not failures, and resending them " "now would deliver the same card twice.", ) # 只挑**这个人最新一行仍未送达**的那些失败行。两个都要: # ① 一个人可能有多条陈旧的 failed 行(前几轮重发留下的源行 —— ``_resend_one`` # 从不改源行的状态),不去重的话一次点击就给他克隆好几封; # ② 一个已经靠上一次重发送达的人,他那条陈旧 failed 源行还在,再点一次就会 # 再给他发一封。 tally_rows = (await db.execute( select( NotificationOutbox.id, NotificationOutbox.recipient_person_id, NotificationOutbox.status, NotificationOutbox.created_at, ).where( NotificationOutbox.related_resource_type == BATCH_RESOURCE_TYPE, NotificationOutbox.related_resource_id == str(batch_id), ) )).all() latest_status: dict[str, tuple] = {} failed_rows: dict[str, uuid.UUID] = {} for row_id, pid, status, created_at in tally_rows: key = str(pid) rank = (created_at or datetime.min, 1 if status == STATUS_DELIVERED else 0) if key not in latest_status or rank > latest_status[key][0]: latest_status[key] = (rank, status) if status in (STATUS_FAILED, STATUS_DEAD): # 同一个人多条失败行只取一条重发。 failed_rows.setdefault(key, row_id) ids = [ row_id for key, row_id in failed_rows.items() if latest_status[key][1] not in (STATUS_DELIVERED, STATUS_PARTIAL) ] trace_id = str(getattr(request.state, "trace_id", "") or "") result = {"resent": 0, "skipped": 0, "new_ids": [], "skipped_detail": []} if ids: result = await NotificationService(db).resend_outbox_rows( ids, actor_user_id=current_user.id, trace_id=trace_id, ) # **无论有没有重发**都按 outbox 真值重算台账。``_resend_one`` 的语义是「克隆 # 一行新的」而不是改旧行,所以源行永久停在 ``failed``;不重算的话发送记录页会 # 永远显示「失败 N」、「只重发失败项」按钮永远亮着,HR 每点一次员工就多收一封。 # 而「一个都不用重发」恰恰是台账最需要被纠正的那种情况 —— 上一次重发已经把人 # 送到了,只有这张表还在说失败。``batch_delivery_tally`` 按**人**取最新一行。 fresh = await db.get(HrLeaveReminderBatch, batch_id, populate_existing=True) if fresh is not None: svc.apply_tally(fresh, await svc.batch_delivery_tally(db, batch_id)) # 从 ``failed`` 也要能翻回来:集成挂了 ⇒ delivered 0 ⇒ 状态 failed; # HR 修好飞书、点「只重发失败项」、全部送达 —— 若只认 partial→completed, # 页面会继续显示「失败」+ 那条红色「No recipient could be delivered」, # HR 很可能据此把整轮圈人重跑一遍。 if fresh.failed == 0 and fresh.status in ( BATCH_STATUS_PARTIAL, BATCH_STATUS_FAILED, ): fresh.status = BATCH_STATUS_COMPLETED fresh.error_summary = None await AuditService(db).log( trace_id=trace_id, actor_user_id=current_user.id, actor_type="user", action="hr.leave_expiry_reminder.resend_failed", resource_type="hr_leave_reminder_batch", resource_id=str(batch_id), status="success", detail={"candidates": len(ids), "resent": result.get("resent", 0)}, ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) await db.commit() return success_response(data=result, request=request) __all__ = ["router"]