"""安全宣贯:活动 + 逐人台账。 ## 这是在解决什么 596 建了素材和曝光台账,但「把哪份素材,发给哪些人,走哪个渠道,什么时候发, 要不要回执」这件事还没有对象。这两张表就是那个对象。 | 表 | 是什么 | |---|---| | `secaw_campaigns` | 一次宣贯:素材 × 受众 × 渠道 × 时机 × 强度 | | `secaw_campaign_targets` | 逐人台账:发给了谁、他打开没有、确认没有 | ## 两档强度是一个列,不是两张表 `mandatory=True` 的活动进 `/my/todos`、有 `due_at`、会被催办、算进覆盖率, 并且走 `CAT_SECURITY`(员工关不掉);`mandatory=False` 是纯运营触达,走 新的 `CAT_AWARENESS`(可关、INFO 桶、不驱动红点)。**不分这两档的话,所有 宣贯都会变成员工关不掉的推送**,而那正是一个渠道被静音的方式。 ## 投递状态刻意不在这张表里 `secaw_campaign_targets` 有 `queued_at` / `opened_at` / `acknowledged_at`, **没有 `delivered_at`**。投递真值只有一个来源 —— `notification_outbox` 的那 一行(它自己带重试阶梯、`dead` 终态和离职闸),镜像一列过来就是造第二个真 值,而两个真值一定会漂。读活动详情时按 `(related_resource_type='secaw_campaign', related_resource_id, recipient_person_id)` join 回 outbox 取,这个形状在 `ix_notification_outbox_related_resource` (迁移 509)上有索引。 ## `(campaign_id, person_id)` 唯一 重发只补发「还没确认的人」,而重发不能给同一个人开第二行台账 —— 否则 「发给了多少人」会随重发次数膨胀。唯一约束把重入变成 UPDATE。 ## `material_version` / `lang` / `person_snapshot` 都是写入时的快照 素材之后还会被编辑、人之后还会换部门。「他确认的时候看到的是哪一版」和 「他当时属于哪个部门」是这张表存在的理由,现算等于答错。 ## `secaw_impressions.campaign_id` 596 留了 `campaign_card` 这个 surface 但没有列可以说「是哪一次活动」。补一列 (可空,历史行与非活动曝光都是 NULL),这样 PR4 的覆盖率能按活动切。 Revision ID: 597 Revises: 596 ⚠️ 迁移号要对着 `origin/main` 抢,而且每次 push 之前重新抢一次 —— 本仓踩过的 号是 462 / 472 / 582 / 587,两个分支同时拿同一个号,合进 main 之后链上出现 两个 head,后端每个测试分片和镜像构建一起红。 """ from __future__ import annotations import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for revision = "597" down_revision = "596" branch_labels = None depends_on = None CAMPAIGNS = "secaw_campaigns" TARGETS = "secaw_campaign_targets" IMPRESSIONS = "secaw_impressions" MATERIALS = "secaw_materials" JOB_KEY = "security_awareness.campaign_dispatch" _JOB_DESC = ( "把到点的安全宣贯活动发出去。每 10 分钟看一次 `secaw_campaigns` 里 " "`status='scheduled'` 且 `scheduled_at <= now()` 的行,展开受众、逐人建" "台账、经通知 outbox 发飞书卡片 / 邮件 / 站内铃铛。\n" "**只发一次**:活动状态在发送前先置 `sending`(行锁 + 状态条件更新)," "所以两个 tick 撞上同一行时第二个什么都不做;逐人台账的 " "`(campaign_id, person_id)` 唯一约束让重入变成 UPDATE 而不是第二行。\n" "受众在**发送那一刻**展开成快照 —— 活动发出后的人员变动不改写已发历史。" ) def _has_table(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :n" ), {"n": name}, ).scalar() is not None def _has_column(bind, table: str, column: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.columns " "WHERE table_schema = DATABASE() " "AND table_name = :t AND column_name = :c" ), {"t": table, "c": column}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() # FK 列的类型一律向 id_type_for 要 —— TiDB 3780 只认「和父列完全一致」。 person_fk = id_type_for("identity_persons") if not _has_table(bind, CAMPAIGNS): op.create_table( CAMPAIGNS, sa.Column("id", sa.String(36), primary_key=True), # 刻意没有 ondelete:素材只归档不删除,而一次已经发出去的活动 # 指向的那份素材是证据。 sa.Column( "material_id", id_type_for(MATERIALS), sa.ForeignKey(f"{MATERIALS}.id"), nullable=False, ), # 发出去那一刻素材的版本。未发送时为 NULL。 sa.Column("material_version", sa.Integer()), # 管理员自己看的活动名(单语,内部标识,不面向员工)。 sa.Column("name", sa.String(200), nullable=False), # app.core.person_filters.FilterConfig 的形状 + 本域自己的 # exclude_person_ids 叠加层(共享引擎没有排除维度,而它同时是 # 拉取面的授权边界,不该为一个内容功能去放宽)。 sa.Column("audience_config", sa.JSON()), # ["feishu", "email", "inbox"] 的子集。 sa.Column("channels", sa.JSON(), nullable=False), sa.Column( "mandatory", sa.Boolean(), nullable=False, server_default="0", ), # impression | open | ack —— 哪一档算「完成」。 sa.Column( "completion_mode", sa.String(16), nullable=False, server_default="ack", ), # draft | scheduled | sending | sent | closed | cancelled sa.Column( "status", sa.String(16), nullable=False, server_default="draft", ), sa.Column("scheduled_at", sa.DateTime(timezone=True)), sa.Column("due_at", sa.DateTime(timezone=True)), sa.Column("sent_at", sa.DateTime(timezone=True)), sa.Column("closed_at", sa.DateTime(timezone=True)), # 最近一次发送/补发的失败摘要。绿灯不是证据,一次半途失败的 # 群发必须在管理端看得见。 sa.Column("last_dispatch_error", sa.String(500)), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "created_by_person_id", person_fk, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), ), sa.Column( "updated_by_person_id", person_fk, sa.ForeignKey("identity_persons.id", ondelete="SET NULL"), ), ) # cron 每 10 分钟问的那个问题。 op.create_index( "ix_secaw_campaigns_due", CAMPAIGNS, ["status", "scheduled_at"], ) op.create_index( "ix_secaw_campaigns_material", CAMPAIGNS, ["material_id"], ) if not _has_table(bind, TARGETS): op.create_table( TARGETS, sa.Column("id", sa.String(36), primary_key=True), sa.Column( "campaign_id", id_type_for(CAMPAIGNS), sa.ForeignKey(f"{CAMPAIGNS}.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "person_id", person_fk, sa.ForeignKey("identity_persons.id", ondelete="CASCADE"), nullable=False, ), # 写入时的快照 —— 人之后会换部门,而「他当时属于哪个部门」是 # 按部门看覆盖率时唯一说得通的口径。 sa.Column("person_snapshot", sa.JSON()), sa.Column( "material_version", sa.Integer(), nullable=False, server_default="1", ), sa.Column( "lang", sa.String(8), nullable=False, server_default="zh", ), # 我们把它交给 outbox 的时刻。**投递状态不在这张表里** —— # 见模块文档。 sa.Column("queued_at", sa.DateTime(timezone=True)), sa.Column("opened_at", sa.DateTime(timezone=True)), sa.Column("acknowledged_at", sa.DateTime(timezone=True)), # web | mobile | mcp —— 确认是从哪个面点的。 sa.Column("ack_source", sa.String(16)), # 补发次数。重发只补「还没确认的人」,而次数要看得见:一个被 # 催了四次的人和一个被催了一次的人不是同一件事。 sa.Column( "resend_count", sa.Integer(), nullable=False, server_default="0", ), sa.Column("last_sent_at", sa.DateTime(timezone=True)), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.UniqueConstraint( "campaign_id", "person_id", name="uq_secaw_target_campaign_person", ), ) # 活动详情的漏斗与逐人表。 op.create_index( "ix_secaw_targets_campaign_ack", TARGETS, ["campaign_id", "acknowledged_at"], ) # 「我还欠哪些宣贯」—— /my/todos 的那一格。 op.create_index( "ix_secaw_targets_person_ack", TARGETS, ["person_id", "acknowledged_at"], ) # 596 留了 campaign_card 这个 surface 却没有列说「是哪一次活动」。 if _has_table(bind, IMPRESSIONS) and not _has_column( bind, IMPRESSIONS, "campaign_id", ): op.add_column( IMPRESSIONS, sa.Column( "campaign_id", id_type_for(CAMPAIGNS), sa.ForeignKey(f"{CAMPAIGNS}.id", ondelete="SET NULL"), nullable=True, ), ) # ── 调度 cron ──────────────────────────────────────────────── if _has_table(bind, "scheduled_jobs"): bind.execute( sa.text( "INSERT INTO scheduled_jobs " "(id, name, display_name, description, handler_key, " " trigger_type, cron_expr, timezone, enabled, config, " " run_count, failure_count, created_at, updated_at) " "SELECT :id, :name, :display, :desc, :handler, 'cron', " ":cron, 'Asia/Shanghai', 1, :config, 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " "(SELECT 1 FROM scheduled_jobs WHERE handler_key = :handler)", ), { "id": str(uuid.uuid4()), "name": JOB_KEY, "display": "安全宣贯 · 发送到点的活动", "desc": _JOB_DESC, "handler": JOB_KEY, # 每 10 分钟。宣贯不是分钟级业务,而更密的 tick 只会让一次 # 卡住的群发更快地重试同一件事。 "cron": "*/10 * * * *", "config": "{}", }, ) def downgrade() -> None: bind = op.get_bind() if _has_table(bind, "scheduled_jobs"): bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"), {"k": JOB_KEY}, ) if _has_table(bind, IMPRESSIONS) and _has_column( bind, IMPRESSIONS, "campaign_id", ): # MySQL / TiDB refuse to drop a column that is still part of a # foreign key (1828), and the constraint name here is # auto-generated (``secaw_impressions_ibfk_N``) — so look it up # rather than guessing. Caught by actually replaying # upgrade → downgrade → upgrade against an empty database; the # first version of this downgrade failed halfway, after it had # already deleted the cron row. for (fk_name,) in bind.execute( sa.text( "SELECT DISTINCT constraint_name " "FROM information_schema.key_column_usage " "WHERE table_schema = DATABASE() AND table_name = :t " "AND column_name = :c AND referenced_table_name IS NOT NULL" ), {"t": IMPRESSIONS, "c": "campaign_id"}, ).all(): op.drop_constraint(fk_name, IMPRESSIONS, type_="foreignkey") op.drop_column(IMPRESSIONS, "campaign_id") # 子表先删:MySQL / TiDB 都不允许先删被 FK 指着的父表。 if _has_table(bind, TARGETS): op.drop_table(TARGETS) if _has_table(bind, CAMPAIGNS): op.drop_table(CAMPAIGNS)