"""ORM rows owned by the security_awareness domain (安全宣贯). Six tables: * :class:`SecawMaterial` — the content library. A comic, an article or a link, carried in up to three languages. (migration 596) * :class:`SecawImpression` — the exposure ledger. Append-only; one row per (person, material, surface) exposure outside the dedupe window. (migration 596; gains ``campaign_id`` in 597) * :class:`SecawCampaign` — one act of 宣贯: material × audience × channel × moment × strength. (migration 597) * :class:`SecawCampaignTarget` — the per-person ledger for a campaign: who we sent it to, who opened it, who acknowledged. (migration 597) * :class:`SecawMaterialQuestion` — quiz questions attached to a material, reusing the existing secquiz_questions bank. (600) * :class:`SecawQuizAttempt` — every go at a material's quiz, passed or not. (migration 600) **Why the ledger exists at all.** The security tips that rotate in the mascot bubble today record nothing, so the honest answer to "prove this employee was told about phishing" is currently "we can't". Everything in this domain is built around making that answerable: the material carries a version, the exposure carries a timestamp and a surface, and neither is ever destroyed by an edit — an admin retires a material by archiving it, not by deleting the evidence that it was shown. Later PRs in this theme add the ambient surface (PR3) and a compliance control that reads coverage off these ledgers (PR4). """ from __future__ import annotations import uuid from datetime import datetime from typing import Any from sqlalchemy import ( JSON, Boolean, DateTime, ForeignKey, Index, Integer, String, UniqueConstraint, ) from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy.sql import func from app.core.database import Base from app.core.sql_types import GUID as Uuid from app.domains.security_awareness.catalog import ( KIND_ARTICLE, STATUS_DRAFT, ) class SecawMaterial(Base): """One piece of security-awareness content. **Localized content lives in one JSON column, not 18 flat columns.** Three languages × (title, summary, body, image list, link, cover) is eighteen mostly-NULL columns whose only relationship — "these six belong to the same language" — the schema could not express anyway. The JSON column carries that grouping explicitly:: { "zh": { "title": "识破钓鱼邮件的五个细节", "summary": "……", # optional "body": "## 正文…", # markdown, kind=article "image_keys": ["awareness_material/2026/09/…/p1.png", …], "link_url": "https://…", # kind=link "cover_key": "awareness_material/…/cover.png" }, "en": { … }, "ja": { … } } A language key is present only when that language has content, so "which languages does this material exist in" is ``localized.keys()`` rather than six NULL checks. The shape is validated in ``schemas.LocalizedContent`` and normalised in the service before it reaches this column — nothing writes here directly. ``zh`` is always present (the fallback chain terminates there). ``image_keys`` / ``cover_key`` hold **S3 object keys**, never URLs. URLs are minted per read as short-lived presigned GETs, which is what keeps the bucket private; a stored URL would either expire in the column or, worse, be a permanent public link. ``version`` increments on every *content* edit (localized payload, kind, category). It is not decoration: a campaign records the version it went out with, so "the article said something different when I confirmed it" has an answer. Purely operational edits (pin, order, status) do not bump it. """ __tablename__ = "secaw_materials" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( # The employee library read: published rows, newest first. Index("ix_secaw_materials_live", "status", "published_at"), # The admin console's category filter and the coverage report's # per-category rollup. Index("ix_secaw_materials_category", "category", "status"), # Ambient-surface pick order (pinned first, then display order). Index("ix_secaw_materials_pinned", "is_pinned", "display_order"), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) # comic | article | link — see catalog.MATERIAL_KINDS. kind: Mapped[str] = mapped_column( String(16), nullable=False, default=KIND_ARTICLE, server_default=KIND_ARTICLE, ) # One of catalog.MATERIAL_CATEGORIES. Closed list on purpose. category: Mapped[str] = mapped_column( String(32), nullable=False, default="other", server_default="other", ) # Language-agnostic free tags, for search and for "more like this". # NULL coerces to [] at the schema layer (legacy-row convention used # across this codebase — see platform_policy_notices.keywords). tags: Mapped[list[str] | None] = mapped_column(JSON, nullable=True) # See the class docstring. Never written to directly — the service # normalises through ``schemas.LocalizedContent``. localized: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) # Languages in ``localized`` that a MODEL wrote, not a human # (migration 607). ``["en", "ja"]``; NULL / [] means every block is # human-authored. # # Why a column instead of a flag inside the block: the write path # (``schemas.MaterialWrite.normalized_localized``) whitelists the # keys it persists, so a marker smuggled into the JSON would be # dropped by any admin save — including a ``kind``-only edit, which # round-trips every language through the same validator. The badge # would then disappear for reasons unrelated to anybody reviewing # the translation. Here it is cleared deliberately: per-language on a # real content edit (``service.update``) and by the explicit # "reviewed" endpoint. mt_langs: Mapped[list[str] | None] = mapped_column(JSON, nullable=True) # draft | published | archived — see catalog.MATERIAL_STATUSES. status: Mapped[str] = mapped_column( String(16), nullable=False, default=STATUS_DRAFT, server_default=STATUS_DRAFT, ) # Sticks to the top of the employee library and is preferred by the # ambient surface. Independent of ``display_order`` so an admin can # promote one material without renumbering the whole list. is_pinned: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False, server_default="0", ) # Lower renders first within the pinned / unpinned group. display_order: Mapped[int] = mapped_column( Integer, nullable=False, default=0, server_default="0", ) # Admin override for the "约 N 分钟" hint. NULL → derived from the # content length at read time (see service._estimate_minutes). estimated_minutes: Mapped[int | None] = mapped_column( Integer, nullable=True, ) # Bumped on content edits only. See the class docstring. version: Mapped[int] = mapped_column( Integer, nullable=False, default=1, server_default="1", ) # First transition into ``published``. Not reset by a later # unpublish/republish cycle — "when did employees first get this" # has one answer, and the audit log carries the cycle. published_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) archived_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) created_by_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) updated_by_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) def __repr__(self) -> str: # pragma: no cover - debug helper return f"" class SecawImpression(Base): """One recorded exposure of one material to one person. Append-only. Nothing updates or deletes a row here — this is the table an auditor reads when asking "show me that the workforce was actually reached", and a ledger that gets rewritten is not evidence. Rows are deduped by the service inside ``catalog.IMPRESSION_DEDUPE_MINUTES``: re-rendering the same ambient bubble on every route change is not a new exposure, and without the window the table would grow without bound while making "shown N times" meaningless. ``material_id`` deliberately has **no** ``ondelete="CASCADE"``: materials are archived, never deleted (see :class:`SecawMaterial`), and a cascade here would quietly convert a mis-click on a delete button into the destruction of the coverage record. ``person_id`` does cascade — a person row is only ever hard-deleted during a data correction, at which point their impressions are personal data we are removing on purpose. """ __tablename__ = "secaw_impressions" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( # The dedupe probe and the per-person "have I seen this" read. Index( "ix_secaw_impr_person_material", "person_id", "material_id", "occurred_at", ), # Per-material reach rollups for the admin console. Index("ix_secaw_impr_material", "material_id", "occurred_at"), # PR3's frequency gate: "how much has this person been shown # today". Kept here so the gate never needs a schema change. Index("ix_secaw_impr_person_time", "person_id", "occurred_at"), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) person_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="CASCADE"), nullable=False, ) material_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("secaw_materials.id"), nullable=False, ) # detail | bubble | campaign_card — see catalog.IMPRESSION_SURFACES. surface: Mapped[str] = mapped_column(String(24), nullable=False) # The material version the person actually saw. Copied at write time # because the material can be edited afterwards, and "which text did # they see" is exactly what a dispute is about. material_version: Mapped[int] = mapped_column( Integer, nullable=False, default=1, server_default="1", ) # Which localized block was rendered for them. Answers "was this # employee shown the English copy" without re-deriving their language # preference months later (it may have changed since). lang: Mapped[str] = mapped_column( String(8), nullable=False, default="zh", server_default="zh", ) # Which campaign produced this exposure, when one did. NULL for the # library / ambient surfaces and for every row written before PR2 — # "not attributable to a campaign" is a real answer, not a gap. campaign_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("secaw_campaigns.id", ondelete="SET NULL"), nullable=True, ) occurred_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) def __repr__(self) -> str: # pragma: no cover - debug helper return f"" class SecawCampaign(Base): """One act of 宣贯: a material, an audience, a channel, a moment. **Two strengths, one column.** ``mandatory`` decides whether this is work an employee owes (lands in ``/my/todos``, has a ``due_at``, gets chased, counts toward coverage, and pushes through the immutable ``security`` notification category) or a piece of operational outreach (mutable ``awareness`` category, info bucket, no red dot). Without that split every awareness push becomes something an employee cannot turn off — which is exactly how a channel gets muted, and then the mandatory ones stop being read too. **The audience is expanded at send time, into snapshots.** People change departments; a campaign that re-resolved its audience on every read would silently rewrite who it was sent to. ``audience_config`` describes the intent, ``secaw_campaign_targets`` records the fact. **``status`` is also the send lock.** The transition into ``sending`` is a conditional UPDATE, so two dispatch ticks landing on the same row cannot both send it. """ __tablename__ = "secaw_campaigns" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( # The dispatch cron's question, every ten minutes. Index("ix_secaw_campaigns_due", "status", "scheduled_at"), Index("ix_secaw_campaigns_material", "material_id"), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) # No ``ondelete``: a material is archived rather than deleted, and a # campaign that already went out is pointing at evidence. material_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("secaw_materials.id"), nullable=False, ) # The material version this campaign actually went out with. NULL # until it sends. This is the column that answers "the article said # something else when I confirmed". material_version: Mapped[int | None] = mapped_column(Integer) # Admin-facing label. Deliberately single-language and internal — it # never reaches an employee, who sees the material's own title. name: Mapped[str] = mapped_column(String(200), nullable=False) # ``app.core.person_filters.FilterConfig`` shape plus this domain's # own ``exclude_person_ids`` overlay. See ``audience.py`` for why the # exclusion is applied here rather than added to the shared engine. audience_config: Mapped[dict[str, Any] | None] = mapped_column(JSON) # Subset of ``catalog.CAMPAIGN_CHANNELS``. The in-app bell is always # written (every dispatch creates an outbox row) and is therefore not # listed here — it is the channel of last resort, not a choice. channels: Mapped[list[str]] = mapped_column(JSON, nullable=False) mandatory: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False, server_default="0", ) completion_mode: Mapped[str] = mapped_column( String(16), nullable=False, default="ack", server_default="ack", ) status: Mapped[str] = mapped_column( String(16), nullable=False, default="draft", server_default="draft", ) scheduled_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), ) due_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # Last dispatch failure, surfaced in the console. A half-failed # fan-out that looks successful is worse than one that looks broken. last_dispatch_error: Mapped[str | None] = mapped_column(String(500)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) created_by_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) updated_by_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) def __repr__(self) -> str: # pragma: no cover - debug helper return f"" class SecawCampaignTarget(Base): """One person's row in one campaign — who we sent it to, and what they did with it. **There is no ``delivered_at`` here on purpose.** Delivery has exactly one source of truth: the ``notification_outbox`` row, which owns the retry ladder, the ``dead`` terminal state and the leaver gate. Mirroring a delivery column onto this table would create a second truth, and two truths drift. The campaign detail read joins the outbox on ``(related_resource_type, related_resource_id, recipient_person_id)`` — a shape that already has an index (``ix_notification_outbox_related_resource``). ``(campaign_id, person_id)`` is unique so a resend to the people who have not acknowledged updates their row instead of opening a second one; without it "sent to N people" would grow every time somebody pressed 催办. Every snapshot column (``material_version``, ``lang``, ``person_snapshot``) is written once, at send time. Materials get edited and people change departments; re-deriving these at read time would answer a different question than the one being asked. """ __tablename__ = "secaw_campaign_targets" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint( "campaign_id", "person_id", name="uq_secaw_target_campaign_person", ), # The campaign detail funnel + per-person table. Index("ix_secaw_targets_campaign_ack", "campaign_id", "acknowledged_at"), # "What awareness do I still owe" — the /my/todos bucket. Index("ix_secaw_targets_person_ack", "person_id", "acknowledged_at"), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) campaign_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("secaw_campaigns.id", ondelete="CASCADE"), nullable=False, ) person_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="CASCADE"), nullable=False, ) # ``{display_name, work_email, department, country}`` as at send time. person_snapshot: Mapped[dict[str, Any] | None] = mapped_column(JSON) material_version: Mapped[int] = mapped_column( Integer, nullable=False, default=1, server_default="1", ) lang: Mapped[str] = mapped_column( String(8), nullable=False, default="zh", server_default="zh", ) # When we handed it to the outbox. NOT when it was delivered. queued_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) opened_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) acknowledged_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), ) ack_source: Mapped[str | None] = mapped_column(String(16)) # Resends are visible: somebody chased four times and somebody chased # once are not the same situation. resend_count: Mapped[int] = mapped_column( Integer, nullable=False, default=0, server_default="0", ) last_sent_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) def __repr__(self) -> str: # pragma: no cover - debug helper return f"" class SecawMaterialQuestion(Base): """A quiz question attached to a material. A link table rather than a ``material_id`` column on the question: one question can be both a material's companion question and a member of the daily-quiz bank, and a column would lock it to one material. The table also carries the reading order and makes "which materials reuse this question" answerable. """ __tablename__ = "secaw_material_questions" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint( "material_id", "question_id", name="uq_secaw_material_question", ), Index( "ix_secaw_material_questions_material", "material_id", "sort_order", ), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) material_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("secaw_materials.id", ondelete="CASCADE"), nullable=False, ) question_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("secquiz_questions.id", ondelete="CASCADE"), nullable=False, ) sort_order: Mapped[int] = mapped_column( Integer, nullable=False, default=0, server_default="0", ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) created_by_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) class SecawQuizAttempt(Base): """One person's one go at a material's quiz. **Every attempt is kept, including the failed ones.** Storing only "eventually passed" would throw away the most useful signal this domain produces: which questions a lot of people get wrong the first time is direct evidence that a particular passage of the material is not clear, and that is the input to the security team's next edit. ``passed`` and ``answers`` are stored rather than recomputed. A verdict has to use the answer key **as it was at the time** — the questions get edited afterwards, and re-grading an old attempt against a new key would silently rewrite somebody's result. """ __tablename__ = "secaw_quiz_attempts" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( # "have I passed this" per person, and the per-material rollup. Index( "ix_secaw_attempts_person_material", "person_id", "material_id", "passed", ), Index("ix_secaw_attempts_material", "material_id", "created_at"), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) person_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="CASCADE"), nullable=False, ) material_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("secaw_materials.id"), nullable=False, ) material_version: Mapped[int] = mapped_column( Integer, nullable=False, default=1, server_default="1", ) #: 1-based ordinal within (person, material). NOT derived from #: ``created_at``: that column is a second-granularity DATETIME, and #: the single most interesting sequence here — miss, read the #: explanation, retry — happens well inside one second, so ordering #: by time cannot tell the first attempt from the second. The #: per-question first-attempt miss rate (the security team's signal #: that a passage is unclear) depends entirely on getting that right. attempt_no: Mapped[int] = mapped_column( Integer, nullable=False, default=1, server_default="1", ) correct_count: Mapped[int] = mapped_column(Integer, nullable=False) total_count: Mapped[int] = mapped_column(Integer, nullable=False) passed: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False, server_default="0", ) #: ``[{question_id, selected, correct, is_correct}]``. answers: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), )