Edu/services/data-ana/src/data_ana/clickhouse_client.py

"""ClickHouse 客户端（支持降级模式）.

当 settings.clickhouse_host 为空字符串时，get_client() 返回 None，
查询方法在 client 为 None 或查询失败时返回 None（降级模式），
保证服务在 ClickHouse 不可用时仍可启动并响应骨架数据。
"""

import asyncio
from datetime import datetime
from typing import Any

import structlog

from .config import settings

logger = structlog.get_logger(__name__)

_client: Any | None = None
# 标记是否已尝试初始化（避免对失败连接反复重试）
_client_initialized: bool = False


def get_client() -> Any | None:
    """获取 ClickHouse 客户端.

    - 当 clickhouse_host 为空：返回 None（降级模式）
    - 当已初始化但失败：返回 None
    - 当 clickhouse_connect 未安装：返回 None
    """
    global _client, _client_initialized

    if not settings.clickhouse_host:
        # 未配置 ClickHouse，降级模式
        return None

    if _client_initialized:
        return _client

    _client_initialized = True
    try:
        import clickhouse_connect

        kwargs: dict[str, Any] = {
            "host": settings.clickhouse_host,
            "port": settings.clickhouse_port,
            "database": settings.clickhouse_database,
        }
        if settings.clickhouse_user:
            kwargs["username"] = settings.clickhouse_user
        if settings.clickhouse_password:
            kwargs["password"] = settings.clickhouse_password
        _client = clickhouse_connect.get_client(**kwargs)
        logger.info(
            "clickhouse_client_initialized",
            host=settings.clickhouse_host,
            port=settings.clickhouse_port,
            database=settings.clickhouse_database,
        )
    except Exception as exc:  # noqa: BLE001
        # 任何初始化异常都进入降级模式，不抛出
        logger.warning("clickhouse_client_init_failed_degraded", error=str(exc))
        _client = None

    return _client


async def close_client() -> None:
    """关闭客户端."""
    global _client, _client_initialized
    if _client is not None:
        try:
            await asyncio.to_thread(_client.close)
        except Exception as exc:  # noqa: BLE001
            logger.warning("clickhouse_client_close_failed", error=str(exc))
        finally:
            _client = None
            _client_initialized = False


async def query_dashboard(student_id: str) -> dict | None:
    """查询学生学情看板（宽表 student_dashboard_view）.

    返回 None 表示降级模式（ClickHouse 不可用或查询失败）。
    """
    client = get_client()
    if client is None:
        return None

    try:
        result = await asyncio.to_thread(
            client.query,
            "SELECT student_id, class_id, exam_id, subject_id, score, "
            "rank_in_class, knowledge_point_id, mastery_level, error_count, "
            "last_updated "
            "FROM student_dashboard_view "
            "WHERE student_id = {sid:String} "
            "ORDER BY last_updated DESC "
            "LIMIT 50",
            parameters={"sid": student_id},
        )
        rows = result.result_rows
    except Exception as exc:  # noqa: BLE001
        logger.warning("query_dashboard_failed_degraded", error=str(exc), student_id=student_id)
        return None

    columns = [
        "student_id",
        "class_id",
        "exam_id",
        "subject_id",
        "score",
        "rank_in_class",
        "knowledge_point_id",
        "mastery_level",
        "error_count",
        "last_updated",
    ]
    records = [dict(zip(columns, row, strict=True)) for row in rows]
    return {
        "studentId": student_id,
        "records": records,
        "total": len(records),
    }


async def query_class_performance(class_id: str) -> dict | None:
    """查询班级成绩分析（聚合 student_dashboard_view）.

    返回 None 表示降级模式。
    """
    client = get_client()
    if client is None:
        return None

    try:
        # 平均分、参考人数、及格率（>=60）
        result = await asyncio.to_thread(
            client.query,
            "SELECT "
            "  count() AS total_students, "
            "  avg(score) AS average_score, "
            "  countIf(score >= 60) / count() AS pass_rate "
            "FROM student_dashboard_view "
            "WHERE class_id = {cid:String}",
            parameters={"cid": class_id},
        )
        agg_rows = result.result_rows
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "query_class_performance_failed_degraded",
            error=str(exc),
            class_id=class_id,
        )
        return None

    if not agg_rows:
        return {
            "classId": class_id,
            "averageScore": 0.0,
            "passRate": 0.0,
            "totalStudents": 0,
        }

    total_students, average_score, pass_rate = agg_rows[0]
    return {
        "classId": class_id,
        "averageScore": float(average_score) if average_score is not None else 0.0,
        "passRate": float(pass_rate) if pass_rate is not None else 0.0,
        "totalStudents": int(total_students) if total_students is not None else 0,
    }


