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 <cursoragent@cursor.com>

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}")

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