"""安全每日一题 service — daily question, answering, streaks, leaderboard,
compliance coverage, admin CRUD, and AI question generation.

Read/write assembly for the quiz domain. Streaks and the leaderboard are
computed from ``secquiz_answers``; the question-of-the-day is a per-date
assignment (cron-filled, lazily ensured on read so the feature never shows an
empty state). The AI generator reuses the org's existing LLM connector and is a
graceful no-op when no key is configured.
"""
from __future__ import annotations

import logging
import uuid
from datetime import date, datetime, timedelta
from typing import Any

from sqlalchemy import and_, case, func, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import is_transaction_aborting_db_error
from app.core.locale import Lang
from app.core.working_hours import resolve_timezone
from app.domains.security_quiz import localization
from app.domains.security_quiz.models import (
    QTYPE_BOOLEAN,
    QTYPE_MULTIPLE,
    QTYPE_SINGLE,
    QTYPES,
    QUESTION_ACTIVE,
    SOURCE_AI,
    SOURCE_MANUAL,
    SecQuizAnswer,
    SecQuizDaily,
    SecQuizQuestion,
)
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)


def _local_today(person: IdentityPerson) -> date:
    return datetime.now(resolve_timezone(person)).date()


def _normalize_keys(value: str | None) -> str:
    """Sorted, unique, uppercase option keys. 'b a a' -> 'AB'; 'D' -> 'D'.
    The canonical form for comparing a selection against the correct answer,
    independent of input order or duplicates."""
    return "".join(sorted({c for c in (value or "").upper() if c.isalpha()}))


def _qtype_of(q: SecQuizQuestion) -> str:
    """Tolerate pre-qtype rows (NULL/empty) by defaulting to single-choice."""
    return getattr(q, "qtype", None) or QTYPE_SINGLE


def _public_question(q: SecQuizQuestion, lang: Lang) -> dict[str, Any]:
    """Question shape WITHOUT the answer — what an unanswered player sees.

    Resolved into ONE language server-side (see ``localization``), never
    handed over as a per-language map: the option list the reader picks
    from and the ``correct_key`` the grader compares against have to come
    from the same resolution, and a client free to choose could grade
    against options it did not render.
    """
    view = localization.render(q, lang)
    return {
        "id": str(q.id),
        "prompt": view["prompt"],
        "options": view["options"],
        "qtype": _qtype_of(q),
        "category": q.category,
        "difficulty": q.difficulty,
        "lang": view["lang"],
        # True when this question has no copy in the reader's language and
        # they are looking at a fallback. The page says so rather than
        # leaving a Chinese question on an English screen unexplained.
        "is_fallback_lang": view["is_fallback_lang"],
    }


def _answer_reveal(q: SecQuizQuestion, lang: Lang) -> dict[str, Any]:
    view = localization.render(q, lang)
    return {
        "correct_key": q.correct_key,
        "explanation": view["explanation"],
        "explanation_lang": view["lang"],
    }


# ── Daily assignment ────────────────────────────────────────────────


