From 1303fae1e0a3ebc49050208ecb88f8fa648f8070 Mon Sep 17 00:00:00 2001 From: wuzilong <704541015@qq.com> Date: Sun, 13 Sep 2026 11:44:24 +0800 Subject: [PATCH] feat(pl): add driver-plan service for personnel-scoped history Resolve plans by linked driver/escort personnel and reuse public driver page/feedback helpers for login-state miniprogram APIs. Co-authored-by: Cursor --- .../app/services/pl_driver_plan_service.py | 500 ++++++++++++++++++ backend/tests/test_pl_driver_plan_service.py | 30 ++ 2 files changed, 530 insertions(+) create mode 100644 backend/app/services/pl_driver_plan_service.py create mode 100644 backend/tests/test_pl_driver_plan_service.py diff --git a/backend/app/services/pl_driver_plan_service.py b/backend/app/services/pl_driver_plan_service.py new file mode 100644 index 000000000..74d51acec --- /dev/null +++ b/backend/app/services/pl_driver_plan_service.py @@ -0,0 +1,500 @@ +"""运输计划 · 司机/押运登录态（history / page / feedback / upload）""" +from __future__ import annotations + +import json +import os +from datetime import datetime +from decimal import Decimal +from typing import Any, Optional, Tuple + +from fastapi import UploadFile +from sqlalchemy import and_, func, or_ +from sqlalchemy.orm import Session + +from app.core.attachment_biz import resolve_upload_folder +from app.core.config import settings +from app.core.storage import storage +from app.crud import oss as oss_svc +from app.models.pl_models import ( + PlPersonnel, + PlPlan, + PlPlanProduct, + PlPlanTransport, + PlTransport, +) +from app.schemas.pl.transport import ( + PublicTransportDriverStopFeedback, + PublicTransportFeedback, + PublicTransportHistoryPageRead, + PublicTransportHistoryProductRead, + PublicTransportHistoryRecordRead, + PublicTransportPageRead, +) +from app.services.pl_feedback_brand_qty import ( + persist_transport_product_feedback, + require_brand_qtys_if_needed, +) +from app.services.pl_product_display import ( + load_crm_product_alias_maps, + resolve_product_display_names, +) +from app.services.pl_public_transport_helpers import ( + CRM_PRODUCT_UNIT, + _history_status_label, + _parse_history_date, + _resolve_driver_link_flags, + _resolve_unit_label, +) +from app.services.pl_transport_stop_feedback_service import ( + load_driver_stop_segments, + save_driver_stop_feedback, +) +from app.utils.excel_export import load_dict_label_maps + + +def transport_visible_to_personnel(transport: PlTransport, personnel_id: str) -> bool: + pid = (personnel_id or "").strip() + if not pid: + return False + return (transport.driver_id or "").strip() == pid or (transport.escort_id or "").strip() == pid + + +def resolve_personnel_for_user( + db: Session, *, tenant_no: str, user_id: str +) -> Optional[PlPersonnel]: + uid = (user_id or "").strip() + if not uid: + return None + return ( + db.query(PlPersonnel) + .filter( + PlPersonnel.tenant_no == tenant_no, + PlPersonnel.sys_user_id == uid, + PlPersonnel.del_flag == 0, + ) + .first() + ) + + +def resolve_transport_for_driver_plan( + db: Session, *, tenant_no: str, plan_id: str, personnel_id: str +) -> Tuple[Optional[PlPlan], Optional[PlTransport]]: + """本租户计划 + 运输，且当前人员是司机或押运。""" + plan = ( + db.query(PlPlan) + .filter(PlPlan.id == plan_id, PlPlan.tenant_no == tenant_no, PlPlan.del_flag == 0) + .first() + ) + if not plan: + return None, None + link = ( + db.query(PlPlanTransport) + .filter(PlPlanTransport.plan_id == plan.id, PlPlanTransport.del_flag == 0) + .order_by(PlPlanTransport.create_time.desc()) + .first() + ) + if not link or not link.transport_id: + return plan, None + transport = ( + db.query(PlTransport) + .filter( + PlTransport.id == link.transport_id, + PlTransport.tenant_no == tenant_no, + PlTransport.del_flag == 0, + ) + .first() + ) + if not transport or not transport_visible_to_personnel(transport, personnel_id): + return plan, None + return plan, transport + + +def _build_driver_url(token: str) -> str: + origin = (settings.PUBLIC_WEB_ORIGIN or "").rstrip("/") + return f"{origin}/#/public/pl/transport/{token}" + + +def _format_file_size(size_bytes: int) -> str: + if size_bytes < 1024: + return f"{size_bytes} B" + if size_bytes < 1024 * 1024: + return f"{size_bytes / 1024:.1f} KB" + return f"{size_bytes / (1024 * 1024):.1f} MB" + + +def build_driver_history_page( + db: Session, + *, + tenant_no: str, + user_id: str, + date_str: str, +) -> Tuple[Optional[PublicTransportHistoryPageRead], Optional[str]]: + day = _parse_history_date(date_str) + if day is None: + return None, "日期格式无效，请使用 YYYY-MM-DD" + + personnel = resolve_personnel_for_user(db, tenant_no=tenant_no, user_id=user_id) + if not personnel: + return ( + PublicTransportHistoryPageRead( + license_plate=None, date=day.isoformat(), records=[] + ), + None, + ) + + personnel_id = personnel.id + rows = ( + db.query(PlPlan, PlTransport) + .join( + PlPlanTransport, + and_( + PlPlanTransport.plan_id == PlPlan.id, + PlPlanTransport.del_flag == 0, + ), + ) + .join( + PlTransport, + and_( + PlTransport.id == PlPlanTransport.transport_id, + PlTransport.del_flag == 0, + ), + ) + .filter( + PlPlan.tenant_no == tenant_no, + PlPlan.del_flag == 0, + PlPlan.transport_status == "arranged", + PlTransport.tenant_no == tenant_no, + or_( + PlTransport.driver_id == personnel_id, + PlTransport.escort_id == personnel_id, + ), + or_( + and_(PlPlan.pickup_time.isnot(None), func.date(PlPlan.pickup_time) == day), + and_( + PlPlan.delivery_time.isnot(None), + func.date(PlPlan.delivery_time) == day, + ), + ), + ) + .order_by( + PlPlan.pickup_time.desc(), + PlPlan.plan_date.desc(), + PlPlan.create_time.desc(), + ) + .all() + ) + + seen_plan: set[str] = set() + plan_rows: list[tuple[PlPlan, PlTransport]] = [] + for plan, tr in rows: + if plan.id in seen_plan: + continue + seen_plan.add(plan.id) + plan_rows.append((plan, tr)) + + # Lazy import to reuse public history stop builder without circular import at module load. + from app.api.pl.public_transport import _build_history_stop_reads + + products_by_plan: dict[str, list[PublicTransportHistoryProductRead]] = {} + if plan_rows: + plan_ids = [plan.id for plan, _ in plan_rows] + product_rows = ( + db.query(PlPlanProduct) + .filter( + PlPlanProduct.plan_id.in_(plan_ids), + PlPlanProduct.del_flag == 0, + ) + .order_by(PlPlanProduct.order_num.asc(), PlPlanProduct.create_time.asc()) + .all() + ) + crm_product_ids = { + (p.product_id or "").strip() for p in product_rows if (p.product_id or "").strip() + } + standard_by_id, alias_options_by_id = load_crm_product_alias_maps( + db, tenant_no=tenant_no, product_ids=crm_product_ids + ) + unit_map = load_dict_label_maps(db, [CRM_PRODUCT_UNIT], tenant_no=tenant_no).get( + CRM_PRODUCT_UNIT, {} + ) + for prod in product_rows: + product_id = (prod.product_id or "").strip() + display_name, _alias_name = resolve_product_display_names( + saved_product_name=prod.product_name, + product_id=product_id, + standard_by_id=standard_by_id, + alias_options_by_id=alias_options_by_id, + ) + products_by_plan.setdefault(prod.plan_id, []).append( + PublicTransportHistoryProductRead( + product_name=display_name or prod.product_name or "—", + model=prod.model, + package_spec=prod.package_spec, + quantity=prod.quantity + or prod.inbound_estimate_qty + or prod.outbound_estimate_qty, + unit=_resolve_unit_label(unit_map, prod.unit), + ) + ) + + records: list[PublicTransportHistoryRecordRead] = [] + for plan, tr in plan_rows: + voided, archived = _resolve_driver_link_flags(plan, tr) + tok = (tr.driver_access_token or "").strip() + records.append( + PublicTransportHistoryRecordRead( + plan_id=plan.id, + plan_no=plan.plan_no or "", + transport_id=tr.id, + driver_access_token=tok or None, + driver_url=_build_driver_url(tok) if tok else None, + license_plate=tr.license_plate, + pickup_time=plan.pickup_time, + delivery_time=plan.delivery_time, + pickup_address=plan.pickup_address, + destination_address=plan.destination_address or plan.destination_name, + products=products_by_plan.get(plan.id) or [], + stops=_build_history_stop_reads( + db, tenant_no=tenant_no, plan=plan, transport=tr + ), + plan_voided=voided, + plan_archived=archived, + status_label=_history_status_label(voided, archived), + ) + ) + + return ( + PublicTransportHistoryPageRead( + license_plate=None, + date=day.isoformat(), + records=records, + ), + None, + ) + + +def build_driver_page_for_plan( + db: Session, *, tenant_no: str, plan_id: str, personnel_id: str +) -> Tuple[Optional[PublicTransportPageRead], Optional[str]]: + plan, transport = resolve_transport_for_driver_plan( + db, tenant_no=tenant_no, plan_id=plan_id, personnel_id=personnel_id + ) + if not plan: + return None, "计划不存在" + if not transport: + return None, "无权限查看该运输计划或未安排运输" + from app.api.pl.public_transport import _build_public_page + + return _build_public_page(db, transport), None + + +def save_driver_feedback_for_plan( + db: Session, + *, + tenant_no: str, + plan_id: str, + personnel_id: str, + payload: PublicTransportFeedback, +) -> Tuple[bool, Optional[str]]: + plan, transport = resolve_transport_for_driver_plan( + db, tenant_no=tenant_no, plan_id=plan_id, personnel_id=personnel_id + ) + if not plan: + return False, "计划不存在" + if not transport: + return False, "无权限操作该运输计划" + from app.api.pl.public_transport import _build_public_page + + page = _build_public_page(db, transport) + if page.plan_voided: + return False, "运输计划已作废，无法提交回执" + if page.plan_archived: + return False, "运输计划已归档，无法提交回执" + + if payload.feedback_images is not None: + transport.feedback_images = json.dumps(payload.feedback_images) + if payload.feedback_remark is not None: + transport.feedback_remark = payload.feedback_remark + + if payload.products: + products_payload = [ + { + "planProductId": item.plan_product_id, + "actualQty": item.actual_qty, + "brandQtys": [b.model_dump(by_alias=True) for b in (item.brand_qtys or [])], + } + for item in payload.products + ] + try: + persist_transport_product_feedback( + db, + tenant_no=transport.tenant_no, + transport=transport, + plan_id=plan.id, + products=products_payload, + user_id=None, + enforce_brand=True, + ) + except ValueError as exc: + return False, str(exc) + elif payload.feedback_qty is not None: + transport.feedback_qty = payload.feedback_qty + + transport.feedback_time = datetime.now() + return True, None + + +def save_driver_stop_feedback_for_plan( + db: Session, + *, + tenant_no: str, + plan_id: str, + personnel_id: str, + payload: PublicTransportDriverStopFeedback, +) -> Tuple[bool, Optional[str]]: + plan, transport = resolve_transport_for_driver_plan( + db, tenant_no=tenant_no, plan_id=plan_id, personnel_id=personnel_id + ) + if not plan: + return False, "计划不存在" + if not transport: + return False, "无权限操作该运输计划" + from app.api.pl.public_transport import _build_public_page + + page = _build_public_page(db, transport) + if page.plan_voided: + return False, "运输计划已作废，无法提交回执" + if page.plan_archived: + return False, "运输计划已归档，无法提交回执" + + stop_type = (payload.stop_type or "").strip() + stop_id = (payload.stop_id or "").strip() + if stop_type not in ("load", "delivery") or not stop_id: + return False, "停靠段参数无效" + + segments = load_driver_stop_segments( + db, + tenant_no=transport.tenant_no, + plan_id=plan.id, + transport_id=transport.id, + ) + segment = next( + (row for row in segments if row.stop_type == stop_type and row.stop_id == stop_id), + None, + ) + if not segment: + return False, "停靠段不存在或不可回传" + + products_payload: Optional[list[dict]] = None + if payload.products is not None: + allowed_products = {row.stop_product_id: row for row in segment.products} + plan_products = ( + db.query(PlPlanProduct) + .filter( + PlPlanProduct.plan_id == plan.id, + PlPlanProduct.tenant_no == transport.tenant_no, + PlPlanProduct.del_flag == 0, + ) + .all() + ) + plan_products_by_id = {row.id: row for row in plan_products} + products_payload = [] + for item in payload.products: + stop_product_id = (item.stop_product_id or "").strip() + stop_row = allowed_products.get(stop_product_id) + if not stop_row: + return False, "产品不属于当前停靠段" + actual_qty = Decimal(str(item.actual_qty or 0)) + brand_qtys = [b.model_dump(by_alias=True) for b in (item.brand_qtys or [])] + plan_product = plan_products_by_id.get(stop_row.plan_product_id) + try: + require_brand_qtys_if_needed( + brand_mode=plan_product.brand_mode if plan_product else None, + required_brand_id=plan_product.brand_id if plan_product else None, + actual_qty=actual_qty, + rows=brand_qtys, + enforce=True, + ) + except ValueError as exc: + return False, str(exc) + row_payload: dict = { + "stopProductId": stop_product_id, + "actualQty": float(actual_qty), + } + if brand_qtys: + row_payload["brandQtys"] = brand_qtys + products_payload.append(row_payload) + + save_driver_stop_feedback( + db, + tenant_no=transport.tenant_no, + transport_id=transport.id, + plan_id=plan.id, + stop_type=stop_type, + stop_id=stop_id, + attachment_urls=payload.attachment_urls or [], + products=products_payload, + remark=payload.remark, + ) + return True, None + + +def upload_driver_attachment_for_plan( + db: Session, + *, + tenant_no: str, + plan_id: str, + personnel_id: str, + file: UploadFile, + folder: str = "", +) -> Tuple[Optional[dict[str, Any]], Optional[Tuple[int, str]]]: + plan, transport = resolve_transport_for_driver_plan( + db, tenant_no=tenant_no, plan_id=plan_id, personnel_id=personnel_id + ) + if not plan: + return None, (404, "计划不存在") + if not transport: + return None, (403, "无权限操作该运输计划") + from app.api.pl.public_transport import _build_public_page + + page = _build_public_page(db, transport) + if page.plan_voided: + return None, (400, "运输计划已作废，无法上传附件") + if page.plan_archived: + return None, (400, "运输计划已归档，无法上传附件") + if not file or not file.filename: + return None, (400, "请选择文件") + + try: + effective_folder = resolve_upload_folder( + folder=folder or "", biz_type="pl_transport_feedback" + ) + file_content = file.file.read() + full_url = storage.upload_file(file_content, file.filename, effective_folder) + base_url = settings.MINIO_BASE_URL.rstrip("/") if settings.MINIO_BASE_URL else "" + if base_url and full_url.startswith(base_url + "/"): + file_url = full_url[len(base_url) + 1 :] + else: + file_url = full_url + + record = oss_svc.create_oss( + db, + tenant_no=transport.tenant_no, + url=file_url, + original_name=file.filename, + file_suffix=os.path.splitext(file.filename)[1].lstrip(".") if file.filename else "", + file_size=len(file_content), + file_type=file.content_type or "", + biz_type="pl_transport_feedback", + user_id=None, + ext1=effective_folder or None, + ) + db.commit() + return { + "url": full_url, + "name": file.filename, + "size": _format_file_size(len(file_content)), + "ossId": record.id if record else None, + }, None + except Exception as exc: # noqa: BLE001 + db.rollback() + return None, (500, f"上传失败: {exc}") diff --git a/backend/tests/test_pl_driver_plan_service.py b/backend/tests/test_pl_driver_plan_service.py new file mode 100644 index 000000000..7b2b54f57 --- /dev/null +++ b/backend/tests/test_pl_driver_plan_service.py @@ -0,0 +1,30 @@ +# backend/tests/test_pl_driver_plan_service.py +from datetime import date, datetime + +from app.api.pl.public_transport import _history_matches_date +from app.services.pl_driver_plan_service import transport_visible_to_personnel + + +def test_history_date_or_logic(): + day = date(2026, 9, 13) + assert _history_matches_date(datetime(2026, 9, 13, 9, 0), None, day) is True + assert _history_matches_date(None, datetime(2026, 9, 13, 18, 0), day) is True + assert ( + _history_matches_date( + datetime(2026, 9, 12, 9, 0), datetime(2026, 9, 14, 18, 0), day + ) + is False + ) + + +def test_visible_as_driver_or_escort(): + class T: + def __init__(self, driver_id=None, escort_id=None): + self.driver_id = driver_id + self.escort_id = escort_id + + assert transport_visible_to_personnel(T(driver_id="p1"), "p1") is True + assert transport_visible_to_personnel(T(escort_id="p1"), "p1") is True + assert transport_visible_to_personnel(T(driver_id="p2", escort_id="p3"), "p1") is False + assert transport_visible_to_personnel(T(), "p1") is False + assert transport_visible_to_personnel(T(driver_id="p1"), "") is False