"""知识题库 service — bank overview, admin CRUD, and AI generation. The seed ships a curated base; this is how the bank grows to "a lot": IT admins fire AI generation per category (reusing the org LLM connector, graceful no-op without a key) and can edit/archive results. All admin endpoints gate on global_admin. """ from __future__ import annotations import logging import uuid from typing import Any from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException, NotFoundError from app.domains.arcade.trivia_models import ( QUESTION_ACTIVE, QUESTION_ARCHIVED, SOURCE_AI, SOURCE_MANUAL, TRIVIA_CATEGORIES, ArcadeTriviaQuestion, ) from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) # The only statuses the game / list filters recognise. An unrecognised # status (an admin typo) would silently pull a question out of rotation # with no error, so admin mutations are whitelisted here. _VALID_STATUSES = frozenset({QUESTION_ACTIVE, QUESTION_ARCHIVED}) # Human-readable category names for the AI prompt (the model writes better # questions when told the domain in natural language). _CATEGORY_ZH = { "common_sense": "生活常识", "history": "历史", "science": "科学", "geography": "地理", "chinese_culture": "中华传统文化", "life": "生活百科与健康", } def _normalize_keys(value: str | None) -> str: return "".join(sorted({c for c in (value or "").upper() if c.isalpha()})) def serialize(q: ArcadeTriviaQuestion) -> dict[str, Any]: return { "id": str(q.id), "prompt": q.prompt, "options": q.options, "correct_key": q.correct_key, "explanation": q.explanation, "category": q.category, "difficulty": q.difficulty, "status": q.status, "source": q.source, "created_at": q.created_at.isoformat() if q.created_at else None, } async def bank_overview(db: AsyncSession) -> dict[str, Any]: """Active question counts per category — powers the admin dashboard and the frontend category picker (a category with < MIN_BANK is disabled).""" rows = (await db.execute( select(ArcadeTriviaQuestion.category, func.count()) .where(ArcadeTriviaQuestion.status == QUESTION_ACTIVE) .group_by(ArcadeTriviaQuestion.category) )).all() counts = {c: int(n or 0) for c, n in rows} total = int((await db.scalar( select(func.count()).select_from(ArcadeTriviaQuestion) .where(ArcadeTriviaQuestion.status == QUESTION_ACTIVE) )) or 0) return { "total_active": total, "by_category": [ {"category": c, "active": counts.get(c, 0)} for c in TRIVIA_CATEGORIES ], } async def list_questions( db: AsyncSession, *, category: str | None = None, status: str | None = None, ) -> list[dict[str, Any]]: stmt = select(ArcadeTriviaQuestion).order_by(ArcadeTriviaQuestion.created_at.desc()) if category: stmt = stmt.where(ArcadeTriviaQuestion.category == category) if status: stmt = stmt.where(ArcadeTriviaQuestion.status == status) rows = (await db.execute(stmt.limit(500))).scalars().all() return [serialize(q) for q in rows] def _validate(prompt: str, options: list, correct_key: str, category: str) -> str: keys = [o.get("key") for o in (options or [])] if not prompt or not prompt.strip(): raise AppException(code="ARCADE_TRIVIA_BAD", message="Prompt is required.", status_code=422) if category not in TRIVIA_CATEGORIES: raise AppException(code="ARCADE_TRIVIA_BAD", message="Unknown category.", status_code=422) if len(options or []) < 2: raise AppException(code="ARCADE_TRIVIA_BAD", message="At least 2 options required.", status_code=422) if len(keys) != len(set(keys)): raise AppException(code="ARCADE_TRIVIA_BAD", message="Option keys must be unique.", status_code=422) norm = _normalize_keys(correct_key) if len(norm) != 1 or norm not in keys: raise AppException( code="ARCADE_TRIVIA_BAD", message="correct_key must be exactly one of the option keys.", status_code=422, ) return norm async def create_question( db: AsyncSession, *, actor: IdentityPerson, prompt: str, options: list, correct_key: str, explanation: str | None, category: str, difficulty: str = "normal", source: str = SOURCE_MANUAL, ) -> dict[str, Any]: correct_key = _validate(prompt, options, correct_key, category) q = ArcadeTriviaQuestion( id=uuid.uuid4(), prompt=prompt.strip(), options=options, correct_key=correct_key, explanation=explanation, category=category, difficulty=difficulty or "normal", status=QUESTION_ACTIVE, source=source, created_by=actor.id, ) db.add(q) await db.flush() await db.commit() await db.refresh(q) return serialize(q) async def set_status(db: AsyncSession, question_id: uuid.UUID, status: str) -> dict[str, Any]: status = (status or "").strip() if status not in _VALID_STATUSES: raise AppException( code="ARCADE_TRIVIA_BAD", message="status must be one of: active, archived.", status_code=422, ) q = await db.get(ArcadeTriviaQuestion, question_id) if q is None: raise NotFoundError("arcade_trivia_question", str(question_id)) q.status = status await db.flush() await db.commit() await db.refresh(q) return serialize(q) async def generate_ai_trivia( db: AsyncSession, *, actor: IdentityPerson, category: str, count: int = 10, ) -> dict[str, Any]: """Generate fresh trivia via the org LLM connector. Graceful no-op when no LLM is configured. Inserts source='ai', ACTIVE (immediately playable; admins can edit/archive). Mirrors secquiz's generator.""" if category not in TRIVIA_CATEGORIES: raise AppException(code="ARCADE_TRIVIA_BAD", message="Unknown category.", status_code=422) count = max(1, min(count, 20)) from app.domains.support.copilot import _resolve_llm 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"} cat_zh = _CATEGORY_ZH.get(category, category) system = ( "你是一位面向企业员工的趣味知识竞赛出题专家。请生成简体中文的" f"「{cat_zh}」类单选题,用于轻松的知识小游戏。要求:题目有趣、" "老少皆宜、答案客观无争议、避免冷僻偏门与政治敏感内容;每题恰好" "4 个选项、只有一个正确答案,并给出一句简短解析。" ) user = ( f"请生成 {count} 道「{cat_zh}」单选题。严格输出 JSON 对象,形如:" '{"questions":[{"prompt":"...","options":[{"key":"A","text":"..."},' '{"key":"B","text":"..."},{"key":"C","text":"..."},{"key":"D","text":"..."}],' '"correct_key":"A","explanation":"..."}]}' ) 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="arcade.trivia") payload = await client.chat_json(system=system, user=user, max_tokens=3500, temperature=0.8) except Exception as exc: # noqa: BLE001 — LLM failure degrades to no-op logger.warning("arcade trivia 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") # Reuse the SAME validation as manual create so the AI path can't # slip in a question the grader can never mark correct — e.g. a # non-alpha correct_key that _normalize_keys collapses to "" (which # would score every player wrong forever and poison the 科举 board). if not isinstance(options, list): continue try: norm_key = _validate(prompt, options, correct_key or "", category) except AppException: continue db.add(ArcadeTriviaQuestion( id=uuid.uuid4(), prompt=prompt, options=options, correct_key=norm_key, explanation=(item.get("explanation") or None), category=category, 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__ = [ "bank_overview", "list_questions", "create_question", "set_status", "generate_ai_trivia", "serialize", ]