async def ensure_today_assignment(
    db: AsyncSession, quiz_date: date,
) -> SecQuizDaily | None:
    """Get-or-create the question-of-the-day. Picks the active question used
    least recently (rotation); falls back to any active question. Race-safe via
    a nested savepoint + the UNIQUE(quiz_date) guard. Returns None only when the
    bank has no active questions."""
    existing = await db.scalar(
        select(SecQuizDaily).where(SecQuizDaily.quiz_date == quiz_date)
    )
    if existing is not None:
        return existing

    # Active questions never (or least recently) assigned come first.
    last_used_sq = (
        select(SecQuizDaily.question_id, func.max(SecQuizDaily.quiz_date).label("last_used"))
        .group_by(SecQuizDaily.question_id)
        .subquery()
    )
    picked = await db.scalar(
        select(SecQuizQuestion.id)
        .outerjoin(last_used_sq, last_used_sq.c.question_id == SecQuizQuestion.id)
        .where(SecQuizQuestion.status == QUESTION_ACTIVE)
        .order_by(last_used_sq.c.last_used.is_(None).desc(), last_used_sq.c.last_used.asc())
        .limit(1)
    )
    if picked is None:
        return None

    row = SecQuizDaily(id=uuid.uuid4(), quiz_date=quiz_date, question_id=picked)
    try:
        async with db.begin_nested():
            db.add(row)
            await db.flush()
    except IntegrityError as exc:
        if is_transaction_aborting_db_error(exc):
            # txn-aborting (deadlock 1213 / lock-wait 1205): the transaction is
            # already rolled back server-side — re-raise so the caller retries on
            # a fresh session instead of querying a poisoned transaction.
            raise
        # Either we lost the race (another worker already assigned this date) or
        # the insert genuinely violated a constraint (e.g. a UUID-format/FK
        # mismatch). Read the winner; if there's none, this was NOT a race —
        # re-raise so the failure surfaces (a swallowed None here would leave the
        # day silently unassigned, the empty-state bug this domain just had).
        winner = await db.scalar(
            select(SecQuizDaily).where(SecQuizDaily.quiz_date == quiz_date)
        )
        if winner is None:
            raise
        return winner
    return row


# ── Player surface ──────────────────────────────────────────────────


async def get_today(
    db: AsyncSession, person: IdentityPerson, lang: Lang | None = None,
) -> dict[str, Any]:
    """Today's question, rendered in the reader's language.

    ``lang`` is resolved by the caller — the router from ``?lang=`` /
    ``Accept-Language`` / the person's stored preference. Passing None
    falls back to the person's own preference, which is what the MCP
    surface wants: there is no browser there to ask.
    """
    lang = lang or localization.reader_lang(person)
    today = _local_today(person)
    assignment = await ensure_today_assignment(db, today)
    if assignment is None:
        return {"quiz_date": today.isoformat(), "available": False, "answered": False, "question": None}

    question = await db.get(SecQuizQuestion, assignment.question_id)
    if question is None:
        return {"quiz_date": today.isoformat(), "available": False, "answered": False, "question": None}

    answer = await db.scalar(
        select(SecQuizAnswer).where(
            SecQuizAnswer.person_id == person.id,
            SecQuizAnswer.quiz_date == today,
        )
    )
    out: dict[str, Any] = {
        "quiz_date": today.isoformat(),
        "available": True,
        "answered": answer is not None,
        "question": _public_question(question, lang),
    }
    if answer is not None:
        out["result"] = {
            "selected_key": answer.selected_key,
            "is_correct": answer.is_correct,
            **_answer_reveal(question, lang),
        }
        out["stats"] = await compute_stats(db, person.id, today)
    return out


