"""安全问卷 · 乙方侧客户问卷 + 答案库 + 证据文档库 — 5 tables + due cron. Tables: * auditor_answer_library_entries — reusable bilingual Q&A knowledge base * auditor_evidence_documents — our security artifacts (cert/policy/report) * auditor_inbound_questionnaires — a customer's questionnaire we must fill * auditor_inbound_questions — per-question rows (AI-drafted + reviewed) * auditor_inbound_versions — archived exported versions (accountability) person columns are plain id columns (NO FK) — TiDB-FK-safe. Intra-feature parent refs (questionnaire→question / →version) carry real same-charset FKs (reuse identity_persons' id type/charset). Seeds the daily inbound due-date reminder cron. Idempotent. Revision ID: 376 Revises: 375 """ 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 = "376" down_revision = "375" branch_labels = None depends_on = None LIB = "auditor_answer_library_entries" DOC = "auditor_evidence_documents" IQ = "auditor_inbound_questionnaires" IQQ = "auditor_inbound_questions" IQV = "auditor_inbound_versions" JOB = "auditor.inbound_questionnaire_due_reminder" 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, LIB): op.create_table( LIB, sa.Column("id", pid, primary_key=True), sa.Column("category", sa.String(64), nullable=False, server_default="general"), sa.Column("question_zh", sa.Text, nullable=False), sa.Column("question_en", sa.Text, nullable=False), sa.Column("answer_zh", sa.Text, nullable=False), sa.Column("answer_en", sa.Text, nullable=False), sa.Column("tags", sa.JSON), sa.Column("owner_person_id", pid), sa.Column("status", sa.String(16), nullable=False, server_default="active"), sa.Column("usage_count", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("last_reviewed_at", sa.DateTime(timezone=True)), sa.Column("review_interval_months", sa.Integer, nullable=False, server_default=sa.text("12")), sa.Column("source_doc_ids", sa.JSON), sa.Column("created_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()), **kw, ) op.create_index("idx_auditor_answer_lib_category", LIB, ["category", "status"]) op.create_index("idx_auditor_answer_lib_owner", LIB, ["owner_person_id"]) if not _table_exists(bind, DOC): op.create_table( DOC, sa.Column("id", pid, primary_key=True), sa.Column("title", sa.String(255), nullable=False), sa.Column("category", sa.String(32), nullable=False, server_default="other"), sa.Column("description", sa.Text), 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("version", sa.String(32)), sa.Column("owner_person_id", pid), sa.Column("valid_until", sa.Date), sa.Column("is_shareable", sa.Boolean, nullable=False, server_default=sa.text("1")), sa.Column("status", sa.String(16), nullable=False, server_default="active"), sa.Column("uploaded_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()), **kw, ) op.create_index("idx_auditor_evidence_doc_category", DOC, ["category", "status"]) op.create_index("idx_auditor_evidence_doc_valid", DOC, ["valid_until"]) if not _table_exists(bind, IQ): op.create_table( IQ, sa.Column("id", pid, primary_key=True), sa.Column("customer_name", sa.String(255), nullable=False), sa.Column("title", sa.String(255), nullable=False), sa.Column("due_date", sa.Date), sa.Column("owner_person_id", pid), sa.Column("status", sa.String(16), nullable=False, server_default="received"), sa.Column("source_object_key", sa.String(512)), sa.Column("source_filename", sa.String(255)), sa.Column("notes", sa.Text), sa.Column("completed_at", sa.DateTime(timezone=True)), sa.Column("exported_at", sa.DateTime(timezone=True)), sa.Column("reminder_stage", sa.String(16)), sa.Column("reminder_sent_at", sa.DateTime(timezone=True)), sa.Column("created_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()), **kw, ) op.create_index("idx_auditor_inbound_status", IQ, ["status"]) op.create_index("idx_auditor_inbound_due", IQ, ["due_date"]) op.create_index("idx_auditor_inbound_owner", IQ, ["owner_person_id"]) if not _table_exists(bind, IQQ): op.create_table( IQQ, sa.Column("id", pid, primary_key=True), sa.Column("questionnaire_id", pid, nullable=False), sa.Column("seq", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("section", sa.String(255)), sa.Column("question_text", sa.Text, nullable=False), sa.Column("matched_library_entry_id", pid), sa.Column("match_confidence", sa.Integer), sa.Column("answer_text", sa.Text), sa.Column("status", sa.String(16), nullable=False, server_default="pending"), sa.Column("reviewed_by_person_id", pid), sa.Column("reviewed_at", sa.DateTime(timezone=True)), 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( ["questionnaire_id"], [f"{IQ}.id"], name="fk_inbound_question_questionnaire", ondelete="CASCADE", ), **kw, ) op.create_index("idx_auditor_inbound_q_questionnaire", IQQ, ["questionnaire_id", "seq"]) if not _table_exists(bind, IQV): op.create_table( IQV, sa.Column("id", pid, primary_key=True), sa.Column("questionnaire_id", pid, nullable=False), sa.Column("version_no", sa.Integer, nullable=False, server_default=sa.text("1")), sa.Column("object_key", sa.String(512), nullable=False), sa.Column("filename", sa.String(255), nullable=False), sa.Column("note", sa.Text), sa.Column("exported_by_person_id", pid), sa.Column("exported_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( ["questionnaire_id"], [f"{IQ}.id"], name="fk_inbound_version_questionnaire", ondelete="CASCADE", ), sa.UniqueConstraint("questionnaire_id", "version_no", name="uq_auditor_inbound_version_no"), **kw, ) op.create_index("idx_auditor_inbound_version_q", IQV, ["questionnaire_id"]) 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 5 * * *', 'UTC', 1, '{}', 0, 0, NOW(), NOW()) """ ), { "id": str(uuid.uuid4()), "name": JOB, "display_name": "安全问卷 · 客户问卷截止提醒", "description": ( "每天 05:00 UTC 按 due_date 分阶段(提前 7 天 / 3 天 / 到期 / 逾期)" "提醒未完成客户问卷的负责人;每阶段仅提醒一次;仅内部推送。" ), }, ) def downgrade() -> None: bind = op.get_bind() bind.execute(sa.text("DELETE FROM scheduled_jobs WHERE name = :n"), {"n": JOB}) op.drop_table(IQV) op.drop_table(IQQ) op.drop_table(IQ) op.drop_table(DOC) op.drop_table(LIB)