Pine
866f82dbc4
新增 courses 表（app/training/db.py SEED_COURSES），经 training_admin_bridge 桥接 app.training 的课程/活动(events)/报名(bookings)/测评(tests) 数据为统一运营视图； schemas 增 Course/Activity/Booking/Test 请求模型；rbac_operator.py 增四件套端点 （含 create/status/patch），走 operator 角色 + action:* 权限码 + write_audit。
139 lines
4.9 KiB
Python
139 lines
4.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""运营端 → 培训子应用数据桥接（课程 / 活动 / 报名 / 测评）。
|
|
|
|
培训业务数据存于 app.training（独立 serverdata/data/opc.db）；运营端 /admin/*
|
|
端点在此聚合读写，避免在平台应用重建一套课程/活动模型。
|
|
|
|
仅作适度封装（list / create / status / patch），状态与原 mock 语义保持一致；
|
|
创建统一生成 id（前缀 C- / E- / B- / T-），时间用 UTC ISO。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from app.training import db as tdb
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
# ── 课程（courses 表）────────────────────────────────────────────────────
|
|
|
|
def list_courses(status: str | None = None) -> list[dict]:
|
|
rows = tdb.list_all("courses", "created_at DESC")
|
|
if status:
|
|
return [r for r in rows if r.get("status") == status]
|
|
return rows
|
|
|
|
|
|
def create_course(data: dict) -> dict:
|
|
row = dict(data)
|
|
row["id"] = tdb.gen_id("C-")
|
|
row.setdefault("created_at", _now())
|
|
row.setdefault("status", "draft")
|
|
tdb.insert("courses", row)
|
|
return tdb.fetch_by_id("courses", row["id"])
|
|
|
|
|
|
def set_course_status(course_id: str, status: str) -> dict | None:
|
|
if tdb.fetch_by_id("courses", course_id) is None:
|
|
return None
|
|
tdb.update_row("courses", course_id, {"status": status})
|
|
return tdb.fetch_by_id("courses", course_id)
|
|
|
|
|
|
# ── 活动（events 表）─────────────────────────────────────────────────────
|
|
|
|
def list_activities(status: str | None = None) -> list[dict]:
|
|
rows = tdb.list_all("events", "start_at DESC")
|
|
if status:
|
|
return [r for r in rows if r.get("status") == status]
|
|
return rows
|
|
|
|
|
|
def set_activity_status(event_id: str, status: str) -> dict | None:
|
|
if tdb.fetch_by_id("events", event_id) is None:
|
|
return None
|
|
tdb.update_row("events", event_id, {"status": status})
|
|
return tdb.fetch_by_id("events", event_id)
|
|
|
|
|
|
# ── 报名（bookings 表）───────────────────────────────────────────────────
|
|
|
|
def list_bookings(status: str | None = None, audit_status: str | None = None) -> list[dict]:
|
|
rows = tdb.list_all("bookings", "created_at DESC")
|
|
if status:
|
|
rows = [r for r in rows if r.get("status") == status]
|
|
if audit_status:
|
|
rows = [r for r in rows if r.get("audit_status") == audit_status]
|
|
# 映射为运营端 Booking 视图
|
|
return [pack_booking(r) for r in rows]
|
|
|
|
|
|
def pack_booking(r: dict) -> dict:
|
|
return {
|
|
"id": r.get("id"),
|
|
"user_id": r.get("username"),
|
|
"user_name": r.get("name"),
|
|
"target_type": "event",
|
|
"target_id": r.get("event_id"),
|
|
"target_title": r.get("event_title"),
|
|
"status": r.get("status"),
|
|
"audit_status": r.get("audit_status"),
|
|
"note": r.get("question"),
|
|
"created_at": r.get("created_at"),
|
|
}
|
|
|
|
|
|
def update_booking(booking_id: str, patch: dict) -> dict | None:
|
|
if tdb.fetch_by_id("bookings", booking_id) is None:
|
|
return None
|
|
allowed = {k: v for k, v in patch.items() if k in ("status", "audit_status", "note", "question")}
|
|
if "note" in allowed:
|
|
allowed["question"] = allowed.pop("note")
|
|
if allowed:
|
|
tdb.update_row("bookings", booking_id, allowed)
|
|
return pack_booking(tdb.fetch_by_id("bookings", booking_id))
|
|
|
|
|
|
# ── 测评（tests 表：OPC 适配度测评记录）─────────────────────────────────────
|
|
|
|
def list_tests() -> list[dict]:
|
|
rows = tdb.list_all("tests", "created_at DESC")
|
|
return [pack_test(r) for r in rows]
|
|
|
|
|
|
def pack_test(r: dict) -> dict:
|
|
return {
|
|
"id": r.get("id"),
|
|
"title": r.get("persona") or "OPC 适配度测评",
|
|
"category": r.get("type_code"),
|
|
"pass_score": 0,
|
|
"question_count": 30,
|
|
"status": "completed",
|
|
"adapt_level": r.get("adapt_level"),
|
|
"tracks": r.get("tracks"),
|
|
"created_at": r.get("created_at"),
|
|
}
|
|
|
|
|
|
def pack_test_if_exists(test_id: str) -> dict | None:
|
|
row = tdb.fetch_by_id("tests", test_id)
|
|
return pack_test(row) if row else None
|
|
|
|
|
|
def create_test(data: dict) -> dict:
|
|
row = {
|
|
"id": tdb.gen_id("T-"),
|
|
"created_at": _now(),
|
|
"username": "",
|
|
"type_code": data.get("category", "opc"),
|
|
"persona": data.get("title", "OPC 适配度测评"),
|
|
"adapt_level": "",
|
|
"tracks": data.get("tracks", ""),
|
|
"version": "",
|
|
}
|
|
tdb.insert("tests", row)
|
|
return pack_test(tdb.fetch_by_id("tests", row["id"]))
|