async def submit_answer(
    db: AsyncSession, person: IdentityPerson, selected_key: str,
    lang: Lang | None = None,
) -> dict[str, Any]:
    lang = lang or localization.reader_lang(person)
    today = _local_today(person)
    assignment = await ensure_today_assignment(db, today)
    if assignment is None:
        return {"available": False}
    question = await db.get(SecQuizQuestion, assignment.question_id)
    if question is None:
        return {"available": False}

    # Idempotent: an existing answer wins (one shot per day).
    existing = await db.scalar(
        select(SecQuizAnswer).where(
            SecQuizAnswer.person_id == person.id,
            SecQuizAnswer.quiz_date == today,
        )
    )
    if existing is not None:
        return {
            "available": True,
            "already_answered": True,
            "result": {
                "selected_key": existing.selected_key,
                "is_correct": existing.is_correct,
                **_answer_reveal(question, lang),
            },
            "stats": await compute_stats(db, person.id, today),
        }

    valid_keys = {o.get("key") for o in (question.options or [])}
    qtype = _qtype_of(question)
    selected_norm = _normalize_keys(selected_key)
    bad = (
        not selected_norm
        or any(c not in valid_keys for c in selected_norm)
        or (qtype != QTYPE_MULTIPLE and len(selected_norm) != 1)
    )
    if bad:
        from app.core.exceptions import AppException
        raise AppException(
            code="SECQUIZ_BAD_OPTION",
            message="Selected option is not one of the question's choices.",
            status_code=422,
        )

    # Multiple-choice is correct only on an exact set match (order-insensitive);
    # single/boolean reduce to a single-key compare via the same normalisation.
    is_correct = selected_norm == _normalize_keys(question.correct_key)
    row = SecQuizAnswer(
        id=uuid.uuid4(), person_id=person.id, quiz_date=today,
        question_id=question.id, selected_key=selected_norm, is_correct=is_correct,
    )
    try:
        async with db.begin_nested():
            db.add(row)
            await db.flush()
    except IntegrityError as exc:
        if is_transaction_aborting_db_error(exc):
            # txn-aborting (deadlock 1213 / lock-wait 1205): the transaction is
            # already rolled back server-side — re-raise so the caller retries on
            # a fresh session instead of querying a poisoned transaction.
            raise
        # Raced with another concurrent submit — return the persisted winner.
        winner = await db.scalar(
            select(SecQuizAnswer).where(
                SecQuizAnswer.person_id == person.id,
                SecQuizAnswer.quiz_date == today,
            )
        )
        if winner is not None:
            return {
                "available": True,
                "already_answered": True,
                "result": {
                    "selected_key": winner.selected_key,
                    "is_correct": winner.is_correct,
                    **_answer_reveal(question, lang),
                },
                "stats": await compute_stats(db, person.id, today),
            }
        raise

    return {
        "available": True,
        "already_answered": False,
        "result": {
            "selected_key": selected_norm,
            "is_correct": is_correct,
            **_answer_reveal(question, lang),
        },
        "stats": await compute_stats(db, person.id, today),
    }


async def compute_stats(
    db: AsyncSession, person_id: uuid.UUID, today: date,
) -> dict[str, Any]:
    rows = (await db.execute(
        select(SecQuizAnswer.quiz_date, SecQuizAnswer.is_correct)
        .where(SecQuizAnswer.person_id == person_id)
        .order_by(SecQuizAnswer.quiz_date)
    )).all()
    total_answered = len(rows)
    total_correct = sum(1 for _, ok in rows if ok)
    correct_dates = sorted({d for d, ok in rows if ok})

    # Longest run of consecutive calendar dates answered correctly.
    longest = 0
    run = 0
    prev: date | None = None
    for d in correct_dates:
        run = run + 1 if (prev is not None and (d - prev).days == 1) else 1
        longest = max(longest, run)
        prev = d

    # Current streak: trailing run ending at today or yesterday.
    current = 0
    if correct_dates and (today - correct_dates[-1]).days <= 1:
        cd = set(correct_dates)
        expected = correct_dates[-1]
        while expected in cd:
            current += 1
            expected -= timedelta(days=1)

    return {
        "total_answered": total_answered,
        "total_correct": total_correct,
        "current_streak": current,
        "longest_streak": longest,
    }


# ── Leaderboard ─────────────────────────────────────────────────────


