"""How many approvals does the company run, and who answers for each one. Both questions were unanswerable outside the Feishu approval admin console. The second one is unanswerable *inside* it too: an approval definition has no owner field. Verified live against the tenant — ``GET /approval/v4/approvals/{code}`` returns ``approval_name`` / ``status`` / ``form`` / ``node_list`` / ``viewers``, and the documented ``approval_admin_ids`` comes back null. So ownership is established here, and ``owner_source`` records how each name got into the box. A machine-derived candidate and a person who confirmed they own it are different facts, and a single non-null owner column would flatten them into one — which is how a register reports 100% coverage on the strength of its own guesses. Two group columns for the same reason. ``source_group`` is what the Feishu instance API reports; ``category`` is what Helix publishes. They genuinely disagree in this tenant: the console files 域名DNS解析申请 under ``IT Services & Security`` while the API answers ``IT Equipment`` for the same definition. Publishing the API value alone shows everyone a taxonomy no administrator recognises; overwriting it destroys the record of what upstream actually said. The nightly sweep is seeded ENABLED, unlike the bounty mail poller: the scopes it needs (``approval:approval.list:readonly`` + ``approval:approval:readonly``) were granted on the Feishu app on 2026-08-26 and verified working, so it has nothing to wait for. Revision ID: 553 Revises: 552 """ 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 = "553" down_revision = "552" branch_labels = None depends_on = None FLOWS = "registry_approval_flows" EVENTS = "registry_approval_flow_events" HANDLER_KEY = "approval_registry.feishu_sync" _DESC = ( "每晚同步飞书审批流程台账:①按在职员工扇出 instances/query 发现全公司" "在用的审批定义(名称+分组,分组只认这个接口 —— tasks/search 的 group " "字段实测不可信);②对已知 code 逐个 GET 审批定义刷新名称与状态" "(不订阅事件,那是别的团队的流程);③按 tasks/search 统计实际审批人," "占比过半且样本足够才把负责人填为「推断·待确认」,人工确认过的一律不覆盖。" "增量跑 30 天窗口(飞书上限);全量重扫在台账页手动触发。" ) def upgrade() -> None: conn = op.get_bind() if sa.inspect(conn).has_table(FLOWS): return op.create_table( FLOWS, sa.Column("id", id_type_for("identity_persons"), primary_key=True), # Feishu's approval definition code — the natural key, upper-cased # on write so a hand-typed lower-case copy cannot slip past the # UNIQUE index and register the same flow twice. sa.Column("approval_code", sa.String(128), nullable=False), sa.Column("name", sa.String(255), nullable=False), # What upstream says the group is. Read-only provenance. sa.Column("source_group", sa.String(128), nullable=True), # What Helix publishes. Seeded from source_group, then owned by IT. sa.Column("category", sa.String(128), nullable=True), # active | inactive | retired sa.Column( "status", sa.String(16), nullable=False, server_default="active", ), # The raw upstream string, so a Feishu status we do not map yet is # visible instead of silently collapsing to active. sa.Column("feishu_status", sa.String(32), nullable=True), sa.Column("description", sa.Text(), nullable=True), sa.Column( "owner_person_id", id_type_for("identity_persons"), nullable=True, ), # unassigned | inferred | manual | imported | confirmed sa.Column( "owner_source", sa.String(16), nullable=False, server_default="unassigned", ), sa.Column("owner_department", sa.String(255), nullable=True), sa.Column("owner_assigned_at", sa.DateTime(timezone=True), nullable=True), sa.Column("owner_confirmed_at", sa.DateTime(timezone=True), nullable=True), # Evidence behind an inferred owner (approver, share, sample size, # window). Kept so a human judging the suggestion can see what it # rests on instead of being asked to trust a name. sa.Column("inference_meta", sa.JSON(), nullable=True), sa.Column( "published", sa.Boolean(), nullable=False, server_default="1", ), # discovered | manual | imported sa.Column( "source", sa.String(16), nullable=False, server_default="discovered", ), sa.Column( "observed_instances", sa.Integer(), nullable=False, server_default="0", ), sa.Column("last_instance_at", sa.DateTime(timezone=True), nullable=True), sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=True), sa.Column("last_discovered_at", sa.DateTime(timezone=True), nullable=True), sa.Column("last_synced_at", sa.DateTime(timezone=True), nullable=True), sa.Column("last_sync_error", sa.String(512), nullable=True), sa.Column("created_by", id_type_for("identity_persons"), nullable=True), 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.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), sa.ForeignKeyConstraint( ["owner_person_id"], ["identity_persons.id"], ondelete="SET NULL", ), sa.ForeignKeyConstraint( ["created_by"], ["identity_persons.id"], ondelete="SET NULL", ), **table_kw_for("identity_persons"), ) op.create_index( "uq_registry_approval_flows_code", FLOWS, ["approval_code"], unique=True, ) op.create_index( "ix_registry_approval_flows_owner", FLOWS, ["owner_person_id", "deleted_at"], ) op.create_index( "ix_registry_approval_flows_status", FLOWS, ["status", "deleted_at"], ) op.create_index( "ix_registry_approval_flows_owner_source", FLOWS, ["owner_source", "status"], ) op.create_index( "ix_registry_approval_flows_published", FLOWS, ["published", "status"], ) op.create_table( EVENTS, sa.Column("id", id_type_for("identity_persons"), primary_key=True), sa.Column("flow_id", id_type_for("identity_persons"), nullable=False), sa.Column("event_type", sa.String(32), nullable=False), sa.Column( "actor_person_id", id_type_for("identity_persons"), nullable=True, ), # Snapshot so the trail still reads correctly after the actor # leaves. NULL means the platform itself acted (a sweep). sa.Column("actor_name", sa.String(255), nullable=True), sa.Column("body", sa.Text(), nullable=True), sa.Column("meta", sa.JSON(), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP"), ), sa.ForeignKeyConstraint( ["flow_id"], [f"{FLOWS}.id"], ondelete="CASCADE", ), sa.ForeignKeyConstraint( ["actor_person_id"], ["identity_persons.id"], ondelete="SET NULL", ), **table_kw_for("identity_persons"), ) op.create_index( "ix_registry_approval_flow_events_flow", EVENTS, ["flow_id", "created_at"], ) 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": HANDLER_KEY, "display": "审批流程台账 · 每晚同步飞书", "desc": _DESC, "handler": HANDLER_KEY, # 03:30 Asia/Shanghai — after the nightly identity syncs have # settled the roster this sweep fans out over, and outside the # 03:00 cluster that already contends for the pool. "cron": "30 3 * * *", "config": _json.dumps({"window_days": 30}), }, ) def downgrade() -> None: conn = op.get_bind() conn.execute( sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"), {"k": HANDLER_KEY}, ) # Children before parents. ``DROP TABLE`` takes the table's own # indexes and FKs with it, so there is no index-before-FK 1553 dance # here — but the events table must go first, because its FK points at # the flows table. if sa.inspect(conn).has_table(EVENTS): op.drop_table(EVENTS) if sa.inspect(conn).has_table(FLOWS): op.drop_table(FLOWS)