chore(data-ana): 更新工作进度跟踪与ISSUE状态，补建基础设施脚本，清理遗留代码
- workline: 追加 §6 实现进度跟踪（P2-P5 已完成，P6 未开始） - issue: 更新 §0.2/§0.3 核查表，proto 文件已全部补全 - 新增 scripts/clickhouse_ddl.sql（5 宽表建表脚本） - 新增 scripts/seed_clickhouse.py（mock 种子数据脚本） - 删除遗留代码 clickhouse_client.py 和 health.py
This commit is contained in:
SpecialX
87
services/data-ana/scripts/clickhouse_ddl.sql
Normal file
87
services/data-ana/scripts/clickhouse_ddl.sql
Normal file
@@ -0,0 +1,87 @@
|
||||
-- data-ana ClickHouse DDL：5 宽表建表脚本
|
||||
-- 对齐 02-architecture-design.md §3 DDL 设计
|
||||
-- 引擎：ReplacingMergeTree（幂等消费保证）+ MergeTree（历史快照）
|
||||
-- 使用方式：clickhouse-client --multiquery < scripts/clickhouse_ddl.sql
|
||||
|
||||
CREATE DATABASE IF NOT EXISTS edu_analytics;
|
||||
|
||||
-- §3.1 学生学情宽表
|
||||
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 DateTime64(3, 'UTC')
|
||||
)
|
||||
ENGINE = ReplacingMergeTree(last_updated)
|
||||
PARTITION BY toYYYYMM(last_updated)
|
||||
ORDER BY (student_id, exam_id, knowledge_point_id)
|
||||
SETTINGS index_granularity = 8192;
|
||||
|
||||
-- §3.2 学生错题本
|
||||
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 DateTime64(3, 'UTC'),
|
||||
content String
|
||||
)
|
||||
ENGINE = ReplacingMergeTree(last_error_time)
|
||||
PARTITION BY toYYYYMM(last_error_time)
|
||||
ORDER BY (student_id, question_id);
|
||||
|
||||
-- §3.3 知识点掌握度历史快照
|
||||
CREATE TABLE IF NOT EXISTS edu_analytics.mastery_snapshot
|
||||
(
|
||||
student_id String,
|
||||
knowledge_point_id String,
|
||||
subject_id String,
|
||||
mastery_level Float32,
|
||||
calculated_at DateTime64(3, 'UTC'),
|
||||
calculation_method LowCardinality(String)
|
||||
)
|
||||
ENGINE = MergeTree
|
||||
PARTITION BY toYYYYMM(calculated_at)
|
||||
ORDER BY (student_id, knowledge_point_id, calculated_at);
|
||||
|
||||
-- §3.4 AI 用量计费记录
|
||||
CREATE TABLE IF NOT EXISTS edu_analytics.ai_usage_log
|
||||
(
|
||||
request_id String,
|
||||
user_id String,
|
||||
provider LowCardinality(String),
|
||||
model LowCardinality(String),
|
||||
prompt_tokens UInt32,
|
||||
completion_tokens UInt32,
|
||||
total_tokens UInt32,
|
||||
latency_ms UInt32,
|
||||
success Boolean,
|
||||
cost_cents UInt32,
|
||||
occurred_at DateTime64(3, 'UTC')
|
||||
)
|
||||
ENGINE = ReplacingMergeTree(occurred_at)
|
||||
PARTITION BY toYYYYMM(occurred_at)
|
||||
ORDER BY (request_id);
|
||||
|
||||
-- §3.5 学生考勤记录
|
||||
CREATE TABLE IF NOT EXISTS edu_analytics.attendance_logs
|
||||
(
|
||||
student_id String,
|
||||
class_id String,
|
||||
attendance_date Date,
|
||||
status LowCardinality(String),
|
||||
recorded_by String,
|
||||
remark String DEFAULT '',
|
||||
occurred_at DateTime64(3, 'UTC')
|
||||
)
|
||||
ENGINE = ReplacingMergeTree(occurred_at)
|
||||
PARTITION BY toYYYYMM(attendance_date)
|
||||
ORDER BY (student_id, class_id, attendance_date);
|
||||
241
services/data-ana/scripts/seed_clickhouse.py
Normal file
241
services/data-ana/scripts/seed_clickhouse.py
Normal file
@@ -0,0 +1,241 @@
|
||||
"""data-ana ClickHouse 种子数据脚本.
|
||||
|
||||
生成 mock 数据并写入 ClickHouse 5 张宽表：
|
||||
- student_dashboard_view: 30 学生 × 5 考试 × 3 学科 + 30 学生 × 10 作业
|
||||
- student_errors: 每生 2~5 条错题
|
||||
- mastery_snapshot: 30 学生 × 6 知识点 × 3 历史快照
|
||||
- ai_usage_log: 80 条 AI 用量记录
|
||||
- attendance_logs: 30 学生 × 30 天（跳过周末）
|
||||
|
||||
使用方式：python scripts/seed_clickhouse.py
|
||||
环境变量：DATA_ANA_CLICKHOUSE_HOST / DATA_ANA_CLICKHOUSE_PORT / DATA_ANA_CLICKHOUSE_DATABASE
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import os
|
||||
import random
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
import clickhouse_connect
|
||||
|
||||
DATABASE = os.environ.get("DATA_ANA_CLICKHOUSE_DATABASE", "edu_analytics")
|
||||
HOST = os.environ.get("DATA_ANA_CLICKHOUSE_HOST", "localhost")
|
||||
PORT = int(os.environ.get("DATA_ANA_CLICKHOUSE_PORT", "8123"))
|
||||
USERNAME = os.environ.get("DATA_ANA_CLICKHOUSE_USERNAME", "default")
|
||||
PASSWORD = os.environ.get("DATA_ANA_CLICKHOUSE_PASSWORD", "")
|
||||
|
||||
SUBJECTS = ["math", "chinese", "english"]
|
||||
CLASSES = ["class-001", "class-002"]
|
||||
KNOWLEDGE_POINTS = {
|
||||
"math": ["kp-math-001", "kp-math-002", "kp-math-003"],
|
||||
"chinese": ["kp-cn-001", "kp-cn-002"],
|
||||
"english": ["kp-en-001", "kp-en-002"],
|
||||
}
|
||||
PROVIDERS = ["openai", "anthropic", "baichuan", "local"]
|
||||
MODELS = ["gpt-4o", "claude-3-sonnet", "baichuan-2", "local-llama"]
|
||||
BATCH_SIZE = 500
|
||||
SEED = 42
|
||||
|
||||
|
||||
def generate_students() -> list[dict[str, str]]:
|
||||
students: list[dict[str, str]] = []
|
||||
for i in range(30):
|
||||
class_id = CLASSES[i % 2]
|
||||
students.append({
|
||||
"student_id": f"stu-{i + 1:03d}",
|
||||
"class_id": class_id,
|
||||
})
|
||||
return students
|
||||
|
||||
|
||||
def generate_dashboard_rows(
|
||||
students: list[dict[str, str]], rng: random.Random
|
||||
) -> list[list[Any]]:
|
||||
rows: list[list[Any]] = []
|
||||
base_time = datetime.now(UTC) - timedelta(days=30)
|
||||
for s in students:
|
||||
for exam_idx in range(5):
|
||||
exam_id = f"exam-{exam_idx + 1:03d}"
|
||||
for subject in SUBJECTS:
|
||||
score = round(rng.uniform(55, 98), 1)
|
||||
rank = rng.randint(1, 30)
|
||||
for kp in KNOWLEDGE_POINTS[subject]:
|
||||
mastery = round(rng.uniform(0.2, 0.95), 3)
|
||||
error_count = rng.randint(0, 5)
|
||||
ts = base_time + timedelta(
|
||||
days=exam_idx * 6, hours=rng.randint(0, 23)
|
||||
)
|
||||
rows.append([
|
||||
s["student_id"], s["class_id"], exam_id, subject,
|
||||
score, rank, kp, mastery, error_count, ts,
|
||||
])
|
||||
for hw_idx in range(10):
|
||||
hw_id = f"hw-{hw_idx + 1:03d}"
|
||||
for subject in SUBJECTS:
|
||||
score = round(rng.uniform(60, 100), 1)
|
||||
ts = base_time + timedelta(days=hw_idx * 3, hours=rng.randint(0, 23))
|
||||
rows.append([
|
||||
s["student_id"], s["class_id"], hw_id, subject,
|
||||
score, rng.randint(1, 30), "kp-hw", 0.0, 0, ts,
|
||||
])
|
||||
return rows
|
||||
|
||||
|
||||
def generate_error_rows(
|
||||
students: list[dict[str, str]], rng: random.Random
|
||||
) -> list[list[Any]]:
|
||||
rows: list[list[Any]] = []
|
||||
for s in students:
|
||||
count = rng.randint(2, 5)
|
||||
for _ in range(count):
|
||||
subject = rng.choice(SUBJECTS)
|
||||
kp = rng.choice(KNOWLEDGE_POINTS[subject])
|
||||
qid = f"q-{rng.randint(1, 200):03d}"
|
||||
ts = datetime.now(UTC) - timedelta(days=rng.randint(1, 30))
|
||||
rows.append([
|
||||
s["student_id"], qid, kp, rng.randint(1, 4), ts, f"{subject} error",
|
||||
])
|
||||
return rows
|
||||
|
||||
|
||||
def generate_mastery_rows(
|
||||
students: list[dict[str, str]], rng: random.Random
|
||||
) -> list[list[Any]]:
|
||||
rows: list[list[Any]] = []
|
||||
base_time = datetime.now(UTC) - timedelta(days=30)
|
||||
for s in students:
|
||||
for subject in SUBJECTS:
|
||||
for kp in KNOWLEDGE_POINTS[subject]:
|
||||
for snap in range(3):
|
||||
mastery = round(rng.uniform(0.3, 0.9), 3)
|
||||
ts = base_time + timedelta(days=snap * 10)
|
||||
rows.append([
|
||||
s["student_id"], kp, subject, mastery, ts,
|
||||
"weighted_moving_avg",
|
||||
])
|
||||
return rows
|
||||
|
||||
|
||||
def generate_attendance_rows(
|
||||
students: list[dict[str, str]], rng: random.Random
|
||||
) -> list[list[Any]]:
|
||||
rows: list[list[Any]] = []
|
||||
start_date = datetime.now(UTC).date() - timedelta(days=30)
|
||||
for s in students:
|
||||
for day_offset in range(30):
|
||||
d = start_date + timedelta(days=day_offset)
|
||||
if d.weekday() >= 5:
|
||||
continue
|
||||
status = rng.choices(
|
||||
["present", "absent", "late", "leave"],
|
||||
weights=[85, 5, 7, 3],
|
||||
)[0]
|
||||
ts = datetime.combine(d, datetime.min.time()).replace(tzinfo=UTC)
|
||||
rows.append([
|
||||
s["student_id"], s["class_id"], d, status,
|
||||
"teacher-001", "", ts,
|
||||
])
|
||||
return rows
|
||||
|
||||
|
||||
def generate_ai_usage_rows(rng: random.Random) -> list[list[Any]]:
|
||||
rows: list[list[Any]] = []
|
||||
for i in range(80):
|
||||
ts = datetime.now(UTC) - timedelta(
|
||||
days=rng.randint(0, 30), hours=rng.randint(0, 23)
|
||||
)
|
||||
prompt_t = rng.randint(50, 2000)
|
||||
completion_t = rng.randint(20, 1500)
|
||||
rows.append([
|
||||
f"req-{i + 1:04d}", f"stu-{rng.randint(1, 30):03d}",
|
||||
rng.choice(PROVIDERS), rng.choice(MODELS),
|
||||
prompt_t, completion_t, prompt_t + completion_t,
|
||||
rng.randint(100, 5000), rng.random() > 0.05,
|
||||
rng.randint(1, 50), ts,
|
||||
])
|
||||
return rows
|
||||
|
||||
|
||||
def batch_insert(
|
||||
client: Any, table: str, columns: list[str], rows: list[list[Any]]
|
||||
) -> None:
|
||||
if not rows:
|
||||
return
|
||||
for i in range(0, len(rows), BATCH_SIZE):
|
||||
chunk = rows[i : i + BATCH_SIZE]
|
||||
client.insert(f"{DATABASE}.{table}", chunk, column_names=columns)
|
||||
print(f" {table}: {len(rows)} rows inserted")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
rng = random.Random(SEED)
|
||||
print("=" * 70)
|
||||
print("data-ana ClickHouse 种子数据脚本")
|
||||
print(f" Host: {HOST}:{PORT} Database: {DATABASE}")
|
||||
print("=" * 70)
|
||||
|
||||
client = clickhouse_connect.get_client(
|
||||
host=HOST, port=PORT, username=USERNAME, password=PASSWORD,
|
||||
database=DATABASE,
|
||||
)
|
||||
|
||||
students = generate_students()
|
||||
print(f"\n生成 {len(students)} 名学生（2 个班级）")
|
||||
|
||||
print("\n[1/5] student_dashboard_view...")
|
||||
dashboard_rows = generate_dashboard_rows(students, rng)
|
||||
batch_insert(client, "student_dashboard_view", [
|
||||
"student_id", "class_id", "exam_id", "subject_id", "score",
|
||||
"rank_in_class", "knowledge_point_id", "mastery_level",
|
||||
"error_count", "last_updated",
|
||||
], dashboard_rows)
|
||||
|
||||
print("\n[2/5] student_errors...")
|
||||
error_rows = generate_error_rows(students, rng)
|
||||
batch_insert(client, "student_errors", [
|
||||
"student_id", "question_id", "knowledge_point_id", "error_count",
|
||||
"last_error_time", "content",
|
||||
], error_rows)
|
||||
|
||||
print("\n[3/5] mastery_snapshot...")
|
||||
mastery_rows = generate_mastery_rows(students, rng)
|
||||
batch_insert(client, "mastery_snapshot", [
|
||||
"student_id", "knowledge_point_id", "subject_id", "mastery_level",
|
||||
"calculated_at", "calculation_method",
|
||||
], mastery_rows)
|
||||
|
||||
print("\n[4/5] attendance_logs...")
|
||||
attendance_rows = generate_attendance_rows(students, rng)
|
||||
batch_insert(client, "attendance_logs", [
|
||||
"student_id", "class_id", "attendance_date", "status",
|
||||
"recorded_by", "remark", "occurred_at",
|
||||
], attendance_rows)
|
||||
|
||||
print("\n[5/5] ai_usage_log...")
|
||||
ai_rows = generate_ai_usage_rows(rng)
|
||||
batch_insert(client, "ai_usage_log", [
|
||||
"request_id", "user_id", "provider", "model", "prompt_tokens",
|
||||
"completion_tokens", "total_tokens", "latency_ms", "success",
|
||||
"cost_cents", "occurred_at",
|
||||
], ai_rows)
|
||||
|
||||
print("\n" + "=" * 70)
|
||||
print("种子数据写入完成")
|
||||
print(f" student_dashboard_view: {len(dashboard_rows)} rows")
|
||||
print(f" student_errors: {len(error_rows)} rows")
|
||||
print(f" mastery_snapshot: {len(mastery_rows)} rows")
|
||||
print(f" attendance_logs: {len(attendance_rows)} rows")
|
||||
print(f" ai_usage_log: {len(ai_rows)} rows")
|
||||
print(f" Total: {len(dashboard_rows) + len(error_rows) + len(mastery_rows) + len(attendance_rows) + len(ai_rows)} rows")
|
||||
print("=" * 70)
|
||||
|
||||
with contextlib.suppress(Exception):
|
||||
client.close()
|
||||
print("完成。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,352 +0,0 @@
|
||||
"""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
|
||||
@@ -1,35 +0,0 @@
|
||||
"""健康检查端点（data-ana 服务）。
|
||||
|
||||
- GET /healthz：liveness，仅返回进程存活。
|
||||
- GET /readyz：readiness，简化版返回 ok + TODO，待补全 ClickHouse 连通性校验。
|
||||
|
||||
集成说明：在 FastAPI app 中挂载 router：
|
||||
|
||||
from health import router as health_router
|
||||
app.include_router(health_router)
|
||||
"""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
SERVICE_NAME = "data-ana"
|
||||
|
||||
|
||||
@router.get("/healthz")
|
||||
async def healthz() -> dict:
|
||||
return {"status": "ok", "service": SERVICE_NAME}
|
||||
|
||||
|
||||
@router.get("/readyz")
|
||||
async def readyz() -> dict:
|
||||
# TODO: 校验关键依赖
|
||||
# 1. ClickHouse 连通性：clickhouse_client.execute("SELECT 1")
|
||||
# 依赖客户端就绪后再补全检查逻辑，失败时返回 503。
|
||||
return {
|
||||
"status": "ok",
|
||||
"service": SERVICE_NAME,
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
}
|
||||
Reference in New Issue
Block a user