async def get_leaderboard(
    db: AsyncSession, person: IdentityPerson, limit: int = 20,
) -> dict[str, Any]:
    correct_expr = func.sum(case((SecQuizAnswer.is_correct == True, 1), else_=0))  # noqa: E712
    # Exclude departed/soft-deleted people from the board AND from the
    # rank math below — a leaderboard is a headcount-style display, so a
    # high-scorer who has left must not top it forever nor inflate the
    # ``ahead`` count behind everyone else (matches get_coverage /
    # team_battle, which already filter ``employment_status != 'left'``).
    base = (
        select(
            SecQuizAnswer.person_id,
            correct_expr.label("correct"),
            func.count().label("answered"),
        )
        .join(IdentityPerson, IdentityPerson.id == SecQuizAnswer.person_id)
        .where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
        )
        .group_by(SecQuizAnswer.person_id)
        .subquery()
    )
    rows = (await db.execute(
        select(
            base.c.person_id, base.c.correct, base.c.answered,
            IdentityPerson.display_name, IdentityPerson.avatar_url,
            IdentityPerson.employment_status,
        )
        .join(IdentityPerson, IdentityPerson.id == base.c.person_id)
        .where(IdentityPerson.deleted_at.is_(None))
        .order_by(base.c.correct.desc(), base.c.answered.desc())
        .limit(max(1, min(limit, 100)))
    )).all()

    top = [
        {
            "rank": i + 1,
            "person_id": str(pid),
            "display_name": name or "—",
            "avatar_url": avatar,
            "correct": int(correct or 0),
            "answered": int(answered or 0),
            "is_me": pid == person.id,
        }
        for i, (pid, correct, answered, name, avatar, _status) in enumerate(rows)
    ]

    # Caller's own rank. The count of people "ahead" MUST use the same
    # ordering as the displayed board — ``(correct DESC, answered DESC)`` — or
    # a player tied on ``correct`` but with more ``answered`` renders above the
    # caller in ``top`` yet isn't counted here, making my_rank one better than
    # the caller's visible position. Pull both of the caller's dimensions and
    # apply the secondary tiebreak.
    my_row = (await db.execute(
        select(base.c.correct, base.c.answered).where(base.c.person_id == person.id)
    )).first()
    my_correct = int((my_row.correct if my_row else 0) or 0)
    my_answered = int((my_row.answered if my_row else 0) or 0)
    ahead = 0
    if my_correct > 0:
        ahead = await db.scalar(
            select(func.count()).select_from(base).where(
                or_(
                    base.c.correct > my_correct,
                    and_(
                        base.c.correct == my_correct,
                        base.c.answered > my_answered,
                    ),
                )
            )
        )
    my_rank = int(ahead or 0) + 1 if my_correct > 0 else None

    return {
        "top": top,
        "my_rank": my_rank,
        "my_correct": my_correct,
    }


# ── Compliance coverage ─────────────────────────────────────────────


async def language_coverage(db: AsyncSession) -> dict[str, Any]:
    """Per-language readability of the ACTIVE bank.

    Only active questions count: an archived question will never be
    assigned again, so translating it buys nobody anything and letting it
    into the denominator would make the gap look permanently worse than
    it is.

    Computed in Python over a few hundred rows rather than in SQL, because
    "does this block render" is :func:`localization.block_for`'s judgement
    (a present-but-empty prompt does not count, and neither does an option
    list whose keys drifted from the baseline) — and a JSON_EXTRACT
    predicate that answered a slightly different question would make this
    number disagree with what employees actually see.
    """
    rows = (await db.execute(
        select(SecQuizQuestion).where(SecQuizQuestion.status == QUESTION_ACTIVE)
    )).scalars().all()
    total = len(rows)
    per_lang: dict[str, int] = {}
    for lang in localization.QUIZ_LANGS:
        per_lang[lang] = sum(
            1 for q in rows if localization.block_for(q, lang) is not None
        )
    return {
        "active_questions": total,
        "per_lang": per_lang,
        "per_lang_pct": {
            lang: (round(100.0 * n / total, 1) if total else 0.0)
            for lang, n in per_lang.items()
        },
        # Questions every employee can read in their own language, whichever
        # of the three that is. The honest headline number.
        "fully_localized": sum(
            1 for q in rows
            if all(
                localization.block_for(q, lang) is not None
                for lang in localization.QUIZ_LANGS
            )
        ),
    }


