duxingchen/track
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(backend): 修复并发任务聚合——当前位置汇总全部负责人、最新动态携带操作人、状态聚合补齐 REJECTED

Browse Source
This commit is contained in:
duxingchen
2026-08-13 11:17:10 +08:00
parent 3288223a92
commit e5829c7d06
2 changed files with 44 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/app/schemas/product.py
View File

@ -59,6 +59,8 @@ class ProductResponse(BaseModel):
latest_record_time: datetime | None = None
latest_record_content: str | None = None
latest_record_has_images: bool = False
latest_record_assignee_id: str | None = None # 🔧 最新记录操作人（消除并发张冠李戴）
latest_record_assignee_name: str | None = None
model_config = {"from_attributes": True}

67
backend/app/services/product_service.py
View File

@ -439,6 +439,7 @@ async def get_all_products(
sa_func.max(case(
(Task.status == "WIP", 3),
(Task.status == "PENDING", 2),
(Task.status == "REJECTED", 2),
(Task.status == "COMPLETED", 1),
(Task.status == "ARCHIVED", 1),
else_=0,
@ -452,9 +453,8 @@ async def get_all_products(
for row in task_result:
macro_map[row[0]] = prio_to_status.get(row[1], None)
# 🔧 动态主干状态+位置：只从主干任务中获取最高优先级任务的 task_name + assignee_id
# 🔧 动态主干状态名：只从主干任务中获取最高优先级任务的 task_name（宏观状态名）
overall_names: dict[uuid.UUID, str] = {}
main_assignees: dict[uuid.UUID, str] = {}
if product_ids:
from sqlalchemy import and_, func as sa_func, case as sa_case
main_where = and_(
@ -470,15 +470,13 @@ async def get_all_products(
(Task.status == "COMPLETED", 1),
else_=0,
)
# 子查询：每个产品最高优先级主干任务
max_prio = (
select(Task.product_id, sa_func.max(prio_expr).label("prio"))
.where(main_where)
.group_by(Task.product_id)
).subquery("mp")
# JOIN 回 tasks 拿 task_name + assignee_id（同优先级取最新创建的）
main_stmt = (
select(Task.product_id, Task.task_name, Task.assignee_id)
select(Task.product_id, Task.task_name)
.join(max_prio, and_(
Task.product_id == max_prio.c.product_id,
prio_expr == max_prio.c.prio,
@ -489,20 +487,34 @@ async def get_all_products(
)
main_result = await db.execute(main_stmt)
for row in main_result:
pid, tname, assignee = row[0], row[1], row[2]
overall_names[pid] = tname
if assignee: main_assignees[pid] = assignee
overall_names[row[0]] = row[1]
# 🔧 动态主干的 assignee_id → 查中文姓名
dynamic_location_ids = list(main_assignees.values())
dynamic_name_map = _lookup_display_names(dynamic_location_ids)
# 🔧 当前位置：汇总所有活跃任务（WIP/PENDING，不分主线/分支）的负责人，去重保序
active_assignees_map: dict[uuid.UUID, list[str]] = {}
if product_ids:
active_stmt = (
select(Task.product_id, Task.assignee_id)
.where(
Task.product_id.in_(product_ids),
Task.status.in_(["WIP", "PENDING"]),
Task.assignee_id.isnot(None),
)
.order_by(Task.product_id, Task.created_at)
)
active_result = await db.execute(active_stmt)
for row in active_result:
pid, assignee = row[0], row[1]
lst = active_assignees_map.setdefault(pid, [])
if assignee not in lst:
lst.append(assignee)
# 🔧 合并：静态位置姓名（兜底）+ 动态主干位置姓名（优先）
# 🔧 活跃负责人 + 静态位置（兜底）的 username → 中文姓名（一次批量查）
all_active_ids = [uid for ids in active_assignees_map.values() for uid in ids]
static_location_ids = [p.current_location_id for p in products if p.current_location_id]
merged_location_ids = list(set(static_location_ids + dynamic_location_ids))
merged_location_ids = list(set(all_active_ids + static_location_ids))
merged_name_map = _lookup_display_names(merged_location_ids)
# 🔧 批量查询每个产品活跃任务的最新记录
# 🔧 批量查询每个产品活跃任务的最新记录（含操作人 assignee_id）
latest_record_map: dict[uuid.UUID, tuple] = {}
if product_ids:
from app.models.task import TaskRecord as TR
@ -513,7 +525,7 @@ async def get_all_products(
)
).subquery()
ranked = (
select(TR.task_id, TR.remark, TR.images, TR.created_at, Task.product_id,
select(TR.task_id, TR.remark, TR.images, TR.created_at, Task.product_id, Task.assignee_id,
sa_func.row_number().over(
partition_by=Task.product_id,
order_by=TR.created_at.desc()
@ -522,12 +534,12 @@ async def get_all_products(
.where(Task.id.in_(select(wip_pending_ids.c.id)))
).subquery()
rec_result = await db.execute(
select(ranked.c.product_id, ranked.c.created_at, ranked.c.remark, ranked.c.images)
select(ranked.c.product_id, ranked.c.created_at, ranked.c.remark, ranked.c.images, ranked.c.assignee_id)
.where(ranked.c.rn == 1)
)
for row in rec_result:
has_img = bool(row[3] and row[3] != "[]" and row[3] != "null")
latest_record_map[row[0]] = (row[1], row[2], has_img)
latest_record_map[row[0]] = (row[1], row[2], has_img, row[4])
return [
ProductResponse(
@ -543,21 +555,26 @@ async def get_all_products(
material_type=p.material_type,
parent_product_id=p.parent_product_id,
current_location_id=(
main_assignees.get(p.id) or p.current_location_id
",".join(active_assignees_map.get(p.id, [])) or p.current_location_id
),
current_location_name=(
"仓库" if (main_assignees.get(p.id) or p.current_location_id) == "virtual_warehouse"
else dynamic_name_map.get(main_assignees.get(p.id, ""))
or merged_name_map.get(p.current_location_id) if p.current_location_id
else None
", ".join(merged_name_map.get(uid, uid) for uid in active_assignees_map.get(p.id, []))
if active_assignees_map.get(p.id)
else ("仓库" if p.current_location_id == "virtual_warehouse"
else merged_name_map.get(p.current_location_id) if p.current_location_id else None)
),
macro_status=macro_map.get(p.id) or p.status,
overall_status=overall_names.get(p.id) or p.overall_status,
status=p.status,
created_at=p.created_at,
latest_record_time=latest_record_map.get(p.id, (None, None, False))[0],
latest_record_content=latest_record_map.get(p.id, (None, None, False))[1],
latest_record_has_images=latest_record_map.get(p.id, (None, None, False))[2],
latest_record_time=latest_record_map.get(p.id, (None, None, False, None))[0],
latest_record_content=latest_record_map.get(p.id, (None, None, False, None))[1],
latest_record_has_images=latest_record_map.get(p.id, (None, None, False, None))[2],
latest_record_assignee_id=latest_record_map.get(p.id, (None, None, False, None))[3],
latest_record_assignee_name=(
merged_name_map.get(latest_record_map.get(p.id, (None, None, False, None))[3])
if latest_record_map.get(p.id, (None, None, False, None))[3] else None
),
)
for p in products
]