"""Scheduled-job handlers for asset inventory (盘点). Two handlers: * ``asset.inventory_auto_start`` — fires twice a year (cron: Jan 15 + Jul 15, 09:00 CST). Creates a new ``AssetInventoryCycle`` with empty scope (= company-wide) and starts it. Disabled at seed time so the user explicitly opts in before the half-yearly automation kicks in. * ``asset.inventory_reminder`` — runs hourly. Walks active cycles whose ``started_at + 1d <= now()`` and re-pushes a Feishu card to recipients with pending items, gated by **recipient-local working hours** (Mon–Fri 10:00–17:59 in ``IdentityPerson.country`` → IANA zone, default Asia/Shanghai). Per-recipient daily dedup runs against ``reminded_at`` so the hourly cron only delivers once per local business day. * ``asset.inventory_launch_reconcile`` — runs hourly, seeded ENABLED (unlike the two above, which the admin opts into): re-enqueues launch cards whose ``notification_outbox`` row was never written — the HTTP start paths fan out via ``background.spawn`` in the web container, so a deploy mid-enqueue silently drops every holder not yet recorded, and the outbox-retry worker can't see rows that don't exist. No-op when no active cycle is past the 30-min grace window. Handlers are intentionally thin — orchestration only. Domain logic lives in ``AssetInventoryService``. """ from __future__ import annotations import logging from datetime import UTC, datetime from zoneinfo import ZoneInfo # Cron descriptions in this module reference Asia/Shanghai wall-clock # time (e.g. "1月15日 09:00 CST"). Naming a cycle off ``datetime.now(UTC)`` # during the late-night UTC window when CST is already in the next day # can drop the year/month label by one (verified DST-style boundary # off-by-one). Anchor the label to the same TZ the schedule documents. _CN_TZ = ZoneInfo("Asia/Shanghai") from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.scheduled_jobs.registry import register_scheduled_job logger = logging.getLogger(__name__) @register_scheduled_job( key="asset.inventory_auto_start", display_name="资产盘点 自动触发", description=( "每年 1 月 15 日和 7 月 15 日 09:00 (CST) 自动创建并启动一个" "全公司范围的资产盘点周期。第一次启用前请先用 /admin/asset-inventory " "上的手动按钮跑通完整流程; 默认 disabled。" ), ) async def run_inventory_auto_start( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.asset.inventory_service import AssetInventoryService name_template = config.get("name_template") or "半年度资产盘点 {YYYY}-{MM}" now = datetime.now(_CN_TZ) name = name_template.replace( "{YYYY}", f"{now.year:04d}", ).replace( "{MM}", f"{now.month:02d}", ) service = AssetInventoryService(db) cycle = await service.create_cycle( name=name, description=( "由系统定时任务自动触发的半年度盘点。范围: 全公司活跃资产。" ), scope_filter=None, due_at=None, created_by=None, # system-triggered ) from app.core.exceptions import ConflictError, format_exception_for_log try: cycle = await service.start_cycle( cycle.id, actor_user_id=None, trace_id="asset.inventory_auto_start", ) except ConflictError as exc: # "No eligible assets" is the only legitimate ConflictError path # here (fresh install / empty fixture). Log + return ok-shape so # operators see it on the run history but no oncall page fires. logger.warning( "asset.inventory_auto_start: start_cycle ConflictError: %s", exc, ) return { "cycle_id": str(cycle.id), "name": cycle.name, "started": False, "error": format_exception_for_log(exc), } except Exception as exc: # noqa: BLE001 # Real failures (FK violation, DB outage) MUST raise — this cron # runs only twice a year, so a silent fail = inventory window # missed for 6 months. raise RuntimeError( "asset.inventory_auto_start: start_cycle failed: " f"{format_exception_for_log(exc)}" ) from exc # ``start_cycle`` now performs only the (fast, all-DB) state # transition. Commit it so the cycle-row lock is released, then fan out # the launch cards inline — the worker tolerates the minutes-long drain # that the HTTP paths push to background.spawn. A notification hiccup # must NOT fail the twice-a-year cron: rows enqueued in phase 1 are # durably PENDING and the notification.outbox_retry worker backstops # anything this drain doesn't reach. await db.commit() recipients = 0 try: recipients = await service.dispatch_start_notifications( cycle.id, actor_user_id=None, trace_id="asset.inventory_auto_start", ) except Exception as exc: # noqa: BLE001 logger.exception( "asset.inventory_auto_start: launch fan-out degraded for " "cycle %s — pending outbox rows will be delivered by " "notification.outbox_retry: %s", cycle.id, exc, ) return { "cycle_id": str(cycle.id), "name": cycle.name, "total_items": cycle.total_items, "started": True, "recipients_notified": recipients, } @register_scheduled_job( key="asset.inventory_reminder", display_name="资产盘点 自动提醒", description=( "每小时检查活跃盘点周期; 对开始满 1 天且仍未响应的接收人, " "在 [b]其所在国家的工作时间[/b] (周一至周五 10:00–18:00 本地) " "每天推送一次飞书提醒。时区由 identity_persons.country 解析, " "未知国家回退 Asia/Shanghai。Per-recipient 当地日去重, 同一人" "同一本地日不会重复打扰。默认 disabled — 由 IT 与 auto_start " "一起手动启用。" ), ) async def run_inventory_reminder( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.asset.inventory_service import AssetInventoryService service = AssetInventoryService(db) cycles = await service.cycles_due_for_reminder() if not cycles: return {"cycles_checked": 0, "reminders_sent": 0} total_recipients = 0 failures = 0 for cycle in cycles: # ``send_cycle_reminder`` commits per cycle, deliberately (see the # comment below about not rolling back the cycles that succeeded). # A savepoint would be released by that commit and the block exit # would fail on a freed one; PLATFORM.md #25 calls this must-not-wrap. # savepoint-exempt: send_cycle_reminder commits per cycle try: outcome = await service.send_cycle_reminder( cycle.id, actor_user_id=None, trace_id="asset.inventory_reminder", working_hours_only=True, ) total_recipients += outcome.recipients_notified except Exception as exc: # noqa: BLE001 failures += 1 logger.exception( "asset.inventory_reminder: cycle %s failed: %s", cycle.id, exc, ) continue # Per the scheduled-job handler contract: raise on failure so # ``scheduled_job_runs.status='failed'`` and the in-app failure # notifier fires. A partial loop where *some* cycles succeeded # but at least one threw is still a degraded state worth paging. if failures: # Commit the cycles that DID succeed before raising. The executor # only commits the work session on the success path; an unguarded # raise here would roll back every successful cycle's reminded_at # dedup stamp + outbox rows, so the next tick re-notifies the # healthy cycles AND fails the same one again — a permanent # failing loop that double-sends reminders. Persisting first means # we page on the partial failure without losing the good work. await db.commit() raise RuntimeError( f"asset.inventory_reminder: {failures}/{len(cycles)} cycles " f"failed (see logger.exception above for the per-cycle stacks)" ) return { "cycles_checked": len(cycles), "reminders_sent": total_recipients, } @register_scheduled_job( key="asset.inventory_launch_reconcile", display_name="资产盘点 启动通知补偿", description=( "每小时扫描已开始超过宽限期 (config.grace_minutes, 默认 30) 的活跃" "盘点周期: 对仍有 pending 项、但 notification_outbox 里**完全没有**" "启动通知记录的持有人, 幂等补发启动卡片。堵的洞: HTTP 启动路径的" "群发跑在 web 容器的 background.spawn 里, enqueue 阶段被部署重启" "打断时, 未入库的持有人一张卡都不会有 — outbox_retry 只能重投" "已存在的行, 而重点「开始」会撞 active 状态 409。默认 enabled — " "没有活跃周期时是空转 no-op。" ), ) async def run_inventory_launch_reconcile( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.domains.asset.inventory_service import AssetInventoryService try: grace_minutes = int(config.get("grace_minutes") or 30) except (TypeError, ValueError): grace_minutes = 30 service = AssetInventoryService(db) cycles = await service.cycles_for_launch_reconcile( grace_minutes=grace_minutes, ) if not cycles: return {"cycles_checked": 0, "recipients_reenqueued": 0} total = 0 failures = 0 for cycle in cycles: try: total += await service.reconcile_launch_notifications( cycle.id, actor_user_id=None, trace_id="asset.inventory_launch_reconcile", ) except Exception as exc: # noqa: BLE001 failures += 1 # Roll back so the shared session is usable for the next # cycle (asyncmy raises PendingRollbackError otherwise). await db.rollback() logger.exception( "asset.inventory_launch_reconcile: cycle %s failed: %s", cycle.id, exc, ) continue if failures: # Persist the cycles that DID reconcile before raising — same # partial-failure contract as asset.inventory_reminder: page on # the failure without rolling back the good work. await db.commit() raise RuntimeError( f"asset.inventory_launch_reconcile: {failures}/{len(cycles)} " f"cycles failed (see logger.exception above)" ) return { "cycles_checked": len(cycles), "recipients_reenqueued": total, } @register_scheduled_job( key="asset.borrow_overdue", display_name="资产借出超期 提醒", description=( "每日 09:30 (UTC) 扫描所有 status='borrowed' 且 expected_return_date < " "今天的资产 — 给原归属人 + 借用人各发一条 Feishu 卡片提醒归还。" "借用归还表单 2026-05 起强制要求填 expected_return_date,所以新借出" "都会被这个 cron 监控。idempotent: 同一天对同一笔超期借出只发一次。" ), ) async def run_borrow_overdue( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Notify on borrowed assets past their expected_return_date. Per-day idempotent: looks for an existing ``notifications_outbox`` row this UTC day keyed by template + asset id and skips re-dispatch. """ from sqlalchemy import and_, select from app.domains.asset.models import AssetItem from app.identity.person_models import IdentityPerson from app.notifications.models import NotificationOutbox from app.notifications.service import ( NotificationEnvelope, NotificationService, ) # Use the UTC calendar date (not the worker-local ``date.today()``) so # the per-day idempotency window and ``days_overdue`` are stable # regardless of the worker's TZ — the dedup floor below stamps this # date as UTC, so the date itself must be derived in UTC too. today = datetime.now(UTC).date() today_start = datetime.combine(today, datetime.min.time(), tzinfo=UTC) # Round 2 audit fix-up 2026-05-04: cap the per-tick batch so a # misconfigured cron OR an unforeseen explosion of overdue rows # doesn't blow up worker memory + Feishu API quota in one tick. # Truncated batches re-process tomorrow (the per-day idempotency # check ensures we don't re-notify recipients already alerted). _BATCH_LIMIT = 500 rows = ( await db.execute( select(AssetItem) .where( AssetItem.deleted_at.is_(None), AssetItem.status == "borrowed", AssetItem.expected_return_date.isnot(None), AssetItem.expected_return_date < today, ) .order_by(AssetItem.expected_return_date.asc()) .limit(_BATCH_LIMIT + 1) ) ).scalars().all() truncated = len(rows) > _BATCH_LIMIT if truncated: rows = rows[:_BATCH_LIMIT] logger.warning( "asset.borrow_overdue: hit batch cap (%d) — remaining rows " "will be processed in the next daily tick.", _BATCH_LIMIT, ) if not rows: return {"checked": 0, "notified": 0} notif = NotificationService(db) notified = 0 skipped = 0 for asset in rows: recipient_ids = {asset.borrowed_by, asset.assigned_to} recipient_ids.discard(None) if not recipient_ids: continue # ``days_overdue`` / the "is it overdue?" decision are computed # per-recipient in the BORROWER's local timezone below — the UTC SQL # pre-filter (a harmless 1-day over-scan) only narrows candidates. # Computing against the global UTC date told a borrower west of UTC # "已超期 1 天" while it was still their due date locally # (feedback_cron_must_respect_local_tz). for recipient_id in recipient_ids: # Idempotency probe — has this asset already been notified # today via this template for this recipient? already = ( await db.execute( select(NotificationOutbox.id).where( and_( NotificationOutbox.recipient_person_id == recipient_id, NotificationOutbox.template_key == "asset.borrow_overdue", NotificationOutbox.related_resource_id == str(asset.id), NotificationOutbox.created_at >= today_start, ) ).limit(1) ) ).scalar_one_or_none() if already is not None: skipped += 1 continue recipient = await db.get(IdentityPerson, recipient_id) if recipient is None or recipient.deleted_at is not None: continue # Direct attribute access — IdentityPerson always has # employment_status (memory: feedback_no_getattr_default). # Round 2b audit fix-up 2026-05-04: also skip # ``leaver_pending``. The offboarding flow handles asset # return for leavers; pinging them daily about overdue # borrows is noise — the offboarding admin has the # affordance. if (recipient.employment_status or "").lower() in ( "left", "leaver_pending", ): continue # Overdue is decided in the borrower's local date, not UTC. A # return due "today" isn't overdue until the borrower's local # day has passed it. from app.core.working_hours import resolve_timezone local_today = datetime.now(UTC).astimezone( resolve_timezone(recipient) ).date() if asset.expected_return_date >= local_today: continue days_overdue = (local_today - asset.expected_return_date).days try: from app.notifications.service import ( CardSection, CardStat, LocalizedText, ) asset_label = asset.asset_tag or asset.asset_name or str(asset.id) await notif.dispatch( NotificationEnvelope( recipient=recipient, title=f"⏰ 借用资产已超期 — {asset_label}", body_text=( f"你借用的资产 [b]{asset_label}[/b] 已超过预定归还日期。" f"请尽快通过 [/my/inventory] 完成归还，或联系 IT 顺延。" ), template_key="asset.borrow_overdue", # Placeholders for the registered三语 template # (``notifications/templates/lifecycle.py``). The # envelope's title/body above stay as the no-template # fallback; the bell also re-renders from this context, # so every placeholder the template names must be here. context={ "asset_label": asset_label, "asset_name": asset.asset_name or "—", "expected_return_date": str( asset.expected_return_date ), "days_overdue": days_overdue, }, card_subtitle=LocalizedText( zh=f"预定归还：{asset.expected_return_date}", en=f"Due back: {asset.expected_return_date}", ja=f"返却予定日：{asset.expected_return_date}", ), card_stats=[ CardStat( label=LocalizedText( zh="资产", en="Asset", ja="資産", ), value=asset_label, hint=(asset.asset_name or "—"), ), CardStat( label=LocalizedText( zh="已超期", en="Overdue by", ja="超過日数", ), value=LocalizedText( zh=f"{days_overdue} 天", en=( f"{days_overdue} " f"{'day' if days_overdue == 1 else 'days'}" ), ja=f"{days_overdue} 日", ), trend="▲" if days_overdue > 7 else None, trend_is_good=False, ), CardStat( label=LocalizedText( zh="预定归还日", en="Due return date", ja="返却予定日", ), value=str(asset.expected_return_date), ), CardStat( label=LocalizedText( zh="处理入口", en="Where to act", ja="手続きページ", ), value="/my/inventory", ), ], card_sections=[ CardSection( title=LocalizedText( zh="如何处理", en="What to do", ja="対応方法", ), body=LocalizedText( zh=( "1) 已归还：去 Helix 提交归还登记\n" "2) 还需使用：联系 IT 申请顺延\n" "3) 资产丢失：联系 IT 走丢失流程" ), en=( "1) Already returned: submit the return " "record in Helix\n" "2) Still need it: contact IT to extend " "the due date\n" "3) Asset lost: contact IT to start the " "lost-asset process" ), ja=( "1) 返却済みの場合：Helix で返却登録を" "行ってください\n" "2) 継続利用する場合：IT に返却期限の" "延長を依頼してください\n" "3) 紛失した場合：IT に連絡し、紛失時の" "手続きを行ってください" ), ), tone="warn", ), ], url="/my/inventory", related_resource_type="asset", related_resource_id=asset.id, ), ) notified += 1 # Per-recipient commit. NotificationService.dispatch does # enqueue (an outbox row whose recipient_person_id FK takes # a shared lock on identity_persons) + deliver (20-30s # external Feishu/email sends) in THIS transaction. Without # a per-row commit those FK S-locks accumulate across every # send until the executor's single end-of-run commit — a # borrower's SSO login (which UPDATEs identity_persons # .last_login_at) then blocks behind them and fails with # 1205 lock-wait: a company-wide login brown-out (the exact # #1837 / #1940 incident class, and precisely what # dispatch's own docstring warns fan-outs must avoid). # Committing per row also makes each outbox row durable so # a mid-run crash can't lose already-sent rows and re-send # them next tick. await db.commit() except Exception as exc: # noqa: BLE001 # Roll back THIS row so the shared session is usable for # the next iteration's commit (asyncmy raises # PendingRollbackError on the next statement otherwise). await db.rollback() logger.warning( "asset.borrow_overdue: dispatch failed for %s -> %s: %s", asset.id, recipient_id, exc, ) continue return { "checked": len(rows), "notified": notified, "skipped": skipped, "truncated": truncated, } @register_scheduled_job( key="asset.held_by_leaver_reminder", display_name="离职持有资产 周报", description=( "每周一 09:00 (UTC) 统计 status='held_by_leaver' 的资产数。" "如果 > 0,给每个活跃 IT Admin / global admin 推送一条 in-app 提醒" "(本周内只推一次,基于 notification_outbox 去重)。" "用于催收尚未追回的离职员工硬件 — 处理后在资产详情页点「追回」" "把状态翻回 idle。" ), ) async def run_held_by_leaver_reminder( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Weekly digest to IT admins about unrecovered ex-employee hardware. Per-week idempotent: looks for an existing ``notifications_outbox`` row from the same ISO-week keyed by template + recipient and skips re-dispatch. If the count is 0 for the week, no notification is sent (silence = good news). """ from datetime import timedelta from sqlalchemy import and_, func, select from app.domains.asset.models import ( ASSET_STATUS_HELD_BY_LEAVER, AssetItem, ) from app.notifications.models import NotificationOutbox from app.notifications.service import ( NotificationEnvelope, NotificationService, ) from app.scheduled_jobs.failure_notifier import ( list_admin_recipients, ) held_count = ( await db.execute( select(func.count()).select_from(AssetItem).where( AssetItem.deleted_at.is_(None), AssetItem.status == ASSET_STATUS_HELD_BY_LEAVER, ) ) ).scalar() or 0 if held_count == 0: return {"held_count": 0, "notified": 0, "skipped": 0} sample_rows = ( await db.execute( select(AssetItem) .where( AssetItem.deleted_at.is_(None), AssetItem.status == ASSET_STATUS_HELD_BY_LEAVER, ) .order_by(AssetItem.updated_at.desc()) .limit(5) ) ).scalars().all() sample_labels = [ " / ".join( p for p in (a.asset_tag, a.brand, a.model) if p ) or str(a.id) for a in sample_rows ] # Derive the anchor from the UTC calendar date, NOT the worker-local # ``date.today()`` — the floor below is stamped as UTC and compared # against ``NotificationOutbox.created_at``, which the DB writes in # UTC (``server_default=func.now()``). Mixing the two clocks put the # floor up to a full TZ offset in the FUTURE relative to rows written # right now: on a CST (UTC+8) worker, every Monday 00:00–08:00 local # is still Sunday in UTC, so ``date.today()`` had already rolled to # Monday while ``created_at`` had not — the dedup probe could not see # the row it had just written and the digest re-dispatched to every # IT admin on every tick for those 8 hours. Same reasoning as the # overdue-borrow handler above; see its comment. today = datetime.now(UTC).date() # ISO-week anchor (Monday). Same anchor across the whole tick run # so two recipients getting alerted seconds apart land in the same # dedup window. week_start = today - timedelta(days=today.weekday()) week_start_dt = datetime.combine( week_start, datetime.min.time(), tzinfo=UTC, ) admins = await list_admin_recipients(db) if not admins: logger.warning( "asset.held_by_leaver_reminder: %d held-by-leaver assets " "but no eligible IT admin recipient — skipping send.", held_count, ) return {"held_count": held_count, "notified": 0, "skipped": 0} notif = NotificationService(db) notified = 0 skipped = 0 for admin in admins: already = ( await db.execute( select(NotificationOutbox.id).where( and_( NotificationOutbox.recipient_person_id == admin.id, NotificationOutbox.template_key == "asset.held_by_leaver_reminder", NotificationOutbox.created_at >= week_start_dt, ) ).limit(1) ) ).scalar_one_or_none() if already is not None: skipped += 1 continue sample_block = "\n".join(f" · {s}" for s in sample_labels) body = ( f"当前有 {held_count} 件资产标记为「离职持有」," f"仍归公司所有但物理上由已离职员工保管。\n\n" f"{sample_block}\n\n" "处理建议:确认归还后,在资产详情页点「追回」即可把状态翻回闲置。" ) try: await notif.dispatch( NotificationEnvelope( recipient=admin, title=f"[资产] {held_count} 件离职员工持有的资产待追回", body_text=body, template_key="asset.held_by_leaver_reminder", url="/admin/assets/list", related_resource_type="asset", related_resource_id=None, context={ "held_count": held_count, "samples": sample_labels, # Rendered form of ``samples`` — the template needs a # human-readable block, and a list interpolates as its # Python repr. "sample_block": sample_block, }, ), ) notified += 1 except Exception as exc: # noqa: BLE001 logger.warning( "asset.held_by_leaver_reminder: dispatch failed for " "admin %s: %s", admin.id, exc, ) continue return { "held_count": held_count, "notified": notified, "skipped": skipped, "samples": sample_labels, }