async def get_coverage(db: AsyncSession) -> dict[str, Any]:
    """Security-awareness participation — the auditor's evidence read."""
    # HQ-local date so the 30-day window aligns with the quiz_date values the
    # scanner/answers store (Asia/Shanghai), regardless of the server's wall TZ.
    from zoneinfo import ZoneInfo
    now_hq = datetime.now(ZoneInfo("Asia/Shanghai"))
    today = now_hq.date()
    since_30d = today - timedelta(days=30)
    headcount = await db.scalar(
        select(func.count()).select_from(IdentityPerson).where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
        )
    )
    headcount = int(headcount or 0)
    # Numerators MUST use the same population as the ``headcount`` denominator
    # (active, non-deleted). Counting distinct answerers WITHOUT the leaver
    # filter let departed employees who ever answered inflate coverage — over
    # the quiz's lifetime the all-time % could exceed 100%, and this is the
    # auditor's compliance-evidence read. Join IdentityPerson + mirror the
    # headcount predicate (same as the leaderboard's departed-exclusion).
    participants = await db.scalar(
        select(func.count(func.distinct(SecQuizAnswer.person_id)))
        .select_from(SecQuizAnswer)
        .join(IdentityPerson, IdentityPerson.id == SecQuizAnswer.person_id)
        .where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
        )
    )
    participants = int(participants or 0)
    participants_30d = await db.scalar(
        select(func.count(func.distinct(SecQuizAnswer.person_id)))
        .select_from(SecQuizAnswer)
        .join(IdentityPerson, IdentityPerson.id == SecQuizAnswer.person_id)
        .where(
            SecQuizAnswer.quiz_date >= since_30d,
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
        )
    )
    participants_30d = int(participants_30d or 0)
    total_answers = int(await db.scalar(select(func.count()).select_from(SecQuizAnswer)) or 0)
    return {
        "headcount": headcount,
        # How much of the live bank an en / ja reader can actually read.
        # In the same payload as participation on purpose: "94% of staff
        # answered" is a weaker claim than it looks if a third of them were
        # answering in a language they do not read, and this is the number
        # the security desk cites as awareness evidence.
        "language_coverage": await language_coverage(db),
        "participants_all_time": participants,
        "participants_30d": participants_30d,
        "coverage_30d_pct": round(100.0 * participants_30d / headcount, 1) if headcount else 0.0,
        "coverage_all_time_pct": round(100.0 * participants / headcount, 1) if headcount else 0.0,
        "total_answers": total_answers,
        "generated_at": now_hq.isoformat(),
    }


# ── Admin CRUD ──────────────────────────────────────────────────────


def serialize_question(q: SecQuizQuestion) -> dict[str, Any]:
    return {
        "id": str(q.id),
        "prompt": q.prompt,
        "options": q.options,
        "qtype": _qtype_of(q),
        "correct_key": q.correct_key,
        "explanation": q.explanation,
        "category": q.category,
        "difficulty": q.difficulty,
        "status": q.status,
        "source": q.source,
        # Admin console only: the raw per-language blocks, so the editor can
        # render one tab per language. Employees never see this shape — their
        # read is resolved into ONE language server-side.
        "localized": q.localized or {},
        "available_langs": localization.available_langs(q),
        "created_at": q.created_at.isoformat() if q.created_at else None,
    }


async def list_questions(
    db: AsyncSession, status: str | None = None,
) -> list[dict[str, Any]]:
    stmt = select(SecQuizQuestion).order_by(SecQuizQuestion.created_at.desc())
    if status:
        stmt = stmt.where(SecQuizQuestion.status == status)
    rows = (await db.execute(stmt)).scalars().all()
    return [serialize_question(q) for q in rows]


def _validate_payload(prompt: str, options: list, correct_key: str, qtype: str) -> str:
    """Validate a question payload and return the canonical correct_key.

    single/boolean → correct_key is exactly one option key. multiple → it is one
    or more option keys, returned as the sorted concatenation (e.g. "ABD")."""
    from app.core.exceptions import AppException

    keys = [o.get("key") for o in (options or [])]
    if not prompt or not prompt.strip():
        raise AppException(code="SECQUIZ_BAD", message="Prompt is required.", status_code=422)
    if qtype not in QTYPES:
        raise AppException(code="SECQUIZ_BAD", message="Unknown question type.", status_code=422)
    if len(options or []) < 2:
        raise AppException(code="SECQUIZ_BAD", message="At least 2 options required.", status_code=422)
    if qtype == QTYPE_BOOLEAN and len(options) != 2:
        raise AppException(code="SECQUIZ_BAD", message="Boolean questions need exactly 2 options.", status_code=422)
    if len(keys) != len(set(keys)):
        raise AppException(code="SECQUIZ_BAD", message="Option keys must be unique.", status_code=422)

    norm = _normalize_keys(correct_key)
    if not norm or any(c not in keys for c in norm):
        raise AppException(code="SECQUIZ_BAD", message="correct_key must be one of the options.", status_code=422)
    if qtype != QTYPE_MULTIPLE and len(norm) != 1:
        raise AppException(
            code="SECQUIZ_BAD",
            message="Single/boolean questions need exactly one correct option.",
            status_code=422,
        )
    return norm


