feat(manufacturing): enrich inventory list with customer/user details and filters

The inventory list and detail page previously only showed raw
customer_id / user_list references. list_devices now batch-resolves
customer names and user display names/emails (via Firestore get_all(),
avoiding an N+1 round trip per device) and exposes them as
customer_name / users on DeviceInventoryItem. Search now matches
against device name, console name, customer name, and assigned users
in addition to serial/owner/batch.

Also adds hw_types (multi-select board type) and has_users filters to
GET /manufacturing/devices, and carries console_name through for
display. DeviceInventoryDetail is updated to show and use all of this
(customer name, assigned users, new UI components for signal/charts).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-21 18:34:08 +03:00
co-authored by Claude Sonnet 5
parent b07bdafd85
commit 6686faad6b
4 changed files with 197 additions and 27 deletions
+77 -6
View File
@@ -11,9 +11,11 @@ from shared.firebase import get_db
from shared.exceptions import NotFoundError
from utils.serial_number import generate_serial
from utils.nvs_generator import generate as generate_nvs_binary
from manufacturing.models import BatchCreate, BatchResponse, DeviceInventoryItem, DeviceStatusUpdate, DeviceAssign, ManufacturingStats, RecentActivityItem, BOARD_TYPE_LABELS
from manufacturing.models import BatchCreate, BatchResponse, DeviceInventoryItem, DeviceStatusUpdate, DeviceAssign, ManufacturingStats, RecentActivityItem, BOARD_TYPE_LABELS, DeviceUserSummary
COLLECTION = "devices"
CRM_COLLECTION = "crm_customers"
USERS_COLLECTION = "users"
_BATCH_ID_CHARS = string.ascii_uppercase + string.digits
@@ -45,7 +47,7 @@ def _resolve_user_list(raw_list: list) -> list[str]:
return result
def _doc_to_inventory_item(doc) -> DeviceInventoryItem:
def _doc_to_inventory_item(doc, customer_names: dict | None = None, user_infos: dict | None = None) -> DeviceInventoryItem:
data = doc.to_dict() or {}
created_raw = data.get("created_at")
if isinstance(created_raw, datetime):
@@ -53,6 +55,14 @@ def _doc_to_inventory_item(doc) -> DeviceInventoryItem:
else:
created_str = str(created_raw) if created_raw else None
customer_names = customer_names or {}
user_infos = user_infos or {}
user_ids = _resolve_user_list(data.get("user_list") or [])
users = [user_infos[uid] for uid in user_ids if uid in user_infos]
customer_id = data.get("customer_id")
return DeviceInventoryItem(
id=doc.id,
serial_number=data.get("serial_number", ""),
@@ -64,12 +74,48 @@ def _doc_to_inventory_item(doc) -> DeviceInventoryItem:
owner=data.get("owner"),
assigned_to=data.get("assigned_to"),
device_name=data.get("device_name") or None,
console_name=data.get("console_name") or None,
lifecycle_history=data.get("lifecycle_history") or [],
customer_id=data.get("customer_id"),
user_list=_resolve_user_list(data.get("user_list") or []),
customer_id=customer_id,
customer_name=customer_names.get(customer_id) if customer_id else None,
user_list=user_ids,
users=users,
)
def _bulk_resolve_names(db, docs_data: list[dict]) -> tuple[dict, dict]:
"""Batch-resolve customer names and user summaries referenced across a
list of device docs, so listing N devices costs a handful of get_all()
calls instead of an N+1 Firestore round trip per device."""
customer_ids = {d.get("customer_id") for d in docs_data if d.get("customer_id")}
user_ids = {uid for d in docs_data for uid in _resolve_user_list(d.get("user_list") or [])}
customer_names: dict[str, str] = {}
if customer_ids:
refs = [db.collection(CRM_COLLECTION).document(cid) for cid in customer_ids]
for snap in db.get_all(refs):
if not snap.exists:
continue
cdata = snap.to_dict() or {}
name = " ".join(filter(None, [cdata.get("name"), cdata.get("surname")])) or cdata.get("email") or snap.id
customer_names[snap.id] = name
user_infos: dict[str, DeviceUserSummary] = {}
if user_ids:
refs = [db.collection(USERS_COLLECTION).document(uid) for uid in user_ids]
for snap in db.get_all(refs):
if not snap.exists:
continue
udata = snap.to_dict() or {}
user_infos[snap.id] = DeviceUserSummary(
user_id=snap.id,
display_name=udata.get("display_name") or "",
email=udata.get("email") or "",
)
return customer_names, user_infos
def create_batch(data: BatchCreate) -> BatchResponse:
db = get_db()
existing_sns = _get_existing_sns(db)
@@ -123,6 +169,8 @@ def create_batch(data: BatchCreate) -> BatchResponse:
def list_devices(
status: str | None = None,
hw_type: str | None = None,
hw_types: list[str] | None = None,
has_users: bool | None = None,
search: str | None = None,
limit: int = 100,
offset: int = 0,
@@ -136,7 +184,24 @@ def list_devices(
query = query.where("hw_type", "==", hw_type)
docs = list(query.stream())
items = [_doc_to_inventory_item(doc) for doc in docs]
docs_data = [doc.to_dict() or {} for doc in docs]
if hw_types:
hw_type_set = set(hw_types)
keep = [i for i, d in enumerate(docs_data) if d.get("hw_type") in hw_type_set]
docs = [docs[i] for i in keep]
docs_data = [docs_data[i] for i in keep]
if has_users is not None:
keep = [
i for i, d in enumerate(docs_data)
if bool(d.get("user_list")) == has_users
]
docs = [docs[i] for i in keep]
docs_data = [docs_data[i] for i in keep]
customer_names, user_infos = _bulk_resolve_names(db, docs_data)
items = [_doc_to_inventory_item(doc, customer_names, user_infos) for doc in docs]
if search:
search_lower = search.lower()
@@ -145,6 +210,10 @@ def list_devices(
if search_lower in (item.serial_number or "").lower()
or search_lower in (item.owner or "").lower()
or search_lower in (item.mfg_batch_id or "").lower()
or search_lower in (item.device_name or "").lower()
or search_lower in (item.console_name or "").lower()
or search_lower in (item.customer_name or "").lower()
or any(search_lower in (u.display_name or "").lower() or search_lower in (u.email or "").lower() for u in (item.users or []))
]
return items[offset: offset + limit]
@@ -155,7 +224,9 @@ def get_device_by_sn(sn: str) -> DeviceInventoryItem:
docs = list(db.collection(COLLECTION).where("serial_number", "==", sn).limit(1).stream())
if not docs:
raise NotFoundError("Device")
return _doc_to_inventory_item(docs[0])
doc = docs[0]
customer_names, user_infos = _bulk_resolve_names(db, [doc.to_dict() or {}])
return _doc_to_inventory_item(doc, customer_names, user_infos)
def update_device_status(sn: str, data: DeviceStatusUpdate, set_by: str | None = None) -> DeviceInventoryItem: