641 lines
21 KiB
Python
641 lines
21 KiB
Python
"""运输计划 · 司机/押运登录态（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,
|
||
_accumulate_calendar_day_counts,
|
||
_history_status_label,
|
||
_parse_calendar_month,
|
||
_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 allocate_or_get_plan_copy_seq(
|
||
db: Session, *, tenant_no: str, plan_id: str
|
||
) -> Optional[int]:
|
||
"""
|
||
返回计划的 copy_seq_no；若为空则按租户 MAX(含软删)+1 写入后返回。
|
||
调用方负责 commit。计划不存在返回 None。
|
||
"""
|
||
plan = (
|
||
db.query(PlPlan)
|
||
.filter(PlPlan.id == plan_id, PlPlan.tenant_no == tenant_no)
|
||
.with_for_update()
|
||
.first()
|
||
)
|
||
if not plan:
|
||
return None
|
||
if plan.copy_seq_no is not None:
|
||
return int(plan.copy_seq_no)
|
||
|
||
max_seq = (
|
||
db.query(func.max(PlPlan.copy_seq_no))
|
||
.filter(PlPlan.tenant_no == tenant_no)
|
||
.with_for_update()
|
||
.scalar()
|
||
)
|
||
next_seq = int(max_seq or 0) + 1
|
||
plan.copy_seq_no = next_seq
|
||
db.flush()
|
||
return next_seq
|
||
|
||
|
||
def ensure_copy_seq_for_driver(
|
||
db: Session, *, tenant_no: str, plan_id: str, personnel_id: str
|
||
) -> Tuple[Optional[int], Optional[str]]:
|
||
"""
|
||
司机可见性与 GET 详情一致；通过后分配/返回 copy_seq_no。
|
||
成功返回 (seq, None)；失败返回 (None, err)。
|
||
"""
|
||
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, "无权限查看该运输计划或未安排运输"
|
||
seq = allocate_or_get_plan_copy_seq(db, tenant_no=tenant_no, plan_id=plan.id)
|
||
if seq is None:
|
||
return None, "计划不存在"
|
||
return seq, None
|
||
|
||
|
||
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),
|
||
)
|
||
)
|
||
|
||
from app.services.db.pound_qr_public import build_pound_qrcode_items_for_plan
|
||
|
||
records: list[PublicTransportHistoryRecordRead] = []
|
||
for plan, tr in plan_rows:
|
||
voided, archived = _resolve_driver_link_flags(plan, tr)
|
||
tok = (tr.driver_access_token or "").strip()
|
||
pound_items = build_pound_qrcode_items_for_plan(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
plan_id=plan.id,
|
||
transport_id=tr.id,
|
||
)
|
||
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),
|
||
pound_qrcode_items=pound_items,
|
||
)
|
||
)
|
||
|
||
return (
|
||
PublicTransportHistoryPageRead(
|
||
license_plate=None,
|
||
date=day.isoformat(),
|
||
records=records,
|
||
),
|
||
None,
|
||
)
|
||
|
||
|
||
def build_driver_calendar_counts(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
user_id: str,
|
||
month_str: str,
|
||
) -> Tuple[Optional[dict[str, int]], Optional[str]]:
|
||
"""按月统计司机可见计划条数，键为 YYYY-MM-DD。"""
|
||
bounds = _parse_calendar_month(month_str)
|
||
if bounds is None:
|
||
return None, "月份格式无效，请使用 YYYY-MM"
|
||
month_start, month_end = bounds
|
||
|
||
personnel = resolve_personnel_for_user(db, tenant_no=tenant_no, user_id=user_id)
|
||
if not personnel:
|
||
return {}, None
|
||
|
||
personnel_id = personnel.id
|
||
rows = (
|
||
db.query(PlPlan)
|
||
.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) >= month_start,
|
||
func.date(PlPlan.pickup_time) <= month_end,
|
||
),
|
||
and_(
|
||
PlPlan.delivery_time.isnot(None),
|
||
func.date(PlPlan.delivery_time) >= month_start,
|
||
func.date(PlPlan.delivery_time) <= month_end,
|
||
),
|
||
),
|
||
)
|
||
.all()
|
||
)
|
||
|
||
seen: set[str] = set()
|
||
plans: list[PlPlan] = []
|
||
for plan in rows:
|
||
if plan.id in seen:
|
||
continue
|
||
seen.add(plan.id)
|
||
plans.append(plan)
|
||
|
||
return (
|
||
_accumulate_calendar_day_counts(
|
||
plans,
|
||
month_start=month_start,
|
||
month_end=month_end,
|
||
include_plan_date=False,
|
||
),
|
||
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] if item.brand_qtys is not None else None),
|
||
}
|
||
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]
|
||
if item.brand_qtys is not None else (stop_row.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 item.brand_qtys is not None:
|
||
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}")
|