"""信息安全自评估 — 3 tables + semiannual cycle-opener cron. Tables: * auditor_assessment_cycles — one semiannual report instance * auditor_assessment_sections — materialised sections (cycle→FK) * auditor_assessment_measures — evergreen 整改/措施 worklist person columns carry NO FK to identity_persons (TiDB-FK-safe); cycle→section is a real intra-feature FK (same charset). The ISMS template + per-section data aggregators live in code (assessment_catalog / assessment_data), so no template table. Idempotent. Revision ID: 323 Revises: 322 """ 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 = "323" down_revision = "322" branch_labels = None depends_on = None CYCLE_TABLE = "auditor_assessment_cycles" SECTION_TABLE = "auditor_assessment_sections" MEASURE_TABLE = "auditor_assessment_measures" JOB = "auditor.assessment_cycle_opener_semiannual" 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 def upgrade() -> None: bind = op.get_bind() pid = id_type_for("identity_persons") kw = table_kw_for("identity_persons") if not _table_exists(bind, CYCLE_TABLE): op.create_table( CYCLE_TABLE, sa.Column("id", pid, primary_key=True), sa.Column( "template_key", sa.String(64), nullable=False, server_default="isms_default", ), sa.Column("period_key", sa.String(16), nullable=False), sa.Column("period_start", sa.Date, nullable=False), sa.Column("period_end", sa.Date, nullable=False), sa.Column("report_date", sa.Date), sa.Column("title", sa.String(255), nullable=False), sa.Column( "status", sa.String(16), nullable=False, server_default="draft", ), sa.Column("intro_narrative", sa.Text), sa.Column("conclusion_narrative", sa.Text), sa.Column( "overall_conclusion", sa.String(16), nullable=False, server_default="not_assessed", ), sa.Column("created_by_person_id", pid), sa.Column("finalized_at", sa.DateTime(timezone=True)), sa.Column("finalized_by_person_id", pid), 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("period_key", name="uq_auditor_assessment_period"), **kw, ) op.create_index("idx_auditor_assessment_status", CYCLE_TABLE, ["status"]) if not _table_exists(bind, SECTION_TABLE): op.create_table( SECTION_TABLE, sa.Column("id", pid, primary_key=True), sa.Column("cycle_id", pid, nullable=False), sa.Column("section_key", sa.String(64), nullable=False), sa.Column("parent_key", sa.String(64)), sa.Column("title", sa.String(255), nullable=False), sa.Column( "sort_order", sa.Integer, nullable=False, server_default=sa.text("0"), ), sa.Column( "has_table", sa.Boolean, nullable=False, server_default=sa.text("0"), ), sa.Column("narrative", sa.Text), sa.Column( "conclusion", sa.String(16), nullable=False, server_default="not_assessed", ), sa.Column( "conclusion_applicable", sa.Boolean, nullable=False, server_default=sa.text("1"), ), sa.Column("table_columns", sa.JSON), sa.Column("data_table", sa.JSON), sa.Column("auto_source", sa.String(64)), sa.Column("auto_data", sa.JSON), sa.Column("auto_refreshed_at", sa.DateTime(timezone=True)), sa.Column("control_key", sa.String(64)), sa.Column( "ai_drafted", sa.Boolean, 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( ["cycle_id"], [f"{CYCLE_TABLE}.id"], name="fk_assessment_section_cycle", ondelete="CASCADE", ), sa.UniqueConstraint( "cycle_id", "section_key", name="uq_auditor_assessment_section", ), **kw, ) op.create_index( "idx_auditor_assessment_section_cycle", SECTION_TABLE, ["cycle_id"], ) if not _table_exists(bind, MEASURE_TABLE): op.create_table( MEASURE_TABLE, sa.Column("id", pid, primary_key=True), sa.Column("item", sa.String(255), nullable=False), sa.Column("content", sa.Text), sa.Column("measures", sa.Text), sa.Column("owner_person_id", pid), sa.Column( "status", sa.String(16), nullable=False, server_default="pending", ), sa.Column("note", sa.Text), sa.Column( "sort_order", sa.Integer, nullable=False, server_default=sa.text("0"), ), sa.Column( "archived", sa.Boolean, 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(), ), **kw, ) op.create_index( "idx_auditor_assessment_measure_status", MEASURE_TABLE, ["status", "archived"], ) 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 2 1 1,7 *', 'UTC', 1, '{}', 0, 0, NOW(), NOW()) """ ), { "id": str(uuid.uuid4()), "name": JOB, "display_name": "安全员 · 信息安全自评估半年开盘", "description": ( "每年 1/1 与 7/1 02:00 UTC 为刚结束的半年开一份自评估报告草稿" "(ISMS 模板 + 真值自动填表);该周期已有则幂等跳过。" ), }, ) def downgrade() -> None: bind = op.get_bind() bind.execute(sa.text("DELETE FROM scheduled_jobs WHERE name = :n"), {"n": JOB}) op.drop_table(MEASURE_TABLE) op.drop_table(SECTION_TABLE) op.drop_table(CYCLE_TABLE)