"""The ambient surface — what the mascot bubble shows, and how often. The bubble already existed and already rotated admin-written security tips. Two things were missing and both are the reason this module is here: 1. **It recorded nothing.** A tip rotating past 500 people produced zero evidence that anybody was reached. The read endpoint did not even take a person id. 2. **It could not carry a comic.** ``platform_security_tips`` is a 500-character bilingual string; the security team's actual content is artwork and articles. So this module answers "what should THIS person see in the bubble right now" from the material library, records the exposure, and — the part that matters most for whether the feature survives contact with the workforce — **rations it**. ## The rationing rules, and why each one exists * **A small per-person daily set, stable within the day.** Recomputing the pick on every page load would make the bubble reshuffle as somebody navigates, which reads as noise rather than as a message. The set is derived deterministically from ``(person, date)``. * **A cap on how many NEW materials a person meets per day** (default 1). The point of an ambient surface is repetition of one idea, not a firehose. * **A cap on how many times one material is shown to one person** (default 3). After that it has either worked or it has not, and a fourth showing only teaches them to ignore the bubble. * **Anything they have opened is dropped.** They did the thing; still nudging them about it is the fastest way to make the nudge worthless. ## What it is NOT It is not a campaign. Nothing here pushes, nothing here is required, nothing here appears in a todo list. If the security team needs somebody to actually read something, that is a campaign with an acknowledgement — this is the surface for the other 95% of awareness, where the honest goal is "seen in passing, repeatedly". """ from __future__ import annotations import hashlib import logging import uuid from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.security_awareness.catalog import ( STATUS_PUBLISHED, SURFACE_BUBBLE, SURFACE_DETAIL, ) from app.domains.security_awareness.models import SecawImpression, SecawMaterial from app.domains.security_awareness.presign import presign_material_keys from app.domains.security_awareness.service import pick_block logger = logging.getLogger(__name__) # ── Settings keys (platform_settings) ─────────────────────────────── SETTING_ENABLED = "security_awareness.bubble.enabled" SETTING_DAILY_NEW = "security_awareness.bubble.daily_new_materials" SETTING_MAX_REPEATS = "security_awareness.bubble.max_repeats_per_material" DEFAULT_ENABLED = True #: How many DIFFERENT materials one person may meet in the bubble in a #: day. One, by default. An ambient surface earns its keep by repeating #: one idea, and an admin who wants to say five things this week should #: say them on five days. DEFAULT_DAILY_NEW = 1 #: How many times one material may be shown to one person, ever. After #: three it has either worked or it has not. DEFAULT_MAX_REPEATS = 3 #: Ceiling on what the endpoint will return in one response, regardless #: of settings. The bubble rotates through this list client-side, and a #: list longer than a few items is a feed, not an ambient nudge. MAX_ROTATION = 3 #: Hard ceiling on the pool we consider. The library is curated content #: (tens of items), and the ranking below is Python-side. POOL_CAP = 200 @dataclass(slots=True) class BubbleItem: """One thing the bubble can show.""" material_id: uuid.UUID kind: str category: str title: str #: The one-line hook. The summary when there is one, else the title — #: the bubble is a single line of text next to a thumbnail, so a #: 500-character summary would be truncated into nonsense anyway and #: the service picks rather than the CSS. hook: str cover_url: str | None lang: str is_fallback_lang: bool @dataclass(slots=True) class BubbleSettings: enabled: bool daily_new_materials: int max_repeats_per_material: int async def load_settings(db: AsyncSession) -> BubbleSettings: """Read the three knobs, coercing anything unusable to the default. A malformed setting must not take the bubble down: this is ambient decoration on every authenticated page, and an exception here would surface as a broken app shell. """ from app.integrations.service import get_platform_setting async def _int(key: str, default: int) -> int: raw = await get_platform_setting(key, default, session=db) try: value = int(raw) except (TypeError, ValueError): logger.warning("secaw.bubble.bad_setting key=%s raw=%r", key, raw) return default return max(0, min(value, 50)) raw_enabled = await get_platform_setting( SETTING_ENABLED, DEFAULT_ENABLED, session=db, ) return BubbleSettings( enabled=bool(raw_enabled), daily_new_materials=await _int(SETTING_DAILY_NEW, DEFAULT_DAILY_NEW), max_repeats_per_material=await _int( SETTING_MAX_REPEATS, DEFAULT_MAX_REPEATS, ), ) def _day_key(person_id: uuid.UUID, day: str, material_id: uuid.UUID) -> str: """Deterministic ordering key for (person, day, material). A stable hash rather than ``random``: the pick must be the same for every request this person makes today — otherwise the bubble reshuffles as they navigate, which reads as noise rather than as a message — but it must differ between people, so the whole company is not staring at the same card on the same morning. """ return hashlib.sha256( f"{person_id}:{day}:{material_id}".encode() ).hexdigest() class AmbientService: def __init__(self, db: AsyncSession): self.db = db async def bubble_feed( self, *, person_id: uuid.UUID, lang: str, now: datetime | None = None, ) -> dict[str, Any]: """What this person's bubble should rotate through right now. Returns ``{"enabled": bool, "items": [...]}``. An empty ``items`` with ``enabled: true`` is a normal, expected state — it means this person is up to date, and the client falls back to the legacy tips rather than showing an empty bubble. """ settings = await load_settings(self.db) if not settings.enabled: return {"enabled": False, "items": []} now = now or datetime.now(UTC) day = now.date().isoformat() rows = (await self.db.execute( select(SecawMaterial) .where(SecawMaterial.status == STATUS_PUBLISHED) .limit(POOL_CAP) )).scalars().all() pool = [r for r in rows if self._renderable(r)] if not pool: return {"enabled": True, "items": []} seen = await self._exposure_map(person_id, [r.id for r in pool]) opened = await self._opened_set(person_id, [r.id for r in pool]) today_ids = await self._shown_today(person_id, now) eligible: list[SecawMaterial] = [] for row in pool: # Opened it → they did the thing. Still nudging them about it # is the fastest way to make the nudge worthless. if row.id in opened: continue if seen.get(row.id, 0) >= settings.max_repeats_per_material: continue eligible.append(row) # Anything already shown TODAY stays in the rotation regardless of # the daily-new cap — the cap limits how many new ideas a person # meets, not how long today's idea stays on screen. Without this # the bubble would go blank the moment the cap was reached. carried = [r for r in eligible if r.id in today_ids] fresh = [r for r in eligible if r.id not in today_ids] room = max(0, settings.daily_new_materials - len(carried)) # Pinned first (an admin promoting something means it), then the # per-person stable hash so the whole company is not looking at # the same card on the same morning. fresh.sort(key=lambda r: ( not r.is_pinned, _day_key(person_id, day, r.id), )) chosen = (carried + fresh[:room])[:MAX_ROTATION] if not chosen: return {"enabled": True, "items": []} cover_urls = await self._cover_urls(chosen, lang) return { "enabled": True, "items": [ self._to_item(row, lang, cover_urls) for row in chosen ], } # ── Internals ─────────────────────────────────────────────────── @staticmethod def _renderable(row: SecawMaterial) -> bool: for block in (row.localized or {}).values(): if isinstance(block, dict) and block.get("title"): return True return False async def _exposure_map( self, person_id: uuid.UUID, material_ids: list[uuid.UUID], ) -> dict[uuid.UUID, int]: """How many times each material has been shown to this person in the bubble, ever.""" if not material_ids: return {} rows = (await self.db.execute( select(SecawImpression.material_id, func.count()) .where( SecawImpression.person_id == person_id, SecawImpression.material_id.in_(material_ids), SecawImpression.surface == SURFACE_BUBBLE, ) .group_by(SecawImpression.material_id) )).all() return {mid: int(n) for mid, n in rows} async def _opened_set( self, person_id: uuid.UUID, material_ids: list[uuid.UUID], ) -> set[uuid.UUID]: if not material_ids: return set() rows = (await self.db.execute( select(SecawImpression.material_id) .where( SecawImpression.person_id == person_id, SecawImpression.material_id.in_(material_ids), SecawImpression.surface == SURFACE_DETAIL, ) .distinct() )).scalars().all() return set(rows) async def _shown_today( self, person_id: uuid.UUID, now: datetime, ) -> set[uuid.UUID]: """Materials this person already met in the bubble today. Window is the last 24 hours rather than a calendar day: "today" differs per timezone across the workforce, and a UTC midnight would reset the cap in the middle of a Shanghai afternoon. """ since = now - timedelta(hours=24) rows = (await self.db.execute( select(SecawImpression.material_id) .where( SecawImpression.person_id == person_id, SecawImpression.surface == SURFACE_BUBBLE, SecawImpression.occurred_at >= since, ) .distinct() )).scalars().all() return set(rows) async def _cover_urls( self, rows: list[SecawMaterial], lang: str, ) -> dict[uuid.UUID, str | None]: key_by_material: dict[uuid.UUID, str] = {} for row in rows: _lang, block, _fb = pick_block(row.localized, lang) key = block.get("cover_key") if not key: images = block.get("image_keys") or [] key = images[0] if images else None if key: key_by_material[row.id] = key url_map = await presign_material_keys(self.db, key_by_material.values()) return {mid: url_map.get(k) for mid, k in key_by_material.items()} @staticmethod def _to_item( row: SecawMaterial, want_lang: str, cover_urls: dict[uuid.UUID, str | None], ) -> dict[str, Any]: lang, block, is_fallback = pick_block(row.localized, want_lang) title = str(block.get("title") or "") summary = block.get("summary") return { "material_id": str(row.id), "kind": row.kind, "category": row.category, "title": title, # The bubble is one line next to a thumbnail. The SERVICE # picks which string that line is, rather than leaving the # CSS to truncate a 500-character summary into nonsense. "hook": (summary or title), "cover_url": cover_urls.get(row.id), "lang": lang, "is_fallback_lang": is_fallback, # The bubble is where the language gap was actually reported # (an English UI showing a Chinese hook, 2026-09-05). It has # no room for a notice, so it renders a small marker: a # fallback block gets the language it is actually in, and a # machine-translated block gets an "AI" chip. Both beat the # previous behaviour, which was to show Chinese and say # nothing. "machine_translated": lang in (row.mt_langs or []), }