Pine
541c60e64f
- alembic 0043: tests 表增加 answers 列（TEXT，JSON 格式逐题答案） - report_test(POST /api/tests): 接收并存储 answers 字段 - get_test: 返回 answers + answer_list（带题目文本/part/选项/用户选择的逐题列表）+ 答题进度 - pack_test: 增加 answers 字段
392 lines
13 KiB
Python
392 lines
13 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
|
|
|
|
import json
|
|
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)
|
|
|
|
|
|
def update_course(course_id: str, patch: dict) -> dict | None:
|
|
if tdb.fetch_by_id("courses", course_id) is None:
|
|
return None
|
|
if patch:
|
|
tdb.update_row("courses", course_id, patch)
|
|
return tdb.fetch_by_id("courses", course_id)
|
|
|
|
|
|
def delete_course(course_id: str) -> bool:
|
|
if tdb.fetch_by_id("courses", course_id) is None:
|
|
return False
|
|
tdb.delete_row("courses", course_id)
|
|
return True
|
|
|
|
|
|
# ── 活动（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 create_activity(data: dict) -> dict:
|
|
row = dict(data)
|
|
row["id"] = tdb.gen_id("E-")
|
|
row.setdefault("mode", "offline")
|
|
row.setdefault("status", "open")
|
|
row.setdefault("duration_min", 90)
|
|
tdb.insert("events", row)
|
|
return tdb.fetch_by_id("events", row["id"])
|
|
|
|
|
|
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)
|
|
|
|
|
|
def update_activity(event_id: str, patch: dict) -> dict | None:
|
|
if tdb.fetch_by_id("events", event_id) is None:
|
|
return None
|
|
if patch:
|
|
tdb.update_row("events", event_id, patch)
|
|
return tdb.fetch_by_id("events", event_id)
|
|
|
|
|
|
def delete_activity(event_id: str) -> bool:
|
|
if tdb.fetch_by_id("events", event_id) is None:
|
|
return False
|
|
tdb.delete_row("events", event_id)
|
|
return True
|
|
|
|
|
|
# ── 报名（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"),
|
|
"pay_status": r.get("pay_status") or "",
|
|
"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 "audit_status" in allowed and "status" not in allowed:
|
|
if allowed["audit_status"] == "approved":
|
|
allowed["status"] = "confirmed" # 审核通过 → 已确认（待参加）
|
|
elif allowed["audit_status"] == "rejected":
|
|
allowed["status"] = "rejected" # 驳回 → 已驳回（终态，不计入有效报名）
|
|
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"),
|
|
"username": r.get("username") or "",
|
|
"pass_score": 0,
|
|
"question_count": 30,
|
|
"status": "completed",
|
|
"adapt_index": r.get("adapt_index"),
|
|
"adapt_level": r.get("adapt_level"),
|
|
"tracks": _parse_json(r.get("tracks")),
|
|
"version": r.get("version"),
|
|
"answers": _parse_json(r.get("answers")),
|
|
"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"]))
|
|
|
|
|
|
def _parse_json(raw):
|
|
if raw is None:
|
|
return None
|
|
if isinstance(raw, (dict, list)):
|
|
return raw
|
|
try:
|
|
return json.loads(raw)
|
|
except (TypeError, ValueError):
|
|
return raw
|
|
|
|
|
|
# ── 调研（survey_logs 表）─────────────────────────────────────────────────
|
|
|
|
def list_surveys() -> list[dict]:
|
|
rows = tdb.list_all("survey_logs", "created_at DESC")
|
|
return [
|
|
{
|
|
"id": r.get("id"),
|
|
"username": r.get("username"),
|
|
"source": r.get("source"),
|
|
"answers": _parse_json(r.get("answers")),
|
|
"created_at": r.get("created_at"),
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
# ── 政策（policy_logs 表）─────────────────────────────────────────────────
|
|
|
|
def list_policies() -> list[dict]:
|
|
rows = tdb.list_all("policy_logs", "created_at DESC")
|
|
return [
|
|
{
|
|
"id": r.get("id"),
|
|
"username": r.get("username"),
|
|
"answers": _parse_json(r.get("answers")),
|
|
"policies_count": r.get("policies_count"),
|
|
"subsidies_count": r.get("subsidies_count"),
|
|
"loans_count": r.get("loans_count"),
|
|
"summary": r.get("summary"),
|
|
"created_at": r.get("created_at"),
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
# ── 流程（plan_logs 表）───────────────────────────────────────────────────
|
|
|
|
def list_plans() -> list[dict]:
|
|
rows = tdb.list_all("plan_logs", "created_at DESC")
|
|
return [
|
|
{
|
|
"id": r.get("id"),
|
|
"username": r.get("username"),
|
|
"region": r.get("region"),
|
|
"status": r.get("status"),
|
|
"need_park": r.get("need_park"),
|
|
"has_staff": r.get("has_staff"),
|
|
"steps_count": r.get("steps_count"),
|
|
"created_at": r.get("created_at"),
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
# ── 通用详情 / 删除（各模块统一）──────────────────────────────────────────
|
|
|
|
def get_test(test_id: str) -> dict | None:
|
|
row = tdb.fetch_by_id("tests", test_id)
|
|
if not row:
|
|
return None
|
|
d = pack_test(row)
|
|
# 逐题答案：匹配题目文本，供 admin 详情页展示
|
|
answers = d.get("answers") or {}
|
|
if isinstance(answers, dict) and answers:
|
|
from ..training.opc_engine import current_questions
|
|
version = row.get("version") or "full"
|
|
qs = current_questions(version)
|
|
answer_list = []
|
|
for q in qs:
|
|
qid = q["id"]
|
|
choice = answers.get(qid) if qid in answers else answers.get(str(qid))
|
|
answer_list.append({
|
|
"id": qid,
|
|
"part": q.get("part", ""),
|
|
"question": q.get("question", ""),
|
|
"A": q.get("A", ""),
|
|
"B": q.get("B", ""),
|
|
"user_choice": choice or "",
|
|
})
|
|
d["answer_list"] = answer_list
|
|
d["answered_count"] = sum(1 for a in answer_list if a["user_choice"] in ("A", "B"))
|
|
d["total_count"] = len(answer_list)
|
|
else:
|
|
d["answer_list"] = []
|
|
d["answered_count"] = 0
|
|
d["total_count"] = 0
|
|
return d
|
|
|
|
|
|
def delete_test(test_id: str) -> bool:
|
|
if tdb.fetch_by_id("tests", test_id) is None:
|
|
return False
|
|
tdb.delete_row("tests", test_id)
|
|
return True
|
|
|
|
|
|
def get_booking(booking_id: str) -> dict | None:
|
|
row = tdb.fetch_by_id("bookings", booking_id)
|
|
if not row:
|
|
return None
|
|
d = pack_booking(row)
|
|
# 详情页补充完整字段
|
|
d["contact"] = row.get("contact") or ""
|
|
d["status_label"] = row.get("status_label") or ""
|
|
d["want"] = row.get("want") or ""
|
|
d["topics"] = _parse_json(row.get("topics"))
|
|
d["question"] = row.get("question") or ""
|
|
d["source"] = row.get("source") or ""
|
|
d["event_start"] = row.get("event_start") or ""
|
|
d["checkin_at"] = row.get("checkin_at") or ""
|
|
d["order_no"] = row.get("order_no") or ""
|
|
d["appeal_status"] = row.get("appeal_status") or ""
|
|
d["appeal_reason"] = row.get("appeal_reason") or ""
|
|
d["appeal_at"] = row.get("appeal_at") or ""
|
|
return d
|
|
|
|
|
|
def delete_booking(booking_id: str) -> bool:
|
|
if tdb.fetch_by_id("bookings", booking_id) is None:
|
|
return False
|
|
tdb.delete_row("bookings", booking_id)
|
|
return True
|
|
|
|
|
|
def get_survey(survey_id: str) -> dict | None:
|
|
row = tdb.fetch_by_id("survey_logs", survey_id)
|
|
if not row:
|
|
return None
|
|
return {
|
|
"id": row.get("id"),
|
|
"username": row.get("username") or "",
|
|
"source": row.get("source") or "",
|
|
"answers": _parse_json(row.get("answers")),
|
|
"created_at": row.get("created_at"),
|
|
}
|
|
|
|
|
|
def delete_survey(survey_id: str) -> bool:
|
|
if tdb.fetch_by_id("survey_logs", survey_id) is None:
|
|
return False
|
|
tdb.delete_row("survey_logs", survey_id)
|
|
return True
|
|
|
|
|
|
def get_policy(policy_id: str) -> dict | None:
|
|
row = tdb.fetch_by_id("policy_logs", policy_id)
|
|
if not row:
|
|
return None
|
|
return {
|
|
"id": row.get("id"),
|
|
"username": row.get("username") or "",
|
|
"answers": _parse_json(row.get("answers")),
|
|
"policies_count": row.get("policies_count"),
|
|
"subsidies_count": row.get("subsidies_count"),
|
|
"loans_count": row.get("loans_count"),
|
|
"summary": row.get("summary") or "",
|
|
"created_at": row.get("created_at"),
|
|
}
|
|
|
|
|
|
def delete_policy(policy_id: str) -> bool:
|
|
if tdb.fetch_by_id("policy_logs", policy_id) is None:
|
|
return False
|
|
tdb.delete_row("policy_logs", policy_id)
|
|
return True
|
|
|
|
|
|
def get_plan(plan_id: str) -> dict | None:
|
|
row = tdb.fetch_by_id("plan_logs", plan_id)
|
|
if not row:
|
|
return None
|
|
return {
|
|
"id": row.get("id"),
|
|
"username": row.get("username") or "",
|
|
"region": row.get("region") or "",
|
|
"status": row.get("status") or "",
|
|
"need_park": row.get("need_park"),
|
|
"has_staff": row.get("has_staff"),
|
|
"steps_count": row.get("steps_count"),
|
|
"created_at": row.get("created_at"),
|
|
}
|
|
|
|
|
|
def delete_plan(plan_id: str) -> bool:
|
|
if tdb.fetch_by_id("plan_logs", plan_id) is None:
|
|
return False
|
|
tdb.delete_row("plan_logs", plan_id)
|
|
return True
|