627 lines
26 KiB
Python
627 lines
26 KiB
Python
from datetime import datetime
|
||
import json
|
||
from urllib import error, request
|
||
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
from sqlalchemy.orm import Session
|
||
|
||
from backend.app.core.config import get_settings
|
||
from backend.app.core.error_codes import ErrorCode
|
||
from backend.app.core.exceptions import AppException
|
||
from backend.app.repositories.logistics_repository import LogisticsRepository
|
||
from backend.app.repositories.order_repository import OrderRepository
|
||
from backend.app.services.arrears_service import arrears_service
|
||
from backend.app.services.audit_service import audit_service
|
||
from backend.app.services.demo_store import demo_store
|
||
from backend.app.services.file_service import file_service
|
||
|
||
|
||
class LogisticsService:
|
||
def __init__(self) -> None:
|
||
self.settings = get_settings()
|
||
self.logistics_repository = LogisticsRepository()
|
||
self.order_repository = OrderRepository()
|
||
self.trace_records: dict[int, list[dict]] = {}
|
||
|
||
def list_tasks(
|
||
self,
|
||
session: Session | None = None,
|
||
filters: dict | None = None,
|
||
current_user: dict | None = None,
|
||
) -> dict:
|
||
if session is not None:
|
||
try:
|
||
tasks = self.logistics_repository.list_tasks_by_filters(
|
||
session,
|
||
self.normalize_task_filters(filters, current_user),
|
||
)
|
||
return {
|
||
"total": len(tasks),
|
||
"page_no": 1,
|
||
"page_size": 20,
|
||
"list": [self._map_task_summary(task, session) for task in tasks],
|
||
}
|
||
except SQLAlchemyError:
|
||
pass
|
||
return demo_store.list_logistics_tasks()
|
||
|
||
def create_task(
|
||
self,
|
||
payload: dict,
|
||
session: Session | None = None,
|
||
current_user: dict | None = None,
|
||
) -> dict:
|
||
if session is not None:
|
||
try:
|
||
order = self.order_repository.get_order(session, payload["order_id"])
|
||
if order is None:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="订单不存在", status_code=404)
|
||
if order.order_status not in {"pending_factory", "approved"}:
|
||
raise AppException(code=ErrorCode.INVALID_STATUS, message="当前状态不允许创建司机任务", status_code=400)
|
||
|
||
existing_task = self.logistics_repository.get_task_by_order_id(session, order.id)
|
||
if existing_task is not None and existing_task.status != "canceled":
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="该订单已存在有效司机任务", status_code=400)
|
||
|
||
task = self.logistics_repository.create_task(
|
||
session,
|
||
{
|
||
"task_no": f"LT{datetime.now().strftime('%Y%m%d%H%M%S')}",
|
||
"order_id": order.id,
|
||
"driver_id": payload["driver_id"],
|
||
"factory_id": payload.get("factory_id") or order.factory_id,
|
||
"pickup_address": payload["pickup_address"],
|
||
"delivery_address": payload["delivery_address"],
|
||
"pickup_content": payload["pickup_content"],
|
||
"quantity": payload["quantity"],
|
||
"status": "pending",
|
||
"created_by": current_user.get("user_id") if current_user else None,
|
||
"remark": payload.get("remark"),
|
||
},
|
||
)
|
||
|
||
# 司机任务创建后，订单进入待司机接单阶段。
|
||
self.order_repository.update_order_status(session, order, "pending_driver")
|
||
audit_service.write_log(
|
||
session,
|
||
{
|
||
"operate_type": "logistics_task_create",
|
||
"biz_type": "logistics_task",
|
||
"biz_id": task.id,
|
||
"before_value": None,
|
||
"after_value": self._map_task_detail(task, session),
|
||
"remark": f"创建司机任务 {task.task_no}",
|
||
},
|
||
)
|
||
session.commit()
|
||
return self._map_task_detail(task, session)
|
||
except AppException:
|
||
session.rollback()
|
||
raise
|
||
except SQLAlchemyError:
|
||
session.rollback()
|
||
|
||
return demo_store.create_logistics_task(payload)
|
||
|
||
def get_task(
|
||
self,
|
||
task_id: int,
|
||
session: Session | None = None,
|
||
current_user: dict | None = None,
|
||
) -> dict:
|
||
if session is not None:
|
||
try:
|
||
task = self.logistics_repository.get_task(session, task_id)
|
||
if task is None:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="司机任务不存在", status_code=404)
|
||
self._ensure_task_access(task, current_user)
|
||
return self._map_task_detail(task, session)
|
||
except SQLAlchemyError:
|
||
pass
|
||
task = demo_store.get_logistics_task(task_id)
|
||
if task is None:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="司机任务不存在", status_code=404)
|
||
return task
|
||
|
||
def list_driver_tasks(self, session: Session | None = None, current_user: dict | None = None) -> dict:
|
||
return self.list_tasks(session=session, filters={}, current_user=current_user)
|
||
|
||
def cancel_task(
|
||
self,
|
||
task_id: int,
|
||
payload: dict | None = None,
|
||
session: Session | None = None,
|
||
current_user: dict | None = None,
|
||
) -> dict:
|
||
operate_payload = payload or {}
|
||
if session is not None:
|
||
try:
|
||
task = self.logistics_repository.get_task(session, task_id)
|
||
if task is None:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="司机任务不存在", status_code=404)
|
||
self._ensure_task_access(task, current_user)
|
||
if task.status not in {"pending", "accepted"}:
|
||
raise AppException(code=ErrorCode.INVALID_STATUS, message="当前任务状态不允许取消", status_code=400)
|
||
|
||
before_status = task.status
|
||
self.logistics_repository.update_task_status(session, task, "canceled")
|
||
task.canceled_at = datetime.now()
|
||
task.canceled_by = current_user.get("user_id") if current_user else None
|
||
task.cancel_reason = (operate_payload.get("cancel_reason") or "").strip() or None
|
||
session.add(task)
|
||
|
||
order = self.order_repository.get_order(session, task.order_id)
|
||
if order is not None and order.order_status == "pending_driver":
|
||
self.order_repository.update_order_status(session, order, "pending_factory")
|
||
|
||
audit_service.write_log(
|
||
session,
|
||
{
|
||
"operate_type": "logistics_task_cancel",
|
||
"biz_type": "logistics_task",
|
||
"biz_id": task.id,
|
||
"before_value": {"task_status": before_status},
|
||
"after_value": {
|
||
"task_status": task.status,
|
||
"canceled_at": task.canceled_at.strftime("%Y-%m-%d %H:%M:%S") if task.canceled_at else None,
|
||
"cancel_reason": task.cancel_reason,
|
||
},
|
||
"remark": operate_payload.get("remark") or f"取消司机任务 {task.task_no}",
|
||
},
|
||
)
|
||
session.commit()
|
||
return {"task_id": task.id, "status": task.status, "order_id": task.order_id}
|
||
except AppException:
|
||
session.rollback()
|
||
raise
|
||
except SQLAlchemyError:
|
||
session.rollback()
|
||
|
||
return {"task_id": task_id, "status": "canceled", "order_id": 0}
|
||
|
||
def get_driver_task(self, task_id: int, session: Session | None = None, current_user: dict | None = None) -> dict:
|
||
return self.get_task(task_id, session=session, current_user=current_user)
|
||
|
||
def accept_task(
|
||
self,
|
||
task_id: int,
|
||
payload: dict | None = None,
|
||
session: Session | None = None,
|
||
current_user: dict | None = None,
|
||
) -> dict:
|
||
result = self._change_task_status(task_id, "pending", "accepted", "accepted", session, current_user, payload or {})
|
||
self._append_trace(
|
||
result["order_id"],
|
||
{
|
||
"task_id": task_id,
|
||
"node_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"node_desc": "司机已接单",
|
||
"node_type": "accepted",
|
||
"source_platform": "driver",
|
||
"remark": (payload or {}).get("remark"),
|
||
},
|
||
)
|
||
return result
|
||
|
||
def pickup_task(
|
||
self,
|
||
task_id: int,
|
||
payload: dict | None = None,
|
||
session: Session | None = None,
|
||
current_user: dict | None = None,
|
||
) -> dict:
|
||
operate_payload = payload or {}
|
||
if not operate_payload.get("photo_files"):
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="确认揽货至少上传一张照片", status_code=400)
|
||
result = self._change_task_status(
|
||
task_id,
|
||
"accepted",
|
||
"picked_up",
|
||
"picked_up",
|
||
session,
|
||
current_user,
|
||
operate_payload,
|
||
)
|
||
self._append_trace(
|
||
result["order_id"],
|
||
{
|
||
"task_id": task_id,
|
||
"node_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"node_desc": "司机已揽货",
|
||
"node_type": "picked_up",
|
||
"source_platform": "driver",
|
||
"remark": operate_payload.get("remark"),
|
||
"photo_files": operate_payload.get("photo_files", []),
|
||
"video_files": operate_payload.get("video_files", []),
|
||
},
|
||
)
|
||
return result
|
||
|
||
def deliver_task(
|
||
self,
|
||
task_id: int,
|
||
payload: dict | None = None,
|
||
session: Session | None = None,
|
||
current_user: dict | None = None,
|
||
) -> dict:
|
||
operate_payload = payload or {}
|
||
result = self._change_task_status(
|
||
task_id,
|
||
"picked_up",
|
||
"delivered",
|
||
"delivered",
|
||
session,
|
||
current_user,
|
||
operate_payload,
|
||
)
|
||
self._append_trace(result["order_id"], self._build_trace_payload(task_id, "司机已送达", "delivered", operate_payload))
|
||
return result
|
||
|
||
def get_trace(
|
||
self,
|
||
order_id: int,
|
||
session: Session | None = None,
|
||
current_user: dict | None = None,
|
||
) -> dict:
|
||
if session is not None:
|
||
try:
|
||
self._ensure_trace_access(order_id, session, current_user)
|
||
traces = self.logistics_repository.list_traces_by_order_id(session, order_id)
|
||
trace_list = [
|
||
{
|
||
"task_id": item.task_id,
|
||
"node_time": item.node_time.strftime("%Y-%m-%d %H:%M:%S") if item.node_time else "",
|
||
"node_desc": item.node_desc,
|
||
"node_type": item.node_type,
|
||
"source_platform": item.source_platform,
|
||
"remark": item.remark,
|
||
}
|
||
for item in traces
|
||
]
|
||
third_party_trace_list = self._load_third_party_traces(order_id)
|
||
if trace_list or third_party_trace_list:
|
||
return {
|
||
"order_id": order_id,
|
||
"trace_list": self._merge_trace_list(trace_list, third_party_trace_list),
|
||
}
|
||
except SQLAlchemyError:
|
||
pass
|
||
|
||
return {
|
||
"order_id": order_id,
|
||
"trace_list": self._merge_trace_list(
|
||
self.trace_records.get(
|
||
order_id,
|
||
[
|
||
{
|
||
"node_time": "2026-05-14 12:00:00",
|
||
"node_desc": "司机已揽货",
|
||
"node_type": "picked_up",
|
||
"source_platform": "manual",
|
||
}
|
||
],
|
||
),
|
||
self._load_third_party_traces(order_id),
|
||
),
|
||
}
|
||
|
||
def create_trace(
|
||
self,
|
||
order_id: int,
|
||
payload: dict,
|
||
session: Session | None = None,
|
||
current_user: dict | None = None,
|
||
) -> dict:
|
||
if session is not None:
|
||
try:
|
||
self._ensure_trace_access(order_id, session, current_user)
|
||
node_time = datetime.strptime(payload["node_time"], "%Y-%m-%d %H:%M:%S")
|
||
self.logistics_repository.create_trace(
|
||
session,
|
||
{
|
||
"order_id": order_id,
|
||
"task_id": payload.get("task_id"),
|
||
"node_time": node_time,
|
||
"node_desc": payload["node_desc"],
|
||
"node_type": payload.get("node_type"),
|
||
"source_platform": payload.get("source_platform"),
|
||
"remark": payload.get("remark"),
|
||
},
|
||
)
|
||
session.commit()
|
||
return {"order_id": order_id, "created": True}
|
||
except ValueError as exc:
|
||
session.rollback()
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="物流节点时间格式错误", status_code=400) from exc
|
||
except SQLAlchemyError:
|
||
session.rollback()
|
||
|
||
self._append_trace(order_id, payload)
|
||
return {"order_id": order_id, "created": True}
|
||
|
||
def _change_task_status(
|
||
self,
|
||
task_id: int,
|
||
current_status: str,
|
||
target_status: str,
|
||
order_status: str,
|
||
session: Session | None = None,
|
||
current_user: dict | None = None,
|
||
payload: dict | None = None,
|
||
) -> dict:
|
||
operate_payload = payload or {}
|
||
if session is not None:
|
||
try:
|
||
task = self.logistics_repository.get_task(session, task_id)
|
||
if task is None:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="司机任务不存在", status_code=404)
|
||
self._ensure_task_access(task, current_user)
|
||
if task.status != current_status:
|
||
raise AppException(code=ErrorCode.INVALID_STATUS, message="当前任务状态不允许执行该操作", status_code=400)
|
||
|
||
self.logistics_repository.update_task_status(session, task, target_status)
|
||
order = self.order_repository.get_order(session, task.order_id)
|
||
if order is not None:
|
||
self.order_repository.update_order_status(session, order, order_status)
|
||
self.logistics_repository.create_trace(
|
||
session,
|
||
{
|
||
"order_id": task.order_id,
|
||
"task_id": task.id,
|
||
"node_time": datetime.now(),
|
||
"node_desc": self._status_desc(target_status),
|
||
"node_type": target_status,
|
||
"source_platform": "driver",
|
||
"remark": operate_payload.get("remark"),
|
||
},
|
||
)
|
||
self._persist_task_attachments(session, task, target_status, operate_payload, current_user)
|
||
audit_service.write_log(
|
||
session,
|
||
{
|
||
"operate_type": f"driver_task_{target_status}",
|
||
"biz_type": "logistics_task",
|
||
"biz_id": task.id,
|
||
"before_value": {"task_status": current_status, "order_status": order.order_status},
|
||
"after_value": {
|
||
"task_status": task.status,
|
||
"order_status": order_status,
|
||
"photo_count": len(operate_payload.get("photo_files", [])),
|
||
"video_count": len(operate_payload.get("video_files", [])),
|
||
},
|
||
"remark": operate_payload.get("remark") or f"司机任务状态变更为 {target_status}",
|
||
},
|
||
)
|
||
if target_status == "delivered":
|
||
arrears_service.sync_order_arrears(session, order)
|
||
session.commit()
|
||
return {"task_id": task.id, "status": task.status, "order_id": task.order_id}
|
||
except AppException:
|
||
session.rollback()
|
||
raise
|
||
except SQLAlchemyError:
|
||
session.rollback()
|
||
|
||
task = demo_store.get_logistics_task(task_id)
|
||
return {
|
||
"task_id": task_id,
|
||
"status": target_status,
|
||
"order_id": task["order_id"] if task else 0,
|
||
}
|
||
|
||
def normalize_task_filters(self, filters: dict | None, current_user: dict | None = None) -> dict:
|
||
normalized = dict(filters or {})
|
||
if current_user and current_user.get("role_code") == "driver":
|
||
normalized["driver_id"] = current_user.get("user_id")
|
||
return normalized
|
||
|
||
def _ensure_task_access(self, task: object, current_user: dict | None) -> None:
|
||
if not current_user:
|
||
return
|
||
role_code = current_user.get("role_code")
|
||
if role_code in {"admin", "manager"}:
|
||
return
|
||
if role_code == "driver":
|
||
if task.driver_id != current_user.get("user_id"):
|
||
raise AppException(code=ErrorCode.FORBIDDEN, message="仅可访问本人司机任务", status_code=403)
|
||
return
|
||
raise AppException(code=ErrorCode.FORBIDDEN, message="无权限访问", status_code=403)
|
||
|
||
def _ensure_trace_access(self, order_id: int, session: Session, current_user: dict | None) -> None:
|
||
if not current_user:
|
||
return
|
||
role_code = current_user.get("role_code")
|
||
if role_code in {"admin", "manager"}:
|
||
return
|
||
if role_code == "driver":
|
||
task = self.logistics_repository.get_task_by_order_id(session, order_id)
|
||
if task is None or task.driver_id != current_user.get("user_id"):
|
||
raise AppException(code=ErrorCode.FORBIDDEN, message="仅可访问本人订单轨迹", status_code=403)
|
||
return
|
||
raise AppException(code=ErrorCode.FORBIDDEN, message="无权限访问", status_code=403)
|
||
|
||
def _map_task_summary(self, task, session: Session) -> dict:
|
||
order = self.order_repository.get_order(session, task.order_id)
|
||
supplier = self.order_repository.get_supplier(session, task.factory_id)
|
||
return {
|
||
"task_id": task.id,
|
||
"task_no": task.task_no,
|
||
"order_id": task.order_id,
|
||
"order_no": order.order_no if order else "",
|
||
"driver_id": task.driver_id,
|
||
"driver_name": f"司机{task.driver_id}",
|
||
"factory_id": task.factory_id,
|
||
"factory_name": supplier.supplier_name if supplier else "",
|
||
"pickup_address": task.pickup_address,
|
||
"delivery_address": task.delivery_address,
|
||
"pickup_content": task.pickup_content,
|
||
"quantity": float(task.quantity or 0),
|
||
"status": task.status,
|
||
"remark": task.remark,
|
||
"created_at": task.created_at.strftime("%Y-%m-%d %H:%M:%S") if task.created_at else "",
|
||
}
|
||
|
||
def _map_task_detail(self, task, session: Session) -> dict:
|
||
summary = self._map_task_summary(task, session)
|
||
order = self.order_repository.get_order(session, task.order_id)
|
||
summary["salesman_name"] = ""
|
||
summary["customer_name"] = order.customer_name if order else ""
|
||
return summary
|
||
|
||
def _append_trace(self, order_id: int, payload: dict) -> None:
|
||
trace_list = self.trace_records.setdefault(order_id, [])
|
||
trace_list.append(payload)
|
||
|
||
def _status_desc(self, status: str) -> str:
|
||
if status == "accepted":
|
||
return "司机已接单"
|
||
if status == "picked_up":
|
||
return "司机已揽货"
|
||
if status == "delivered":
|
||
return "司机已送达"
|
||
return "物流状态更新"
|
||
|
||
def _build_trace_payload(self, task_id: int, node_desc: str, node_type: str, payload: dict | None) -> dict:
|
||
return {
|
||
"task_id": task_id,
|
||
"node_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"node_desc": node_desc,
|
||
"node_type": node_type,
|
||
"source_platform": "driver",
|
||
"remark": (payload or {}).get("remark"),
|
||
"photo_files": (payload or {}).get("photo_files", []),
|
||
"video_files": (payload or {}).get("video_files", []),
|
||
}
|
||
|
||
def _persist_task_attachments(
|
||
self,
|
||
session: Session,
|
||
task,
|
||
target_status: str,
|
||
payload: dict,
|
||
current_user: dict | None,
|
||
) -> None:
|
||
file_payloads = [
|
||
*[
|
||
{
|
||
"biz_type": "logistics_task",
|
||
"biz_id": task.id,
|
||
"file_name": item["file_name"],
|
||
"file_url": item["file_url"],
|
||
"file_type": item.get("file_type") or "image/jpeg",
|
||
"file_size": item.get("file_size"),
|
||
}
|
||
for item in payload.get("photo_files", [])
|
||
],
|
||
*[
|
||
{
|
||
"biz_type": "logistics_task",
|
||
"biz_id": task.id,
|
||
"file_name": item["file_name"],
|
||
"file_url": item["file_url"],
|
||
"file_type": item.get("file_type") or "video/mp4",
|
||
"file_size": item.get("file_size"),
|
||
}
|
||
for item in payload.get("video_files", [])
|
||
],
|
||
]
|
||
if not file_payloads:
|
||
return
|
||
|
||
# 司机操作附件统一落到现有附件表，便于后续审计和详情展示复用。
|
||
for attachment_payload in file_payloads:
|
||
repository_payload = file_service.build_attachment_payload(
|
||
attachment_payload,
|
||
created_by=current_user.get("user_id") if current_user else None,
|
||
require_file_size=False,
|
||
)
|
||
file_service.repository.create_attachment(session, repository_payload)
|
||
|
||
audit_service.write_log(
|
||
session,
|
||
{
|
||
"operate_type": "logistics_task_attachment_create",
|
||
"biz_type": "logistics_task",
|
||
"biz_id": task.id,
|
||
"before_value": None,
|
||
"after_value": {
|
||
"task_status": target_status,
|
||
"photo_count": len(payload.get("photo_files", [])),
|
||
"video_count": len(payload.get("video_files", [])),
|
||
},
|
||
"remark": f"司机任务 {task.task_no} 附件留痕",
|
||
},
|
||
)
|
||
|
||
def _load_third_party_traces(self, order_id: int) -> list[dict]:
|
||
provider = (self.settings.logistics_trace_provider or "internal").strip().lower()
|
||
if provider == "internal":
|
||
return []
|
||
return self._fetch_remote_traces(order_id)
|
||
|
||
def _fetch_remote_traces(self, order_id: int) -> list[dict]:
|
||
provider = (self.settings.logistics_trace_provider or "third_party").strip().lower()
|
||
if not self.settings.logistics_trace_endpoint.strip() or not self.settings.logistics_trace_path.strip():
|
||
raise AppException(code=ErrorCode.THIRD_PARTY_FAILED, message="第三方物流轨迹配置不完整", status_code=400)
|
||
|
||
path = self.settings.logistics_trace_path.strip()
|
||
if "{order_id}" in path:
|
||
path = path.replace("{order_id}", str(order_id))
|
||
endpoint = self.settings.logistics_trace_endpoint.strip().rstrip("/")
|
||
if not path.startswith("/"):
|
||
path = f"/{path}"
|
||
url = f"{endpoint}{path}"
|
||
req = request.Request(
|
||
url=url,
|
||
headers={
|
||
"Authorization": f"APPCODE {self.settings.logistics_trace_app_code.strip()}",
|
||
"Content-Type": "application/json; charset=UTF-8",
|
||
},
|
||
method="GET",
|
||
)
|
||
try:
|
||
with request.urlopen(req, timeout=20) as response:
|
||
payload = json.loads(response.read().decode("utf-8") or "{}")
|
||
except error.HTTPError as exc:
|
||
detail = exc.read().decode("utf-8", errors="ignore")
|
||
raise AppException(
|
||
code=ErrorCode.THIRD_PARTY_FAILED,
|
||
message=f"第三方物流轨迹调用失败：HTTP {exc.code} {detail}".strip(),
|
||
status_code=400,
|
||
) from exc
|
||
except error.URLError as exc:
|
||
raise AppException(
|
||
code=ErrorCode.THIRD_PARTY_FAILED,
|
||
message=f"第三方物流轨迹网络请求失败：{exc.reason}",
|
||
status_code=400,
|
||
) from exc
|
||
except json.JSONDecodeError as exc:
|
||
raise AppException(code=ErrorCode.THIRD_PARTY_FAILED, message="第三方物流轨迹返回格式异常", status_code=400) from exc
|
||
|
||
rows = payload.get("trace_list") if isinstance(payload, dict) else []
|
||
if not isinstance(rows, list):
|
||
return []
|
||
normalized: list[dict] = []
|
||
for item in rows:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
normalized.append(
|
||
{
|
||
"task_id": item.get("task_id"),
|
||
"node_time": str(item.get("node_time") or ""),
|
||
"node_desc": str(item.get("node_desc") or item.get("content") or ""),
|
||
"node_type": str(item.get("node_type") or "third_party"),
|
||
"source_platform": str(item.get("source_platform") or provider),
|
||
"remark": item.get("remark"),
|
||
}
|
||
)
|
||
return normalized
|
||
|
||
def _merge_trace_list(self, local_trace_list: list[dict], third_party_trace_list: list[dict]) -> list[dict]:
|
||
merged = list(local_trace_list) + list(third_party_trace_list)
|
||
merged.sort(key=lambda item: item.get("node_time") or "")
|
||
return merged
|
||
|
||
|
||
logistics_service = LogisticsService()
|