"""Twice a year, ask the owners whether it is still them. A register decays the moment nobody is asked to stand behind it. 553 recorded who owns each approval flow; this makes somebody say so out loud, on a cadence, and makes the answer measurable. Why a round is a real object rather than a query over ``owner_confirmed_at``: the governance question is 「2026-H2 那轮做完了 吗、完成率多少、谁没回」, and a bare timestamp cannot answer it. A flow registered after the round opened was never asked. A flow whose owner changed mid-round was asked of somebody else. An item's ``owner_person_id`` is therefore a snapshot of **who was asked**, not a live pointer at the current owner. Four item states, and the distinctions are the point: * ``pending`` — asked, no answer yet. * ``confirmed`` — "still me". The only thing that turns a name in a box into coverage. * ``transferred`` — "hand it to X". Counts as answered: the round measures whether a human engaged, not whether the name stayed the same. * ``lapsed`` — the round closed with no answer. Deliberately NOT folded into ``pending``: at close time the difference between "still waiting" and "nobody ever answered" is the entire finding. * ``dropped`` — the flow was archived, or the person asked has left, mid-round. Excluded from the completion rate; chasing an answer about a retired flow is not a governance gap, and counting it as one makes the number meaningless. ``final_stats`` freezes the counts at close so a transfer made in March cannot rewrite December's completion rate — the number is a record of an event, not a view over live rows. Both crons are seeded ENABLED. The opener is daily but only acts inside the seven days after June 1 / December 1: a deploy in the middle of a half-year must not retroactively push a card to every owner in the company (the same storm guard migration 352 established for the half-yearly access self-attestation). Revision ID: 556 Revises: 555 """ from __future__ import annotations import json as _json import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "556" down_revision = "555" branch_labels = None depends_on = None ROUNDS = "registry_flow_confirmation_rounds" ITEMS = "registry_flow_confirmation_items" OPEN_KEY = "approval_registry.owner_confirmation_semiannual" ESCALATE_KEY = "approval_registry.owner_confirmation_escalate" _OPEN_DESC = ( "每日 08:30 运行,仅在边界后 7 天窗口内开盘:6/1 开 H1、12/1 开 H2。" "开盘给每位在册负责人发**一张**飞书卡片(名下九条就是九条列在一张卡上," "不是九张卡),卡片带免登录的「全部确认」按钮(action token,purpose 限定)。" "刻意不追溯历史边界 —— 中途部署的第一件事不该是给全公司负责人补推一轮。" "没有负责人的流程不进这一轮:没有人可问,它们已经被覆盖率读面盯着。" ) _ESCALATE_DESC = ( "每日 09:30。开盘 14 天后仍未回复的,给负责人再推一次并抄送直属上级" "(上级不替他回答;沉默的真实原因往往是「本来就不该是他」)。按人去重。" "同时做收尾:中途失效的项标 dropped(流程已归档 / 被问的人已离职)," "到期的轮次自动关闭、未回复记 lapsed 并冻结完成率。" ) def upgrade() -> None: conn = op.get_bind() if sa.inspect(conn).has_table(ROUNDS): return op.create_table( ROUNDS, sa.Column("id", id_type_for("identity_persons"), primary_key=True), # ``2026-H2`` — the natural key, so a daily cron inside a # seven-day window opens one campaign, not seven. sa.Column("label", sa.String(32), nullable=False), sa.Column( "status", sa.String(16), nullable=False, server_default="open", ), sa.Column( "opened_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP"), ), sa.Column("due_at", sa.DateTime(timezone=True), nullable=True), sa.Column("closed_at", sa.DateTime(timezone=True), nullable=True), # NULL = the boundary cron opened it, not a person. sa.Column("opened_by", id_type_for("identity_persons"), nullable=True), sa.Column("final_stats", sa.JSON(), nullable=True), sa.Column( "notified_owners", sa.Integer(), nullable=False, server_default="0", ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP"), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP"), onupdate=sa.text("CURRENT_TIMESTAMP"), ), sa.ForeignKeyConstraint( ["opened_by"], ["identity_persons.id"], ondelete="SET NULL", ), **table_kw_for("identity_persons"), ) op.create_index( "uq_registry_flow_rounds_label", ROUNDS, ["label"], unique=True, ) op.create_index( "ix_registry_flow_rounds_status", ROUNDS, ["status", "opened_at"], ) op.create_table( ITEMS, sa.Column("id", id_type_for("identity_persons"), primary_key=True), sa.Column("round_id", id_type_for("identity_persons"), nullable=False), sa.Column("flow_id", id_type_for("identity_persons"), nullable=False), # Who was ASKED — not a live pointer at the flow's owner. sa.Column( "owner_person_id", id_type_for("identity_persons"), nullable=True, ), # pending | confirmed | transferred | lapsed | dropped sa.Column( "state", sa.String(16), nullable=False, server_default="pending", ), sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True), # web | mobile | token — "did the card actually work" is a # question the next round will ask. sa.Column("decided_via", sa.String(16), nullable=True), sa.Column( "transferred_to_person_id", id_type_for("identity_persons"), nullable=True, ), sa.Column("note", sa.Text(), nullable=True), sa.Column("notified_at", sa.DateTime(timezone=True), nullable=True), sa.Column("escalated_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP"), ), sa.ForeignKeyConstraint( ["round_id"], [f"{ROUNDS}.id"], ondelete="CASCADE", ), sa.ForeignKeyConstraint( ["flow_id"], ["registry_approval_flows.id"], ondelete="CASCADE", ), sa.ForeignKeyConstraint( ["owner_person_id"], ["identity_persons.id"], ondelete="SET NULL", ), sa.ForeignKeyConstraint( ["transferred_to_person_id"], ["identity_persons.id"], ondelete="SET NULL", ), **table_kw_for("identity_persons"), ) op.create_index( "uq_registry_flow_items_round_flow", ITEMS, ["round_id", "flow_id"], unique=True, ) op.create_index( "ix_registry_flow_items_round_state", ITEMS, ["round_id", "state"], ) op.create_index( "ix_registry_flow_items_owner", ITEMS, ["owner_person_id", "state"], ) for key, display, desc, cron, config in ( ( OPEN_KEY, "审批流程台账 · 半年度负责人确认开盘", _OPEN_DESC, "30 8 * * *", {"window_days": 7, "due_days": 30}, ), ( ESCALATE_KEY, "审批流程台账 · 确认逾期升级", _ESCALATE_DESC, "30 9 * * *", {"after_days": 14}, ), ): conn.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": key, "display": display, "desc": desc, "handler": key, "cron": cron, "config": _json.dumps(config), }, ) def downgrade() -> None: conn = op.get_bind() conn.execute( sa.text("DELETE FROM scheduled_jobs WHERE handler_key IN (:a, :b)"), {"a": OPEN_KEY, "b": ESCALATE_KEY}, ) # Children before parents: the items table FKs the rounds table. if sa.inspect(conn).has_table(ITEMS): op.drop_table(ITEMS) if sa.inspect(conn).has_table(ROUNDS): op.drop_table(ROUNDS)