feat: 新增效能分析看板（个人能力图谱 + 设备人员耗时对比）
- 前端：AnalyticsDashboard 页面 + BaseEChart 封装 + analyticsApi，路由与导航 - 后端：/analytics 系列接口（capability/flow/options/device-records） - 能力图谱单机颗粒度、流转对比按人堆叠识别瓶颈、点击下钻备注、筛选动态联动 - 依赖：引入 echarts，移除 echarts-for-react
This commit is contained in:
501
backend/app/services/analytics_service.py
Normal file
501
backend/app/services/analytics_service.py
Normal file
@ -0,0 +1,501 @@
|
||||
"""效能分析服务 — 个人能力图谱 + 设备流转对比（ECharts 数据源）
|
||||
|
||||
数据来源：
|
||||
- Task 工序/任务节点，携带 assignee_id、received_at、completed_at → 计算单台耗时。
|
||||
- Product 设备身份证(sn)、规格型号(spec_model)、物料名，用于筛选与展示。
|
||||
- TaskRecord 备注时间线（本模块当前仅作耗时主数据补充，可按需在后续下钻中引入）。
|
||||
|
||||
耗时口径：
|
||||
- 单台真实耗时 = (completed_at or now) - (received_at or created_at)，单位小时。
|
||||
- naive datetime 按 UTC 处理，统一换算北京时间。
|
||||
"""
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from sqlalchemy import select, func
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
# ============================================================
|
||||
# Schemas（与前端 src/services/analyticsApi.ts 对齐）
|
||||
# ============================================================
|
||||
|
||||
class CapabilityDataPoint(BaseModel):
|
||||
value: float | None # 该人员在该设备上的总耗时（未触及为 None，真实 0 为 0.0）
|
||||
spec_model: str # 规格型号
|
||||
status: str # 该设备当前状态 WIP/PENDING/COMPLETED/—
|
||||
first_received_at: str # 最早接收时间 YYYY-MM-DD HH:mm
|
||||
last_completed_at: str # 最后完成时间 YYYY-MM-DD HH:mm（无则空串）
|
||||
|
||||
|
||||
class CapabilityDevice(BaseModel):
|
||||
product_sn: str # 身份证
|
||||
external_serial: str | None # 产品序列号（业务序列号）
|
||||
spec_model: str # 规格型号
|
||||
|
||||
|
||||
class CapabilitySeries(BaseModel):
|
||||
name: str # 人员姓名
|
||||
assignee_id: str # 人员ID
|
||||
data: list[CapabilityDataPoint] # 与 categories 严格对齐，未触及设备补 0
|
||||
|
||||
|
||||
class CapabilityResponse(BaseModel):
|
||||
categories: list[str] # X 轴：设备身份证（按时间升序）
|
||||
devices: list[CapabilityDevice] # 与 categories 对齐的设备元数据
|
||||
series: list[CapabilitySeries]
|
||||
|
||||
|
||||
class FlowDevice(BaseModel):
|
||||
product_sn: str
|
||||
external_serial: str | None
|
||||
material_name: str
|
||||
spec_model: str
|
||||
|
||||
|
||||
class FlowSeries(BaseModel):
|
||||
name: str # 工序节点名
|
||||
data: list[float | None] # 每台设备在该工序的总耗时（未经过为 None）
|
||||
|
||||
|
||||
class FlowResponse(BaseModel):
|
||||
devices: list[FlowDevice]
|
||||
series: list[FlowSeries]
|
||||
|
||||
|
||||
class AssigneeOption(BaseModel):
|
||||
id: str
|
||||
name: str
|
||||
|
||||
|
||||
class AnalyticsOptions(BaseModel):
|
||||
assignees: list[AssigneeOption]
|
||||
spec_models: list[str]
|
||||
|
||||
|
||||
class DeviceRecord(BaseModel):
|
||||
task_name: str # 工序名
|
||||
assignee_name: str # 负责人姓名
|
||||
status: str # 任务状态
|
||||
remark: str | None # 备注
|
||||
images: list[str] # 图片 URL 列表
|
||||
created_at: str # 记录时间 ISO
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 时间工具
|
||||
# ============================================================
|
||||
|
||||
def _to_bj(dt: datetime | None) -> datetime | None:
|
||||
"""naive datetime 按 UTC 处理，统一换算为北京时间。"""
|
||||
from app.core.time_utils import BEIJING_TZ
|
||||
if not dt:
|
||||
return None
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=timezone.utc).astimezone(BEIJING_TZ)
|
||||
else:
|
||||
dt = dt.astimezone(BEIJING_TZ)
|
||||
return dt
|
||||
|
||||
|
||||
def _duration_hours(start: datetime | None, end: datetime) -> float:
|
||||
"""计算单台耗时（小时），无开始时间返回 0。"""
|
||||
if not start:
|
||||
return 0.0
|
||||
return (end - start).total_seconds() / 3600
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 个人能力图谱 — 单台设备耗时对比（分组柱状图，X=设备身份证）
|
||||
# ============================================================
|
||||
|
||||
async def get_capability_profile(
|
||||
db: AsyncSession,
|
||||
assignee_ids: list[str] | None = None,
|
||||
spec_models: list[str] | None = None,
|
||||
since: datetime | None = None,
|
||||
until: datetime | None = None,
|
||||
) -> CapabilityResponse:
|
||||
"""
|
||||
个人能力图谱（单机颗粒度）：X 轴 = 设备身份证，每个负责人一条柱状 series。
|
||||
按 (负责人, 设备) 分组，累加该人在该设备所有工序的耗时，
|
||||
series.data 与 categories 严格对齐，未触及设备补 0。
|
||||
耗时口径：(coalesce(completed_at, now) - coalesce(received_at, created_at))。
|
||||
"""
|
||||
from app.models.task import (
|
||||
Task, TASK_STATUS_WIP, TASK_STATUS_PENDING, TASK_STATUS_COMPLETED,
|
||||
)
|
||||
from app.models.product import Product
|
||||
from app.core.time_utils import get_beijing_time
|
||||
|
||||
now = get_beijing_time()
|
||||
|
||||
stmt = (
|
||||
select(
|
||||
Task.assignee_id, Task.status,
|
||||
Task.received_at, Task.created_at, Task.completed_at,
|
||||
Product.serial_number, Product.external_serial, Product.spec_model,
|
||||
)
|
||||
.join(Product, Task.product_id == Product.id)
|
||||
.where(
|
||||
Task.status.in_([TASK_STATUS_WIP, TASK_STATUS_PENDING, TASK_STATUS_COMPLETED]),
|
||||
Task.assignee_id.isnot(None),
|
||||
)
|
||||
)
|
||||
if assignee_ids:
|
||||
stmt = stmt.where(Task.assignee_id.in_(assignee_ids))
|
||||
if spec_models:
|
||||
stmt = stmt.where(Product.spec_model.in_(spec_models))
|
||||
|
||||
# 时间交集（与 people-history 口径一致）
|
||||
start_expr = func.coalesce(Task.received_at, Task.created_at)
|
||||
end_expr = func.coalesce(Task.completed_at, now)
|
||||
if until:
|
||||
stmt = stmt.where(start_expr <= until)
|
||||
if since:
|
||||
stmt = stmt.where(end_expr >= since)
|
||||
|
||||
result = await db.execute(stmt)
|
||||
rows = result.all()
|
||||
|
||||
# 设备元数据 + 时间基准；人员×设备 耗时聚合
|
||||
device_meta: dict[str, dict] = {}
|
||||
agg: dict[tuple[str, str], dict] = {}
|
||||
for row in rows:
|
||||
assignee = row[0]
|
||||
status = row[1]
|
||||
start = _to_bj(row[2] or row[3]) # received_at or created_at
|
||||
end = _to_bj(row[4]) if row[4] else now # completed_at or now
|
||||
sn = row[5]
|
||||
ext = row[6]
|
||||
spec = row[7] or "未知型号"
|
||||
|
||||
meta = device_meta.setdefault(sn, {"external_serial": ext, "spec_model": spec, "earliest": None})
|
||||
if start and (meta["earliest"] is None or start < meta["earliest"]):
|
||||
meta["earliest"] = start
|
||||
|
||||
entry = agg.setdefault((assignee, sn), {
|
||||
"hours": 0.0, "prio": 9,
|
||||
"first_start": None, "last_completed": None,
|
||||
})
|
||||
entry["hours"] += _duration_hours(start, end)
|
||||
if start and (entry["first_start"] is None or start < entry["first_start"]):
|
||||
entry["first_start"] = start
|
||||
completed_dt = _to_bj(row[4]) if row[4] else None
|
||||
if completed_dt and (entry["last_completed"] is None or completed_dt > entry["last_completed"]):
|
||||
entry["last_completed"] = completed_dt
|
||||
prio = 0 if status == TASK_STATUS_WIP else (1 if status == TASK_STATUS_PENDING else 2)
|
||||
entry["prio"] = min(entry["prio"], prio)
|
||||
|
||||
if not agg:
|
||||
return CapabilityResponse(categories=[], devices=[], series=[])
|
||||
|
||||
# X 轴：按最早接收/创建时间升序的设备身份证
|
||||
categories = sorted(device_meta.keys(), key=lambda s: device_meta[s]["earliest"] or now)
|
||||
devices = [
|
||||
CapabilityDevice(
|
||||
product_sn=sn,
|
||||
external_serial=device_meta[sn]["external_serial"],
|
||||
spec_model=device_meta[sn]["spec_model"],
|
||||
)
|
||||
for sn in categories
|
||||
]
|
||||
|
||||
# 人员姓名映射
|
||||
raw_ids = list({a for (a, _) in agg.keys()})
|
||||
name_map: dict[str, str] = {}
|
||||
if raw_ids:
|
||||
from app.services.mom_cache import get_display_names
|
||||
name_map = get_display_names(raw_ids)
|
||||
|
||||
STATUS_CODE = {0: "WIP", 1: "PENDING", 2: "COMPLETED"}
|
||||
|
||||
by_assignee: dict[str, dict[str, dict]] = {}
|
||||
for (a, sn), entry in agg.items():
|
||||
by_assignee.setdefault(a, {})[sn] = entry
|
||||
|
||||
series: list[CapabilitySeries] = []
|
||||
for a in sorted(by_assignee.keys()):
|
||||
sn_entries = by_assignee[a]
|
||||
data: list[CapabilityDataPoint] = []
|
||||
for sn in categories:
|
||||
e = sn_entries.get(sn)
|
||||
if e:
|
||||
data.append(CapabilityDataPoint(
|
||||
value=round(e["hours"], 1),
|
||||
spec_model=device_meta[sn]["spec_model"],
|
||||
status=STATUS_CODE[e["prio"]],
|
||||
first_received_at=e["first_start"].strftime("%Y-%m-%d %H:%M") if e["first_start"] else "",
|
||||
last_completed_at=e["last_completed"].strftime("%Y-%m-%d %H:%M") if e["last_completed"] else "",
|
||||
))
|
||||
else:
|
||||
data.append(CapabilityDataPoint(
|
||||
value=None,
|
||||
spec_model=device_meta[sn]["spec_model"],
|
||||
status="—",
|
||||
first_received_at="",
|
||||
last_completed_at="",
|
||||
))
|
||||
series.append(CapabilitySeries(
|
||||
name=name_map.get(a, a),
|
||||
assignee_id=a,
|
||||
data=data,
|
||||
))
|
||||
series.sort(key=lambda s: s.name)
|
||||
return CapabilityResponse(categories=categories, devices=devices, series=series)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 流转对比 — 设备各工序耗时（堆叠柱状）
|
||||
# ============================================================
|
||||
|
||||
async def get_flow_compare(
|
||||
db: AsyncSession,
|
||||
product_sns: list[str] | None = None,
|
||||
spec_models: list[str] | None = None,
|
||||
) -> FlowResponse:
|
||||
"""
|
||||
查询每台设备上各操作人的累计耗时，拼装为堆叠柱状图：
|
||||
x 轴 = 设备身份证，每个「人员」一个 series，值为该员工在这台设备上的总耗时，
|
||||
谁的色块最长谁就是该设备的瓶颈。
|
||||
|
||||
设备来源：
|
||||
- 传 product_sns：按给定身份证；
|
||||
- 仅传 spec_models：这些型号下最近有流转的 20 台设备；
|
||||
- 都未传：返回空。
|
||||
|
||||
排序：人员按「最早介入该设备的真实时间」（received_at/created_at 最小值）升序堆叠。
|
||||
未参与某设备的人员返回 None（null），不补 0。
|
||||
"""
|
||||
from app.models.task import Task, TASK_STATUS_WIP, TASK_STATUS_PENDING, TASK_STATUS_COMPLETED
|
||||
from app.models.product import Product
|
||||
from app.core.time_utils import get_beijing_time
|
||||
|
||||
now = get_beijing_time()
|
||||
|
||||
if product_sns:
|
||||
product_rows = (await db.execute(
|
||||
select(
|
||||
Product.id, Product.serial_number, Product.external_serial,
|
||||
Product.material_name, Product.spec_model,
|
||||
).where(Product.serial_number.in_(product_sns))
|
||||
)).all()
|
||||
elif spec_models:
|
||||
latest_subq = (
|
||||
select(
|
||||
Task.product_id,
|
||||
func.max(func.coalesce(Task.received_at, Task.created_at)).label("latest"),
|
||||
)
|
||||
.group_by(Task.product_id)
|
||||
.subquery()
|
||||
)
|
||||
product_rows = (await db.execute(
|
||||
select(
|
||||
Product.id, Product.serial_number, Product.external_serial,
|
||||
Product.material_name, Product.spec_model,
|
||||
)
|
||||
.join(latest_subq, latest_subq.c.product_id == Product.id)
|
||||
.where(Product.spec_model.in_(spec_models))
|
||||
.order_by(latest_subq.c.latest.desc())
|
||||
.limit(20)
|
||||
)).all()
|
||||
else:
|
||||
return FlowResponse(devices=[], series=[])
|
||||
|
||||
if not product_rows:
|
||||
return FlowResponse(devices=[], series=[])
|
||||
|
||||
devices = [
|
||||
FlowDevice(
|
||||
product_sn=r[1], external_serial=r[2],
|
||||
material_name=r[3] or "", spec_model=r[4] or "",
|
||||
)
|
||||
for r in product_rows
|
||||
]
|
||||
# 显式传入身份证时按输入顺序排列
|
||||
if product_sns:
|
||||
order = {sn: i for i, sn in enumerate(product_sns)}
|
||||
devices.sort(key=lambda d: order.get(d.product_sn, len(order)))
|
||||
|
||||
id_to_sn = {r[0]: r[1] for r in product_rows}
|
||||
sn_to_index = {d.product_sn: i for i, d in enumerate(devices)}
|
||||
product_ids = [r[0] for r in product_rows]
|
||||
|
||||
# ── 这些设备的全部任务（按操作人聚合） ──
|
||||
task_rows = (await db.execute(
|
||||
select(
|
||||
Task.product_id, Task.assignee_id,
|
||||
Task.received_at, Task.created_at, Task.completed_at,
|
||||
)
|
||||
.where(
|
||||
Task.product_id.in_(product_ids),
|
||||
Task.status.in_([TASK_STATUS_WIP, TASK_STATUS_PENDING, TASK_STATUS_COMPLETED]),
|
||||
Task.assignee_id.isnot(None),
|
||||
)
|
||||
)).all()
|
||||
|
||||
# ── 聚合：人员(assignee_id) → {设备索引: 累计耗时}，同时记录人员最早介入时间 ──
|
||||
agg: dict[str, dict[int, float]] = {}
|
||||
person_earliest: dict[str, datetime] = {}
|
||||
for row in task_rows:
|
||||
pid, assignee_id, received_at, created_at, completed_at = row
|
||||
sn = id_to_sn.get(pid)
|
||||
if sn is None or sn not in sn_to_index:
|
||||
continue
|
||||
idx = sn_to_index[sn]
|
||||
start = _to_bj(received_at or created_at)
|
||||
end = _to_bj(completed_at) if completed_at else now
|
||||
agg.setdefault(assignee_id, {}).setdefault(idx, 0.0)
|
||||
agg[assignee_id][idx] += _duration_hours(start, end)
|
||||
if start and (person_earliest.get(assignee_id) is None or start < person_earliest[assignee_id]):
|
||||
person_earliest[assignee_id] = start
|
||||
|
||||
# 翻译人员姓名
|
||||
raw_ids = list(agg.keys())
|
||||
name_map: dict[str, str] = {}
|
||||
if raw_ids:
|
||||
from app.services.mom_cache import get_display_names
|
||||
name_map = get_display_names(raw_ids)
|
||||
|
||||
# ── 按人员最早介入时间升序排序（从下到上 = 谁先接手谁后接手） ──
|
||||
ordered_persons = sorted(agg.keys(), key=lambda p: person_earliest.get(p) or now)
|
||||
|
||||
series = [
|
||||
FlowSeries(
|
||||
name=name_map.get(p, p),
|
||||
data=[
|
||||
round(agg[p][i], 1) if i in agg[p] else None
|
||||
for i in range(len(devices))
|
||||
],
|
||||
)
|
||||
for p in ordered_persons
|
||||
]
|
||||
return FlowResponse(devices=devices, series=series)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 设备备注记录 — 单台设备的全部备注/照片（弹窗下钻）
|
||||
# ============================================================
|
||||
|
||||
async def get_device_records(
|
||||
db: AsyncSession,
|
||||
product_sn: str,
|
||||
assignee_ids: list[str] | None = None,
|
||||
) -> list[DeviceRecord]:
|
||||
"""查询某台设备（身份证）的所有任务备注记录，含图片，按时间倒序。
|
||||
可选按负责人过滤（assignee_ids）。"""
|
||||
import json
|
||||
from app.models.task import Task, TaskRecord
|
||||
from app.models.product import Product
|
||||
|
||||
product_id = await db.scalar(
|
||||
select(Product.id).where(Product.serial_number == product_sn)
|
||||
)
|
||||
if not product_id:
|
||||
return []
|
||||
|
||||
stmt = (
|
||||
select(
|
||||
TaskRecord.remark, TaskRecord.images, TaskRecord.created_at,
|
||||
Task.task_name, Task.assignee_id, Task.status,
|
||||
)
|
||||
.join(Task, TaskRecord.task_id == Task.id)
|
||||
.where(Task.product_id == product_id)
|
||||
)
|
||||
if assignee_ids:
|
||||
stmt = stmt.where(Task.assignee_id.in_(assignee_ids))
|
||||
stmt = stmt.order_by(TaskRecord.created_at.desc())
|
||||
result = await db.execute(stmt)
|
||||
rows = result.all()
|
||||
|
||||
raw_ids = list({r[4] for r in rows if r[4]})
|
||||
name_map: dict[str, str] = {}
|
||||
if raw_ids:
|
||||
from app.services.mom_cache import get_display_names
|
||||
name_map = get_display_names(raw_ids)
|
||||
|
||||
records: list[DeviceRecord] = []
|
||||
for row in rows:
|
||||
remark, images_raw, created, task_name, assignee, status = row
|
||||
try:
|
||||
images = json.loads(images_raw) if images_raw else []
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
images = []
|
||||
if not isinstance(images, list):
|
||||
images = []
|
||||
records.append(DeviceRecord(
|
||||
task_name=task_name or "",
|
||||
assignee_name=name_map.get(assignee or "", assignee or ""),
|
||||
status=status or "",
|
||||
remark=remark,
|
||||
images=images,
|
||||
created_at=_to_bj(created).isoformat() if created else "",
|
||||
))
|
||||
return records
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 下拉选项 — 负责人 + 规格型号
|
||||
# ============================================================
|
||||
|
||||
async def get_analytics_options(
|
||||
db: AsyncSession,
|
||||
assignee_ids: list[str] | None = None,
|
||||
spec_models: list[str] | None = None,
|
||||
) -> AnalyticsOptions:
|
||||
"""返回筛选栏选项，支持动态联动：
|
||||
- 传入 spec_models：只返回碰过这些型号的人；
|
||||
- 传入 assignee_ids：只返回这些人处理过的型号。"""
|
||||
from app.models.task import Task
|
||||
from app.models.product import Product
|
||||
|
||||
# 负责人：从 tasks 去重（可选按 spec_models 精确过滤）
|
||||
if spec_models:
|
||||
assignee_stmt = (
|
||||
select(Task.assignee_id)
|
||||
.join(Product, Task.product_id == Product.id)
|
||||
.where(
|
||||
Task.assignee_id.isnot(None),
|
||||
Product.spec_model.in_(spec_models),
|
||||
)
|
||||
.distinct()
|
||||
)
|
||||
else:
|
||||
assignee_stmt = (
|
||||
select(Task.assignee_id)
|
||||
.where(Task.assignee_id.isnot(None))
|
||||
.distinct()
|
||||
)
|
||||
assignee_rows = (await db.execute(assignee_stmt)).all()
|
||||
distinct_assignees = sorted({r[0] for r in assignee_rows if r[0]})
|
||||
|
||||
name_map: dict[str, str] = {}
|
||||
if distinct_assignees:
|
||||
from app.services.mom_cache import get_display_names
|
||||
name_map = get_display_names(distinct_assignees)
|
||||
|
||||
assignees = [
|
||||
AssigneeOption(id=aid, name=name_map.get(aid, aid))
|
||||
for aid in distinct_assignees
|
||||
]
|
||||
|
||||
# 规格型号：从 products 去重（可选按 assignee_ids 过滤）
|
||||
if assignee_ids:
|
||||
spec_stmt = (
|
||||
select(Product.spec_model)
|
||||
.join(Task, Task.product_id == Product.id)
|
||||
.where(
|
||||
Product.spec_model.isnot(None),
|
||||
func.trim(Product.spec_model) != "",
|
||||
Task.assignee_id.in_(assignee_ids),
|
||||
)
|
||||
)
|
||||
else:
|
||||
spec_stmt = (
|
||||
select(Product.spec_model)
|
||||
.where(Product.spec_model.isnot(None), func.trim(Product.spec_model) != "")
|
||||
)
|
||||
spec_rows = (await db.execute(spec_stmt.distinct())).all()
|
||||
spec_models = sorted({r[0] for r in spec_rows if r[0]})
|
||||
|
||||
return AnalyticsOptions(assignees=assignees, spec_models=spec_models)
|
||||
Reference in New Issue
Block a user