"""Procurement reminders — daily cron handler (PR 3). Scope: one daily run that: 1. **Contract expiry** — every contract with ``status ∈ {active, signed}`` whose ``expiry_date`` falls within ``renew_notice_days`` (or ``category.default_reminder_days`` if NULL) emits one in-app notification per (contract, owner). Recipients: business_owner_id + procurement_owner_id (fallback to it_admin distribution list when both are NULL). 2. **Budget thresholds** — every active budget whose locked % rises above 80% or 95% emits one notification to owner_person_id. Edge-triggered: the highest tier already alerted is persisted in ``procurement_budgets.last_alerted_threshold``, so a pool that sits at 85% alerts once (on the run it crossed 80%) and never re-nags; the marker resets downward when utilization falls below a tier so a genuine future re-crossing alerts again. 3. **Lingering submitted POs** — every PO in ``status='submitted'`` for more than 5 days without an ``external_po_number`` filled in (Helian hasn't approved) emits one nag to procurement_owner_id. Dedup: each (kind, target_id, recipient_id) pair pushes at most one notification per ISO week. The ISO-week bucket + target id + recipient (+ threshold) are encoded into a ``coalesce_key`` string stamped into the outbox row's ``context``; before dispatching, ``_push_inbox`` looks back for an existing ``procurement.reminder.*`` row to the same recipient carrying the identical key and skips if found. (There is no ``coalesce_key`` column on ``notification_outbox`` — the dedup is a Python look-back over a tiny per-recipient row set, DB-agnostic across MySQL / TiDB.) Writes are confined to ``notification_outbox`` / ``audit_logs`` rows the dispatcher creates, plus the ``procurement_budgets.last_alerted_threshold`` edge-trigger marker (so budget alerts fire on transition, not every run). No other domain rows are mutated. """ from __future__ import annotations import logging import uuid from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any from sqlalchemy import func, select from sqlalchemy.exc import OperationalError from sqlalchemy.ext.asyncio import AsyncSession from app.domains.procurement.models import ( ProcurementBudget, ProcurementBudgetConsumption, ProcurementCategory, ProcurementContract, ProcurementOrder, ) from app.scheduled_jobs.registry import register_scheduled_job logger = logging.getLogger(__name__) # One template key per reminder kind. The three reminders share nothing # but their delivery mechanics — a contract expiry, a budget threshold and # a late PO are three different sentences — and a ``template_key`` resolves # to exactly one piece of copy, so a single shared key could never render # all three. They used to share the (unregistered) ``procurement.reminder`` # key, which is precisely why every recipient got the Chinese title/body # the call site hardcoded regardless of their language. # # ``_LEGACY_REMINDER_TEMPLATE_KEY`` stays in the dedup look-back so the # first run after this ships doesn't re-nag anyone whose same-week # reminder was written under the old key. _TEMPLATE_CONTRACT_EXPIRY = "procurement.reminder.contract_expiry" _TEMPLATE_BUDGET_THRESHOLD = "procurement.reminder.budget_threshold" _TEMPLATE_ORDER_LATE = "procurement.reminder.order_late" _LEGACY_REMINDER_TEMPLATE_KEY = "procurement.reminder" _REMINDER_TEMPLATE_KEYS: tuple[str, ...] = ( _TEMPLATE_CONTRACT_EXPIRY, _TEMPLATE_BUDGET_THRESHOLD, _TEMPLATE_ORDER_LATE, _LEGACY_REMINDER_TEMPLATE_KEY, ) @register_scheduled_job( key="procurement.reminders", display_name="采购提醒(合同到期 / 预算阈值 / 汇联易迟批)", description=( "每日 06:30 UTC 跑一次,扫描三类:(1) ``procurement_contracts`` " "里在 ``renew_notice_days``(或 category 默认值)窗口内到期的合同 " "— 推送给 business_owner + procurement_owner;(2) " "``procurement_budgets`` 里 locked 占用首次升破 80% / 95% 的预算池 — " "推送给 owner(edge-triggered,占用持平不重复打扰);(3) " "``procurement_orders`` 里 status='submitted' " "超过 5 天但 external_po_number 仍未填的 PO — 推送给 procurement_" "owner 提示汇联易侧迟批。同 (kind, target_id, recipient) 一周内只" "推送一次,避免噪音。" ), ) async def run_procurement_reminders( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: today = date.today() iso_week_key = today.strftime("%G-W%V") counts = { "contracts_scanned": 0, "contract_reminders_dispatched": 0, "budgets_scanned": 0, "budget_alerts_dispatched": 0, "orders_scanned": 0, "order_nags_dispatched": 0, } # ── (1) Contracts ──────────────────────────────────────────────── contract_rows = await db.execute( select(ProcurementContract, ProcurementCategory) .outerjoin( ProcurementCategory, ProcurementCategory.id == ProcurementContract.category_id, ) .where( ProcurementContract.expiry_date.is_not(None), ProcurementContract.status.in_(["active", "signed"]), ) ) for contract, category in contract_rows: counts["contracts_scanned"] += 1 days = contract.renew_notice_days if days is None and category is not None: days = category.default_reminder_days if days is None: continue cutoff = today + timedelta(days=days) if contract.expiry_date is None or contract.expiry_date > cutoff: continue recipients: list[uuid.UUID] = [] if contract.business_owner_id is not None: recipients.append(contract.business_owner_id) if contract.procurement_owner_id is not None: recipients.append(contract.procurement_owner_id) if not recipients: continue # no one to notify; admin can pin owners later days_until = (contract.expiry_date - today).days for r in recipients: sent = await _push_inbox( db, recipient_id=r, coalesce_key=f"procurement.contract.expiry:{contract.id}:{r}:{iso_week_key}", template_key=_TEMPLATE_CONTRACT_EXPIRY, title=f"合同即将到期:{contract.contract_number}", body=( f"{contract.title}(供应商 ID {contract.supplier_id})" f"将在 {days_until} 天后(于 {contract.expiry_date})到期。" f"如需续约请尽快与采购联系。" ), template_context={ "contract_number": contract.contract_number, "contract_title": contract.title, "supplier_id": str(contract.supplier_id), "days_until": days_until, "expiry_date": contract.expiry_date.isoformat(), }, action_url=f"/admin/procurement/contracts/{contract.id}", ) if sent: counts["contract_reminders_dispatched"] += 1 # ── (2) Budgets ────────────────────────────────────────────────── budgets = list((await db.execute( select(ProcurementBudget) .where(ProcurementBudget.is_active.is_(True)) )).scalars()) for budget in budgets: counts["budgets_scanned"] += 1 # Per-budget SAVEPOINT — _push_inbox's notification dispatch # flushes the shared session; one bad flush would leave it in # PendingRollbackError so the NEXT budget's autoflushed query # raises and the rest of the batch is silently dropped. The # savepoint isolates this budget; we log + continue on failure. try: async with db.begin_nested(): if budget.owner_person_id is None: continue locks = await db.scalar( select(func.coalesce(func.sum(ProcurementBudgetConsumption.amount), 0)) .where( ProcurementBudgetConsumption.budget_id == str(budget.id), ProcurementBudgetConsumption.event == "lock", ) ) or 0 releases = await db.scalar( select(func.coalesce(func.sum(ProcurementBudgetConsumption.amount), 0)) .where( ProcurementBudgetConsumption.budget_id == str(budget.id), ProcurementBudgetConsumption.event == "release", ) ) or 0 # Decimal money math (never float) so a pool sitting exactly # on a tier boundary alerts identically to the Decimal-based # API/UI (compute_budget_balance). locked = Decimal(locks or 0) - Decimal(releases or 0) total = Decimal(budget.total_amount or 0) if total <= 0: continue pct = (locked / total) * 100 current_tier = 95 if pct >= 95 else (80 if pct >= 80 else 0) last_tier = budget.last_alerted_threshold or 0 # Edge-triggered: alert ONLY when utilization rises above # the highest tier already alerted to this owner. A pool # that merely sits at 85% is notified once (the run it # crossed 80%), never again — no perpetual weekly nag. # When utilization falls below a previously-alerted tier # we reset the marker downward (no notification) so a # genuine future re-crossing alerts again. if current_tier <= last_tier: if current_tier < last_tier: budget.last_alerted_threshold = current_tier or None continue sent = await _push_inbox( db, recipient_id=budget.owner_person_id, coalesce_key=( f"procurement.budget.threshold:{budget.id}:" f"{current_tier}:{iso_week_key}" ), template_key=_TEMPLATE_BUDGET_THRESHOLD, title=f"预算 {current_tier}% 阈值预警:{budget.department} {budget.year}", body=( f"{budget.year} 年度 {budget.department} 部门 {budget.currency} " f"预算池已占用 {pct:.1f}%({locked:.2f} / {total:.2f})。" f"如需扩容请联系财务 / IT 管理员。" ), # Pre-format the money / percentage here rather than # in the template: a ``{pct:.1f}`` format spec in # admin-editable copy raises ValueError once the # context round-trips through JSON and comes back a # str, and ``_safe_format`` re-raises format-spec # errors (only missing KEYS degrade gracefully). template_context={ "threshold": current_tier, "department": budget.department, "year": budget.year, "currency": budget.currency, "percent": f"{pct:.1f}", "locked": f"{locked:.2f}", "total": f"{total:.2f}", }, action_url="/me/budget", ) # Advance the marker only on a real send so an unreachable # recipient (or transient dispatch failure) re-tries next # run rather than silently losing the alert forever. if sent: budget.last_alerted_threshold = current_tier counts["budget_alerts_dispatched"] += 1 except OperationalError: # Deadlock rolls back the whole transaction server-side — # propagate so the (idempotent) run restarts cleanly. raise except Exception: # noqa: BLE001 logger.exception( "procurement budget alert failed for budget %s", budget.id) continue # ── (3) Lingering POs ──────────────────────────────────────────── cutoff = datetime.now(UTC) - timedelta(days=5) submitted_pos = await db.execute( select(ProcurementOrder).where( ProcurementOrder.status == "submitted", ProcurementOrder.external_po_number.is_(None), ProcurementOrder.requested_at.is_not(None), ProcurementOrder.requested_at < cutoff, ) ) for po in submitted_pos.scalars(): counts["orders_scanned"] += 1 recipient = po.procurement_owner_id or po.applicant_id if recipient is None: continue # ``requested_at`` is ``DateTime(timezone=True)`` but the asyncmy # MySQL/TiDB round-trip strips tzinfo, so a freshly-loaded row # comes back NAIVE. Subtracting it from an aware ``datetime.now(UTC)`` # raises ``TypeError: can't subtract offset-naive and offset-aware`` # — and since this is a ``@register_scheduled_job`` handler, that # raise lands ``scheduled_job_runs.status='failed'`` and pages # on-call on every tick once a single PO is older than the cutoff. # Coerce naive→UTC first, mirroring the sibling handlers # (ai_spending_handlers / fx_rate / owner_confirmation_cards). requested_at = po.requested_at or datetime.now(UTC) if requested_at.tzinfo is None: requested_at = requested_at.replace(tzinfo=UTC) delta = datetime.now(UTC) - requested_at sent = await _push_inbox( db, recipient_id=recipient, coalesce_key=( f"procurement.order.helian_late:{po.id}:{recipient}:" f"{iso_week_key}" ), template_key=_TEMPLATE_ORDER_LATE, title=f"PO {po.order_number} 在汇联易侧已 {delta.days} 天", body=( f"PO {po.order_number} 状态仍是 submitted,external_po_number " f"未填,提示汇联易那边可能尚未批准。请采购同事跟进。" ), template_context={ "order_number": po.order_number, "days": delta.days, }, action_url=f"/admin/procurement/orders/{po.id}", ) if sent: counts["order_nags_dispatched"] += 1 logger.info("procurement.reminders ran: %s", counts) return counts async def _push_inbox( db: AsyncSession, *, recipient_id: uuid.UUID, coalesce_key: str, template_key: str, title: str, body: str, template_context: dict[str, Any], action_url: str | None = None, ) -> bool: """Dispatch one in-app + Feishu reminder, deduped per ISO week. Returns ``True`` if a notification was actually dispatched, ``False`` if it was suppressed as a same-week duplicate (so the caller's ``*_dispatched`` counter reflects real sends). Channel choice — **in-app only** (the ``notification_outbox`` row IS the NotificationCenter bell entry; no Feishu/email push). This cron has been enabled in prod (migration 224) but silently no-op'd until now; making it dispatch must not produce a Feishu/email burst on the first real run (it would fire one push per currently-qualifying contract / budget / PO at once). An empty ``channels`` list lands the reminder in the owner's bell with zero outbound noise — the owner sees it the next time they open the inbox, no interruption. ``coalesce_key`` already embeds the ISO-week bucket + target id + recipient (+ threshold), so it is unique per (kind, target, recipient, week). We stamp it into the outbox row's ``context`` and, before dispatching, look back for an existing row of the ``procurement.reminder*`` key family to the same recipient carrying the identical key — if found, this week's push already happened and we skip. The 30-day ``created_at`` floor keeps the look-back scan small and is safe because the key rotates weekly (a stale key from a prior week can never collide). DB-agnostic: the dedup is evaluated in Python over a tiny per-recipient row set rather than a JSON-path query, so it behaves identically on MySQL and TiDB. The outbox row IS the NotificationCenter inbox entry, so an empty ``channels`` list still surfaces the reminder in the owner's bell. Best-effort — a dispatch failure logs and never aborts the cron. """ from app.identity.person_models import IdentityPerson from app.notifications.models import NotificationOutbox # Per-ISO-week dedup: has this exact (kind, target, recipient, week) # already been pushed? lookback_floor = datetime.now(UTC) - timedelta(days=30) recent_ctx = ( await db.execute( select(NotificationOutbox.context).where( NotificationOutbox.recipient_person_id == recipient_id, # Match the whole key family, not just this reminder's own # key: the coalesce_key already encodes (kind, target, # recipient, week), so a family-wide look-back is exactly # as precise while ALSO seeing rows written under the # pre-split legacy key. NotificationOutbox.template_key.in_(_REMINDER_TEMPLATE_KEYS), NotificationOutbox.created_at >= lookback_floor, ) ) ).scalars().all() for ctx in recent_ctx: if ctx and ctx.get("coalesce_key") == coalesce_key: return False # already pushed this week — suppress person = await db.get(IdentityPerson, recipient_id) if person is None or not person.work_email: return False # unknown / unreachable recipient — nothing to send # Only notify active employees. A contract/budget/PO owner who has # left (or hasn't started) must not keep receiving bell reminders — # the alert would land in a dead inbox no one acts on. Matches the # employment-status guard every sibling recipient-resolving cron uses # (cloud_owner / cmdb_server_owner / asset_inventory / failure_notifier). if person.employment_status != "active": return False try: from app.notifications.service import ( NotificationEnvelope, NotificationService, ) await NotificationService(db).dispatch( NotificationEnvelope( recipient=person, title=title, body_text=body, channels=[], # in-app only — see _push_inbox docstring template_key=template_key, url=action_url, context={ # Placeholders the template interpolates. Spread # FIRST so a caller can never shadow ``coalesce_key`` # — that field is the dedup marker this function # reads back on the next run, and a template variable # colliding with it would silently disable dedup. **template_context, "source": "procurement.reminders", "coalesce_key": coalesce_key, }, related_resource_type="procurement_reminder", ), trace_id="procurement.reminders", ) return True except Exception: # noqa: BLE001 # Best-effort — a transient dispatch error must not abort the # cron or skip the remaining reminders. The dispatch path # isolates its own outbox INSERT in a savepoint, so the session # stays usable. logger.exception( "failed to dispatch procurement reminder; coalesce_key=%s", coalesce_key, ) return False