"""转岗复核包裹:把「转岗要看什么」从「只有云授权」扩成这个人真正持有的一切。 迁移 565 让转岗第一次真的会触发。这两张表是它触发进去的地方。 在此之前,一次转岗复核的全部内容是**云授权**。拿生产量过,那只是一个人实际 持有物的一小块 —— 一位资深工程师名下是 **238 项、跨 10 个类别**(44 个域名、 111 条 DNS 记录、18 个软件负责人、19 个 IT 联系人、13 个坐席、9 条云授权、 8 个审批流程负责人、3 个邮件组…),转岗之后它们全都还指着他。离职域早就把这 26 类枚举得清清楚楚并逐条要求交接;转岗域只看其中一类。 **包裹是问题,不是动作。** issue #623 的既定策略在这里一个字没改:转岗**不 自动回收任何权限**(法务在一次季中误回收之后要求必须有人做决定)。这两张表 只负责把「该问的都问到」和「谁答了什么」记下来。 两个刻意的形状: * **`(person_id, effective_date)` 唯一** —— 和 role_change 生命周期事件同一个 键。协调器被重放时不会为同一次移动再开一份包裹。 * **`(review_id, resource_type, resource_id)` 唯一** —— 重新扫描是幂等的: 第二次扫到同样的持有物不会重复入库,也不会冲掉已经做过的决定。 `resource_type` 用的是**离职域那套词表**(`RES_*`),不是新造一套。两边用同 一种语言描述「一个人持有什么」,才可能用一条结构性测试挡住「离职域加了一类、 转岗域悄悄没有」—— 那正是 565 刚在这条路的另一头修掉的那类 bug。 `move_kind` 是确定性派生的,不来自模型:「同部门换了个经理」和「从研发去了 销售」不是对同样 238 件事的同一个问题,把两者当成一回事正是复核界面训练人闭 眼点确认的方式。 Revision ID: 568 Revises: 567 """ from __future__ import annotations import json as _json import uuid import sqlalchemy as sa from alembic import op revision = "568" down_revision = "567" branch_labels = None depends_on = None REVIEWS = "hr_role_change_reviews" ITEMS = "hr_role_change_review_items" BACKFILL_JOB = "role_change.review_backfill" NOTIFY_JOB = "role_change.review_notify" _BACKFILL_DESC = ( "每天 03:40 (Asia/Shanghai) 复核最近 90 天内的 role_change 生命周期事件:" "凡是跨部门或换经理、却没有对应复核包裹的,补建一份。包裹由协调器内联构建," "而协调器按平台惯例自吞异常 —— 一次失败完全隐形。内联构建保证「移动落地即有" "包裹」,这个扫描保证「没有一次移动没有包裹」。只改职级的移动不建包裹。" "按 (person_id, effective_date) 幂等,重复运行空转,不覆盖任何已做出的决定。" ) _NOTIFY_DESC = ( "每天 09:30 (Asia/Shanghai) 把尚未通知过的开放复核包裹推给复核人 —— 接手这位" "同事的新经理。为什么不是 IT:管理员看到「这个人有 9 条云授权」无从判断他还" "需不需要,唯一知道的人是刚把他接过来的那位。新经理已离职或没有经理时回落给" "在职 global_admin。卡片是摘要(总量 + 类别分布 + 模型初判),处理在页面上按" "类别整批做;每份包裹只通知一次。" ) def _has_table(conn: sa.engine.Connection, name: str) -> bool: # DATABASE()-qualified: an unqualified information_schema probe # matches a same-named table in ANY schema, and this box runs # hundreds of helix_test_* schemas side by side. return bool( conn.execute( sa.text( "SELECT COUNT(*) FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :t", ), {"t": name}, ).scalar() ) def upgrade() -> None: conn = op.get_bind() if not _has_table(conn, REVIEWS): op.create_table( REVIEWS, sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("person_id", sa.CHAR(36), nullable=False), sa.Column("lifecycle_event_id", sa.CHAR(36), nullable=True), sa.Column("effective_date", sa.Date(), nullable=False), sa.Column("move_kind", sa.String(24), nullable=False), sa.Column( "status", sa.String(16), nullable=False, server_default="open", ), sa.Column("before_dept_code", sa.String(64)), sa.Column("after_dept_code", sa.String(64)), sa.Column("before_dept_path", sa.String(512)), sa.Column("after_dept_path", sa.String(512)), sa.Column("before_manager_person_id", sa.CHAR(36)), sa.Column("after_manager_person_id", sa.CHAR(36)), sa.Column("before_job_title", sa.String(128)), sa.Column("after_job_title", sa.String(128)), sa.Column("reviewer_person_id", sa.CHAR(36)), sa.Column("notified_at", sa.DateTime(timezone=True)), sa.Column( "total_items", sa.Integer(), nullable=False, server_default="0", ), sa.Column( "reviewable_items", sa.Integer(), nullable=False, server_default="0", ), sa.Column( "decided_items", sa.Integer(), nullable=False, server_default="0", ), sa.Column("failed_scan_types", sa.JSON()), sa.Column("ai_summary_zh", sa.Text()), sa.Column("ai_summary_en", sa.Text()), sa.Column("ai_model", sa.String(64)), sa.Column("ai_generated_at", sa.DateTime(timezone=True)), sa.Column("ai_failure_code", sa.String(64)), 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"), ), sa.Column("resolved_at", sa.DateTime(timezone=True)), ) # One packet per move. Same key the lifecycle event is unique on, # so a replayed coordinator call cannot mint a second packet. op.create_index( "uq_hr_role_change_reviews_person_date", REVIEWS, ["person_id", "effective_date"], unique=True, ) op.create_index( "ix_hr_role_change_reviews_status", REVIEWS, ["status", "created_at"], ) op.create_index( "ix_hr_role_change_reviews_reviewer", REVIEWS, ["reviewer_person_id"], ) if not _has_table(conn, ITEMS): op.create_table( ITEMS, sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("review_id", sa.CHAR(36), nullable=False), sa.Column("resource_type", sa.String(32), nullable=False), sa.Column("resource_id", sa.CHAR(36), nullable=False), sa.Column("label", sa.String(512), nullable=False), sa.Column("disposition", sa.String(16), nullable=False), sa.Column("detail", sa.JSON()), sa.Column("decision", sa.String(16)), sa.Column("decided_by_person_id", sa.CHAR(36)), sa.Column("decided_at", sa.DateTime(timezone=True)), sa.Column("decision_note", sa.String(500)), sa.Column("ai_recommendation", sa.String(16)), sa.Column("ai_reason", sa.String(500)), 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"), ), ) # Re-scanning a packet is idempotent on this key: the second scan # re-finds the same holdings and must neither duplicate them nor # lose the decisions already recorded against them. op.create_index( "uq_hr_rc_review_items_resource", ITEMS, ["review_id", "resource_type", "resource_id"], unique=True, ) op.create_index( "ix_hr_rc_review_items_pending", ITEMS, ["review_id", "decision"], ) # FKs added separately and best-effort: TiDB is strict about column # type parity, and an install whose identity_persons.id predates the # CHAR(36) convention would otherwise fail the whole migration for a # constraint that is a convenience, not a correctness requirement # here (both writers resolve the person first). for name, table, cols, ref in ( ("fk_hr_rc_reviews_person", REVIEWS, ["person_id"], "identity_persons.id"), ( "fk_hr_rc_reviews_reviewer", REVIEWS, ["reviewer_person_id"], "identity_persons.id", ), ( "fk_hr_rc_reviews_event", REVIEWS, ["lifecycle_event_id"], "hr_lifecycle_events.id", ), ( "fk_hr_rc_items_review", ITEMS, ["review_id"], f"{REVIEWS}.id", ), ( "fk_hr_rc_items_decider", ITEMS, ["decided_by_person_id"], "identity_persons.id", ), ): ref_table, ref_col = ref.split(".") try: op.create_foreign_key(name, table, ref_table, cols, [ref_col]) except Exception: # noqa: BLE001 — see comment above pass # Both crons seeded ENABLED. Neither can storm: the backfill writes # no notification at all, and the notifier sends at most one card per # packet ever (``notified_at``). Seeding them disabled would mean the # first real move after deploy silently reaches nobody, which is the # state this whole feature exists to end. for key, display, desc, cron, cfg in ( ( BACKFILL_JOB, "转岗复核包裹补建", _BACKFILL_DESC, "40 3 * * *", {"lookback_days": 90, "limit": 50}, ), ( # 09:30 — after the 03:40 backfill has had a night to build # whatever the inline path missed, and inside working hours so # the card lands when the manager can act on it. NOTIFY_JOB, "转岗复核 · 通知新经理", _NOTIFY_DESC, "30 9 * * *", {"limit": 50}, ), ): 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(cfg), }, ) def downgrade() -> None: conn = op.get_bind() if sa.inspect(conn).has_table("scheduled_jobs"): # Before the tables: a downgraded image whose worker still finds # an enabled row for a handler this revision introduced fails its # boot-time registration assertion. conn.execute( sa.text( "DELETE FROM scheduled_jobs WHERE handler_key IN (:a, :b)", ), {"a": BACKFILL_JOB, "b": NOTIFY_JOB}, ) if _has_table(conn, ITEMS): op.drop_table(ITEMS) if _has_table(conn, REVIEWS): op.drop_table(REVIEWS)