"""安全问卷 · 甲方侧供应商安全评估 — 5 tables + seed templates + review cron. Tables: * auditor_sq_templates — bilingual questionnaire templates * auditor_sq_template_questions — questions in a template * auditor_vendor_assessments — one assessment per vendor * auditor_vendor_assessment_answers — materialised answers (snapshots) * auditor_vendor_assessment_attachments — evidence / returned workbook person/supplier-shaped columns are plain id columns (NO FK) — TiDB-FK-safe + cross-domain (supplier lives in procurement). Intra-feature parent refs carry real same-charset FKs (all reuse identity_persons' id type/charset). Seeds two system templates (ISO 27001:2022 + 27701 full / lite) from the inlined ``SEED_TEMPLATES`` below (kept self-contained — a migration must not import a live domain module) and the daily review-reminder cron. Idempotent. Revision ID: 374 Revises: 373 """ from __future__ import annotations import json import uuid from typing import Any import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "374" down_revision = "373" branch_labels = None depends_on = None TMPL = "auditor_sq_templates" TQ = "auditor_sq_template_questions" ASSESS = "auditor_vendor_assessments" ANS = "auditor_vendor_assessment_answers" ATT = "auditor_vendor_assessment_attachments" JOB = "auditor.vendor_assessment_review_reminder" # ── Seed content (inlined — a migration must be self-contained and must # NOT import a live domain module: the alembic chain guard imports every # migration WITHOUT a DB env, and importing app.domains.auditor.* pulls # in Settings() and fails; a frozen migration also must survive a future # refactor/deletion of that module). ISO/IEC 27001:2022 Annex A + 27701, # bilingual. See docs/ for the questionnaire rationale. _SEC_GOV = ("governance", "治理与合规", "Governance & Compliance") _SEC_ORG = ("org_controls", "组织控制 (A.5)", "Organizational Controls (A.5)") _SEC_PEOPLE = ("people_controls", "人员控制 (A.6)", "People Controls (A.6)") _SEC_PHYS = ("physical_controls", "物理控制 (A.7)", "Physical Controls (A.7)") _SEC_TECH = ("tech_controls", "技术控制 (A.8)", "Technological Controls (A.8)") _SEC_DATA = ( "data_privacy", "数据保护与隐私 (ISO 27701)", "Data Protection & Privacy (ISO 27701)", ) _SEC_BCP = ("continuity", "业务连续性", "Business Continuity") _SEC_INC = ("incident", "事件管理", "Incident Management") _SEC_SUP = ("supply_chain", "供应链安全", "Supply-Chain Security") def _yn( code: str, section: tuple[str, str, str], zh: str, en: str, *, weight: int = 2, na: bool = False, evidence: bool = False, guidance_zh: str | None = None, guidance_en: str | None = None, ) -> dict[str, Any]: return { "code": code, "section_key": section[0], "section_zh": section[1], "section_en": section[2], "text_zh": zh, "text_en": en, "guidance_zh": guidance_zh, "guidance_en": guidance_en, "answer_type": "yes_no_na" if na else "yes_no", "options": None, "weight": weight, "expected_answer": "yes", "requires_evidence": evidence, } def _text( code: str, section: tuple[str, str, str], zh: str, en: str, ) -> dict[str, Any]: return { "code": code, "section_key": section[0], "section_zh": section[1], "section_en": section[2], "text_zh": zh, "text_en": en, "guidance_zh": None, "guidance_en": None, "answer_type": "text", "options": None, "weight": 0, "expected_answer": None, "requires_evidence": False, } def _scale( code: str, section: tuple[str, str, str], zh: str, en: str, *, weight: int = 3, ) -> dict[str, Any]: return { "code": code, "section_key": section[0], "section_zh": section[1], "section_en": section[2], "text_zh": zh, "text_en": en, "guidance_zh": "1=未建立 5=成熟且持续改进", "guidance_en": "1=none, 5=mature & continuously improving", "answer_type": "scale_1_5", "options": None, "weight": weight, "expected_answer": None, "requires_evidence": False, } _CERT_OPTIONS = [ {"value": "iso27001", "label_zh": "ISO/IEC 27001", "label_en": "ISO/IEC 27001", "score": 1.0}, {"value": "soc2", "label_zh": "SOC 2 Type II", "label_en": "SOC 2 Type II", "score": 1.0}, {"value": "both", "label_zh": "两者皆有", "label_en": "Both", "score": 1.0}, {"value": "in_progress", "label_zh": "认证进行中", "label_en": "In progress", "score": 0.5}, {"value": "none", "label_zh": "无", "label_en": "None", "score": 0.0}, ] _FULL_QUESTIONS: list[dict[str, Any]] = [ { "code": "GOV-1", "section_key": _SEC_GOV[0], "section_zh": _SEC_GOV[1], "section_en": _SEC_GOV[2], "text_zh": "贵公司当前持有哪些信息安全认证?", "text_en": "Which information-security certifications does your organization currently hold?", "guidance_zh": None, "guidance_en": None, "answer_type": "single_choice", "options": _CERT_OPTIONS, "weight": 4, "expected_answer": None, "requires_evidence": True, }, _yn("GOV-2", _SEC_GOV, "是否有正式的、经管理层批准并定期评审的信息安全策略?", "Do you have a formal information-security policy that is management-approved and reviewed periodically?", weight=3, evidence=True), _yn("GOV-3", _SEC_GOV, "是否指定了负责信息安全的专职人员或团队(如 CISO)?", "Have you designated a dedicated person or team responsible for information security (e.g. a CISO)?", weight=2), _yn("GOV-4", _SEC_GOV, "是否每年至少开展一次独立的第三方安全审计或渗透测试?", "Do you undergo an independent third-party security audit or penetration test at least annually?", weight=3, evidence=True), _yn("GOV-5", _SEC_GOV, "是否维护并定期更新信息资产清单及数据分类分级制度?", "Do you maintain and regularly update an information-asset inventory and a data-classification scheme?", weight=2), _yn("ORG-1", _SEC_ORG, "是否建立了正式的风险评估与风险处置流程,并定期执行?", "Do you have a formal risk-assessment and risk-treatment process that is executed periodically?", weight=3, evidence=True), _yn("ORG-2", _SEC_ORG, "是否对所有员工和承包商签署保密协议(NDA)?", "Do all employees and contractors sign a confidentiality / non-disclosure agreement?", weight=2), _yn("ORG-3", _SEC_ORG, "是否定义了访问控制策略,遵循最小权限与按需知密原则?", "Do you have an access-control policy based on least-privilege and need-to-know?", weight=3), _yn("ORG-4", _SEC_ORG, "是否对特权账号进行专门管理与定期复核?", "Are privileged accounts specially managed and reviewed on a regular cadence?", weight=3, evidence=True), _yn("PEO-1", _SEC_PEOPLE, "入职前是否对员工进行背景调查(在法律允许范围内)?", "Do you perform pre-employment background screening (where legally permitted)?", weight=2), _yn("PEO-2", _SEC_PEOPLE, "是否为全体员工提供年度信息安全意识培训?", "Do you provide annual security-awareness training to all staff?", weight=2, evidence=True), _yn("PEO-3", _SEC_PEOPLE, "离职或转岗时是否有及时撤销访问权限的正式流程?", "Is there a formal process to promptly revoke access on termination or role change?", weight=3), _yn("PHY-1", _SEC_PHYS, "数据中心 / 办公场所是否有物理访问控制(门禁、访客登记、监控)?", "Do your data-centre / office facilities have physical access controls (badge access, visitor logs, CCTV)?", weight=2, na=True), _yn("PHY-2", _SEC_PHYS, "是否使用了具备物理与环境安全认证的托管/云基础设施?", "Do you use hosting / cloud infrastructure with recognized physical & environmental security certifications?", weight=2, na=True), _yn("TEC-1", _SEC_TECH, "静态数据是否加密(如 AES-256)?", "Is data encrypted at rest (e.g. AES-256)?", weight=4, evidence=True), _yn("TEC-2", _SEC_TECH, "传输中数据是否强制使用 TLS 1.2 及以上加密?", "Is data in transit encrypted using TLS 1.2 or above?", weight=4), _yn("TEC-3", _SEC_TECH, "是否对所有管理及远程访问强制启用多因素认证(MFA)?", "Is multi-factor authentication (MFA) enforced for all administrative and remote access?", weight=4, evidence=True), _yn("TEC-4", _SEC_TECH, "是否有正式的补丁与漏洞管理流程,并有明确的修复时限?", "Do you have a formal patch- and vulnerability-management process with defined remediation SLAs?", weight=3, evidence=True), _yn("TEC-5", _SEC_TECH, "是否集中收集安全日志并保留至少 6 个月?", "Are security logs centrally collected and retained for at least 6 months?", weight=2), _yn("TEC-6", _SEC_TECH, "是否部署了端点保护(EDR/防病毒)并保持更新?", "Is endpoint protection (EDR / anti-malware) deployed and kept up to date?", weight=2), _yn("TEC-7", _SEC_TECH, "开发是否遵循安全开发生命周期(SSDLC),含代码审查与依赖扫描?", "Does development follow a secure SDLC including code review and dependency scanning?", weight=3, na=True), _yn("TEC-8", _SEC_TECH, "生产环境与测试/开发环境是否隔离,且测试不使用真实客户数据?", "Are production and test/dev environments segregated, with no real customer data used in testing?", weight=3), _scale("TEC-9", _SEC_TECH, "请自评贵公司整体技术安全防护的成熟度。", "Please self-rate the overall maturity of your technical security controls."), _yn("DAT-1", _SEC_DATA, "是否遵循适用的数据保护法规(如 GDPR / PIPL / CCPA)?", "Do you comply with applicable data-protection regulations (e.g. GDPR / PIPL / CCPA)?", weight=4, evidence=True), _yn("DAT-2", _SEC_DATA, "是否指定了数据保护负责人(DPO)或等效角色?", "Have you appointed a Data Protection Officer (DPO) or equivalent?", weight=2, na=True), _yn("DAT-3", _SEC_DATA, "是否维护数据处理活动记录,并明确数据留存与销毁策略?", "Do you maintain records of processing activities and defined data retention & destruction policies?", weight=3), _yn("DAT-4", _SEC_DATA, "客户数据是否会跨境传输?若是,是否有合规的传输机制?", "Is customer data transferred across borders? If so, are lawful transfer mechanisms in place?", weight=2, na=True), _yn("DAT-5", _SEC_DATA, "是否支持在合同终止后按约定返还或安全销毁客户数据?", "Can you return or securely destroy customer data on contract termination as agreed?", weight=3), _yn("DAT-6", _SEC_DATA, "是否会将客户数据委托给次级处理者(subprocessor)?是否对其进行安全评估?", "Do you engage subprocessors for customer data, and do you security-assess them?", weight=2, na=True), _yn("BCP-1", _SEC_BCP, "是否有文档化并定期演练的业务连续性/灾难恢复计划(BCP/DRP)?", "Do you have a documented and regularly tested business-continuity / disaster-recovery plan?", weight=3, evidence=True), _yn("BCP-2", _SEC_BCP, "是否定义了恢复时间目标(RTO)与恢复点目标(RPO)?", "Have you defined Recovery Time Objectives (RTO) and Recovery Point Objectives (RPO)?", weight=2), _yn("BCP-3", _SEC_BCP, "关键数据是否定期备份,且备份经过恢复测试?", "Are critical data backed up regularly, with backups periodically restore-tested?", weight=3), _yn("INC-1", _SEC_INC, "是否有正式的安全事件响应流程与响应团队?", "Do you have a formal security-incident response process and response team?", weight=3, evidence=True), _yn("INC-2", _SEC_INC, "发生涉及客户数据的安全事件时,是否承诺在约定时限内(如 72 小时)通知客户?", "Will you notify us of a security incident involving our data within an agreed timeframe (e.g. 72 hours)?", weight=4), _yn("INC-3", _SEC_INC, "是否保留事件记录并进行事后复盘(RCA)?", "Do you keep incident records and conduct post-incident root-cause analysis?", weight=2), _yn("SUP-1", _SEC_SUP, "是否对关键供应商/次级处理者进行安全尽职调查?", "Do you conduct security due diligence on critical suppliers / subprocessors?", weight=2), _yn("SUP-2", _SEC_SUP, "是否在与供应商的合同中包含安全与保密条款?", "Do your supplier contracts include security and confidentiality clauses?", weight=2), _text("SUP-3", _SEC_SUP, "请列出承载或处理客户数据的主要次级处理者及其所在地区。", "Please list the main subprocessors that host or process customer data and their regions."), _text("GEN-1", _SEC_GOV, "如有任何补充说明或已知的安全限制/例外,请在此说明。", "Please note any additional context or known security limitations / exceptions here."), ] _LITE_QUESTIONS: list[dict[str, Any]] = [ { "code": "L-1", "section_key": _SEC_GOV[0], "section_zh": _SEC_GOV[1], "section_en": _SEC_GOV[2], "text_zh": "贵公司当前持有哪些信息安全认证?", "text_en": "Which information-security certifications does your organization currently hold?", "guidance_zh": None, "guidance_en": None, "answer_type": "single_choice", "options": _CERT_OPTIONS, "weight": 4, "expected_answer": None, "requires_evidence": True, }, _yn("L-2", _SEC_GOV, "是否有管理层批准的信息安全策略?", "Do you have a management-approved information-security policy?", weight=3), _yn("L-3", _SEC_GOV, "是否每年进行独立的安全审计或渗透测试?", "Do you undergo an annual independent security audit or penetration test?", weight=3, evidence=True), _yn("L-4", _SEC_ORG, "是否遵循最小权限的访问控制,并对特权账号定期复核?", "Do you enforce least-privilege access control and periodically review privileged accounts?", weight=3), _yn("L-5", _SEC_PEOPLE, "离职时是否及时撤销访问权限?", "Is access promptly revoked upon termination?", weight=3), _yn("L-6", _SEC_PEOPLE, "是否为员工提供年度安全意识培训?", "Do you provide annual security-awareness training?", weight=2), _yn("L-7", _SEC_TECH, "静态与传输中数据是否均加密?", "Is data encrypted both at rest and in transit?", weight=4, evidence=True), _yn("L-8", _SEC_TECH, "管理与远程访问是否强制 MFA?", "Is MFA enforced for administrative and remote access?", weight=4), _yn("L-9", _SEC_TECH, "是否有补丁与漏洞管理流程及修复时限?", "Do you have patch/vulnerability management with remediation SLAs?", weight=3), _yn("L-10", _SEC_TECH, "生产与测试环境是否隔离,测试不使用真实客户数据?", "Are production and test environments segregated with no real customer data in test?", weight=2), _yn("L-11", _SEC_DATA, "是否遵循适用的数据保护法规(GDPR/PIPL/CCPA 等)?", "Do you comply with applicable data-protection regulations (GDPR/PIPL/CCPA)?", weight=4), _yn("L-12", _SEC_DATA, "合同终止后能否按约定返还或销毁客户数据?", "Can you return or destroy customer data on contract termination as agreed?", weight=3), _yn("L-13", _SEC_DATA, "是否维护数据留存与销毁策略?", "Do you maintain data retention & destruction policies?", weight=2), _yn("L-14", _SEC_BCP, "是否有经测试的备份与灾难恢复能力?", "Do you have tested backup and disaster-recovery capabilities?", weight=3, evidence=True), _yn("L-15", _SEC_INC, "是否有安全事件响应流程?", "Do you have a security-incident response process?", weight=3), _yn("L-16", _SEC_INC, "涉及客户数据的事件是否承诺在约定时限内通知?", "Will incidents involving our data be notified within an agreed timeframe?", weight=4), _yn("L-17", _SEC_ORG, "员工与承包商是否签署保密协议?", "Do employees and contractors sign confidentiality agreements?", weight=2), _yn("L-18", _SEC_SUP, "是否对承载客户数据的次级处理者进行安全评估?", "Do you security-assess subprocessors that host customer data?", weight=2, na=True), _scale("L-19", _SEC_TECH, "请自评贵公司整体安全防护成熟度。", "Please self-rate your overall security maturity."), _text("L-20", _SEC_GOV, "补充说明(可选)。", "Additional context (optional)."), ] SEED_TEMPLATES: list[dict[str, Any]] = [ { "key": "vendor_iso27001_full", "name_zh": "供应商安全评估问卷(完整版 · ISO 27001:2022 + ISO 27701)", "name_en": "Vendor Security Assessment — Full (ISO 27001:2022 + ISO 27701)", "description_zh": "面向关键/高风险供应商的完整安全评估问卷,覆盖治理、组织/人员/物理/技术控制、数据保护与隐私、业务连续性、事件管理与供应链。", "description_en": "Full security assessment for critical / high-risk vendors covering governance, organizational/people/physical/technical controls, data protection & privacy, continuity, incident management and supply chain.", "tier": "full", "sort_order": 10, "questions": _FULL_QUESTIONS, }, { "key": "vendor_lite", "name_zh": "供应商安全评估问卷(精简版)", "name_en": "Vendor Security Assessment — Lite", "description_zh": "面向中/低风险供应商的精简安全评估问卷(约 20 个核心问题)。", "description_en": "Short security assessment for medium / low-risk vendors (~20 essential questions).", "tier": "lite", "sort_order": 20, "questions": _LITE_QUESTIONS, }, ] 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, TMPL): op.create_table( TMPL, sa.Column("id", pid, primary_key=True), sa.Column("key", sa.String(64), nullable=False), sa.Column("name_zh", sa.String(255), nullable=False), sa.Column("name_en", sa.String(255), nullable=False), sa.Column("description_zh", sa.Text), sa.Column("description_en", sa.Text), sa.Column("kind", sa.String(16), nullable=False, server_default="vendor"), sa.Column("tier", sa.String(16)), sa.Column("version", sa.Integer, nullable=False, server_default=sa.text("1")), sa.Column("is_system", sa.Boolean, nullable=False, server_default=sa.text("0")), 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_sq_template_key"), **kw, ) op.create_index("idx_auditor_sq_template_enabled", TMPL, ["enabled", "sort_order"]) if not _table_exists(bind, TQ): op.create_table( TQ, sa.Column("id", pid, primary_key=True), sa.Column("template_id", pid, nullable=False), sa.Column("code", sa.String(32), nullable=False), sa.Column("section_key", sa.String(64), nullable=False), sa.Column("section_zh", sa.String(255), nullable=False), sa.Column("section_en", sa.String(255), nullable=False), sa.Column("text_zh", sa.Text, nullable=False), sa.Column("text_en", sa.Text, nullable=False), sa.Column("guidance_zh", sa.Text), sa.Column("guidance_en", sa.Text), sa.Column("answer_type", sa.String(24), nullable=False), sa.Column("options", sa.JSON), sa.Column("weight", sa.Integer, nullable=False, server_default=sa.text("1")), sa.Column("expected_answer", sa.String(64)), sa.Column("requires_evidence", sa.Boolean, nullable=False, server_default=sa.text("0")), 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.ForeignKeyConstraint( ["template_id"], [f"{TMPL}.id"], name="fk_sq_template_question_template", ondelete="CASCADE", ), sa.UniqueConstraint("template_id", "code", name="uq_auditor_sq_template_question_code"), **kw, ) op.create_index("idx_auditor_sq_tmpl_question_tmpl", TQ, ["template_id", "sort_order"]) if not _table_exists(bind, ASSESS): op.create_table( ASSESS, sa.Column("id", pid, primary_key=True), sa.Column("template_id", pid), sa.Column("template_key_snapshot", sa.String(64)), sa.Column("template_name_snapshot", sa.String(255)), sa.Column("supplier_id", pid), sa.Column("vendor_name", sa.String(255), nullable=False), sa.Column("vendor_contact_name", sa.String(128)), sa.Column("vendor_contact_email", sa.String(255)), sa.Column("tier", sa.String(16), nullable=False, server_default="medium"), sa.Column("status", sa.String(16), nullable=False, server_default="draft"), sa.Column("score_earned", sa.Integer), sa.Column("score_max", sa.Integer), sa.Column("score_pct", sa.Integer), sa.Column("risk_rating", sa.String(16)), sa.Column("rating_overridden", sa.Boolean, nullable=False, server_default=sa.text("0")), sa.Column("summary", sa.Text), sa.Column("owner_person_id", pid), sa.Column("reviewer_person_id", pid), sa.Column("due_at", sa.DateTime(timezone=True)), sa.Column("sent_at", sa.DateTime(timezone=True)), sa.Column("answered_at", sa.DateTime(timezone=True)), sa.Column("completed_at", sa.DateTime(timezone=True)), sa.Column("next_review_date", sa.Date), sa.Column("review_period_months", sa.Integer), sa.Column("review_reminder_stage", sa.String(16)), sa.Column("review_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_vendor_assess_status", ASSESS, ["status"]) op.create_index("idx_auditor_vendor_assess_review", ASSESS, ["next_review_date"]) op.create_index("idx_auditor_vendor_assess_supplier", ASSESS, ["supplier_id"]) op.create_index("idx_auditor_vendor_assess_owner", ASSESS, ["owner_person_id"]) if not _table_exists(bind, ANS): op.create_table( ANS, sa.Column("id", pid, primary_key=True), sa.Column("assessment_id", pid, nullable=False), sa.Column("question_code", sa.String(32), nullable=False), sa.Column("section_key", sa.String(64), nullable=False), sa.Column("section_zh", sa.String(255), nullable=False), sa.Column("section_en", sa.String(255), nullable=False), sa.Column("text_zh", sa.Text, nullable=False), sa.Column("text_en", sa.Text, nullable=False), sa.Column("guidance_zh", sa.Text), sa.Column("guidance_en", sa.Text), sa.Column("answer_type", sa.String(24), nullable=False), sa.Column("options", sa.JSON), sa.Column("weight", sa.Integer, nullable=False, server_default=sa.text("1")), sa.Column("expected_answer", sa.String(64)), sa.Column("requires_evidence", sa.Boolean, nullable=False, server_default=sa.text("0")), sa.Column("sort_order", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("answer_value", sa.Text), sa.Column("answer_comment", sa.Text), sa.Column("auto_score", sa.Integer), sa.Column("max_score", sa.Integer), sa.Column("is_scorable", sa.Boolean, nullable=False, server_default=sa.text("0")), sa.Column("review_flag", sa.String(16)), sa.Column("reviewer_note", sa.Text), sa.Column("finding_id", pid), 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( ["assessment_id"], [f"{ASSESS}.id"], name="fk_vendor_answer_assessment", ondelete="CASCADE", ), sa.UniqueConstraint("assessment_id", "question_code", name="uq_auditor_vendor_answer_code"), **kw, ) op.create_index("idx_auditor_vendor_answer_assess", ANS, ["assessment_id", "sort_order"]) if not _table_exists(bind, ATT): op.create_table( ATT, sa.Column("id", pid, primary_key=True), sa.Column("assessment_id", pid, nullable=False), sa.Column("answer_id", pid), sa.Column("kind", sa.String(24), nullable=False, server_default="evidence"), 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( ["assessment_id"], [f"{ASSESS}.id"], name="fk_vendor_att_assessment", ondelete="CASCADE", ), **kw, ) op.create_index("idx_auditor_vendor_att_assess", ATT, ["assessment_id"]) op.create_index("idx_auditor_vendor_att_answer", ATT, ["answer_id"]) # ── Seed system templates + questions (idempotent) ────────────── for tspec in SEED_TEMPLATES: row = bind.execute( sa.text(f"SELECT id FROM {TMPL} WHERE `key` = :k LIMIT 1"), {"k": tspec["key"]}, ).first() if row is None: tmpl_id = str(uuid.uuid4()) bind.execute( sa.text( f""" INSERT INTO {TMPL} (id, `key`, name_zh, name_en, description_zh, description_en, kind, tier, version, is_system, enabled, sort_order, created_at, updated_at) VALUES (:id, :key, :name_zh, :name_en, :description_zh, :description_en, 'vendor', :tier, 1, 1, 1, :sort_order, NOW(), NOW()) """ ), { "id": tmpl_id, "key": tspec["key"], "name_zh": tspec["name_zh"], "name_en": tspec["name_en"], "description_zh": tspec.get("description_zh"), "description_en": tspec.get("description_en"), "tier": tspec.get("tier"), "sort_order": tspec.get("sort_order", 0), }, ) else: tmpl_id = row[0] for order, q in enumerate(tspec["questions"], start=1): exists = bind.execute( sa.text( f"SELECT 1 FROM {TQ} WHERE template_id = :t AND code = :c LIMIT 1" ), {"t": tmpl_id, "c": q["code"]}, ).scalar() if exists: continue bind.execute( sa.text( f""" INSERT INTO {TQ} (id, template_id, code, section_key, section_zh, section_en, text_zh, text_en, guidance_zh, guidance_en, answer_type, options, weight, expected_answer, requires_evidence, sort_order, created_at, updated_at) VALUES (:id, :template_id, :code, :section_key, :section_zh, :section_en, :text_zh, :text_en, :guidance_zh, :guidance_en, :answer_type, :options, :weight, :expected_answer, :requires_evidence, :sort_order, NOW(), NOW()) """ ), { "id": str(uuid.uuid4()), "template_id": tmpl_id, "code": q["code"], "section_key": q["section_key"], "section_zh": q["section_zh"], "section_en": q["section_en"], "text_zh": q["text_zh"], "text_en": q["text_en"], "guidance_zh": q.get("guidance_zh"), "guidance_en": q.get("guidance_en"), "answer_type": q["answer_type"], "options": json.dumps(q["options"], ensure_ascii=False) if q.get("options") else None, "weight": q.get("weight", 1), "expected_answer": q.get("expected_answer"), "requires_evidence": 1 if q.get("requires_evidence") else 0, "sort_order": order * 10, }, ) # ── Seed daily review-reminder 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 4 * * *', 'UTC', 1, '{}', 0, 0, NOW(), NOW()) """ ), { "id": str(uuid.uuid4()), "name": JOB, "display_name": "安全问卷 · 供应商复评到期提醒", "description": ( "每天 04:00 UTC 按 next_review_date 分阶段(30/14/到期/逾期)" "提醒供应商安全评估负责人复评;每阶段仅提醒一次;仅内部推送。" ), }, ) def downgrade() -> None: bind = op.get_bind() bind.execute(sa.text("DELETE FROM scheduled_jobs WHERE name = :n"), {"n": JOB}) op.drop_table(ATT) op.drop_table(ANS) op.drop_table(ASSESS) op.drop_table(TQ) op.drop_table(TMPL)