140 lines
4.4 KiB
Python
140 lines
4.4 KiB
Python
from typing import Any, Literal
|
||
|
||
from django.db.models import Q
|
||
|
||
from apps.wpmw.models import Wpr, WprDefect
|
||
from mcp_server.context import authenticated_factory_user
|
||
from mcp_server.tools.common import json_safe_result, validate_result_size
|
||
|
||
|
||
WprLocation = Literal["all", "workshop", "warehouse", "unassigned"]
|
||
|
||
|
||
def _base_queryset():
|
||
return Wpr.objects.select_related(
|
||
"material",
|
||
"material_start",
|
||
"wm",
|
||
"mb",
|
||
"wpr_from",
|
||
)
|
||
|
||
|
||
def _wpr_summary(wpr: Wpr) -> dict[str, Any]:
|
||
material = wpr.material
|
||
return {
|
||
"id": str(wpr.id),
|
||
"number": wpr.number,
|
||
"number_out": wpr.number_out,
|
||
"version": wpr.version,
|
||
"state": wpr.state,
|
||
"state_name": wpr.get_state_display(),
|
||
"material": {
|
||
"id": str(material.id),
|
||
"name": material.name,
|
||
"model": material.model,
|
||
"specification": material.specification,
|
||
},
|
||
"workshop_batch": wpr.wm.batch if wpr.wm_id else None,
|
||
"warehouse_batch": wpr.mb.batch if wpr.mb_id else None,
|
||
"create_time": wpr.create_time,
|
||
"update_time": wpr.update_time,
|
||
}
|
||
|
||
|
||
def search_wprs(
|
||
query: str = "",
|
||
state: int | None = None,
|
||
material_id: str | None = None,
|
||
batch: str = "",
|
||
location: WprLocation = "all",
|
||
limit: int = 20,
|
||
) -> dict[str, Any]:
|
||
"""按编号、物料或批次搜索单件产品；仅提供只读摘要。"""
|
||
authenticated_factory_user()
|
||
if location not in {"all", "workshop", "warehouse", "unassigned"}:
|
||
raise ValueError(f"不支持的 WPR 位置：{location}")
|
||
safe_limit = max(1, min(limit, 100))
|
||
queryset = _base_queryset()
|
||
if query.strip():
|
||
keyword = query.strip()
|
||
queryset = queryset.filter(
|
||
Q(number__icontains=keyword)
|
||
| Q(number_out__icontains=keyword)
|
||
| Q(material__name__icontains=keyword)
|
||
| Q(material__model__icontains=keyword)
|
||
| Q(material__specification__icontains=keyword)
|
||
)
|
||
if state is not None:
|
||
queryset = queryset.filter(state=state)
|
||
if material_id:
|
||
queryset = queryset.filter(material_id=material_id)
|
||
if batch.strip():
|
||
queryset = queryset.filter(
|
||
Q(wm__batch__icontains=batch.strip())
|
||
| Q(mb__batch__icontains=batch.strip())
|
||
)
|
||
if location == "workshop":
|
||
queryset = queryset.filter(wm__isnull=False)
|
||
elif location == "warehouse":
|
||
queryset = queryset.filter(mb__isnull=False)
|
||
elif location == "unassigned":
|
||
queryset = queryset.filter(wm__isnull=True, mb__isnull=True)
|
||
|
||
items = [
|
||
_wpr_summary(wpr)
|
||
for wpr in queryset.distinct().order_by("number", "create_time")[:safe_limit]
|
||
]
|
||
result = json_safe_result({"items": items, "limit": safe_limit})
|
||
validate_result_size(result)
|
||
return result
|
||
|
||
|
||
def get_wpr(identifier: str) -> dict[str, Any]:
|
||
"""按 WPR ID、内部编号或对外编号读取单件详情。"""
|
||
authenticated_factory_user()
|
||
lookup = Q(number=identifier) | Q(number_out=identifier)
|
||
if identifier.isdigit():
|
||
lookup |= Q(pk=identifier)
|
||
wpr = _base_queryset().filter(lookup).order_by("-version", "-update_time").first()
|
||
if wpr is None:
|
||
raise ValueError(f"未找到 WPR：{identifier}")
|
||
|
||
result = _wpr_summary(wpr)
|
||
material_start = wpr.material_start
|
||
result.update(
|
||
{
|
||
"material_start": (
|
||
{
|
||
"id": str(material_start.id),
|
||
"name": material_start.name,
|
||
"model": material_start.model,
|
||
"specification": material_start.specification,
|
||
}
|
||
if material_start
|
||
else None
|
||
),
|
||
"wpr_from": (
|
||
{"id": str(wpr.wpr_from.id), "number": wpr.wpr_from.number}
|
||
if wpr.wpr_from_id
|
||
else None
|
||
),
|
||
"oinfo": wpr.oinfo,
|
||
"data": wpr.data,
|
||
"pre_info": wpr.pre_info,
|
||
"defects": list(
|
||
WprDefect.objects.filter(wpr=wpr)
|
||
.select_related("defect")
|
||
.values("defect_id", "defect__name", "is_main")
|
||
),
|
||
}
|
||
)
|
||
result = json_safe_result(result)
|
||
validate_result_size(result)
|
||
return result
|
||
|
||
|
||
def register_wpr_tools(server) -> None:
|
||
server.tool()(search_wprs)
|
||
server.tool()(get_wpr)
|