shanxia2.0/backend/app/services/pl_driver_plan_service.py

"""运输计划 · 司机/押运登录态（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}")