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