feat(data-ana): 完善学情诊断服务并添加ClickHouse降级模式

config.py ClickHouse连接改可选+加DEV_MODE/kafka_brokers

clickhouse_client.py 降级模式: host为空时返回None

main.py 端点先查ClickHouse降级返回骨架数据+新增errorbook

新增clickhouse-init.sql创建宽表和错题表

Gateway添加/analytics路由

scripts/clickhouse-init.sql

@@ -0,0 +1,37 @@
+-- ClickHouse 数据库初始化脚本
+-- 适用服务：data-ana（数据分析）
+-- 表结构：student_dashboard_view（学生学情宽表）/ student_errors（错题本）
+-- 与 services/data-ana/src/data_ana/clickhouse_client.py 中的查询字段对齐
+--
+-- 使用方式（启用 ClickHouse 时执行一次）：
+--   clickhouse-client --multiquery < scripts/clickhouse-init.sql
+-- 注意：ClickHouse 为可选依赖，未配置时 data-ana 服务进入降级模式。
+
+-- 数据库
+CREATE DATABASE IF NOT EXISTS edu_analytics;
+
+-- 学生学情宽表（考试/班级/知识点维度）
+CREATE TABLE IF NOT EXISTS edu_analytics.student_dashboard_view (
+  student_id String,
+  class_id String,
+  exam_id String,
+  subject_id String,
+  score Float64,
+  rank_in_class UInt32,
+  knowledge_point_id String,
+  mastery_level Float32,
+  error_count UInt32,
+  last_updated DateTime
+) ENGINE = MergeTree()
+ORDER BY (student_id, class_id, exam_id);
+
+-- 学生错题表（错题本）
+CREATE TABLE IF NOT EXISTS edu_analytics.student_errors (
+  student_id String,
+  question_id String,
+  knowledge_point_id String,
+  error_count UInt32,
+  last_error_time DateTime,
+  content String
+) ENGINE = MergeTree()
+ORDER BY (student_id, knowledge_point_id);

services/api-gateway/internal/config/config.go

