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

"""数据分析服务入口.

支持 ClickHouse 降级模式：当 CLICKHOUSE_HOST 未配置或不可达时，
查询端点返回骨架数据，服务仍可启动与响应。

支持 CDC 消费者：当 KAFKA_BROKERS 配置时，
后台启动 aiokafka 消费者，监听 Debezium CDC 事件写入 ClickHouse。
"""

import asyncio
import contextlib
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from datetime import UTC, datetime

import structlog
from fastapi import APIRouter, FastAPI
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from prometheus_client import make_asgi_app

from .cdc_consumer import run_consumer as run_cdc_consumer
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__)

# CDC 消费者后台任务句柄
_cdc_task: asyncio.Task | None = None

# 日志级别映射
_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_bound_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.

    endpoint 从 settings.otel_endpoint 读取（不硬编码）。
    """
    provider = TracerProvider()
    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) -> AsyncGenerator[None, None]:
    """应用生命周期.

    1. 初始化 logger（structlog）
    2. 初始化 OTel tracer（endpoint 从 config 读）
    3. 触发 ClickHouse 客户端惰性初始化（不阻塞启动，失败进入降级模式）
    4. 若配置了 kafka_brokers，后台启动 CDC 消费者任务
    5. 关闭时停止 CDC 任务并释放 ClickHouse 客户端
    """
    global _cdc_task
    logger = init_logger()
    init_tracer()
    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,
        kafka_cdc_topics=settings.kafka_cdc_topics,
    )
    # 启动 CDC 消费者后台任务（若未配置 kafka_brokers，run_consumer 内部直接返回）
    _cdc_task = asyncio.create_task(run_cdc_consumer())
    yield
    logger.info("data_ana_service_stopping")
    # 取消 CDC 任务
    if _cdc_task is not None and not _cdc_task.done():
        _cdc_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await _cdc_task
    await close_client()


app = FastAPI(
    title="Data Analytics Service",
    version="0.1.0",
    lifespan=lifespan,
)

# OpenTelemetry FastAPI 自动埋点（HTTP 请求/响应 span）
FastAPIInstrumentor.instrument_app(app)

# Prometheus 指标
app.mount("/metrics", make_asgi_app())

# 业务路由
router = APIRouter()


@app.get("/healthz")
async def healthz() -> dict:
    """健康检查（liveness）.

    只要进程存活即返回 ok，不依赖 ClickHouse。
    """
    return {"status": "ok", "service": "data-ana"}


@app.get("/readyz")
async def readyz() -> dict:
    """就绪检查（readiness）.

    ClickHouse 为可选依赖：
    - 已配置且可达：ready=true
    - 未配置：ready=true，degraded=true（降级模式仍可服务）
    - 已配置但不可达：ready=false

    CDC 消费者状态附加在响应中：
    - cdc_consumer: running / disabled / failed
    """
    cdc_status = "disabled"
    if _cdc_task is not None:
        if _cdc_task.done():
            cdc_status = "failed"
        elif not settings.kafka_brokers:
            cdc_status = "disabled"
        else:
            cdc_status = "running"

    if not settings.clickhouse_host:
        return {
            "status": "ok",
            "service": "data-ana",
            "ready": True,
            "degraded": True,
            "clickhouse": "not_configured",
            "cdc_consumer": cdc_status,
            "kafka_brokers": settings.kafka_brokers or None,
            "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",
        "cdc_consumer": cdc_status,
        "kafka_brokers": settings.kafka_brokers or None,
        "timestamp": datetime.now(UTC).isoformat(),
    }


@router.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}}


@router.get("/analytics/student/{student_id}/weakness")
async def student_weakness(student_id: str) -> dict:
    """学生薄弱知识点分析.

    优先查 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": weak_points,
                "records": result["records"],
                "total": result["total"],
                "degraded": False,
            },
        }


@router.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,
            },
        }


app.include_router(router)