"""GRC spine — control catalog + multi-framework crosswalk + CCM. Revision ID: 250 Revises: 249 Create Date: 2026-06-04 Redesigns the 安全合规 menu around a first-class ``Control`` entity (the model every GRC platform — Vanta / Drata / SailPoint — is built on). Five tables: * ``compliance_frameworks`` — SOC2 / ISO27001 / … standards * ``compliance_requirements`` — clauses within a framework (CC6.1, A.8.5) * ``compliance_controls`` — first-class controls (1:1 w/ detective jobs) * ``compliance_control_requirements`` — crosswalk (control ↔ requirement, M:N) * ``compliance_control_status`` — latest evaluated status per control (CCM) Catalog DATA (frameworks/requirements/controls/mappings) is NOT seeded here — it's reconciled idempotently from ``compliance_catalog.py`` by ``compliance_service.sync_compliance_catalog`` (run by the recompute cron and lazily on first read), so editing the catalog never needs a migration. This migration only creates the schema + seeds the recompute scheduled-job row. TiDB-FK strictness: all id / inter-table FK columns are ``CHAR(36)`` (matching the ``GUID`` type), ``owner_person_id`` uses ``id_type_for`` to match ``identity_persons.id`` exactly, and every table inherits ``identity_persons``' charset/collation via ``table_kw_for`` so FKs aren't rejected with error 3780. Tables are guarded with information_schema and a DROP-if-VARCHAR-id pre-flight so a re-run after a bad local cut converges. """ 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 = "250" down_revision = "249" branch_labels = None depends_on = None _TABLES = [ "compliance_control_status", "compliance_control_requirements", "compliance_controls", "compliance_requirements", "compliance_frameworks", ] _RECOMPUTE_HANDLER = "compliance.control_status_recompute" _RECOMPUTE_DESC = ( "每小时(:15)重算所有合规控制项的状态(持续控制监控 CCM):先从 " "compliance_catalog 幂等同步框架/条款/控制项/crosswalk,再按每个控制项的" "支撑信号(检测类定时任务的新鲜度 + 它产出的指标)评估 pass/fail/stale/" "not_configured,upsert 进 compliance_control_status。喂给「合规总览」看板" "和各框架(SOC2/ISO27001/…)就绪度。与各检测任务正交:那些采集数据,这个" "把数据投影成控制项状态与框架条款覆盖。" ) def _table_exists(name: str) -> bool: return op.get_bind().execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n LIMIT 1" ), {"n": name}, ).scalar() is not None def _id_is_varchar(name: str) -> bool: row = op.get_bind().execute( sa.text( "SELECT DATA_TYPE FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n " "AND COLUMN_NAME = 'id' LIMIT 1" ), {"n": name}, ).scalar() return bool(row) and str(row).lower() == "varchar" def upgrade() -> None: bind = op.get_bind() persons_id = id_type_for("identity_persons") table_kw = table_kw_for("identity_persons") # ── 0. Re-run convergence: drop any prior bad-id cut (empty tables, # DB is sole data source, so dropping loses nothing). ──────── for name in _TABLES: if _id_is_varchar(name): op.execute(f"DROP TABLE IF EXISTS `{name}`") def _ts_cols(): return ( sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False, ), ) # ── 1. compliance_frameworks ────────────────────────────────────── if not _table_exists("compliance_frameworks"): op.create_table( "compliance_frameworks", sa.Column("id", sa.CHAR(36), nullable=False), sa.Column("key", sa.String(32), nullable=False), sa.Column("name", sa.String(128), nullable=False), sa.Column("version", sa.String(32), nullable=True), sa.Column("description", sa.Text(), nullable=True), 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")), *_ts_cols(), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("key", name="uq_compliance_framework_key"), **table_kw, ) op.create_index( "idx_compliance_framework_enabled", "compliance_frameworks", ["enabled"], ) # ── 2. compliance_requirements ──────────────────────────────────── if not _table_exists("compliance_requirements"): op.create_table( "compliance_requirements", sa.Column("id", sa.CHAR(36), nullable=False), sa.Column("framework_id", sa.CHAR(36), nullable=False), sa.Column("ref", sa.String(64), nullable=False), sa.Column("title", sa.String(255), nullable=False), sa.Column("description", sa.Text(), nullable=True), sa.Column("category", sa.String(128), nullable=True), sa.Column("sort_order", sa.Integer(), nullable=False, server_default=sa.text("0")), *_ts_cols(), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint( ["framework_id"], ["compliance_frameworks.id"], ondelete="CASCADE", name="fk_compliance_req_framework", ), sa.UniqueConstraint( "framework_id", "ref", name="uq_compliance_req_framework_ref", ), **table_kw, ) op.create_index( "idx_compliance_req_framework", "compliance_requirements", ["framework_id"], ) # ── 3. compliance_controls ──────────────────────────────────────── if not _table_exists("compliance_controls"): op.create_table( "compliance_controls", sa.Column("id", sa.CHAR(36), nullable=False), sa.Column("key", sa.String(64), nullable=False), sa.Column("name", sa.String(128), nullable=False), sa.Column("category", sa.String(32), nullable=False), sa.Column("description", sa.Text(), nullable=True), sa.Column("signal_type", sa.String(32), nullable=False, server_default=sa.text("'scheduled_job'")), sa.Column("signal_ref", sa.String(128), nullable=True), sa.Column("owner_person_id", persons_id, nullable=True), sa.Column("config", sa.JSON(), nullable=True), 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")), *_ts_cols(), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint( ["owner_person_id"], ["identity_persons.id"], ondelete="SET NULL", name="fk_compliance_control_owner", ), sa.UniqueConstraint("key", name="uq_compliance_control_key"), **table_kw, ) op.create_index( "idx_compliance_control_category", "compliance_controls", ["category"], ) op.create_index( "idx_compliance_control_enabled", "compliance_controls", ["enabled"], ) # ── 4. compliance_control_requirements (crosswalk) ──────────────── if not _table_exists("compliance_control_requirements"): op.create_table( "compliance_control_requirements", sa.Column("id", sa.CHAR(36), nullable=False), sa.Column("control_id", sa.CHAR(36), nullable=False), sa.Column("requirement_id", sa.CHAR(36), nullable=False), sa.Column("note", sa.String(255), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False, ), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint( ["control_id"], ["compliance_controls.id"], ondelete="CASCADE", name="fk_compliance_ctrl_req_control", ), sa.ForeignKeyConstraint( ["requirement_id"], ["compliance_requirements.id"], ondelete="CASCADE", name="fk_compliance_ctrl_req_requirement", ), sa.UniqueConstraint( "control_id", "requirement_id", name="uq_compliance_ctrl_req", ), **table_kw, ) op.create_index( "idx_compliance_ctrl_req_control", "compliance_control_requirements", ["control_id"], ) op.create_index( "idx_compliance_ctrl_req_requirement", "compliance_control_requirements", ["requirement_id"], ) # ── 5. compliance_control_status ────────────────────────────────── if not _table_exists("compliance_control_status"): op.create_table( "compliance_control_status", sa.Column("id", sa.CHAR(36), nullable=False), sa.Column("control_id", sa.CHAR(36), nullable=False), sa.Column("status", sa.String(16), nullable=False, server_default=sa.text("'unknown'")), sa.Column("score", sa.Integer(), nullable=True), sa.Column("summary", sa.String(512), nullable=True), sa.Column("detail", sa.JSON(), nullable=True), sa.Column("evaluated_at", sa.DateTime(timezone=True), nullable=False), sa.Column("source_fresh_at", sa.DateTime(timezone=True), nullable=True), *_ts_cols(), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint( ["control_id"], ["compliance_controls.id"], ondelete="CASCADE", name="fk_compliance_control_status_control", ), sa.UniqueConstraint( "control_id", name="uq_compliance_control_status_control", ), **table_kw, ) op.create_index( "idx_compliance_control_status_status", "compliance_control_status", ["status"], ) # ── 6. Seed the CCM recompute scheduled job ─────────────────────── if _table_exists("scheduled_jobs"): bind.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, 'UTC', 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": _RECOMPUTE_HANDLER, "display": "合规 · 控制项状态重算(CCM)", "desc": _RECOMPUTE_DESC, "handler": _RECOMPUTE_HANDLER, "cron": "15 * * * *", "config": _json.dumps({}), }, ) def downgrade() -> None: bind = op.get_bind() if _table_exists("scheduled_jobs"): bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"), {"k": _RECOMPUTE_HANDLER}, ) for name in _TABLES: op.execute(f"DROP TABLE IF EXISTS `{name}`")