@@ -15,6 +15,7 @@ type Config struct {
 	TeacherBffURL     string
 	CoreEduServiceURL string
 	ContentServiceURL string
+	DataAnaServiceURL string
 	OTLPEndpoint      string
 	LogLevel          string
 	DevMode           bool
@@ -31,6 +32,7 @@ func Load() *Config {
 		TeacherBffURL:     getEnv("TEACHER_BFF_URL", "http://localhost:3003"),
 		CoreEduServiceURL: getEnv("CORE_EDU_SERVICE_URL", "http://localhost:3004"),
 		ContentServiceURL: getEnv("CONTENT_SERVICE_URL", "http://localhost:3005"),
+		DataAnaServiceURL: getEnv("DATA_ANA_SERVICE_URL", "http://localhost:3006"),
 		OTLPEndpoint:      getEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318"),
 		LogLevel:          getEnv("LOG_LEVEL", "info"),
 		DevMode:           getEnvBool("DEV_MODE", false),

services/api-gateway/main.go

@@ -108,6 +108,15 @@ func main() {
 		api.Any("/knowledge-points/*path", contentHandler)
 		api.Any("/questions", contentHandler)
 		api.Any("/questions/*path", contentHandler)
+
+		// data-ana 服务路由（学情诊断/错题本）
+		dataAnaProxy, err := proxy.NewProxy(cfg.DataAnaServiceURL)
+		if err != nil {
+			log.Fatalf("failed to create data-ana proxy: %v", err)
+		}
+		dataAnaHandler := proxy.ProxyHandler(dataAnaProxy)
+		api.Any("/analytics", dataAnaHandler)
+		api.Any("/analytics/*path", dataAnaHandler)
 	}
 
 	srv := &http.Server{

services/data-ana/pyproject.toml

@@ -11,6 +11,7 @@ dependencies = [
     "pydantic-settings>=2.5.0",
     "opentelemetry-api>=1.27.0",
     "opentelemetry-sdk>=1.27.0",
+    "opentelemetry-exporter-otlp>=1.27.0",
     "opentelemetry-instrumentation-fastapi>=0.48b0",
     "prometheus-client>=0.20.0",
     "structlog>=24.4.0",

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

@@ -1,27 +1,218 @@
-"""ClickHouse 客户端."""
+"""ClickHouse 客户端（支持降级模式）.
 
-import clickhouse_connect
+当 settings.clickhouse_host 为空字符串时，get_client() 返回 None，
+查询方法在 client 为 None 或查询失败时返回 None（降级模式），
+保证服务在 ClickHouse 不可用时仍可启动并响应骨架数据。
+"""
+
+from typing import Any
+
+import structlog
 
 from .config import settings
 
-_client = None
+logger = structlog.get_logger(__name__)
+
+_client: Any | None = None
+# 标记是否已尝试初始化（避免对失败连接反复重试）
+_client_initialized: bool = False
 
 
-def get_client():
+def get_client() -> Any | None:
-    """获取 ClickHouse 客户端."""
+    """获取 ClickHouse 客户端.
-    global _client
+
-    if _client is None:
+    - 当 clickhouse_host 为空：返回 None（降级模式）
-        _client = clickhouse_connect.get_client(
+    - 当已初始化但失败：返回 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
+    global _client, _client_initialized
-    if _client:
+    if _client is not None:
-        _client.close()
+        try:
-        _client = None
+            _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:
+        rows = 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},
+        ).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）
+        agg_rows = 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},
+        ).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:
+        rows = 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},
+        ).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:
+        client.query("SELECT 1")
+        return True
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("clickhouse_ping_failed", error=str(exc))
+        return False

services/data-ana/src/data_ana/config.py

@@ -1,16 +1,29 @@
 """配置管理."""
+
 from pydantic_settings import BaseSettings
 
 
 class Settings(BaseSettings):
-    """应用配置."""
+    """应用配置.
+
+    ClickHouse 连接参数为可选：当 clickhouse_host 为空字符串时，
+    服务进入降级模式（查询方法返回 None / 空数据），保证服务可启动。
+    """
 
     port: int = 3006
-    clickhouse_host: str = "localhost"
+    # ClickHouse 连接（可选：留空则降级模式）
+    clickhouse_host: str = ""
     clickhouse_port: int = 8123
     clickhouse_database: str = "edu_analytics"
+    clickhouse_user: str = ""
+    clickhouse_password: str = ""
+    # 可观测性
     otel_endpoint: str = "http://localhost:4318"
     log_level: str = "info"
+    # 开发模式开关（"true"/"false"）
+    dev_mode: str = "false"
+    # Kafka brokers（CDC 消费预留，暂不实现）
+    kafka_brokers: str = "localhost:9092"
 
     model_config = {"env_file": ".env", "env_prefix": ""}
 

services/data-ana/src/data_ana/main.py

@@ -1,6 +1,11 @@
-"""数据分析服务入口."""
+"""数据分析服务入口.
+
+支持 ClickHouse 降级模式：当 CLICKHOUSE_HOST 未配置或不可达时，
+查询端点返回骨架数据，服务仍可启动与响应。
+"""
 
 from contextlib import asynccontextmanager
+from datetime import UTC, datetime
 
 import structlog
 from fastapi import FastAPI
@@ -10,25 +15,90 @@ from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
 from prometheus_client import make_asgi_app
 
-logger = structlog.get_logger()
+from .clickhouse_client import (
+    close_client,
+    query_class_performance,
+    query_dashboard,
+    query_student_errors,
+)
+from .clickhouse_client import ping as ch_ping
+from .config import settings
+
+_logger: structlog.stdlib.BoundLogger | None = None
 tracer = trace.get_tracer(__name__)
 
+# 日志级别映射
+_LOG_LEVELS: dict[str, int] = {
+    "DEBUG": 10,
+    "INFO": 20,
+    "WARNING": 30,
+    "ERROR": 40,
+    "CRITICAL": 50,
+}
+
+
+def init_logger() -> structlog.stdlib.BoundLogger:
+    """初始化 structlog logger.
+
+    根据配置的 log_level 设置日志级别。
+    """
+    global _logger
+    level = _LOG_LEVELS.get(settings.log_level.upper(), 20)
+    structlog.configure(
+        wrapper_class=structlog.make_filtering_logger(level),
+        processors=[
+            structlog.contextvars.merge_contextvars,
+            structlog.processors.add_log_level,
+            structlog.processors.TimeStamper(fmt="iso"),
+            structlog.dev.ConsoleRenderer(),
+        ],
+        cache_logger_on_first_use=True,
+    )
+    _logger = structlog.get_logger(__name__)
+    return _logger
+
+
+def get_logger() -> structlog.stdlib.BoundLogger:
+    """获取已初始化的 logger（未初始化时自动初始化）."""
+    global _logger
+    if _logger is None:
+        return init_logger()
+    return _logger
+
 
 def init_tracer() -> None:
-    """初始化 OpenTelemetry."""
+    """初始化 OpenTelemetry.
+
+    endpoint 从 settings.otel_endpoint 读取（不硬编码）。
+    """
     provider = TracerProvider()
-    exporter = OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")
+    endpoint = settings.otel_endpoint.rstrip("/")
+    exporter = OTLPSpanExporter(endpoint=f"{endpoint}/v1/traces")
     provider.add_span_processor(BatchSpanProcessor(exporter))
     trace.set_tracer_provider(provider)
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    """应用生命周期."""
+    """应用生命周期.
+
+    1. 初始化 logger（structlog）
+    2. 初始化 OTel tracer（endpoint 从 config 读）
+    3. 触发 ClickHouse 客户端惰性初始化（不阻塞启动，失败进入降级模式）
+    4. 关闭时释放 ClickHouse 客户端
+    """
+    logger = init_logger()
     init_tracer()
-    logger.info("data-ana service starting")
+    logger.info(
+        "data_ana_service_starting",
+        port=settings.port,
+        dev_mode=settings.dev_mode,
+        clickhouse_configured=bool(settings.clickhouse_host),
+        kafka_brokers=settings.kafka_brokers,
+    )
     yield
-    logger.info("data-ana service stopping")
+    logger.info("data_ana_service_stopping")
+    await close_client()
 
 
 app = FastAPI(
@@ -42,36 +112,142 @@ app.mount("/metrics", make_asgi_app())
 
 
 @app.get("/healthz")
-async def healthz():
+async def healthz() -> dict:
-    """健康检查."""
+    """健康检查（liveness）.
+
+    只要进程存活即返回 ok，不依赖 ClickHouse。
+    """
     return {"status": "ok", "service": "data-ana"}
 
 
-@app.get("/analytics/class/{class_id}/performance")
+@app.get("/readyz")
-async def class_performance(class_id: str):
+async def readyz() -> dict:
-    """班级成绩分析."""
+    """就绪检查（readiness）.
-    with tracer.start_as_current_span("class_performance"):
+
-        # P4 骨架：从 ClickHouse 查询分析数据
+    ClickHouse 为可选依赖：
+    - 已配置且可达：ready=true
+    - 未配置：ready=true，degraded=true（降级模式仍可服务）
+    - 已配置但不可达：ready=false
+    """
+    if not settings.clickhouse_host:
         return {
-            "success": True,
+            "status": "ok",
-            "data": {
+            "service": "data-ana",
-                "classId": class_id,
+            "ready": True,
-                "averageScore": 0,
+            "degraded": True,
-                "passRate": 0,
+            "clickhouse": "not_configured",
-                "message": "P4 skeleton - ClickHouse integration pending",
+            "timestamp": datetime.now(UTC).isoformat(),
-            },
         }
 
+    ch_ok = await ch_ping()
+    return {
+        "status": "ok" if ch_ok else "degraded",
+        "service": "data-ana",
+        "ready": ch_ok,
+        "degraded": not ch_ok,
+        "clickhouse": "ok" if ch_ok else "unreachable",
+        "timestamp": datetime.now(UTC).isoformat(),
+    }
+
+
+@app.get("/analytics/class/{class_id}/performance")
+async def class_performance(class_id: str) -> dict:
+    """班级成绩分析.
+
+    优先查 ClickHouse；降级时返回骨架数据。
+    """
+    logger = get_logger()
+    with tracer.start_as_current_span("class_performance") as span:
+        span.set_attribute("class_id", class_id)
+        result = await query_class_performance(class_id)
+        if result is None:
+            logger.info("class_performance_degraded", class_id=class_id)
+            return {
+                "success": True,
+                "data": {
+                    "classId": class_id,
+                    "averageScore": 0,
+                    "passRate": 0,
+                    "totalStudents": 0,
+                    "message": "ClickHouse unavailable - skeleton data",
+                    "degraded": True,
+                },
+            }
+        return {"success": True, "data": {**result, "degraded": False}}
+
 
 @app.get("/analytics/student/{student_id}/weakness")
-async def student_weakness(student_id: str):
+async def student_weakness(student_id: str) -> dict:
-    """学生薄弱知识点分析."""
+    """学生薄弱知识点分析.
-    with tracer.start_as_current_span("student_weakness"):
+
+    优先查 ClickHouse；降级时返回骨架数据。
+    """
+    logger = get_logger()
+    with tracer.start_as_current_span("student_weakness") as span:
+        span.set_attribute("student_id", student_id)
+        result = await query_dashboard(student_id)
+        if result is None:
+            logger.info("student_weakness_degraded", student_id=student_id)
+            return {
+                "success": True,
+                "data": {
+                    "studentId": student_id,
+                    "weakPoints": [],
+                    "message": "ClickHouse unavailable - skeleton data",
+                    "degraded": True,
+                },
+            }
+
+        # 从宽表提取薄弱知识点：mastery_level < 0.6 视为薄弱
+        weak_points = [
+            {
+                "knowledgePointId": r["knowledge_point_id"],
+                "masteryLevel": r["mastery_level"],
+                "errorCount": r["error_count"],
+            }
+            for r in result["records"]
+            if r.get("mastery_level") is not None and r["mastery_level"] < 0.6
+        ]
         return {
             "success": True,
             "data": {
                 "studentId": student_id,
-                "weakPoints": [],
+                "weakPoints": weak_points,
-                "message": "P4 skeleton - weakness analysis pending",
+                "records": result["records"],
+                "total": result["total"],
+                "degraded": False,
+            },
+        }
+
+
+@app.get("/analytics/student/{student_id}/errorbook")
+async def student_errorbook(student_id: str) -> dict:
+    """学生错题本.
+
+    优先查 ClickHouse；降级时返回空列表。
+    """
+    logger = get_logger()
+    with tracer.start_as_current_span("student_errorbook") as span:
+        span.set_attribute("student_id", student_id)
+        result = await query_student_errors(student_id)
+        if result is None:
+            logger.info("student_errorbook_degraded", student_id=student_id)
+            return {
+                "success": True,
+                "data": {
+                    "studentId": student_id,
+                    "errors": [],
+                    "total": 0,
+                    "message": "ClickHouse unavailable - empty errorbook",
+                    "degraded": True,
+                },
+            }
+        return {
+            "success": True,
+            "data": {
+                "studentId": student_id,
+                "errors": result,
+                "total": len(result),
+                "degraded": False,
             },
         }