"""Intelligence Center persistence kernel (feature remains disabled). Revision ID: 472 Revises: 471 """ import json import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "472" down_revision = "471" branch_labels = None depends_on = None def _has_table(name: str) -> bool: return op.get_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 _has_index(table: str, name: str) -> bool: return op.get_bind().execute( sa.text( "SELECT 1 FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t " "AND INDEX_NAME = :n LIMIT 1" ), {"t": table, "n": name}, ).scalar() is not None def _reset_if_stale(name: str, want_collation: str | None) -> None: """Drop a leftover table from a *failed* run of this same migration. ``upgrade()`` only executes while ``alembic_version < 472``, so any of these four tables already in the schema can only be debris from a run that died partway — TiDB DDL is non-transactional, so the tables it managed to create before the failure stay behind. Skipping such a table is **not** enough. The failed run created it without an explicit charset, so it carries the *server default* (``utf8mb4_bin`` on prod TiDB) while this run builds its siblings to match ``identity_persons`` (``utf8mb4_unicode_ci``). A child FK into the stale table then hits the very error 3780 this migration is being repaired for — the second time, in a place the first fix didn't look. Refuses to drop a table holding rows: that would mean the premise above is wrong, and losing data silently is worse than stopping. """ if not want_collation: return current = op.get_bind().execute( sa.text( "SELECT TABLE_COLLATION FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n" ), {"n": name}, ).scalar() if current is None or current == want_collation: return rows = op.get_bind().execute( sa.text(f"SELECT COUNT(*) FROM {name}"), # noqa: S608 - literal ).scalar() if rows: raise RuntimeError( f"{name} exists with collation {current} (want {want_collation}) " f"and holds {rows} rows — refusing to drop. Convert it by hand." ) op.drop_table(name) def upgrade() -> None: # TiDB 3780 防线(2026-08-16 生产实测)。``identity_persons.id`` 在生产 # 上是 ``char(36) utf8mb4_unicode_ci``;新表不显式指定 charset 就继承 # **服务器默认**,FK 两侧 collation 不一致 → "Referencing column ... and # referenced column ... in foreign key constraint 'fk_2' are # incompatible"。**本地 MySQL 8 两侧 collation 恰好相同,复现不了** —— # 这类只在 TiDB 上炸,已是同一 bug 类的第三次(见 # ``migration_helpers.id_type_for`` 的注释)。 person_id = id_type_for("identity_persons") table_kw = table_kw_for("identity_persons") # 每步存在性守卫。首次部署恰好在建完 ``intelligence_snapshots`` 之后、 # 在 ``intelligence_analyses`` 的 FK 上失败:TiDB DDL 非事务,所以第一张 # 表留在库里而 ``alembic_version`` 仍是 471 —— 不守卫,重跑就撞 # "table already exists",这条链会永久卡住(它挡的不只是本迁移,是它 # 之后的每一个迁移)。 for _leftover in ( "intelligence_generation_leases", "intelligence_recommendations", "intelligence_analyses", "intelligence_snapshots", ): _reset_if_stale(_leftover, table_kw.get("mysql_collate")) if not _has_table("intelligence_snapshots"): op.create_table( "intelligence_snapshots", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("subject_type", sa.String(24), nullable=False), sa.Column("subject_id", sa.String(128), nullable=False), sa.Column("audience_key", sa.String(128), nullable=False), sa.Column("score_version", sa.String(32), nullable=False), sa.Column("facts_hash", sa.String(64), nullable=False), sa.Column("overall_score", sa.Numeric(5, 2), nullable=False), sa.Column("confidence_score", sa.Numeric(5, 2), nullable=False), sa.Column("grade", sa.String(24), nullable=False), sa.Column("dimensions", sa.JSON(), nullable=False), sa.Column("aggregate_facts", sa.JSON(), nullable=False), sa.Column("reference_map", sa.JSON(), nullable=False), sa.Column("data_as_of", sa.DateTime(timezone=True), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.UniqueConstraint( "subject_type", "subject_id", "audience_key", "score_version", "facts_hash", name="uq_intel_snapshot_identity", ), **table_kw, ) if not _has_index("intelligence_snapshots", "ix_intel_snapshot_latest"): op.create_index("ix_intel_snapshot_latest", "intelligence_snapshots", ["subject_type", "subject_id", "audience_key", "created_at"]) if not _has_index("intelligence_snapshots", "ix_intel_snapshot_data_as_of"): op.create_index("ix_intel_snapshot_data_as_of", "intelligence_snapshots", ["data_as_of"]) if not _has_table("intelligence_analyses"): op.create_table( "intelligence_analyses", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("snapshot_id", sa.CHAR(36), nullable=False), sa.Column("subject_type", sa.String(24), nullable=False), sa.Column("subject_id", sa.String(128), nullable=False), sa.Column("audience_key", sa.String(128), nullable=False), sa.Column("analysis_kind", sa.String(32), nullable=False), sa.Column("generation_key", sa.String(64), nullable=False), sa.Column("status", sa.String(16), nullable=False), sa.Column("summary_zh", sa.Text()), sa.Column("summary_en", sa.Text()), sa.Column("conclusions", sa.JSON()), sa.Column("risks", sa.JSON()), sa.Column("opportunities", sa.JSON()), sa.Column("forecasts", sa.JSON()), sa.Column("recommendations", sa.JSON()), sa.Column("data_gaps", sa.JSON()), sa.Column("model_used", sa.String(128)), sa.Column("input_token_count", sa.Integer()), sa.Column("output_token_count", sa.Integer()), sa.Column("generated_by_user_id", person_id), sa.Column("data_as_of", sa.DateTime(timezone=True), nullable=False), sa.Column("generated_at", sa.DateTime(timezone=True)), sa.Column("failure_code", sa.String(64)), sa.Column("failure_message", sa.String(512)), sa.Column("cache_reused", sa.Boolean(), nullable=False, server_default=sa.false()), sa.Column("forced", sa.Boolean(), nullable=False, server_default=sa.false()), 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), sa.ForeignKeyConstraint(["snapshot_id"], ["intelligence_snapshots.id"], ondelete="RESTRICT"), sa.ForeignKeyConstraint(["generated_by_user_id"], ["identity_persons.id"], ondelete="SET NULL"), sa.UniqueConstraint("generation_key", name="uq_intel_analysis_generation_key"), **table_kw, ) if not _has_index("intelligence_analyses", "ix_intel_analysis_latest"): op.create_index("ix_intel_analysis_latest", "intelligence_analyses", ["subject_type", "subject_id", "audience_key", "generated_at"]) if not _has_index("intelligence_analyses", "ix_intel_analysis_status_created"): op.create_index("ix_intel_analysis_status_created", "intelligence_analyses", ["status", "created_at"]) if not _has_index("intelligence_analyses", "ix_intel_analysis_generator_date"): op.create_index("ix_intel_analysis_generator_date", "intelligence_analyses", ["generated_by_user_id", "created_at"]) if not _has_table("intelligence_recommendations"): op.create_table( "intelligence_recommendations", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("snapshot_id", sa.CHAR(36), nullable=False), sa.Column("analysis_id", sa.CHAR(36)), sa.Column("subject_type", sa.String(24), nullable=False), sa.Column("subject_id", sa.String(128), nullable=False), sa.Column("audience_key", sa.String(128), nullable=False), sa.Column("dedupe_key", sa.String(64), nullable=False), sa.Column("domain", sa.String(32), nullable=False), sa.Column("severity", sa.String(16), nullable=False), sa.Column("priority", sa.Integer(), nullable=False), sa.Column("title_zh", sa.String(256), nullable=False), sa.Column("title_en", sa.String(256), nullable=False), sa.Column("summary_zh", sa.Text(), nullable=False), sa.Column("summary_en", sa.Text(), nullable=False), sa.Column("evidence_refs", sa.JSON(), nullable=False), sa.Column("action_contract", sa.JSON()), sa.Column("expected_monthly_savings", sa.Numeric(18, 4)), sa.Column("expected_annual_savings", sa.Numeric(18, 4)), sa.Column("confirmed_savings", sa.Numeric(18, 4)), sa.Column("realized_savings", sa.Numeric(18, 4)), sa.Column("currency", sa.String(8)), sa.Column("status", sa.String(16), nullable=False), sa.Column("ignored_reason", sa.String(1000)), sa.Column("assigned_to_person_id", person_id), sa.Column("last_changed_by_user_id", person_id), sa.Column("first_seen_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("last_seen_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), sa.Column("viewed_at", sa.DateTime(timezone=True)), sa.Column("resolved_at", sa.DateTime(timezone=True)), sa.Column("ignored_at", sa.DateTime(timezone=True)), 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), sa.ForeignKeyConstraint(["snapshot_id"], ["intelligence_snapshots.id"], ondelete="RESTRICT"), sa.ForeignKeyConstraint(["analysis_id"], ["intelligence_analyses.id"], ondelete="SET NULL"), sa.ForeignKeyConstraint(["assigned_to_person_id"], ["identity_persons.id"], ondelete="SET NULL"), sa.ForeignKeyConstraint(["last_changed_by_user_id"], ["identity_persons.id"], ondelete="SET NULL"), sa.UniqueConstraint("dedupe_key", name="uq_intel_recommendation_dedupe_key"), **table_kw, ) if not _has_index("intelligence_recommendations", "ix_intel_recommendation_status_severity"): op.create_index("ix_intel_recommendation_status_severity", "intelligence_recommendations", ["status", "severity"]) if not _has_index("intelligence_recommendations", "ix_intel_recommendation_subject"): op.create_index("ix_intel_recommendation_subject", "intelligence_recommendations", ["subject_type", "subject_id"]) if not _has_index("intelligence_recommendations", "ix_intel_recommendation_domain_created"): op.create_index("ix_intel_recommendation_domain_created", "intelligence_recommendations", ["domain", "created_at"]) if not _has_table("intelligence_generation_leases"): op.create_table( "intelligence_generation_leases", sa.Column("lease_key", sa.String(191), primary_key=True), sa.Column("owner_token", sa.String(64), nullable=False), sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), 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), **table_kw, ) if not _has_index("intelligence_generation_leases", "ix_intel_lease_expires"): op.create_index("ix_intel_lease_expires", "intelligence_generation_leases", ["expires_at"]) op.get_bind().execute( sa.text( "INSERT INTO platform_settings (`key`, value, description, created_at, updated_at) " "VALUES (:key, :value, :description, NOW(), NOW()) " "ON DUPLICATE KEY UPDATE description = VALUES(description)" ), { "key": "intelligence.enabled", "value": json.dumps(False), "description": "Master release gate for the complete Intelligence Center product.", }, ) def downgrade() -> None: op.get_bind().execute( sa.text("DELETE FROM platform_settings WHERE `key` = :key"), {"key": "intelligence.enabled"}, ) for index in ["ix_intel_lease_expires"]: op.drop_index(index, table_name="intelligence_generation_leases") op.drop_table("intelligence_generation_leases") for index in ["ix_intel_recommendation_domain_created", "ix_intel_recommendation_subject", "ix_intel_recommendation_status_severity"]: op.drop_index(index, table_name="intelligence_recommendations") op.drop_table("intelligence_recommendations") for index in ["ix_intel_analysis_generator_date", "ix_intel_analysis_status_created", "ix_intel_analysis_latest"]: op.drop_index(index, table_name="intelligence_analyses") op.drop_table("intelligence_analyses") for index in ["ix_intel_snapshot_data_as_of", "ix_intel_snapshot_latest"]: op.drop_index(index, table_name="intelligence_snapshots") op.drop_table("intelligence_snapshots")