"""Seed the daily access-drift scan (observe-only by default).

Revision ID: 499
Revises: 498

证据查询的同一套确定性逻辑反过来用就是探测器:离职者仍持有的访问、完成了却没有
证据归档的离职、持有区间台账里同一 key 的多个开区间、有角色却没有台账区间
(时点查询看不见它)。跑成定时任务,就把「等下个审计季才发现」变成「出事那周就有单」。

**默认 observe_only:只算不开单。** 把一个新探测器放到多年累积的存量上,第一天就
可能刷出一个没人看的队列 —— 而没人看的队列比没有队列更糟,它训练所有人忽略这个
入口。先用真实生产数据量一遍首日条数,确认不成风暴,再置
config={"observe_only": false} 开单。这个开关刻意做成「缺省即观察」:缺省值等于
「全开」正是首次部署变成工单风暴的方式。

开单后走既有 findings 工单流(按 (control_key, dedup_key) 去重、项目消失即自动关单),
推送只到 auditor / global_admin —— 漂移单是 IT 的活,不触达员工。

排 04:10:在 03:xx 那批任务与 04:xx 成本任务之间,且晚于离职相关的夜间流程,
读到的是当天已经收敛过的状态而不是半程状态。只读:不打厂商接口、不调模型。
"""

import uuid

import sqlalchemy as sa

from alembic import op

revision = "499"
down_revision = "498"
branch_labels = None
depends_on = None

_KEY = "auditor.access_drift_scan"
_DISPLAY = "访问漂移扫描(默认只观察)"
_DESCRIPTION = (
    "每天 04:10(Asia/Shanghai)用证据查询同一套确定性逻辑反过来做探测:离职者仍"
    "持有的访问、完成了却没有证据归档的离职、持有区间台账里同一 key 的多个开区间、"
    "以及有角色却没有台账区间(时点查询看不见它)。\n"
    "【默认 observe_only=true:只算不开单】把一个新探测器放到多年累积的存量上,"
    "第一天就可能刷出一个没人看的队列 —— 而没人看的队列比没有队列更糟。先用真实"
    "生产数据量一遍首日条数,确认不会成风暴,再置 "
    'config={"observe_only": false} 开单。\n'
    "开单后走既有 findings 工单流(按 (control_key, dedup_key) 去重、消失即自动关单),"
    "推送只到 auditor / global_admin,不触达员工。\n"
    "只读:不打厂商接口、不调用模型。\n"
    'config={"grace_days":N} 覆写离职宽限期(默认 7 天)。'
)
# 04:10:在 03:xx 那批任务与 04:xx 成本任务之间,且晚于离职相关的夜间流程 ——
# 读到的是当天已经收敛过的状态,而不是半程状态。
_CRON = "10 4 * * *"


def upgrade() -> None:
    op.get_bind().execute(
        sa.text(
            "INSERT INTO scheduled_jobs "
            "(id, name, handler_key, display_name, description, "
            " trigger_type, cron_expr, timezone, enabled, config, "
            " created_at, updated_at) "
            "SELECT :id, :name, :k, :dn, :desc, 'cron', :cron, "
            " 'Asia/Shanghai', 1, '{}', NOW(), NOW() "
            "FROM DUAL WHERE NOT EXISTS ("
            "  SELECT 1 FROM scheduled_jobs WHERE handler_key = :k)"
        ),
        {
            "id": str(uuid.uuid4()),
            "name": _KEY,
            "k": _KEY,
            "dn": _DISPLAY,
            "desc": _DESCRIPTION,
            "cron": _CRON,
        },
    )


def downgrade() -> None:
    op.get_bind().execute(
        sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"),
        {"k": _KEY},
    )