"""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