Pine
648 lines
24 KiB
Python
648 lines
24 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
|
|
answers = _parse_json(row.get("answers"))
|
|
# 匹配调研题目定义，返回问题→答案可读格式
|
|
answer_list = _match_survey_answers(answers)
|
|
return {
|
|
"id": row.get("id"),
|
|
"username": row.get("username") or "",
|
|
"source": row.get("source") or "",
|
|
"answers": answers,
|
|
"answer_list": answer_list,
|
|
"answered_count": sum(1 for a in answer_list if a.get("answer_text")),
|
|
"total_count": len(answer_list),
|
|
"created_at": row.get("created_at"),
|
|
}
|
|
|
|
|
|
def _match_survey_answers(answers: dict) -> list[dict]:
|
|
"""将 answers JSON 匹配调研题目定义，返回可读的问题→答案列表。"""
|
|
try:
|
|
from app.training.survey_data import SURVEY_QUESTIONS
|
|
except Exception:
|
|
return []
|
|
result = []
|
|
for q in SURVEY_QUESTIONS:
|
|
qid = q["id"]
|
|
raw = answers.get(qid) if qid in answers else answers.get(str(qid))
|
|
answer_text = _format_answer(q, raw)
|
|
result.append({
|
|
"id": qid,
|
|
"part": q.get("part", ""),
|
|
"question": q.get("label", ""),
|
|
"type": q.get("type", ""),
|
|
"raw_value": raw,
|
|
"answer_text": answer_text,
|
|
})
|
|
return result
|
|
|
|
|
|
def _format_answer(question: dict, raw) -> str:
|
|
"""根据题型格式化答案为可读文本。"""
|
|
if raw is None or raw == "":
|
|
return ""
|
|
qtype = question.get("type", "")
|
|
options = {o.get("value"): o.get("label", str(o.get("value"))) for o in question.get("options", [])}
|
|
if qtype == "multi" and isinstance(raw, list):
|
|
return "、".join(options.get(v, str(v)) for v in raw if v)
|
|
if qtype == "likert":
|
|
likert_labels = {"1": "非常不同意", "2": "不同意", "3": "一般", "4": "同意", "5": "非常同意"}
|
|
return likert_labels.get(str(raw), str(raw))
|
|
if qtype == "text":
|
|
return str(raw)
|
|
# single 或其他
|
|
return options.get(raw, str(raw))
|
|
|
|
|
|
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
|
|
answers = _parse_json(row.get("answers"))
|
|
answer_list = _match_policy_answers(answers)
|
|
return {
|
|
"id": row.get("id"),
|
|
"username": row.get("username") or "",
|
|
"answers": answers,
|
|
"answer_list": answer_list,
|
|
"answered_count": sum(1 for a in answer_list if a.get("answer_text")),
|
|
"total_count": len(answer_list),
|
|
"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 _match_policy_answers(answers: dict) -> list[dict]:
|
|
"""将政策测评 answers 匹配题目定义，返回可读格式。"""
|
|
try:
|
|
from app.training.policy_data import PT_QUESTIONS
|
|
except Exception:
|
|
return []
|
|
result = []
|
|
for q in PT_QUESTIONS:
|
|
qid = q["id"]
|
|
raw = answers.get(qid) if qid in answers else answers.get(str(qid))
|
|
options = {o.get("value"): o.get("label", str(o.get("value"))) for o in q.get("options", [])}
|
|
if isinstance(raw, list):
|
|
answer_text = "、".join(options.get(v, str(v)) for v in raw if v)
|
|
else:
|
|
answer_text = options.get(raw, str(raw) if raw not in (None, "") else "")
|
|
result.append({
|
|
"id": qid,
|
|
"question": q.get("label", ""),
|
|
"multi": q.get("multi", False),
|
|
"raw_value": raw,
|
|
"answer_text": answer_text,
|
|
})
|
|
return result
|
|
|
|
|
|
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
|
|
|
|
|
|
# ── 题目定义（供前端映射问题和答案）───────────────────────────────────────
|
|
|
|
def survey_questions() -> list[dict]:
|
|
"""返回调研题目定义（37题），供 admin 详情页/汇总页映射。"""
|
|
try:
|
|
from app.training.survey_data import SURVEY_QUESTIONS
|
|
return [
|
|
{"id": q["id"], "part": q.get("part", ""), "label": q.get("label", ""),
|
|
"type": q.get("type", ""), "options": q.get("options", [])}
|
|
for q in SURVEY_QUESTIONS
|
|
]
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
# ── 汇总统计（各模块）──────────────────────────────────────────────────────
|
|
|
|
def _count_distribution(rows: list[dict], key: str) -> dict:
|
|
"""统计某字段的取值分布。"""
|
|
from collections import Counter
|
|
c = Counter()
|
|
for r in rows:
|
|
v = r.get(key)
|
|
if v not in (None, "", []):
|
|
c[str(v)] += 1
|
|
return dict(c.most_common())
|
|
|
|
|
|
def survey_summary() -> dict:
|
|
"""调研汇总：总提交数 + 各题答案分布。"""
|
|
rows = list_surveys()
|
|
total = len(rows)
|
|
questions = survey_questions()
|
|
# 各题答案分布
|
|
question_stats = []
|
|
for q in questions:
|
|
qid = q["id"]
|
|
qtype = q["type"]
|
|
options = {o.get("value"): o.get("label", str(o.get("value"))) for o in q.get("options", [])}
|
|
counter: dict = {}
|
|
answered = 0
|
|
for r in rows:
|
|
answers = r.get("answers") or {}
|
|
raw = answers.get(qid) if qid in answers else answers.get(str(qid))
|
|
if raw in (None, "", []):
|
|
continue
|
|
answered += 1
|
|
if qtype == "multi" and isinstance(raw, list):
|
|
for v in raw:
|
|
counter[v] = counter.get(v, 0) + 1
|
|
elif qtype == "likert":
|
|
counter[str(raw)] = counter.get(str(raw), 0) + 1
|
|
else:
|
|
counter[str(raw)] = counter.get(str(raw), 0) + 1
|
|
# 转换为带 label 的分布
|
|
distribution = []
|
|
for val, cnt in sorted(counter.items(), key=lambda x: -x[1]):
|
|
distribution.append({
|
|
"value": val,
|
|
"label": options.get(val, val) if qtype != "likert" else {"1": "非常不同意", "2": "不同意", "3": "一般", "4": "同意", "5": "非常同意"}.get(val, val),
|
|
"count": cnt,
|
|
"percent": round(cnt / total * 100, 1) if total else 0,
|
|
})
|
|
question_stats.append({
|
|
"id": qid, "label": q.get("label", ""), "type": qtype,
|
|
"answered": answered, "total": total,
|
|
"distribution": distribution,
|
|
})
|
|
# 来源分布
|
|
source_dist = _count_distribution(rows, "source")
|
|
return {"total": total, "source_distribution": source_dist, "questions": question_stats}
|
|
|
|
|
|
def test_summary() -> dict:
|
|
"""测评汇总：总记录数 + persona/adapt_level 分布 + 平均适配指数。"""
|
|
rows = list_tests()
|
|
total = len(rows)
|
|
persona_dist = _count_distribution(rows, "persona")
|
|
level_dist = _count_distribution(rows, "adapt_level")
|
|
type_dist = _count_distribution(rows, "type_code")
|
|
# 平均适配指数
|
|
indices = [r.get("adapt_index", 0) for r in rows if isinstance(r.get("adapt_index"), (int, float))]
|
|
avg_index = round(sum(indices) / len(indices), 1) if indices else 0
|
|
# 版本分布
|
|
version_dist = _count_distribution(rows, "version")
|
|
return {
|
|
"total": total, "avg_adapt_index": avg_index,
|
|
"persona_distribution": persona_dist,
|
|
"adapt_level_distribution": level_dist,
|
|
"type_distribution": type_dist,
|
|
"version_distribution": version_dist,
|
|
}
|
|
|
|
|
|
def booking_summary() -> dict:
|
|
"""报名汇总：总报名数 + 各活动报名人数 + 状态/审核/支付分布。"""
|
|
rows = list_bookings()
|
|
total = len(rows)
|
|
# 按活动分组
|
|
event_counter: dict = {}
|
|
for r in rows:
|
|
title = r.get("event_title") or r.get("target_title") or "未命名活动"
|
|
event_counter[title] = event_counter.get(title, 0) + 1
|
|
event_dist = [{"event": k, "count": v, "percent": round(v / total * 100, 1) if total else 0}
|
|
for k, v in sorted(event_counter.items(), key=lambda x: -x[1])]
|
|
status_dist = _count_distribution(rows, "status")
|
|
audit_dist = _count_distribution(rows, "audit_status")
|
|
pay_dist = _count_distribution(rows, "pay_status")
|
|
source_dist = _count_distribution(rows, "source")
|
|
return {
|
|
"total": total, "event_distribution": event_dist,
|
|
"status_distribution": status_dist, "audit_distribution": audit_dist,
|
|
"pay_distribution": pay_dist, "source_distribution": source_dist,
|
|
}
|
|
|
|
|
|
def policy_summary() -> dict:
|
|
"""政策测评汇总：总记录数 + 各题答案分布 + 政策/补贴/贷款匹配统计。"""
|
|
rows = list_policies()
|
|
total = len(rows)
|
|
# 政策匹配统计
|
|
policy_counts = [r.get("policies_count", 0) for r in rows if isinstance(r.get("policies_count"), (int, float))]
|
|
subsidy_counts = [r.get("subsidies_count", 0) for r in rows if isinstance(r.get("subsidies_count"), (int, float))]
|
|
loan_counts = [r.get("loans_count", 0) for r in rows if isinstance(r.get("loans_count"), (int, float))]
|
|
# 各题答案分布（政策题 5 道）
|
|
try:
|
|
from app.training.policy_data import PT_QUESTIONS
|
|
except Exception:
|
|
PT_QUESTIONS = []
|
|
question_stats = []
|
|
for q in PT_QUESTIONS:
|
|
qid = q["id"]
|
|
options = {o.get("value"): o.get("label", str(o.get("value"))) for o in q.get("options", [])}
|
|
counter: dict = {}
|
|
answered = 0
|
|
for r in rows:
|
|
answers = r.get("answers") or {}
|
|
raw = answers.get(qid) if qid in answers else answers.get(str(qid))
|
|
if raw in (None, "", []):
|
|
continue
|
|
answered += 1
|
|
if isinstance(raw, list):
|
|
for v in raw:
|
|
counter[v] = counter.get(v, 0) + 1
|
|
else:
|
|
counter[str(raw)] = counter.get(str(raw), 0) + 1
|
|
distribution = [{"value": v, "label": options.get(v, v), "count": c,
|
|
"percent": round(c / total * 100, 1) if total else 0}
|
|
for v, c in sorted(counter.items(), key=lambda x: -x[1])]
|
|
question_stats.append({"id": qid, "label": q.get("label", ""), "answered": answered,
|
|
"total": total, "distribution": distribution})
|
|
return {
|
|
"total": total,
|
|
"avg_policies": round(sum(policy_counts) / len(policy_counts), 1) if policy_counts else 0,
|
|
"avg_subsidies": round(sum(subsidy_counts) / len(subsidy_counts), 1) if subsidy_counts else 0,
|
|
"avg_loans": round(sum(loan_counts) / len(loan_counts), 1) if loan_counts else 0,
|
|
"questions": question_stats,
|
|
}
|
|
|
|
|
|
def plan_summary() -> dict:
|
|
"""流程启动汇总：总记录数 + 区域/状态分布 + 需入驻园区/已有员工比例。"""
|
|
rows = list_plans()
|
|
total = len(rows)
|
|
region_dist = _count_distribution(rows, "region")
|
|
status_dist = _count_distribution(rows, "status")
|
|
need_park_count = sum(1 for r in rows if r.get("need_park"))
|
|
has_staff_count = sum(1 for r in rows if r.get("has_staff"))
|
|
steps_counts = [r.get("steps_count", 0) for r in rows if isinstance(r.get("steps_count"), (int, float))]
|
|
return {
|
|
"total": total,
|
|
"region_distribution": region_dist,
|
|
"status_distribution": status_dist,
|
|
"need_park_count": need_park_count,
|
|
"need_park_percent": round(need_park_count / total * 100, 1) if total else 0,
|
|
"has_staff_count": has_staff_count,
|
|
"has_staff_percent": round(has_staff_count / total * 100, 1) if total else 0,
|
|
"avg_steps": round(sum(steps_counts) / len(steps_counts), 1) if steps_counts else 0,
|
|
}
|