"""Read + CRUD service for the security-awareness material library. Three responsibilities, kept apart on purpose: * **Language resolution.** One place decides which localized block a reader is shown. The frontend never picks, because the impression ledger records what was shown and two independent deciders would make the record a guess. See :func:`pick_block`. * **Presigning.** Object keys never leave this layer; callers get URLs. * **The ledger.** :meth:`SecurityAwarenessService.record_impression` is the only writer, and it is idempotent inside the dedupe window. Mutations are audited under ``security_awareness.material.``. """ from __future__ import annotations import logging import uuid from collections.abc import Sequence from datetime import UTC, datetime, timedelta from math import ceil from typing import Any from pydantic import ValidationError from sqlalchemy import Select, asc, desc, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.domains.security_awareness.catalog import ( BASELINE_LANG, CHARS_PER_MINUTE, IMPRESSION_DEDUPE_MINUTES, KIND_ARTICLE, KIND_COMIC, LANG_FALLBACK, MATERIAL_CATEGORIES, MATERIAL_LANGS, SECONDS_PER_COMIC_PAGE, STATUS_ARCHIVED, STATUS_DRAFT, STATUS_PUBLISHED, SURFACE_DETAIL, ) from app.domains.security_awareness.models import SecawImpression, SecawMaterial from app.domains.security_awareness.presign import ( is_awareness_key, presign_material_keys, ) from app.domains.security_awareness.schemas import ( MaterialCreate, MaterialUpdate, ) from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) # The library is curated content — tens to low hundreds of items, not a # CMS. Reads fetch the published set and filter in Python, which keeps # free-text search identical on MySQL and TiDB without fighting JSON # LIKE portability. Same trade-off (and same reasoning) as # ``policy_notices._LIBRARY_SOFT_CAP``; revisit with a real index only if # the set outgrows it. LIBRARY_SOFT_CAP = 500 # Admin console page size ceiling. Mirrors the frontend's page-size # options — a client asking for more gets clamped rather than refused, # because a 400 on a pagination knob is a worse experience than a # smaller page. ADMIN_MAX_PAGE_SIZE = 100 def normalize_lang(raw: str | None) -> str: """Map anything a client sends (``zh-CN``, ``ja-JP``, ``en,zh;q=0.9``) onto one of the three languages this domain stores. Unknown input resolves to the baseline rather than raising: a reader with an exotic ``Accept-Language`` should get content, not a 422. """ token = (raw or "").strip().lower().replace("_", "-") if not token: return BASELINE_LANG primary = token.split(",")[0].split("-")[0].strip() return primary if primary in MATERIAL_LANGS else BASELINE_LANG def pick_block( localized: dict[str, Any] | None, want: str, ) -> tuple[str, dict[str, Any], bool]: """Resolve ``want`` against a material's localized map. Returns ``(rendered_lang, block, is_fallback)``. ``is_fallback`` is True when the reader did not get the language they asked for — the UI shows a quiet notice in that case, because silently serving Chinese to a Japanese reader looks like a bug on their end. Raises :class:`ValueError` only when the map is empty, which the schema layer already prevents on write; existing rows are guarded by the caller. """ blocks = localized or {} for candidate in LANG_FALLBACK.get(want, (BASELINE_LANG,)): block = blocks.get(candidate) if isinstance(block, dict) and block.get("title"): return candidate, block, candidate != want # Last resort: any language at all. Reached only by a row that # predates the schema guard or was written by a migration. for candidate, block in blocks.items(): if isinstance(block, dict) and block.get("title"): return candidate, block, candidate != want raise ValueError("material has no renderable localized content") def estimate_minutes( kind: str, block: dict[str, Any], override: int | None, ) -> int | None: """The "约 N 分钟" hint. Admin override wins; otherwise derived from the content. Returns ``None`` for a link — we have no idea how long somebody else's page takes to read, and inventing a number would be the kind of confident lie that erodes trust in every other number on the page.""" if override: return override if kind == KIND_ARTICLE: body = block.get("body") or "" if not body: return None return max(1, ceil(len(body) / CHARS_PER_MINUTE)) if kind == KIND_COMIC: pages = len(block.get("image_keys") or []) if not pages: return None return max(1, ceil(pages * SECONDS_PER_COMIC_PAGE / 60)) return None def _searchable_text(row: SecawMaterial) -> str: """Every language's title + summary + the shared tags, lowercased. Search is deliberately cross-language: an admin who tagged a comic in Chinese still finds it by typing "phishing", and an employee reading the English UI still finds 钓鱼 material that only carries an English title. """ parts: list[str] = list(row.tags or []) for block in (row.localized or {}).values(): if not isinstance(block, dict): continue parts.append(str(block.get("title") or "")) parts.append(str(block.get("summary") or "")) return " ".join(p.lower() for p in parts if p) def _matches(row: SecawMaterial, tokens: list[str]) -> bool: """AND over tokens — typing more words narrows, as a reader expects.""" haystack = _searchable_text(row) return all(tok in haystack for tok in tokens) class SecurityAwarenessService: def __init__(self, db: AsyncSession): self.db = db self.audit = AuditService(db) # ════════════════════════════════════════════════════════════════ # Employee reads # ════════════════════════════════════════════════════════════════ async def list_library( self, *, person_id: uuid.UUID, lang: str, category: str | None = None, tag: str | None = None, q: str | None = None, ) -> list[dict[str, Any]]: """The employee learning centre. Published materials only, pinned first, each resolved into ``lang`` with its cover presigned and this reader's own view history attached.""" rows = (await self.db.execute( select(SecawMaterial) .where(SecawMaterial.status == STATUS_PUBLISHED) .order_by( desc(SecawMaterial.is_pinned), asc(SecawMaterial.display_order), desc(SecawMaterial.published_at), desc(SecawMaterial.created_at), ) .limit(LIBRARY_SOFT_CAP) )).scalars().all() if category: rows = [r for r in rows if r.category == category] if tag: needle = tag.strip().lower() rows = [ r for r in rows if any((t or "").lower() == needle for t in (r.tags or [])) ] tokens = [t for t in (q or "").lower().split() if t] if tokens: rows = [r for r in rows if _matches(r, tokens)] viewed = await self._viewed_map(person_id, [r.id for r in rows]) cover_urls = await self._cover_urls(rows, lang) return [ self._to_card(r, lang, cover_urls, viewed) for r in rows if self._renderable(r) ] async def facets( self, *, person_id: uuid.UUID, lang: str, ) -> dict[str, Any]: """Filter options for the learning centre, plus the reader's own unread count. Derived from the **whole** published set, never from a filtered result — chips computed off an already-filtered list vanish the moment you use one, which reads as the app losing options. Skips cover presigning entirely (the facet bar shows no images), which is why this is not just ``len(list_library(...))``: signing forty covers to produce eight integers would put an S3 signing pass on every filter interaction. """ rows = (await self.db.execute( select(SecawMaterial) .where(SecawMaterial.status == STATUS_PUBLISHED) .limit(LIBRARY_SOFT_CAP) )).scalars().all() live = [r for r in rows if self._renderable(r)] viewed = await self._viewed_map(person_id, [r.id for r in live]) counts: dict[str, int] = {} tags: dict[str, int] = {} for row in live: counts[row.category] = counts.get(row.category, 0) + 1 for tag in row.tags or []: tags[tag] = tags.get(tag, 0) + 1 return { "categories": [ {"key": key, "count": counts.get(key, 0)} for key in MATERIAL_CATEGORIES ], "tags": [ {"tag": tag, "count": count} for tag, count in sorted( tags.items(), key=lambda kv: (-kv[1], kv[0]), ) ], "total": len(live), "unread": sum(1 for r in live if r.id not in viewed), } async def get_material_detail( self, *, person_id: uuid.UUID, material_id: uuid.UUID, lang: str, include_unpublished: bool = False, ) -> dict[str, Any]: """One material, fully hydrated for reading. ``include_unpublished`` is the admin preview path — an author has to be able to look at a draft the way an employee will see it, and building a second renderer for that is how previews drift from the real thing. """ row = await self.db.get(SecawMaterial, material_id) if row is None: raise NotFoundError("security awareness material", str(material_id)) if row.status != STATUS_PUBLISHED and not include_unpublished: # Draft / archived materials are not "forbidden", they are # not part of the library — 404 rather than 403 so a stale # bookmark does not read as an access-control decision. raise NotFoundError("security awareness material", str(material_id)) if not self._renderable(row): raise AppException( code="SECAW_MATERIAL_UNRENDERABLE", message="material has no renderable content", status_code=409, ) rendered_lang, block, is_fallback = pick_block(row.localized, lang) keys = list(block.get("image_keys") or []) cover_key = block.get("cover_key") if cover_key: keys.append(cover_key) url_map = await presign_material_keys(self.db, keys) viewed = await self._viewed_map(person_id, [row.id]) card = self._to_card( row, lang, {row.id: url_map.get(cover_key) if cover_key else None}, viewed, ) card.update({ "body": block.get("body"), "image_urls": [ url_map[k] for k in (block.get("image_keys") or []) if k in url_map ], "link_url": block.get("link_url"), "available_langs": [ lg for lg in MATERIAL_LANGS if isinstance((row.localized or {}).get(lg), dict) and (row.localized or {})[lg].get("title") ], }) # Recompute — the card helper derives from the *card* language # pick, and for the detail view we already resolved the block. card["lang"] = rendered_lang card["is_fallback_lang"] = is_fallback is_mt = rendered_lang in (row.mt_langs or []) card["machine_translated"] = is_mt card["machine_translated_from"] = BASELINE_LANG if is_mt else None # Every language whose block a model wrote, not just the one # being rendered — the language switcher on the detail page uses # it to mark the other tabs. card["machine_translated_langs"] = list(row.mt_langs or []) return card async def record_impression( self, *, person_id: uuid.UUID, material_id: uuid.UUID, surface: str, lang: str, ) -> bool: """Append one exposure to the ledger. Returns True when a row was written, False when it collapsed into the dedupe window. Only *published* materials record impressions: an admin previewing their own draft is not workforce reach, and counting it would inflate the coverage number with the author's own clicks. """ row = await self.db.get(SecawMaterial, material_id) if row is None: raise NotFoundError("security awareness material", str(material_id)) if row.status != STATUS_PUBLISHED: return False rendered_lang, _block, _fb = pick_block(row.localized, lang) cutoff = datetime.now(UTC) - timedelta(minutes=IMPRESSION_DEDUPE_MINUTES) recent = (await self.db.execute( select(SecawImpression.id) .where( SecawImpression.person_id == person_id, SecawImpression.material_id == material_id, SecawImpression.surface == surface, SecawImpression.occurred_at >= cutoff, ) .limit(1) )).scalar() if recent is not None: return False self.db.add(SecawImpression( id=uuid.uuid4(), person_id=person_id, material_id=material_id, surface=surface, material_version=row.version, lang=rendered_lang, occurred_at=datetime.now(UTC), )) await self.db.flush() return True # ════════════════════════════════════════════════════════════════ # Admin reads # ════════════════════════════════════════════════════════════════ async def list_admin( self, *, status: str | None = None, category: str | None = None, kind: str | None = None, q: str | None = None, page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: """Paged admin console listing with per-material reach.""" page = max(1, page) page_size = max(1, min(page_size, ADMIN_MAX_PAGE_SIZE)) stmt: Select[Any] = select(SecawMaterial) if status: stmt = stmt.where(SecawMaterial.status == status) if category: stmt = stmt.where(SecawMaterial.category == category) if kind: stmt = stmt.where(SecawMaterial.kind == kind) stmt = stmt.order_by( desc(SecawMaterial.is_pinned), asc(SecawMaterial.display_order), desc(SecawMaterial.created_at), ) rows = list((await self.db.execute(stmt)).scalars().all()) tokens = [t for t in (q or "").lower().split() if t] if tokens: rows = [r for r in rows if _matches(r, tokens)] total = len(rows) start = (page - 1) * page_size page_rows = rows[start:start + page_size] reach = await self._reach_map([r.id for r in page_rows]) # Which rows :meth:`delete` would refuse. Computed here so the # console can withhold the button rather than offering one whose # only outcome is a 409 — the same reason it already withholds it # for a material that has been shown. referenced = await self._referenced_ids([r.id for r in page_rows]) labels = await self._person_labels({ pid for r in page_rows for pid in (r.created_by_person_id, r.updated_by_person_id) if pid }) out: list[dict[str, Any]] = [] for r in page_rows: reached, impressions = reach.get(r.id, (0, 0)) created = labels.get(r.created_by_person_id) or (None, None) updated = labels.get(r.updated_by_person_id) or (None, None) out.append({ "id": r.id, "kind": r.kind, "category": r.category, "tags": list(r.tags or []), "localized": r.localized or {}, # Which language blocks a model wrote and nobody has # reviewed. The console badges them; the "AI 补齐" button # reads the same field to know what is still missing. "mt_langs": list(r.mt_langs or []), "status": r.status, "is_pinned": r.is_pinned, "display_order": r.display_order, "estimated_minutes": r.estimated_minutes, "version": r.version, "published_at": r.published_at, "archived_at": r.archived_at, "created_at": r.created_at, "updated_at": r.updated_at, "created_by_person_id": r.created_by_person_id, "updated_by_person_id": r.updated_by_person_id, # Name and email travel as SEPARATE fields. The platform # spent months un-splitting a "姓名 · 邮箱" string that a # dozen frontends each truncated their own way (#3417); # a new surface does not get to re-create that. "created_by_name": created[0], "created_by_email": created[1], "updated_by_name": updated[0], "updated_by_email": updated[1], "reached_people": reached, "impressions": impressions, # Mirrors :meth:`delete`'s guard exactly. A material that # has been shown, campaigned or answered is evidence, and # archive is the honest control for it. "deletable": impressions == 0 and r.id not in referenced, }) return out, total # ════════════════════════════════════════════════════════════════ # Admin mutations # ════════════════════════════════════════════════════════════════ async def create( self, body: MaterialCreate, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> SecawMaterial: self._assert_keys_owned(body.object_keys()) max_order = (await self.db.execute( select(SecawMaterial.display_order) .order_by(SecawMaterial.display_order.desc()) .limit(1) )).scalar() now = datetime.now(UTC) status = STATUS_PUBLISHED if body.publish else STATUS_DRAFT row = SecawMaterial( id=uuid.uuid4(), kind=body.kind, category=body.category, tags=body.tags, localized=body.normalized_localized(), status=status, is_pinned=body.is_pinned, display_order=(max_order if max_order is not None else -1) + 1, estimated_minutes=body.estimated_minutes, version=1, published_at=now if status == STATUS_PUBLISHED else None, created_at=now, updated_at=now, created_by_person_id=actor_id, updated_by_person_id=actor_id, ) self.db.add(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.material.create", status="success", resource_type="secaw_material", resource_id=str(row.id), detail=self._snapshot(row), ) return row async def update( self, material_id: uuid.UUID, body: MaterialUpdate, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> SecawMaterial: row = await self.db.get(SecawMaterial, material_id) if row is None: raise NotFoundError("security awareness material", str(material_id)) before = self._snapshot(row) # Kept for the machine-translation bookkeeping below: a language # only stops being "a model wrote this" when a human changed # THAT language's block, so we need the pre-edit blocks to # compare against. before_blocks = dict(row.localized or {}) # ``kind`` and ``localized`` are validated together — the "a comic # needs images" rule spans both, so changing either one alone has # to be re-checked against the other's *effective* value. new_kind = body.kind if body.kind is not None else row.kind content_changed = False if body.localized is not None or body.kind is not None: effective = ( body.localized if body.localized is not None else self._rehydrate_localized(row) ) try: merged = MaterialCreate( kind=new_kind, category=( body.category if body.category is not None else row.category ), tags=( body.tags if body.tags is not None else list(row.tags or []) ), localized=effective, estimated_minutes=( body.estimated_minutes if body.estimated_minutes is not None else row.estimated_minutes ), ) except ValidationError as exc: # This model is built HERE, not by FastAPI, so a raw # ``ValidationError`` escaping would surface as a 500 — # FastAPI only converts the ``RequestValidationError`` it # raises itself. A partial update that leaves the material # in an invalid shape (switch to ``comic`` without adding # images) is a client mistake and has to read like one, in # the standard envelope. Caught by # ``test_kind_and_content_are_validated_together``. raise AppException( code="SECAW_MATERIAL_INVALID", message="; ".join( str(e.get("msg", "invalid")) for e in exc.errors() ) or "invalid material payload", status_code=422, ) from exc self._assert_keys_owned(merged.object_keys()) new_localized = merged.normalized_localized() if new_localized != (row.localized or {}) or new_kind != row.kind: row.localized = new_localized row.kind = new_kind content_changed = True if body.category is not None and body.category != row.category: row.category = body.category content_changed = True if body.tags is not None and body.tags != list(row.tags or []): row.tags = body.tags content_changed = True if body.estimated_minutes is not None: row.estimated_minutes = body.estimated_minutes if body.is_pinned is not None: # Operational, not content — deliberately does NOT bump the # version. A campaign's "the text changed after I confirmed" # question must not be muddied by somebody re-pinning a card. row.is_pinned = body.is_pinned if content_changed: row.version += 1 # A human just edited the content. Any language whose block # actually changed is no longer an unreviewed machine # translation — but the ones they did NOT touch still are, # so this is a per-language comparison rather than a blanket # clear. (A ``kind``-only edit rewrites every block through # the validator; comparing values, not identity, is what # keeps that from silently clearing the whole badge set.) kept = [ lang for lang in (row.mt_langs or []) if before_blocks.get(lang) == (row.localized or {}).get(lang) ] row.mt_langs = kept or None row.updated_by_person_id = actor_id row.updated_at = datetime.now(UTC) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.material.update", status="success", resource_type="secaw_material", resource_id=str(row.id), detail={ "before": before, "after": self._snapshot(row), "content_changed": content_changed, }, ) return row async def set_status( self, material_id: uuid.UUID, status: str, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> SecawMaterial: row = await self.db.get(SecawMaterial, material_id) if row is None: raise NotFoundError("security awareness material", str(material_id)) if status == STATUS_PUBLISHED and not self._renderable(row): raise AppException( code="SECAW_MATERIAL_UNRENDERABLE", message=( "cannot publish a material with no renderable content" ), status_code=400, ) before = row.status if before == status: return row now = datetime.now(UTC) row.status = status if status == STATUS_PUBLISHED: # First publish stamps the date; a re-publish after an # unpublish keeps the original — "when did the workforce # first get this" has one answer, and the audit log carries # the cycle. if row.published_at is None: row.published_at = now row.archived_at = None elif status == STATUS_ARCHIVED: row.archived_at = now row.updated_by_person_id = actor_id row.updated_at = now await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action=f"security_awareness.material.{status}", status="success", resource_type="secaw_material", resource_id=str(row.id), detail={"from": before, "to": status}, ) return row async def reorder( self, ordered_ids: list[uuid.UUID], *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> None: rows = list((await self.db.execute(select(SecawMaterial))).scalars().all()) by_id = {r.id: r for r in rows} seen: set[uuid.UUID] = set() order = 0 for mid in ordered_ids: row = by_id.get(mid) if row is None or mid in seen: continue seen.add(mid) row.display_order = order order += 1 # Anything the client did not mention (added in another tab) # keeps its relative sequence at the end rather than colliding # on 0 — same graceful-race handling as policy_notices. for row in sorted(rows, key=lambda r: (r.display_order, r.created_at)): if row.id in seen: continue row.display_order = order order += 1 await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.material.reorder", status="success", resource_type="secaw_material", resource_id="", detail={"ordered": [str(i) for i in ordered_ids]}, ) async def duplicate( self, material_id: uuid.UUID, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> SecawMaterial: """Copy a material into a fresh draft. The main use is versioning content the compliance story depends on: last year's phishing comic must stay exactly as it was shown (its impressions point at it), so "update it for this year" is a copy, not an edit. """ src = await self.db.get(SecawMaterial, material_id) if src is None: raise NotFoundError("security awareness material", str(material_id)) localized = { lang: dict(block) for lang, block in (src.localized or {}).items() if isinstance(block, dict) } max_order = (await self.db.execute( select(SecawMaterial.display_order) .order_by(SecawMaterial.display_order.desc()) .limit(1) )).scalar() now = datetime.now(UTC) row = SecawMaterial( id=uuid.uuid4(), kind=src.kind, category=src.category, tags=list(src.tags or []), localized=localized, # The copy carries the same unreviewed machine translations # it copied — a duplicate is not a review, and dropping the # marker here would launder AI text into "human-authored". mt_langs=list(src.mt_langs or []) or None, status=STATUS_DRAFT, is_pinned=False, display_order=(max_order if max_order is not None else -1) + 1, estimated_minutes=src.estimated_minutes, version=1, created_at=now, updated_at=now, created_by_person_id=actor_id, updated_by_person_id=actor_id, ) self.db.add(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.material.duplicate", status="success", resource_type="secaw_material", resource_id=str(row.id), detail={"source_id": str(src.id)}, ) return row async def mark_translation_reviewed( self, material_id: uuid.UUID, lang: str, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> SecawMaterial: """Clear the "a model wrote this" badge for one language. The badge is cleared automatically when a human EDITS the block, but the most common review outcome is "the translation is fine" — which produces no edit. Without this the badge would be permanent for exactly the translations that turned out good, and a badge that can never be cleared stops being read. Deliberately not a content change: no version bump (nothing an employee reads changed) and no effect on ``updated_at``'s meaning beyond recording who reviewed. """ row = await self.db.get(SecawMaterial, material_id) if row is None: raise NotFoundError("security awareness material", str(material_id)) current = list(row.mt_langs or []) if lang not in current: # Already human-owned. Idempotent: the console may double-fire # on a retry, and a 409 here would be noise, not information. return row remaining = [lg for lg in current if lg != lang] row.mt_langs = remaining or None row.updated_by_person_id = actor_id row.updated_at = datetime.now(UTC) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.material.translation_reviewed", status="success", resource_type="secaw_material", resource_id=str(row.id), detail={"lang": lang, "remaining": remaining}, ) return row async def delete( self, material_id: uuid.UUID, *, actor_id: uuid.UUID, trace_id: str | uuid.UUID, ) -> None: """Hard-delete — allowed **only** while nothing points at the material. Once one impression exists the row is evidence: deleting it would orphan the ledger and quietly shrink a coverage number that somebody may already have reported. The admin console offers archive instead, and says so. Campaigns and quiz attempts are checked as well, and not only for tidiness: both FKs are ``RESTRICT`` (deliberately — a campaign points at the material it sent), so a material that carries a campaign nobody has opened yet would clear an impressions-only guard and then fail in the driver with a 1451, which reaches the admin as a 500 and poisons the transaction. A refusal has to be this layer's decision, with a message that says what to do. """ row = await self.db.get(SecawMaterial, material_id) if row is None: raise NotFoundError("security awareness material", str(material_id)) shown = (await self.db.execute( select(func.count()) .select_from(SecawImpression) .where(SecawImpression.material_id == material_id) )).scalar() or 0 if shown: raise AppException( code="SECAW_MATERIAL_HAS_IMPRESSIONS", message=( "this material has already been shown to employees; " "archive it instead of deleting" ), status_code=409, ) from app.domains.security_awareness.models import ( SecawCampaign, SecawQuizAttempt, ) campaigned = (await self.db.execute( select(func.count()).select_from(SecawCampaign) .where(SecawCampaign.material_id == material_id) )).scalar() or 0 if campaigned: raise AppException( code="SECAW_MATERIAL_HAS_CAMPAIGNS", message=( "this material is used by an awareness campaign; delete " "the campaign first, or archive the material instead" ), status_code=409, ) attempted = (await self.db.execute( select(func.count()).select_from(SecawQuizAttempt) .where(SecawQuizAttempt.material_id == material_id) )).scalar() or 0 if attempted: raise AppException( code="SECAW_MATERIAL_HAS_QUIZ_ATTEMPTS", message=( "somebody has already answered this material's quiz; " "archive it instead of deleting" ), status_code=409, ) snapshot = self._snapshot(row) await self.db.delete(row) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="security_awareness.material.delete", status="success", resource_type="secaw_material", resource_id=str(material_id), detail=snapshot, ) # ════════════════════════════════════════════════════════════════ # Internals # ════════════════════════════════════════════════════════════════ @staticmethod def _renderable(row: SecawMaterial) -> bool: """True when at least one language block carries a title. Guards every read path. The write schema makes this impossible for new rows, but a read that assumes an invariant it does not enforce is how a 500 reaches an employee. """ for block in (row.localized or {}).values(): if isinstance(block, dict) and block.get("title"): return True return False @staticmethod def _rehydrate_localized(row: SecawMaterial) -> dict[str, Any]: """The stored payload, in the shape ``MaterialCreate`` accepts. Stored blocks omit keys they do not use (see ``MaterialWrite.normalized_localized``), so re-validating one has to fill the defaults back in rather than hand Pydantic a dict with missing optional keys and a missing ``image_keys`` list. """ out: dict[str, Any] = {} for lang, block in (row.localized or {}).items(): if not isinstance(block, dict): continue out[lang] = { "title": block.get("title") or "", "summary": block.get("summary"), "body": block.get("body"), "image_keys": list(block.get("image_keys") or []), "link_url": block.get("link_url"), "cover_key": block.get("cover_key"), } return out @staticmethod def _assert_keys_owned(keys: list[str]) -> None: """Refuse any object key that is not an awareness-purpose upload. Object keys arrive in a request body, and a material is a presigning surface: accepting an arbitrary key would let an admin (or anyone who compromised an admin session) turn this domain into a read oracle for every other purpose's bucket prefix — inventory evidence photos, disposal receipts, feed images. """ bad = [k for k in keys if not is_awareness_key(k)] if bad: raise AppException( code="SECAW_INVALID_OBJECT_KEY", message=( "object key does not belong to an awareness-material " f"upload: {bad[0]}" ), status_code=400, ) def _to_card( self, row: SecawMaterial, want_lang: str, cover_urls: dict[uuid.UUID, str | None], viewed: dict[uuid.UUID, datetime], ) -> dict[str, Any]: lang, block, is_fallback = pick_block(row.localized, want_lang) return { "id": row.id, "kind": row.kind, "category": row.category, "tags": list(row.tags or []), "title": block.get("title"), "summary": block.get("summary"), "cover_url": cover_urls.get(row.id), "estimated_minutes": estimate_minutes( row.kind, block, row.estimated_minutes, ), "is_pinned": row.is_pinned, "version": row.version, "published_at": row.published_at, "lang": lang, "is_fallback_lang": is_fallback, # True when the block being rendered was written by a model # (``translation.MaterialTranslator``) and no human has # touched it since. Two different statements to a reader — # ``is_fallback_lang`` says "this is not your language", # this one says "this IS your language, but a machine wrote # it" — and a surface that conflated them would tell an # English reader their English card was Chinese. "machine_translated": lang in (row.mt_langs or []), "machine_translated_from": ( BASELINE_LANG if lang in (row.mt_langs or []) else None ), "viewed_at": viewed.get(row.id), } async def _cover_urls( self, rows: Sequence[SecawMaterial], want_lang: str, ) -> dict[uuid.UUID, str | None]: """Presign one cover per material, in a single batch. A comic with no explicit cover falls back to its first page — an admin should not have to upload the same image twice for the card to look right. """ key_by_material: dict[uuid.UUID, str] = {} for row in rows: if not self._renderable(row): continue _lang, block, _fb = pick_block(row.localized, want_lang) key = block.get("cover_key") if not key and row.kind == KIND_COMIC: 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(key) for mid, key in key_by_material.items() } async def _viewed_map( self, person_id: uuid.UUID, material_ids: list[uuid.UUID], ) -> dict[uuid.UUID, datetime]: """When this reader last OPENED each material. Keyed on the ``detail`` surface only: being shown a card in the ambient bubble is not the same as having read the thing, and the library's read/unread marker must mean the stronger claim. """ if not material_ids: return {} rows = (await self.db.execute( select( SecawImpression.material_id, func.max(SecawImpression.occurred_at), ) .where( SecawImpression.person_id == person_id, SecawImpression.material_id.in_(material_ids), SecawImpression.surface == SURFACE_DETAIL, ) .group_by(SecawImpression.material_id) )).all() return {mid: seen for mid, seen in rows} async def _reach_map( self, material_ids: list[uuid.UUID], ) -> dict[uuid.UUID, tuple[int, int]]: """``{material_id: (distinct_people, total_impressions)}``. Two separate numbers on purpose: "shown 900 times" and "reached 300 people" answer different questions, and an admin console that prints only the first invites exactly the wrong conclusion about coverage. """ if not material_ids: return {} rows = (await self.db.execute( select( SecawImpression.material_id, func.count(func.distinct(SecawImpression.person_id)), func.count(), ) .where(SecawImpression.material_id.in_(material_ids)) .group_by(SecawImpression.material_id) )).all() return {mid: (int(people), int(total)) for mid, people, total in rows} async def _referenced_ids( self, material_ids: list[uuid.UUID], ) -> set[uuid.UUID]: """Material ids that a campaign or a quiz attempt points at. Both foreign keys are RESTRICT on purpose — a sent campaign points at the material it sent, and a graded attempt is evidence of what somebody answered — so these are exactly the rows :meth:`delete` has to refuse. """ if not material_ids: return set() from app.domains.security_awareness.models import ( SecawCampaign, SecawQuizAttempt, ) campaigned = set((await self.db.execute( select(SecawCampaign.material_id) .where(SecawCampaign.material_id.in_(material_ids)) .distinct() )).scalars().all()) attempted = set((await self.db.execute( select(SecawQuizAttempt.material_id) .where(SecawQuizAttempt.material_id.in_(material_ids)) .distinct() )).scalars().all()) return campaigned | attempted async def _person_labels( self, person_ids: set[uuid.UUID], ) -> dict[uuid.UUID, tuple[str | None, str | None]]: """``{person_id: (display_name, work_email)}`` — never joined into one string. See the note in :meth:`list_admin`.""" ids = [pid for pid in person_ids if pid] if not ids: return {} rows = (await self.db.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, ).where(IdentityPerson.id.in_(ids)) )).all() return {pid: (name, email) for pid, name, email in rows} @staticmethod def _snapshot(row: SecawMaterial) -> dict[str, Any]: """Audit detail. Carries the localized *titles* rather than the whole payload — an audit row holding a 60 KB markdown body twice (before + after) makes the log unreadable and the table large, and the material's own version already identifies the content.""" titles = { lang: block.get("title") for lang, block in (row.localized or {}).items() if isinstance(block, dict) } return { "kind": row.kind, "category": row.category, "status": row.status, "tags": list(row.tags or []), "titles": titles, "langs": sorted(titles), "machine_translated_langs": list(row.mt_langs or []), "version": row.version, "is_pinned": row.is_pinned, } __all__ = [ "ADMIN_MAX_PAGE_SIZE", "LIBRARY_SOFT_CAP", "SecurityAwarenessService", "estimate_minutes", "normalize_lang", "pick_block", ]