async def query_student_errors(student_id: str) -> list[dict] | None:
    """查询学生错题本（表 student_errors）.

    返回 None 表示降级模式；返回空列表表示无错题数据。
    """
    client = get_client()
    if client is None:
        return None

    try:
        result = await asyncio.to_thread(
            client.query,
            "SELECT student_id, question_id, knowledge_point_id, error_count, "
            "last_error_time, content "
            "FROM student_errors "
            "WHERE student_id = {sid:String} "
            "ORDER BY last_error_time DESC "
            "LIMIT 100",
            parameters={"sid": student_id},
        )
        rows = result.result_rows
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "query_student_errors_failed_degraded",
            error=str(exc),
            student_id=student_id,
        )
        return None

    columns = [
        "student_id",
        "question_id",
        "knowledge_point_id",
        "error_count",
        "last_error_time",
        "content",
    ]
    return [dict(zip(columns, row, strict=True)) for row in rows]


async def ping() -> bool:
    """ClickHouse 连通性检查（供 /readyz 使用）.

    返回 True 表示可用；False 表示未配置或不可用。
    """
    client = get_client()
    if client is None:
        return False
    try:
        await asyncio.to_thread(client.query, "SELECT 1")
        return True
    except Exception as exc:  # noqa: BLE001
        logger.warning("clickhouse_ping_failed", error=str(exc))
        return False


async def upsert_student_dashboard(
    student_id: str,
    class_id: str,
    exam_id: str,
    subject_id: str,
    score: float,
    rank_in_class: int,
    knowledge_point_id: str,
    mastery_level: float,
    error_count: int,
    last_updated: datetime,
) -> bool:
    """写入/更新学生学情宽表（CDC 消费专用）.

    使用 ReplacingMergeTree 语义：按 ORDER BY 字段去重，保留 last_updated 最大版本。
    返回 True 表示成功；False 表示降级模式或写入失败。
    """
    client = get_client()
    if client is None:
        return False

    try:
        await asyncio.to_thread(
            client.insert,
            "student_dashboard_view",
            [
                [
                    student_id,
                    class_id,
                    exam_id,
                    subject_id,
                    score,
                    rank_in_class,
                    knowledge_point_id,
                    mastery_level,
                    error_count,
                    last_updated,
                ]
            ],
            column_names=[
                "student_id",
                "class_id",
                "exam_id",
                "subject_id",
                "score",
                "rank_in_class",
                "knowledge_point_id",
                "mastery_level",
                "error_count",
                "last_updated",
            ],
        )
        logger.info(
            "student_dashboard_upserted",
            student_id=student_id,
            class_id=class_id,
            exam_id=exam_id,
            score=score,
        )
        return True
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "student_dashboard_upsert_failed_degraded",
            error=str(exc),
            student_id=student_id,
            exam_id=exam_id,
        )
        return False


async def upsert_student_error(
    student_id: str,
    question_id: str,
    knowledge_point_id: str,
    error_count: int,
    last_error_time: datetime,
    content: str,
) -> bool:
    """写入/更新学生错题本（CDC 消费专用）.

    返回 True 表示成功；False 表示降级模式或写入失败。
    """
    client = get_client()
    if client is None:
        return False

    try:
        await asyncio.to_thread(
            client.insert,
            "student_errors",
            [
                [
                    student_id,
                    question_id,
                    knowledge_point_id,
                    error_count,
                    last_error_time,
                    content,
                ]
            ],
            column_names=[
                "student_id",
                "question_id",
                "knowledge_point_id",
                "error_count",
                "last_error_time",
                "content",
            ],
        )
        logger.info(
            "student_error_upserted",
            student_id=student_id,
            question_id=question_id,
            error_count=error_count,
        )
        return True
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "student_error_upsert_failed_degraded",
            error=str(exc),
            student_id=student_id,
            question_id=question_id,
        )
        return False