def _validate_localized(raw: Any, options: list) -> dict[str, Any]:
    """Localized payload → stored shape, or a 422 the editor can read."""
    from app.core.exceptions import AppException

    try:
        return localization.validate_localized(raw, baseline_options=options)
    except localization.LocalizationError as exc:
        raise AppException(
            code="SECQUIZ_BAD_LOCALIZED", message=str(exc), status_code=422,
        ) from exc


async def create_question(
    db: AsyncSession, *, actor: IdentityPerson, prompt: str, options: list,
    correct_key: str, explanation: str | None, category: str, difficulty: str,
    qtype: str = QTYPE_SINGLE, source: str = SOURCE_MANUAL,
    localized: dict | None = None,
) -> dict[str, Any]:
    qtype = qtype or QTYPE_SINGLE
    correct_key = _validate_payload(prompt, options, correct_key, qtype)
    blocks = _validate_localized(localized, options)
    q = SecQuizQuestion(
        id=uuid.uuid4(), prompt=prompt.strip(), options=options, qtype=qtype,
        correct_key=correct_key, explanation=explanation, category=category or "general",
        difficulty=difficulty or "normal", status=QUESTION_ACTIVE, source=source,
        localized=blocks or None, created_by=actor.id,
    )
    db.add(q)
    await db.flush()
    await db.commit()
    # Re-load (async) so serialising the server-default created_at/updated_at
    # doesn't trip a sync lazy-load (MissingGreenlet) post-commit.
    await db.refresh(q)
    return serialize_question(q)


async def update_question(
    db: AsyncSession, question_id: uuid.UUID, **fields: Any,
) -> dict[str, Any]:
    from app.core.exceptions import NotFoundError

    q = await db.get(SecQuizQuestion, question_id)
    if q is None:
        raise NotFoundError("secquiz_question", str(question_id))
    prompt = fields.get("prompt", q.prompt)
    options = fields.get("options", q.options)
    correct_key = fields.get("correct_key", q.correct_key)
    qtype = fields.get("qtype", _qtype_of(q)) or QTYPE_SINGLE
    fields["correct_key"] = _validate_payload(prompt, options, correct_key, qtype)
    fields["qtype"] = qtype

    # Did the BASELINE question change? Compared before anything is
    # assigned, because SQLAlchemy attribute assignment is what we are
    # about to do and after it the old value is gone.
    baseline_changed = (
        prompt != q.prompt
        or options != q.options
        or fields["correct_key"] != q.correct_key
    )

    for attr in ("prompt", "options", "qtype", "correct_key", "explanation", "category", "difficulty", "status"):
        if attr in fields and fields[attr] is not None:
            setattr(q, attr, fields[attr])

    if "localized" in fields:
        # Explicit. ``{}`` / None means "this question has no translations",
        # which is how an admin deletes one.
        blocks = _validate_localized(fields["localized"], q.options)
        q.localized = blocks or None
    elif baseline_changed and q.localized:
        # The Chinese text moved and nobody re-supplied the translations,
        # so the stored en/ja blocks now describe a DIFFERENT question
        # while ``correct_key`` describes this one. Keeping them would
        # grade an English reader against option texts that no longer
        # match the answer key — strictly worse than showing them the
        # Chinese baseline with the fallback notice the read path already
        # renders. Dropped rather than silently retained; the admin
        # console's language tabs are how they come back.
        logger.info(
            "secquiz.localized.invalidated question=%s langs=%s "
            "(baseline prompt/options/correct_key edited without new translations)",
            q.id, sorted(q.localized or {}),
        )
        q.localized = None

    await db.flush()
    await db.commit()
    await db.refresh(q)
    return serialize_question(q)


