duxingchen
394e1e39c3
这三组接口此前是「上帝视角」且**匿名可访问**，现在全部挂 get_data_scope —— 既要求登录、又按业务分组范围过滤。顺带堵上了 AGENTS.md 点名的风险： /dashboard/people-history/export 此前匿名即可批量导出全员工时台账。 dashboard（11 个函数 / 24 处注入） · Product 主体 → product_where()；Task、TaskLog 主体 → 先 join(Product) 再 task_where() · 「卡片数字」与「下钻明细」成对出现的地方用同一谓词，避免「按钮显示 2、点开却是 0 条」 · get_user_operations 的 func.count() 改为 func.count(TaskLog.id) —— 显式化，不依赖 join 形状（当前是 many-to-one 不会放大，但这样写更稳） · unread_notif **刻意不过滤**：Notification.task_id 可空，按 Product 过滤会漏掉 无任务关联的提醒（就地注释说明） · get_my_stats 本轮不动 —— 它按本人归因，语义上不受分组影响 analytics（4 个函数 / 9 条语句） · get_analytics_options 的 4 条独立语句全部处理 —— 它是筛选栏下拉的选项源， 不过滤的话维修组能在下拉里看到生产组的人（最易漏的一处） · get_device_records 的 product_id 查号是安全闸：范围外 SN 查不出 → 直接返回 [] screen（3 个函数） · month_production **不做特判** —— 维修组的「本月生产数」本来就该是 0 · get_wip_distribution 按范围裁剪工序柱子，但坚持「恒 0 才裁、有数必现」， 保证 total == sum(items) 在任何 scope 下都成立 实测（17 个端点）：超管全部 200、匿名全部 401。 造 1 生产 + 1 售后产品后： 超管 products=2 / wip-matrix 2 行 / options 2 个型号 生产组 products=1 / wip-matrix 1 行 / options 1 个型号 维修组 products=1 / wip-matrix 1 行 / options 1 个型号 列表与统计口径一致；未分组用户在过渡期开关下仍走 ungrouped_fallback。 测试数据已还原。
727 lines
28 KiB
Python
727 lines
28 KiB
Python
"""效能分析服务 — 个人能力图谱 + 设备流转对比（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 处理，统一换算北京时间。
|
||
|
||
2026-09 起不再是「上帝视角」：统计函数一律接收调用方解析好的 DataScope，
|
||
查询里按 Product.lifecycle_phase 过滤（谓词只在 data_scope_service 生成，见该模块红线）。
|
||
"""
|
||
from datetime import datetime, timezone
|
||
|
||
from sqlalchemy import select, func
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from pydantic import BaseModel
|
||
|
||
from app.services.data_scope_service import DataScope
|
||
|
||
|
||
# ============================================================
|
||
# 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 # 规格型号
|
||
material_name: 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
|
||
lead_time: float # 设备生命周期总时长（小时）= max_end - min_start
|
||
started_at: str # 设备最早介入时间（T0），格式 MM-DD HH:mm
|
||
total_days: int = 0 # 设备生产总天数（自然天，自最早介入至今）
|
||
total_workdays: int = 0 # 设备生产总天数（工作日，排除周末/节假日）
|
||
|
||
|
||
class FlowSeries(BaseModel):
|
||
name: str # 人员姓名
|
||
data: list[list] # 每项 = [device_index, start_offset, end_offset, task_name, duration]（小时）
|
||
|
||
|
||
class FlowResponse(BaseModel):
|
||
devices: list[FlowDevice]
|
||
series: list[FlowSeries]
|
||
|
||
|
||
class AssigneeOption(BaseModel):
|
||
id: str
|
||
name: str
|
||
|
||
|
||
class SpecModelOption(BaseModel):
|
||
spec_model: str
|
||
material_name: str
|
||
|
||
|
||
class DeviceOption(BaseModel):
|
||
product_sn: str
|
||
external_serial: str | None
|
||
material_name: str
|
||
spec_model: str
|
||
|
||
|
||
class AnalyticsOptions(BaseModel):
|
||
assignees: list[AssigneeOption]
|
||
spec_models: list[SpecModelOption]
|
||
devices: list[DeviceOption]
|
||
|
||
|
||
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, holidays: set = None, mode: str = "workdays") -> float:
|
||
"""计算单台耗时（小时），无开始时间返回 0。
|
||
mode=workdays: 排除周末/节假日的工作小时；mode=natural: 自然小时。"""
|
||
if not start:
|
||
return 0.0
|
||
if mode == "natural":
|
||
return (end - start).total_seconds() / 3600
|
||
from app.core.time_utils import working_duration_hours
|
||
return working_duration_hours(start, end, holidays)
|
||
|
||
|
||
# ============================================================
|
||
# 个人能力图谱 — 单台设备耗时对比（分组柱状图，X=设备身份证）
|
||
# ============================================================
|
||
|
||
async def get_capability_profile(
|
||
db: AsyncSession,
|
||
scope: DataScope,
|
||
assignee_ids: list[str] | None = None,
|
||
spec_models: list[str] | None = None,
|
||
since: datetime | None = None,
|
||
until: datetime | None = None,
|
||
mode: str = "workdays",
|
||
) -> 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.models.holiday import Holiday
|
||
from app.core.time_utils import get_beijing_time
|
||
|
||
now = get_beijing_time()
|
||
|
||
# 读取节假日（排除非工作日）
|
||
hres = await db.execute(select(Holiday.day))
|
||
holidays = {r[0] for r in hres}
|
||
|
||
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,
|
||
Product.material_name,
|
||
)
|
||
.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),
|
||
scope.task_where(), # 🚀 业务分组数据范围（已 join Product）
|
||
)
|
||
)
|
||
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 "未知型号"
|
||
mat = row[8] or ""
|
||
|
||
meta = device_meta.setdefault(sn, {"external_serial": ext, "spec_model": spec, "material_name": mat, "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, holidays, mode)
|
||
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"],
|
||
material_name=device_meta[sn]["material_name"],
|
||
)
|
||
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,
|
||
scope: DataScope,
|
||
product_sns: list[str] | None = None,
|
||
spec_models: list[str] | None = None,
|
||
mode: str = "natural",
|
||
since: datetime | None = None,
|
||
until: datetime | None = None,
|
||
) -> FlowResponse:
|
||
"""
|
||
查询每台设备上各操作人的任务时间区间，拼装为「生命周期时间轴」区间图：
|
||
x 轴 = 设备身份证，y 轴 = 相对该设备 T0（最早介入时间）的小时偏移。
|
||
每个任务一根悬空区间柱（start_offset -> end_offset），并行任务可并排显示。
|
||
|
||
设备来源：
|
||
- 传 product_sns：按给定身份证；
|
||
- 仅传 spec_models：这些型号下最近有流转的 20 台设备；
|
||
- 都未传：返回空。
|
||
|
||
data 每项 = [device_index, task_name, is_main,
|
||
start_nat, end_nat, start_work, end_work,
|
||
duration_total(自然), duration_work(工作日)]（单位小时）。
|
||
前端按 isWorkday 切换选取自然/工作日偏移，即时重绘无需重新请求。
|
||
mode 参数保留兼容（历史调用），不再影响返回内容。
|
||
"""
|
||
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，直接挂 product_where
|
||
# （范围外的身份证查不出来 → 与下钻接口同口径，不会泄露别组设备）
|
||
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), scope.product_where())
|
||
)).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), scope.product_where()) # 🚀 业务分组数据范围（主体 Product）
|
||
.order_by(latest_subq.c.latest.desc())
|
||
.limit(20)
|
||
)).all()
|
||
else:
|
||
return FlowResponse(devices=[], series=[])
|
||
|
||
if not product_rows:
|
||
return FlowResponse(devices=[], series=[])
|
||
|
||
id_to_sn = {r[0]: r[1] for r in product_rows}
|
||
product_ids = [r[0] for r in product_rows]
|
||
|
||
# ── 候选设备的全部任务（含工序名，用于区间图） ──
|
||
# 🚀 业务分组数据范围：product_ids 已来自上方受限结果，这里把口径写全
|
||
# （Task 主体，必须先 join Product 再挂 task_where；结果集不变）
|
||
task_rows = (await db.execute(
|
||
select(
|
||
Task.product_id, Task.assignee_id, Task.task_name,
|
||
Task.received_at, Task.created_at, Task.completed_at,
|
||
Task.task_type,
|
||
)
|
||
.join(Product, Task.product_id == Product.id)
|
||
.where(
|
||
Task.product_id.in_(product_ids),
|
||
Task.status.in_([TASK_STATUS_WIP, TASK_STATUS_PENDING, TASK_STATUS_COMPLETED]),
|
||
Task.assignee_id.isnot(None),
|
||
scope.task_where(),
|
||
)
|
||
)).all()
|
||
|
||
# ── 设备级时间筛选 ──
|
||
# 语义：时间筛选 = "在该时间有活动的设备"，而非"把任务切片只保留该时间"。
|
||
# 设备一旦入选，其任务必须完整返回（完整生命周期），绝不因设备在筛选中无新动作
|
||
# 而把它的历史任务清空。
|
||
if since is not None or until is not None:
|
||
active_pids: set = set()
|
||
for row in task_rows:
|
||
pid = row[0]
|
||
start = _to_bj(row[3] or row[4])
|
||
end = _to_bj(row[5]) if row[5] else now
|
||
if start is None:
|
||
continue
|
||
if since is not None and end < since:
|
||
continue
|
||
if until is not None and start > until:
|
||
continue
|
||
active_pids.add(pid)
|
||
product_rows = [r for r in product_rows if r[0] in active_pids]
|
||
if not product_rows:
|
||
return FlowResponse(devices=[], series=[])
|
||
|
||
# 显式传入身份证时按输入顺序排列
|
||
if product_sns:
|
||
order = {sn: i for i, sn in enumerate(product_sns)}
|
||
product_rows.sort(key=lambda r: order.get(r[1], len(order)))
|
||
devices = [
|
||
FlowDevice(
|
||
product_sn=r[1], external_serial=r[2],
|
||
material_name=r[3] or "", spec_model=r[4] or "",
|
||
lead_time=0.0,
|
||
started_at="",
|
||
)
|
||
for r in product_rows
|
||
]
|
||
sn_to_index = {d.product_sn: i for i, d in enumerate(devices)}
|
||
kept_ids = {r[0] for r in product_rows}
|
||
|
||
# ── 解析为区间记录（完整生命周期，不再按时间切片） ──
|
||
intervals: list[dict] = []
|
||
t0_by_device: dict[int, datetime] = {}
|
||
max_end_by_device: dict[int, datetime] = {}
|
||
for row in task_rows:
|
||
pid, assignee_id, task_name, received_at, created_at, completed_at, task_type = row
|
||
if pid not in kept_ids:
|
||
continue
|
||
sn = id_to_sn[pid]
|
||
idx = sn_to_index[sn]
|
||
start = _to_bj(received_at or created_at)
|
||
end = _to_bj(completed_at) if completed_at else now
|
||
if start is None:
|
||
continue
|
||
is_main = 0 if task_type == "SPAWN" else 1
|
||
intervals.append({
|
||
"idx": idx,
|
||
"assignee_id": assignee_id,
|
||
"task_name": (task_name or "").strip() or "未命名工序",
|
||
"start": start,
|
||
"end": end,
|
||
"is_main": is_main,
|
||
})
|
||
if idx not in t0_by_device or start < t0_by_device[idx]:
|
||
t0_by_device[idx] = start
|
||
if idx not in max_end_by_device or end > max_end_by_device[idx]:
|
||
max_end_by_device[idx] = end
|
||
|
||
# ── 计算每台设备 lead_time（最大 end - 最小 start，小时），重建 devices ──
|
||
lead_time_by_index = {
|
||
idx: round((max_end_by_device[idx] - t0_by_device[idx]).total_seconds() / 3600, 1)
|
||
for idx in t0_by_device
|
||
}
|
||
# 🔧 设备生产总天数（自然天 + 工作日，自最早介入至今）
|
||
import math
|
||
from app.core.time_utils import to_beijing as _tb, working_duration_hours as _wdh
|
||
from app.models.holiday import Holiday as _Holiday
|
||
hres = await db.execute(select(_Holiday.day))
|
||
_holidays = {r[0] for r in hres}
|
||
|
||
def _total_days(t0):
|
||
t0_bj = _tb(t0)
|
||
if not t0_bj:
|
||
return 1, 1
|
||
natural = max(1, math.ceil((now - t0_bj).total_seconds() / 86400))
|
||
workdays = max(1, math.ceil(_wdh(t0_bj, now, _holidays) / 24))
|
||
return natural, workdays
|
||
|
||
days_map = {i: _total_days(t0_by_device[i]) for i in t0_by_device}
|
||
devices = [
|
||
FlowDevice(
|
||
product_sn=d.product_sn, external_serial=d.external_serial,
|
||
material_name=d.material_name, spec_model=d.spec_model,
|
||
lead_time=lead_time_by_index.get(i, 0.0),
|
||
started_at=t0_by_device[i].strftime("%m-%d %H:%M") if i in t0_by_device else "",
|
||
total_days=days_map.get(i, (1, 1))[0],
|
||
total_workdays=days_map.get(i, (1, 1))[1],
|
||
)
|
||
for i, d in enumerate(devices)
|
||
]
|
||
|
||
# ── 按人分组，转为相对 T0 的小时偏移区间 ──
|
||
# 每项同时携带自然天与工作日两套偏移/时长，供前端按 isWorkday 即时切换：
|
||
# [idx, task_name, is_main, start_nat, end_nat, start_work, end_work, dur_total, dur_work]
|
||
from app.core.time_utils import working_duration_hours as _wdh
|
||
from app.models.holiday import Holiday as _Holiday
|
||
hres = await db.execute(select(_Holiday.day))
|
||
_holidays = {r[0] for r in hres}
|
||
by_assignee: dict[str, list[list]] = {}
|
||
for it in intervals:
|
||
t0 = t0_by_device[it["idx"]]
|
||
start_nat = round((it["start"] - t0).total_seconds() / 3600, 1)
|
||
end_nat = round((it["end"] - t0).total_seconds() / 3600, 1)
|
||
start_work = round(_wdh(t0, it["start"], _holidays), 1)
|
||
end_work = round(_wdh(t0, it["end"], _holidays), 1)
|
||
by_assignee.setdefault(it["assignee_id"], []).append(
|
||
[it["idx"], it["task_name"], it["is_main"],
|
||
start_nat, end_nat, start_work, end_work,
|
||
round(end_nat - start_nat, 1), round(end_work - start_work, 1)]
|
||
)
|
||
|
||
# 翻译人员姓名
|
||
raw_ids = list(by_assignee.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)
|
||
|
||
series = [
|
||
FlowSeries(name=name_map.get(a, a), data=by_assignee[a])
|
||
for a in sorted(by_assignee.keys())
|
||
]
|
||
return FlowResponse(devices=devices, series=series)
|
||
|
||
|
||
# ============================================================
|
||
# 设备备注记录 — 单台设备的全部备注/照片（弹窗下钻）
|
||
# ============================================================
|
||
|
||
async def get_device_records(
|
||
db: AsyncSession,
|
||
scope: DataScope,
|
||
product_sn: str,
|
||
assignee_ids: list[str] | None = None,
|
||
) -> list[DeviceRecord]:
|
||
"""查询某台设备（身份证）的所有任务备注记录，含图片，按时间倒序。
|
||
可选按负责人过滤（assignee_ids）。
|
||
|
||
按当前用户的业务分组数据范围过滤 —— 范围外的身份证查不出 product_id，
|
||
直接返回空列表（否则维修组拿生产组的身份证就能读到别组的备注与照片）。
|
||
"""
|
||
import json
|
||
from app.models.task import Task, TaskRecord
|
||
from app.models.product import Product
|
||
|
||
# 🚀 业务分组数据范围：主体 Product，范围外查不出 product_id → 下方直接空返回
|
||
product_id = await db.scalar(
|
||
select(Product.id).where(
|
||
Product.serial_number == product_sn,
|
||
scope.product_where(),
|
||
)
|
||
)
|
||
if not product_id:
|
||
return []
|
||
|
||
# 🚀 业务分组数据范围：product_id 已受限，这里把口径写全（Task 主体，先 join Product）
|
||
stmt = (
|
||
select(
|
||
TaskRecord.remark, TaskRecord.images, TaskRecord.created_at,
|
||
Task.task_name, Task.assignee_id, Task.status,
|
||
)
|
||
.join(Task, TaskRecord.task_id == Task.id)
|
||
.join(Product, Task.product_id == Product.id)
|
||
.where(Task.product_id == product_id, scope.task_where())
|
||
)
|
||
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,
|
||
scope: DataScope,
|
||
assignee_ids: list[str] | None = None,
|
||
spec_models: list[str] | None = None,
|
||
) -> AnalyticsOptions:
|
||
"""返回筛选栏选项，支持动态联动：
|
||
- 传入 spec_models：只返回碰过这些型号的人；
|
||
- 传入 assignee_ids：只返回这些人处理过的型号；
|
||
- devices：根据筛选条件返回关联设备（无筛选则返回最近流转的设备）。
|
||
|
||
⚠️ 本函数是**筛选栏下拉的选项源**，四条语句（负责人 / 型号 / 设备，各含
|
||
带筛选与不带筛选两个分支）必须条条挂上数据范围谓词 —— 漏掉任意一条，
|
||
维修组就能在下拉里看到生产组的人与型号，点开却是空的。
|
||
"""
|
||
from app.models.task import Task
|
||
from app.models.product import Product
|
||
|
||
# ── 负责人：从 tasks 去重（可选按 spec_models 过滤） ──
|
||
if spec_models:
|
||
# 🚀 业务分组数据范围：已 join Product，直接挂 task_where
|
||
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),
|
||
scope.task_where(),
|
||
)
|
||
.distinct()
|
||
)
|
||
else:
|
||
# 🚀 业务分组数据范围：本分支原先没有 join，必须补上 Product 才能挂
|
||
# task_where（tasks 表没有 lifecycle_phase）
|
||
assignee_stmt = (
|
||
select(Task.assignee_id)
|
||
.join(Product, Task.product_id == Product.id)
|
||
.where(
|
||
Task.assignee_id.isnot(None),
|
||
scope.task_where(),
|
||
)
|
||
.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
|
||
]
|
||
|
||
# ── 规格型号 + 物料名（可选按 assignee_ids 过滤；同型号取首个非空物料名） ──
|
||
if assignee_ids:
|
||
# 🚀 业务分组数据范围：主体 Product，直接挂 product_where
|
||
spec_stmt = (
|
||
select(Product.spec_model, Product.material_name)
|
||
.join(Task, Task.product_id == Product.id)
|
||
.where(
|
||
Product.spec_model.isnot(None),
|
||
func.trim(Product.spec_model) != "",
|
||
Task.assignee_id.in_(assignee_ids),
|
||
scope.product_where(),
|
||
)
|
||
.distinct()
|
||
)
|
||
else:
|
||
# 🚀 业务分组数据范围：主体 Product，直接挂 product_where
|
||
spec_stmt = (
|
||
select(Product.spec_model, Product.material_name)
|
||
.where(
|
||
Product.spec_model.isnot(None),
|
||
func.trim(Product.spec_model) != "",
|
||
scope.product_where(),
|
||
)
|
||
.distinct()
|
||
)
|
||
spec_rows = (await db.execute(spec_stmt)).all()
|
||
spec_map: dict[str, str] = {}
|
||
for row in spec_rows:
|
||
sm = row[0]
|
||
mn = (row[1] or "").strip()
|
||
if sm not in spec_map:
|
||
spec_map[sm] = mn
|
||
elif mn and not spec_map[sm]:
|
||
spec_map[sm] = mn
|
||
spec_model_opts = [
|
||
SpecModelOption(spec_model=sm, material_name=spec_map[sm])
|
||
for sm in sorted(spec_map.keys())
|
||
]
|
||
|
||
# ── 设备字典（可选按 assignee_ids / spec_models 过滤；无筛选返回最近流转 100 台） ──
|
||
if assignee_ids or spec_models:
|
||
# 🚀 业务分组数据范围：主体 Product，直接挂 product_where
|
||
# （即使上面 join 了 Task，产品去重后条数仍由 Product 决定）
|
||
device_stmt = select(
|
||
Product.serial_number, Product.external_serial,
|
||
Product.material_name, Product.spec_model,
|
||
).where(scope.product_where())
|
||
if assignee_ids:
|
||
device_stmt = device_stmt.join(
|
||
Task, Task.product_id == Product.id
|
||
).where(Task.assignee_id.in_(assignee_ids))
|
||
if spec_models:
|
||
device_stmt = device_stmt.where(Product.spec_model.in_(spec_models))
|
||
device_stmt = device_stmt.distinct()
|
||
else:
|
||
latest_subq = (
|
||
select(
|
||
Task.product_id,
|
||
func.max(func.coalesce(Task.received_at, Task.created_at)).label("latest"),
|
||
)
|
||
.group_by(Task.product_id)
|
||
.subquery()
|
||
)
|
||
device_stmt = (
|
||
select(
|
||
Product.serial_number, Product.external_serial,
|
||
Product.material_name, Product.spec_model,
|
||
)
|
||
.join(latest_subq, latest_subq.c.product_id == Product.id)
|
||
.where(scope.product_where()) # 🚀 业务分组数据范围（主体 Product）
|
||
.order_by(latest_subq.c.latest.desc())
|
||
.limit(100)
|
||
)
|
||
device_rows = (await db.execute(device_stmt)).all()
|
||
devices = [
|
||
DeviceOption(
|
||
product_sn=r[0] or "",
|
||
external_serial=r[1] or None,
|
||
material_name=r[2] or "",
|
||
spec_model=r[3] or "",
|
||
)
|
||
for r in device_rows
|
||
]
|
||
|
||
return AnalyticsOptions(assignees=assignees, spec_models=spec_model_opts, devices=devices)
|