"""Feishu interactive-card builders for cloud-scope owner governance. Four cards, all going through :class:`app.notifications.service.NotificationService` via :class:`NotificationEnvelope` with rich-card fields populated. Pure formatting layer — no DB, no external calls. Callers (the cron handlers and the assign-owner patch) compose the snapshot data and hand the envelope to ``NotificationService.dispatch``. Layout follows the in-app reference (2x2 stat grid + highlight block + sectioned context + multi-CTA action row), refined for Feishu's card schema (column_set + markdown elements). The rich fields (stat grid / highlights / sections / action buttons) are emitted as :class:`~app.notifications.service.LocalizedText` (zh + en + ja) and resolved PER RECIPIENT by the notification layer, because the template loader renders only ``subject`` / ``body_text`` and has nowhere to put a ``CardStat``. They used to be bilingual (zh + en) and picked here from the recipient's ``preferred_language``, which meant a Japanese reader got the English chrome. The card **title and body**, however, now come from the notification template catalog (``app/notifications/templates/cloud.py``), which is what gives them a Japanese variant — ``_is_en`` below maps ``ja`` onto English, so a Japanese reader used to get an English card. Each builder therefore also returns a ``context`` dict of the sanitized values those templates interpolate (``{scope_name}`` / ``{scope_environment}`` / …); the caller spreads it into the envelope along with everything else. Sanitizing here rather than at the call site keeps :func:`_sanitize_label` the single choke point — a ``scope.display_name`` carrying ``[r]…[/r]`` must not reach the renderer through the context either. """ from __future__ import annotations from datetime import UTC, datetime from decimal import Decimal from typing import Any from app.notifications.service import ( CardAction, CardHighlight, CardSection, CardStat, LocalizedText, ) from app.notifications.text_safety import safe_envelope_text def _sanitize_label(value: str | None, *, limit: int = 80) -> str: """Round-51 S2-Markup: route every user-controlled string that lands in a Feishu card title / body / highlight item through :func:`safe_envelope_text` to strip ``[r]…[/r]`` / ``[b]…[/b]`` markup markers, CR/LF, and Unicode bidi-control chars. HRIS-synced ``display_name`` and admin-set ``scope.name`` / ``scope.environment`` flow into the cards' f-strings directly; a malicious / accidental value like ``张三[r] 紧急通知:点击链接修改密码[/r]`` would otherwise render with the platform's own urgent-alert styling, or a newline in ``scope_display_name`` would survive into the card header. """ return safe_envelope_text(value or "", limit=limit) def _is_en(lang: str | None) -> bool: if not lang: return False norm = lang.lower() return norm.startswith("en") or norm.startswith("ja") def _fmt_usd(amount: Decimal | float | int | None) -> str: if amount is None: return "$0.00" try: val = float(amount) except (TypeError, ValueError): return "$0.00" return f"${val:,.2f}" def _fmt_trend(curr: Decimal | float | None, prev: Decimal | float | None) -> str | None: """Build a ``▲ 12%`` / ``▼ 3%`` trend token comparing two periods. Returns ``None`` when the previous-period value is zero or absent — the cell renders without a trend arrow rather than showing a misleading +∞%. Whole-percent only (the card is for at-a-glance consumption, not finance reconciliation). """ if curr is None or prev is None: return None try: c = float(curr) p = float(prev) except (TypeError, ValueError): return None if p == 0: return None pct = (c - p) / p * 100 if abs(pct) < 1: return None arrow = "▲" if pct > 0 else "▼" return f"{arrow} {abs(pct):.0f}%" def _days_ago_label(ts: datetime | None) -> LocalizedText: """Human-readable "N days ago" / "N 天前", or ``从未`` / ``Never``. Returns a :class:`LocalizedText` so the notification layer picks the reader's language per recipient — the caller no longer decides. """ if ts is None: return LocalizedText(zh="从未", en="Never", ja="なし") # Treat naive datetimes as UTC. SQLAlchemy may return naive # values for ``DateTime(timezone=True)`` columns depending on the # driver; never compute deltas across a tz-aware ↔ tz-naive boundary. if ts.tzinfo is None: ts = ts.replace(tzinfo=UTC) now = datetime.now(UTC) days = max(0, (now - ts).days) if days == 0: return LocalizedText(zh="今天", en="Today", ja="本日") return LocalizedText( zh=f"{days} 天前", en=f"{days} day{'s' if days != 1 else ''} ago", ja=f"{days} 日前", ) def _reminder_subtitle( *, reminder_count: int, first_asked_at: datetime | None, fallback: LocalizedText, ) -> LocalizedText: """Subtitle for a prompt that is being re-sent. An unanswered prompt is re-sent to its owner daily (2026-08-24 policy — IT escalation retired). A daily card that reads identically to the first one looks like a system fault, so a reminder says how long the ask has been open and how many times it has been sent, and states the only thing that stops it: answering. """ if reminder_count <= 0: return fallback day = first_asked_at.date().isoformat() if first_asked_at else None since_zh = f"自 {day} 起" if day else "此前" since_en = f"open since {day}" if day else "still open" since_ja = f"{day} から" if day else "以前から" return LocalizedText( zh=( f"{since_zh}尚未确认 · 第 {reminder_count} 次提醒。" "确认或转移后即不再提醒。" ), en=( f"Still unconfirmed ({since_en}) · reminder #{reminder_count}. " "Confirming or transferring stops these." ), ja=( f"{since_ja}未確認です · {reminder_count} 回目のリマインドです。" "確認または引き継ぎを行うと通知は止まります。" ), ) def build_semiannual_card_fields( *, scope_display_name: str, scope_environment: str, spend_recent_30d: Decimal, spend_prior_30d: Decimal | None, spend_recent_180d: Decimal, spend_prior_180d: Decimal | None, member_count: int, member_count_180d_ago: int | None, last_confirmed_at: datetime | None, top_active_members: list[str], confirm_url: str, transfer_url: str, detail_url: str, lang: str, # Reminder context. ``reminder_count=0`` is the first ask; anything # higher rewrites the subtitle via ``_reminder_subtitle``. reminder_count: int = 0, first_asked_at: datetime | None = None, ) -> dict[str, Any]: """Compose envelope-rich-fields for the semiannual confirmation card. Returns a dict with keys matching :class:`NotificationEnvelope` rich-card field names — caller spreads it into the envelope ctor. """ en = _is_en(lang) # Round-51 S2-Markup: every user-controlled string interpolated # below MUST go through ``_sanitize_label`` first. scope_display_name = _sanitize_label(scope_display_name, limit=120) scope_environment = _sanitize_label(scope_environment, limit=32) top_active_members = [ _sanitize_label(name, limit=60) for name in (top_active_members or []) ] title = ( f"🔔 Cloud subscription semiannual check — {scope_display_name}" if en else f"🔔 云订阅半年确认 — {scope_display_name}" ) # No escalation threat any more (retired 2026-08-24): an unanswered # prompt is re-sent to this same owner every day instead of being # handed to IT after 14 days. The copy must promise what actually # happens. subtitle = _reminder_subtitle( reminder_count=reminder_count, first_asked_at=first_asked_at, fallback=LocalizedText( zh="请确认这个订阅是否仍由你负责，或转给实际负责人。未处理前每天会再提醒一次。", en=( "Confirm you still own this subscription, or transfer it to " "whoever does. Until then you'll get this once a day." ), ja=( "このサブスクリプションの担当者がご自身かを確認し、" "違う場合は実際の担当者に引き継いでください。" "対応いただくまで毎日リマインドします。" ), ), ) body = ( f"You are the owner of this `{scope_environment}` cloud subscription. " f"Future spend on this scope will be attributed to you." if en else f"你是该 `{scope_environment}` 云订阅的 owner，未来花费将记在你名下。" ) stats = [ CardStat( label=LocalizedText( zh="近 30 天花费", en="Last 30 days spend", ja="直近 30 日の費用", ), value=_fmt_usd(spend_recent_30d), trend=_fmt_trend(spend_recent_30d, spend_prior_30d), trend_is_good=False, # rising cost = bad ), CardStat( label=LocalizedText( zh="近 180 天花费", en="Last 180 days spend", ja="直近 180 日の費用", ), value=_fmt_usd(spend_recent_180d), trend=_fmt_trend(spend_recent_180d, spend_prior_180d), trend_is_good=False, ), CardStat( label=LocalizedText( zh="可访问人数", en="People with access", ja="アクセス可能な人数", ), value=f"{member_count}", trend=_fmt_trend(member_count, member_count_180d_ago), trend_is_good=False, # more accessors = more risk ), CardStat( label=LocalizedText( zh="上次确认", en="Last confirmed", ja="前回の確認", ), value=_days_ago_label(last_confirmed_at), ), ] highlights: list[CardHighlight] = [] if top_active_members: # Round-22: label this "recently-added" rather than # "active" / "top accessors" — the underlying query in # ``OwnerConfirmationService._top_member_names`` orders by # ``CmdbCloudAccessGrant.created_at`` desc, which is the # grant-creation timestamp, NOT actual cloud-side usage. # Calling these members "active" or "top accessors" would # mislead the owner into believing the listed people are # currently using the subscription (they may have been # granted access weeks ago and never logged in). The names # are still a useful "who has access" snapshot; just rename # the label so the card matches what the data actually is. highlights.append(CardHighlight( label=LocalizedText( zh="最近添加的成员", en="Recently granted access", ja="最近追加されたメンバー", ), items=[f"[b]{name}[/b]" for name in top_active_members[:3]], )) sections = [ CardSection( title=LocalizedText( zh="你的责任", en="Your responsibility", ja="あなたの責任", ), body=LocalizedText( zh=( "若该订阅不再由你管理，请点 **转给别人**；新 owner 会收到飞书通知，" "并接管未来的花费归属。" ), en=( "If this subscription is no longer yours to manage, " "tap **Transfer to someone else** — the new owner will " "be notified and accept ownership of future spend." ), ja=( "このサブスクリプションを管理していない場合は、**他の人に引き継ぐ** " "をタップしてください。新しいオーナーに Feishu で通知が届き、" "今後の費用の帰属を引き継ぎます。" ), ), tone="warn", ), ] actions = [ CardAction( label=LocalizedText( zh="✅ 确认是我管理", en="✅ I own this", ja="✅ 私が管理しています", ), url=confirm_url, style="primary", ), CardAction( label=LocalizedText( zh="🔄 转给别人", en="🔄 Transfer to someone else", ja="🔄 他の人に引き継ぐ", ), url=transfer_url, ), CardAction( label=LocalizedText( zh="📊 查看详情", en="📊 View details", ja="📊 詳細を見る", ), url=detail_url, ), ] return { "title": title, "body_text": body, "card_subtitle": subtitle, "card_stats": stats, "card_highlights": highlights, "card_sections": sections, "card_actions": actions, "context": { "scope_name": scope_display_name, "scope_environment": scope_environment, }, } def build_zero_spend_card_fields( *, scope_display_name: str, scope_environment: str, period_label: str, spend_last_month: Decimal, spend_recent_90d: Decimal, spend_recent_180d: Decimal, member_count: int, top_members: list[str], scope_created_at: datetime | None, keep_url: str, cleanup_url: str, handover_url: str, detail_url: str, lang: str, reminder_count: int = 0, first_asked_at: datetime | None = None, ) -> dict[str, Any]: """Compose envelope-rich-fields for the zero-spend alert card.""" en = _is_en(lang) # Round-51 S2-Markup: sanitize every user-controlled string. scope_display_name = _sanitize_label(scope_display_name, limit=120) scope_environment = _sanitize_label(scope_environment, limit=32) period_label = _sanitize_label(period_label, limit=32) top_members = [_sanitize_label(n, limit=60) for n in (top_members or [])] title = ( f"⚠️ Zero-spend cloud subscription — {scope_display_name}" if en else f"⚠️ 云订阅零花费提醒 — {scope_display_name}" ) subtitle = _reminder_subtitle( reminder_count=reminder_count, first_asked_at=first_asked_at, fallback=LocalizedText( zh=f"该 `{scope_environment}` 订阅在 {period_label} 花费为 $0。", en=f"This `{scope_environment}` scope had $0 spend in {period_label}.", ja=( f"この `{scope_environment}` サブスクリプションは " f"{period_label} の費用が $0 でした。" ), ), ) body = ( "Pick one of: keep it with a stated reason, clean it up, " "or hand it over to the IT team." if en else "请选择处理方式：写明理由保留 / 由你清理 / 联系 IT 团队接管。" ) age_label = _days_ago_label(scope_created_at) stats = [ CardStat( label=LocalizedText( zh=f"{period_label} 花费", en=f"{period_label} spend", ja=f"{period_label} の費用", ), value=_fmt_usd(spend_last_month), hint=LocalizedText( zh="命中 $0 阈值", en="must be $0 to land this card", ja="$0 のしきい値に該当", ), ), CardStat( label=LocalizedText( zh="近 90 天花费", en="Last 90 days spend", ja="直近 90 日の費用", ), value=_fmt_usd(spend_recent_90d), ), CardStat( label=LocalizedText( zh="近 180 天花费", en="Last 180 days spend", ja="直近 180 日の費用", ), value=_fmt_usd(spend_recent_180d), ), CardStat( label=LocalizedText( zh="订阅创建距今", en="Scope age", ja="サブスクリプション作成からの経過", ), value=age_label, ), ] highlights: list[CardHighlight] = [] if top_members: highlights.append(CardHighlight( label=LocalizedText( zh=f"可访问人员 ({member_count})", en=f"People with access ({member_count})", ja=f"アクセス可能なメンバー ({member_count})", ), items=[f"[b]{n}[/b]" for n in top_members[:3]], )) sections = [ CardSection( title=LocalizedText( zh="三个选项的含义", en="What each choice means", ja="3 つの選択肢の意味", ), body=LocalizedText( zh=( "**保留** — 写明保留理由，订阅保持活跃；如下月仍 $0 花费会再次告警。\n" "**我来清理** — 由你负责走云侧下线流程。\n" "**联系 IT 处理** — IT 团队接手评估并可代为下线。" ), en=( "**Keep** — provide a written reason; this scope stays " "active and the alert will fire again next month if " "spend remains $0.\n" "**Clean up** — you take responsibility for " "decommissioning the cloud-side account.\n" "**Hand to IT** — IT team triages and may decommission " "on your behalf." ), ja=( "**保留する** — 理由を記入していただければ、このサブスクリプションは" "有効なままです。翌月も費用が $0 の場合は再度通知されます。\n" "**自分で整理する** — クラウド側の廃止手続きをご自身で担当します。\n" "**IT に引き継ぐ** — IT チームが内容を確認し、代わりに廃止する場合があります。" ), ), tone="info", ), ] actions = [ CardAction( label=LocalizedText( zh="✅ 还要保留（写理由）", en="✅ Keep (with reason)", ja="✅ 保留する（理由を記入）", ), url=keep_url, style="primary", ), CardAction( label=LocalizedText( zh="🗑 我来清理", en="🗑 I'll clean up", ja="🗑 自分で整理する", ), url=cleanup_url, ), CardAction( label=LocalizedText( zh="🆘 联系 IT 处理", en="🆘 Hand to IT", ja="🆘 IT に引き継ぐ", ), url=handover_url, ), ] return { "title": title, "body_text": body, "card_subtitle": subtitle, "card_stats": stats, "card_highlights": highlights, "card_sections": sections, "card_actions": actions, "context": { "scope_name": scope_display_name, "scope_environment": scope_environment, "period_label": period_label, }, } def build_new_owner_card_fields( *, scope_display_name: str, scope_environment: str, previous_owner_name: str | None, spend_recent_30d: Decimal, member_count: int, next_confirm_due_at: datetime | None, detail_url: str, lang: str, ) -> dict[str, Any]: """Compose envelope-rich-fields for the new-owner transfer-notification card.""" en = _is_en(lang) # Round-51 S2-Markup: sanitize every user-controlled string. scope_display_name = _sanitize_label(scope_display_name, limit=120) scope_environment = _sanitize_label(scope_environment, limit=32) if previous_owner_name is not None: previous_owner_name = _sanitize_label(previous_owner_name, limit=60) title = ( f"📣 You are now the owner — {scope_display_name}" if en else f"📣 你被指定为云订阅 owner — {scope_display_name}" ) prev_zh = previous_owner_name or "前任 owner" prev_en = previous_owner_name or "the previous owner" prev_ja = previous_owner_name or "前任のオーナー" subtitle = LocalizedText( zh=f"{prev_zh} 把这个 `{scope_environment}` 订阅转给了你。", en=f"{prev_en} transferred this `{scope_environment}` scope to you.", ja=( f"{prev_ja} がこの `{scope_environment}` " "サブスクリプションをあなたに引き継ぎました。" ), ) body = ( "Future spend on this subscription will be attributed to you. " "If you don't accept this assignment, ask the previous owner to " "transfer it elsewhere — there is no auto-rollback." if en else "未来花费将记在你名下。如果你不接受这个指派，请联系前任 owner 转给其他人，" "Helix 不会自动回滚。" ) next_due_label: str | LocalizedText if next_confirm_due_at is None: next_due_label = LocalizedText( zh="约 6 个月后", en="~6 months from now", ja="約 6 か月後", ) else: next_due_label = next_confirm_due_at.strftime("%Y-%m-%d") stats = [ CardStat( label=LocalizedText( zh="近 30 天花费", en="Last 30 days spend", ja="直近 30 日の費用", ), value=_fmt_usd(spend_recent_30d), ), CardStat( label=LocalizedText( zh="可访问人数", en="People with access", ja="アクセス可能な人数", ), value=f"{member_count}", ), CardStat( label=LocalizedText(zh="环境", en="Environment", ja="環境"), value=scope_environment, ), CardStat( label=LocalizedText( zh="下次确认到期", en="Next confirmation due", ja="次回確認の期限", ), value=next_due_label, ), ] sections = [ CardSection( title=LocalizedText( zh="你的责任", en="Your responsibility", ja="あなたの責任", ), body=LocalizedText( zh=( "你现在可以无需审批地为该 scope 增删成员；后续治理推送（半年确认、零花费告警）也都会发给你。" ), en=( "You can add/remove members on this scope without approval, " "and will be the recipient of every governance card going " "forward (semiannual confirm, zero-spend alerts)." ), ja=( "このスコープのメンバーは承認なしで追加・削除できます。" "今後のガバナンス通知（半年ごとの確認、費用 $0 のアラート）も" "すべてあなた宛てに届きます。" ), ), tone="info", ), ] return { "title": title, "body_text": body, "card_subtitle": subtitle, "card_stats": stats, "card_sections": sections, "url": detail_url, "context": { "scope_name": scope_display_name, "scope_environment": scope_environment, "previous_owner_name": previous_owner_name or "", }, } # ``build_escalation_card_fields`` lived here until 2026-08-24. IT # escalation was retired in favour of re-sending the prompt to its owner # daily (see ``_reminder_subtitle`` and # ``OwnerConfirmationService.remind_unanswered_owners``), so the builder # had no caller left. Historical ``escalated_to_it`` timeline rows still # render from the DB — the card that announced them is simply never # composed again. __all__ = [ "build_new_owner_card_fields", "build_semiannual_card_fields", "build_zero_spend_card_fields", ]