"""周期性自我检查 — 3 tables + monthly opener cron + seed categories. Tables: * auditor_self_check_categories — recurring check lines (类目) * auditor_self_check_records — one occurrence per (category, period) * auditor_self_check_attachments — evidence files (S3 object keys) person-shaped columns are plain id columns (NO FK to identity_persons) — TiDB-FK-safe convention. Intra-feature parent refs carry real FKs (same charset). All tables reuse identity_persons' id type + charset so the FKs are type-compatible on TiDB. Seeds the monthly opener cron + the initial six check categories from the legacy workbook. Idempotent. Revision ID: 322 Revises: 321 """ from __future__ import annotations import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "322" down_revision = "321" branch_labels = None depends_on = None CAT_TABLE = "auditor_self_check_categories" REC_TABLE = "auditor_self_check_records" ATT_TABLE = "auditor_self_check_attachments" JOB = "auditor.self_check_opener_monthly" def _table_exists(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 # Initial categories (legacy "周期性自我检查" workbook). All quarterly with # evidence required, matching the screenshot台账. SEED_CATEGORIES: list[dict] = [ { "key": "bastion_log_audit", "name": "堡垒机日志审计季度检查", "group_label": "日志审计", "check_item": "堡垒机登录与操作日志是否存在异常操作", "sort_order": 10, }, { "key": "gitee_log_audit", "name": "Gitee 日志审计季度检查", "group_label": "日志审计", "check_item": "Gitee 操作日志是否存在异常访问或操作", "sort_order": 20, }, { "key": "operator_log_audit", "name": "操作员日志审计季度检查", "group_label": "日志审计", "check_item": "各系统操作员账号的操作日志是否存在异常", "sort_order": 30, }, { "key": "antivirus_check", "name": "防病毒季度检查", "group_label": "终端安全", "check_item": "终端防病毒客户端是否全部在线且病毒库为最新", "sort_order": 40, }, { "key": "network_config_check", "name": "网络配置季度检查", "group_label": "网络安全", "check_item": "防火墙/网络设备配置是否符合基线、无未授权变更", "sort_order": 50, }, { "key": "full_disk_encryption", "name": "全盘加密季度检查", "group_label": "终端安全", "check_item": "员工终端是否已启用全盘加密", "sort_order": 60, }, ] def upgrade() -> None: bind = op.get_bind() pid = id_type_for("identity_persons") kw = table_kw_for("identity_persons") if not _table_exists(bind, CAT_TABLE): op.create_table( CAT_TABLE, sa.Column("id", pid, primary_key=True), sa.Column("key", sa.String(64), nullable=False), sa.Column("name", sa.String(128), nullable=False), sa.Column("group_label", sa.String(64)), sa.Column( "periodicity", sa.String(16), nullable=False, server_default="quarterly", ), sa.Column("check_item", sa.Text, nullable=False), sa.Column("guidance", sa.Text), sa.Column("default_checker_person_id", pid), sa.Column( "requires_attachment", sa.Boolean, nullable=False, server_default=sa.text("1"), ), sa.Column( "enabled", sa.Boolean, nullable=False, server_default=sa.text("1"), ), sa.Column( "sort_order", sa.Integer, nullable=False, server_default=sa.text("0"), ), 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.UniqueConstraint("key", name="uq_auditor_self_check_category_key"), **kw, ) op.create_index( "idx_auditor_self_check_cat_enabled", CAT_TABLE, ["enabled", "sort_order"], ) if not _table_exists(bind, REC_TABLE): op.create_table( REC_TABLE, sa.Column("id", pid, primary_key=True), sa.Column("category_id", pid, nullable=False), sa.Column("period_year", sa.Integer, nullable=False), sa.Column("period_label", sa.String(16), nullable=False), sa.Column("period_start", sa.Date, nullable=False), sa.Column("period_end", sa.Date, nullable=False), sa.Column("due_at", sa.DateTime(timezone=True), nullable=False), sa.Column( "status", sa.String(16), nullable=False, server_default="pending", ), sa.Column("check_item_snapshot", sa.Text, nullable=False), sa.Column("outcome", sa.String(16)), sa.Column("assigned_checker_person_id", pid), sa.Column("checker_person_id", pid), sa.Column("completed_at", sa.DateTime(timezone=True)), sa.Column("remark", sa.Text), sa.Column( "attachment_count", sa.Integer, nullable=False, server_default=sa.text("0"), ), 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.ForeignKeyConstraint( ["category_id"], [f"{CAT_TABLE}.id"], name="fk_self_check_record_category", ondelete="CASCADE", ), sa.UniqueConstraint( "category_id", "period_year", "period_label", name="uq_auditor_self_check_record_period", ), **kw, ) op.create_index( "idx_auditor_self_check_rec_status_due", REC_TABLE, ["status", "due_at"], ) op.create_index( "idx_auditor_self_check_rec_category", REC_TABLE, ["category_id"], ) op.create_index( "idx_auditor_self_check_rec_checker", REC_TABLE, ["checker_person_id"], ) if not _table_exists(bind, ATT_TABLE): op.create_table( ATT_TABLE, sa.Column("id", pid, primary_key=True), sa.Column("record_id", pid, nullable=False), sa.Column("object_key", sa.String(512), nullable=False), sa.Column("filename", sa.String(255), nullable=False), sa.Column("content_type", sa.String(128), nullable=False), sa.Column("size_bytes", sa.BigInteger), sa.Column("uploaded_by_person_id", pid), sa.Column( "uploaded_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.ForeignKeyConstraint( ["record_id"], [f"{REC_TABLE}.id"], name="fk_self_check_attachment_record", ondelete="CASCADE", ), **kw, ) op.create_index( "idx_auditor_self_check_att_record", ATT_TABLE, ["record_id"], ) # ── Seed monthly opener cron ──────────────────────────────────── job_exists = bind.execute( sa.text("SELECT 1 FROM scheduled_jobs WHERE name = :n LIMIT 1"), {"n": JOB}, ).scalar() if not job_exists: bind.execute( sa.text( """ INSERT INTO scheduled_jobs (id, name, display_name, description, handler_key, trigger_type, interval_seconds, cron_expr, timezone, enabled, config, run_count, failure_count, created_at, updated_at) VALUES (:id, :name, :display_name, :description, :name, 'cron', NULL, '0 1 1 * *', 'UTC', 1, '{}', 0, 0, NOW(), NOW()) """ ), { "id": str(uuid.uuid4()), "name": JOB, "display_name": "安全员 · 周期自检台账每月自动开单", "description": ( "每月 1 号 01:00 UTC 为每个启用的周期自检类目开当前周期的" "待检查台账;幂等 (类目, 年, 周期标签)。" ), }, ) # ── Seed initial categories ───────────────────────────────────── for spec in SEED_CATEGORIES: exists = bind.execute( sa.text( f"SELECT 1 FROM {CAT_TABLE} WHERE `key` = :k LIMIT 1" ), {"k": spec["key"]}, ).scalar() if exists: continue bind.execute( sa.text( f""" INSERT INTO {CAT_TABLE} (id, `key`, name, group_label, periodicity, check_item, requires_attachment, enabled, sort_order, created_at, updated_at) VALUES (:id, :key, :name, :group_label, 'quarterly', :check_item, 1, 1, :sort_order, NOW(), NOW()) """ ), { "id": str(uuid.uuid4()), "key": spec["key"], "name": spec["name"], "group_label": spec["group_label"], "check_item": spec["check_item"], "sort_order": spec["sort_order"], }, ) def downgrade() -> None: bind = op.get_bind() bind.execute(sa.text("DELETE FROM scheduled_jobs WHERE name = :n"), {"n": JOB}) op.drop_table(ATT_TABLE) op.drop_table(REC_TABLE) op.drop_table(CAT_TABLE)