"""安全每日一题 (security daily quiz). Three tables backing a gamified daily security-awareness quiz: * ``secquiz_questions`` the question bank (seed + AI + manual), editable * ``secquiz_daily`` the deterministic question-of-the-day assignment * ``secquiz_answers`` one answer per person per day (drives streaks + board) All person/question references FK with the parent's charset/collation (``id_type_for`` / ``table_kw_for``) so inline FKs can't trip error 3780 on prod TiDB. Seeds a curated starter bank on first creation so the feature works before any AI generation runs. """ 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 = "313" down_revision = "312" branch_labels = None depends_on = None QUESTIONS = "secquiz_questions" DAILY = "secquiz_daily" ANSWERS = "secquiz_answers" # Curated starter bank (Chinese — question content is data, like KB articles). # source='seed'. Admins can edit/add; the AI generator appends source='ai'. _SEED: list[dict] = [ { "prompt": "收到一封“IT 部门”邮件，要求你点击链接重置密码，下列哪种做法最安全？", "options": [ {"key": "A", "text": "直接点击邮件里的链接并输入账号密码"}, {"key": "B", "text": "不点链接，手动打开公司官方系统或联系 IT 核实"}, {"key": "C", "text": "把密码回复到这封邮件确认身份"}, {"key": "D", "text": "转发给同事让他们也尽快重置"}, ], "correct_key": "B", "explanation": "钓鱼邮件常伪装成 IT。永远不要点邮件里的链接输入凭据；用官方入口或直接联系 IT 核实。", "category": "phishing", }, { "prompt": "下面哪个密码最安全？", "options": [ {"key": "A", "text": "12345678"}, {"key": "B", "text": "password"}, {"key": "C", "text": "公司名+生日"}, {"key": "D", "text": "一串长且随机、且每个站点不同的密码（配合密码管理器）"}, ], "correct_key": "D", "explanation": "长、随机、每站唯一的密码最难被破解；用密码管理器生成并保存。", "category": "password", }, { "prompt": "离开工位时，正确的做法是？", "options": [ {"key": "A", "text": "锁屏（Win+L / Ctrl+Cmd+Q）"}, {"key": "B", "text": "把屏幕调暗就行"}, {"key": "C", "text": "放着不管，反正一会儿就回来"}, {"key": "D", "text": "把密码贴在显示器上方便同事"}, ], "correct_key": "A", "explanation": "离开必锁屏，防止他人趁机访问你的账号与数据。", "category": "device", }, { "prompt": "公共 Wi-Fi 下处理工作，最稳妥的做法是？", "options": [ {"key": "A", "text": "直接连，反正有密码"}, {"key": "B", "text": "连接公司 VPN 后再访问内部系统"}, {"key": "C", "text": "关掉防火墙加快速度"}, {"key": "D", "text": "把文件发到个人网盘再下载"}, ], "correct_key": "B", "explanation": "公共 Wi-Fi 不可信，访问内部资源务必先连公司 VPN 加密流量。", "category": "network", }, { "prompt": "同事在群里发来一个 .exe 附件说是“福利领取工具”，你应该？", "options": [ {"key": "A", "text": "立刻下载运行抢福利"}, {"key": "B", "text": "不运行，先向该同事本人核实，并报告 IT"}, {"key": "C", "text": "转发到更多群让大家一起领"}, {"key": "D", "text": "改个名字再运行就安全了"}, ], "correct_key": "B", "explanation": "账号可能被盗用发恶意文件。陌生可执行文件不要运行，核实来源并报告 IT。", "category": "malware", }, { "prompt": "关于多因素认证（MFA / 2FA），下列说法正确的是？", "options": [ {"key": "A", "text": "开了 MFA 就不用设强密码了"}, {"key": "B", "text": "MFA 太麻烦，能不开就不开"}, {"key": "C", "text": "MFA 在密码泄露时多一道防线，应尽量开启"}, {"key": "D", "text": "把 MFA 验证码截图发群里方便登录"}, ], "correct_key": "C", "explanation": "MFA 是密码之外的第二道防线，强烈建议在所有支持的系统开启；验证码绝不外传。", "category": "account", }, { "prompt": "你不小心把一份含客户信息的文件发错了人，应该？", "options": [ {"key": "A", "text": "假装没发生，删掉自己的发送记录"}, {"key": "B", "text": "第一时间上报 IT / 安全团队，按流程处置"}, {"key": "C", "text": "私下让对方删掉就当没事"}, {"key": "D", "text": "等对方先发现再说"}, ], "correct_key": "B", "explanation": "数据误发属于安全事件，越早上报越能控制影响；隐瞒会让后果更严重。", "category": "data", }, { "prompt": "陌生人尾随你进入需要刷卡的办公区，正确做法是？", "options": [ {"key": "A", "text": "礼貌帮他开门，与人为善"}, {"key": "B", "text": "请对方自行刷卡或到前台登记，不要让其尾随进入"}, {"key": "C", "text": "假装没看见快步走开"}, {"key": "D", "text": "把自己的门禁卡借给他"}, ], "correct_key": "B", "explanation": "尾随（tailgating）是常见的物理入侵手段；请对方走正规登记流程，不确定时联系前台/安保。", "category": "physical", }, ] 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") seed_now = False if not _table_exists(bind, QUESTIONS): op.create_table( QUESTIONS, sa.Column("id", pid, primary_key=True), sa.Column("prompt", sa.Text(), nullable=False), sa.Column("options", sa.JSON(), nullable=False), sa.Column("correct_key", sa.String(8), nullable=False), sa.Column("explanation", sa.Text(), nullable=True), sa.Column("category", sa.String(32), nullable=False, server_default="general"), sa.Column("difficulty", sa.String(16), nullable=False, server_default="normal"), sa.Column("status", sa.String(16), nullable=False, server_default="active"), sa.Column("source", sa.String(16), nullable=False, server_default="seed"), sa.Column("created_by", pid, nullable=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(["created_by"], ["identity_persons.id"], name="fk_secquiz_q_created_by"), **kw, ) op.create_index("ix_secquiz_questions_status", QUESTIONS, ["status"]) seed_now = True if not _table_exists(bind, DAILY): op.create_table( DAILY, sa.Column("id", pid, primary_key=True), sa.Column("quiz_date", sa.Date(), nullable=False), sa.Column("question_id", pid, nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.ForeignKeyConstraint(["question_id"], [f"{QUESTIONS}.id"], name="fk_secquiz_daily_question"), sa.UniqueConstraint("quiz_date", name="uq_secquiz_daily_date"), **kw, ) if not _table_exists(bind, ANSWERS): op.create_table( ANSWERS, sa.Column("id", pid, primary_key=True), sa.Column("person_id", pid, nullable=False), sa.Column("quiz_date", sa.Date(), nullable=False), sa.Column("question_id", pid, nullable=False), sa.Column("selected_key", sa.String(8), nullable=False), sa.Column("is_correct", sa.Boolean(), nullable=False), sa.Column("answered_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.ForeignKeyConstraint(["person_id"], ["identity_persons.id"], name="fk_secquiz_ans_person"), sa.ForeignKeyConstraint(["question_id"], [f"{QUESTIONS}.id"], name="fk_secquiz_ans_question"), sa.UniqueConstraint("person_id", "quiz_date", name="uq_secquiz_ans_person_date"), **kw, ) op.create_index("ix_secquiz_answers_person", ANSWERS, ["person_id", "quiz_date"]) op.create_index("ix_secquiz_answers_date", ANSWERS, ["quiz_date", "is_correct"]) if seed_now: rows = [ { "id": str(uuid.uuid4()), "prompt": q["prompt"], "options": q["options"], "correct_key": q["correct_key"], "explanation": q["explanation"], "category": q["category"], "difficulty": "normal", "status": "active", "source": "seed", } for q in _SEED ] op.bulk_insert( sa.table( QUESTIONS, sa.column("id", pid), sa.column("prompt", sa.Text()), sa.column("options", sa.JSON()), sa.column("correct_key", sa.String(8)), sa.column("explanation", sa.Text()), sa.column("category", sa.String(32)), sa.column("difficulty", sa.String(16)), sa.column("status", sa.String(16)), sa.column("source", sa.String(16)), ), rows, ) # Seed the daily-assignment cron (idempotent; enabled — it's safe + the # player path lazily ensures the assignment anyway). Pre-minted hyphenated # UUID (TiDB UUID()/no-dash pitfalls — see scheduled_jobs UUID notes). job_exists = bind.execute( sa.text("SELECT 1 FROM scheduled_jobs WHERE name = :n LIMIT 1"), {"n": "secquiz.assign_daily"}, ).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, '5 0 * * *', 'UTC', 1, '{}', 0, 0, NOW(), NOW()) """ ), { "id": str(uuid.uuid4()), "name": "secquiz.assign_daily", "display_name": "安全每日一题 · 指派今日题目", "description": ( "每日 00:05 UTC 为 HQ 日期预指派今日安全题(轮换最久未用的" "在线题目)。幂等;玩家访问 /security-quiz/today 时亦会懒加载兜底。" ), }, ) def downgrade() -> None: bind = op.get_bind() bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE name = :n"), {"n": "secquiz.assign_daily"}, ) for name in (ANSWERS, DAILY, QUESTIONS): if _table_exists(bind, name): op.drop_table(name)