async def set_status(
    db: AsyncSession, question_id: uuid.UUID, status: str,
) -> dict[str, Any]:
    return await update_question(db, question_id, status=status)


# ── AI generation ───────────────────────────────────────────────────


async def generate_ai_questions(
    db: AsyncSession, *, actor: IdentityPerson, count: int = 5,
) -> dict[str, Any]:
    """Generate fresh security-awareness questions via the org's LLM connector.

    Graceful no-op when no LLM is configured (or it isn't the bailian/Qwen
    backend the JSON helper targets). Inserts source='ai' questions in ACTIVE
    status so they immediately enter rotation; admins can edit/archive."""
    from app.domains.support.copilot import _resolve_llm

    count = max(1, min(count, 10))
    cfg = await _resolve_llm(db)
    if cfg is None:
        return {"generated": 0, "reason": "llm_not_configured"}
    if cfg.provider != "bailian":
        return {"generated": 0, "reason": "llm_not_bailian"}

    system = (
        "你是企业 IT 安全意识培训出题专家。请生成简体中文的安全意识单选题,"
        "覆盖钓鱼、密码、设备安全、网络、恶意软件、账号(MFA)、数据保护、物理安全等主题。"
        "每题 4 个选项,只有一个正确答案,并给出简短解析(为什么)。"
    )
    user = (
        f"请生成 {count} 道题。严格输出 JSON 对象,形如:"
        '{"questions":[{"prompt":"...","options":[{"key":"A","text":"..."},'
        '{"key":"B","text":"..."},{"key":"C","text":"..."},{"key":"D","text":"..."}],'
        '"correct_key":"A","explanation":"...","category":"phishing"}]}'
    )
    try:
        from app.domains.bailian.llm_client import BailianLLMClient

        client = BailianLLMClient(
            api_key=cfg.key, model=cfg.model, region=cfg.region, timeout=30.0, purpose="security.quiz")
        payload = await client.chat_json(system=system, user=user, max_tokens=3000, temperature=0.7)
    except Exception as exc:  # noqa: BLE001 — LLM failures degrade to no-op
        logger.warning("secquiz AI generation failed: %s", exc)
        return {"generated": 0, "reason": "llm_error"}

    items = payload.get("questions") if isinstance(payload, dict) else None
    if not isinstance(items, list):
        return {"generated": 0, "reason": "bad_response"}

    generated = 0
    for item in items:
        if not isinstance(item, dict):
            continue
        prompt = (item.get("prompt") or "").strip()
        options = item.get("options")
        correct_key = item.get("correct_key")
        if not prompt or not isinstance(options, list) or len(options) < 2:
            continue
        keys = [o.get("key") for o in options if isinstance(o, dict)]
        if correct_key not in keys or len(keys) != len(set(keys)):
            continue
        db.add(SecQuizQuestion(
            id=uuid.uuid4(), prompt=prompt, options=options, correct_key=correct_key,
            explanation=(item.get("explanation") or None),
            category=(item.get("category") or "general")[:32],
            difficulty="normal", status=QUESTION_ACTIVE, source=SOURCE_AI,
            created_by=actor.id,
        ))
        generated += 1

    if generated:
        await db.commit()
    return {"generated": generated, "reason": "ok" if generated else "no_valid_items"}


__all__ = [
    "ensure_today_assignment", "get_today", "submit_answer", "compute_stats",
    "get_leaderboard", "get_coverage", "language_coverage", "list_questions",
    "create_question", "update_question", "set_status", "generate_ai_questions",
    "serialize_question",
]