wecom_it_smart_desk/backend/app/tasks/quiz_generation_task.py

# =============================================================================
# 企微IT智能服务台 — 每日测验题目自动生成定时任务
# =============================================================================
# 说明：每天凌晨 3:00 自动执行：
#   1. 按星期轮转生成 1 个类别的 5 道知识题（7 天一轮）
#   2. 分析近 7 天已解决工单 → 生成 3 道诊断题
#   3. 停用被 >80% 活跃员工答过的陈旧题目
#
# 调度：CronTrigger(hour=3, minute=0)
# 降级：Dify 不可用时记录错误，不影响其他步骤
# =============================================================================

import datetime
import logging

logger = logging.getLogger(__name__)

# 7 个题目类别（按星期轮转：周一=network, 周二=vpn, ...）
QUIZ_CATEGORIES = ["network", "vpn", "email", "system", "printer", "security", "office"]

# 每日知识题生成数量
KNOWLEDGE_QUESTIONS_PER_DAY = 5

# 诊断题生成数量
DIAGNOSTIC_QUESTIONS_PER_BATCH = 3

# 默认诊断题问题类别（当近期无工单数据时使用）
DEFAULT_DIAGNOSTIC_CATEGORIES = ["network_connect", "vpn_auth_fail", "printer_offline"]


async def run_daily_quiz_generation():
    """每日测验题目生成任务。

    执行流程：
    1. 按星期轮转生成 1 个类别的 5 道知识题
    2. 分析近 7 天已解决工单 → 取 top 1 问题类别生成 3 道诊断题
    3. 停用被 >80% 活跃员工答过的陈旧题
    4. 记录汇总日志

    调度：每日 03:00（CronTrigger hour=3, minute=0）
    """
    from app.database import _get_session_factory
    from app.services.quiz_generation_service import get_quiz_generation_service

    logger.info("===== 开始每日测验题目生成 =====")

    factory = _get_session_factory()
    service = get_quiz_generation_service()

    total_generated = 0
    total_errors = 0

    async with factory() as db:
        try:
            # ============================================================
            # 1. 知识题生成（按星期轮转 1 个类别）
            # ============================================================
            today = datetime.date.today()
            weekday = today.weekday()  # 0=Monday, 6=Sunday
            category = QUIZ_CATEGORIES[weekday]

            logger.info(f"今日轮转类别: {category} (weekday={weekday})")

            try:
                result = await service.generate_knowledge_questions_batch(
                    db=db,
                    category=category,
                    count=KNOWLEDGE_QUESTIONS_PER_DAY,
                    is_active=False,  # 定时生成的题目需管理员审批
                )
                total_generated += result["success_count"]
                total_errors += result["failed_count"]
                logger.info(
                    f"知识题生成 [{category}]: "
                    f"成功 {result['success_count']}, 失败 {result['failed_count']}"
                )
                if result["errors"]:
                    logger.warning(f"知识题错误详情: {result['errors'][:3]}")
            except Exception as e:
                logger.error(f"知识题生成 [{category}] 异常: {e}")
                total_errors += KNOWLEDGE_QUESTIONS_PER_DAY

            # ============================================================
            # 2. 诊断题生成（基于近期工单模式）
            # ============================================================
            try:
                # 获取近期工单摘要
                ticket_summaries = await service._get_recent_ticket_summaries(
                    db, days=7, limit=20
                )

                # 提取 top 1 问题类别
                problem_category = _extract_top_problem_category(ticket_summaries)

                logger.info(f"诊断题问题类别: {problem_category}")

                result = await service.generate_diagnostic_questions_batch(
                    db=db,
                    problem_category=problem_category,
                    count=DIAGNOSTIC_QUESTIONS_PER_BATCH,
                    ticket_summaries=[t["summary"] for t in ticket_summaries if t["summary"]],
                    is_active=False,
                )
                total_generated += result["success_count"]
                total_errors += result["failed_count"]
                logger.info(
                    f"诊断题生成 [{problem_category}]: "
                    f"成功 {result['success_count']}, 失败 {result['failed_count']}"
                )
            except Exception as e:
                logger.error(f"诊断题生成异常: {e}")
                total_errors += DIAGNOSTIC_QUESTIONS_PER_BATCH

            # ============================================================
            # 3. 停用陈旧题目
            # ============================================================
            try:
                stale_result = await service.deactivate_stale_questions(
                    db=db, threshold=0.8
                )
                if stale_result["deactivated_count"] > 0:
                    logger.info(
                        f"陈旧题目停用: {stale_result['deactivated_count']} 道 "
                        f"(活跃员工 {stale_result['total_active_employees']} 人)"
                    )
            except Exception as e:
                logger.error(f"陈旧题目停用异常: {e}")

            # 统一提交
            await db.commit()

        except Exception as e:
            await db.rollback()
            logger.error(f"每日题目生成任务异常: {e}", exc_info=True)

    logger.info(
        f"===== 每日题目生成完成: 新增 {total_generated} 道, "
        f"失败 {total_errors} 道 ====="
    )


def _extract_top_problem_category(
    ticket_summaries: list,
) -> str:
    """从工单摘要中提取最高频的问题类别。

    基于 category_hint 字段统计频率。
    降级：如果无数据，返回默认类别。

    Args:
        ticket_summaries: _get_recent_ticket_summaries() 返回的列表

    Returns:
        str: 问题类别标识（如 "vpn_disconnect"）
    """
    if not ticket_summaries:
        # 无工单数据时返回默认
        return DEFAULT_DIAGNOSTIC_CATEGORIES[0]

    # 统计 category_hint 频率
    hint_counts: dict[str, int] = {}
    for ticket in ticket_summaries:
        hint = ticket.get("category_hint", "")
        if hint:
            hint_counts[hint] = hint_counts.get(hint, 0) + 1

    if not hint_counts:
        return DEFAULT_DIAGNOSTIC_CATEGORIES[0]

    # 取最高频的类别
    top_category = max(hint_counts, key=hint_counts.get)

    # 拼接为 problem_category 格式（如 "vpn_disconnect"）
    return f"{top_category}_issue"