shanxia2.0/backend/app/api/pl/public_transport_site.py

"""运输计划 · 现场按段免登录公共 API（一车多卸）"""
from __future__ import annotations

import os
from datetime import date, datetime
from decimal import Decimal
from typing import Optional

from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile
from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session

from app.api.pl.public_transport import (
    CRM_PRODUCT_UNIT,
    _history_status_label,
    _normalize_main_plate,
    _parse_history_date,
    _resolve_driver_link_flags,
    _resolve_unit_label,
)
from app.core.attachment_biz import resolve_upload_folder
from app.core.config import settings
from app.core.database import get_db
from app.core.response import error_response, success_response
from app.core.storage import storage
from app.crud import oss as oss_svc
from app.models.pl_models import (
    PlPlan,
    PlPlanDelivery,
    PlPlanLoad,
    PlPlanProduct,
    PlPlanTransport,
    PlTransport,
)
from app.schemas.pl.transport import (
    PublicTransportHistoryPageRead,
    PublicTransportHistoryProductRead,
    PublicTransportHistoryRecordRead,
    PublicTransportHistoryStopProductRead,
    PublicTransportHistoryStopRead,
    PublicTransportMultiStopSitePageRead,
    PublicTransportSiteStopFeedback,
)
from app.services.pl_product_display import (
    load_crm_product_alias_maps,
    resolve_product_display_names,
)
from app.services.pl_transport_stop_feedback_service import (
    build_public_stop_reads_for_segments,
    load_site_stop_segments,
    save_site_stop_feedback,
)
from app.utils.excel_export import load_dict_label_maps

router = APIRouter(prefix="/public/pl/transport-site", tags=["运输计划-现场按段公共"])


def _get_transport_by_site_token(db: Session, token: str) -> PlTransport:
    transport = (
        db.query(PlTransport)
        .filter(
            PlTransport.site_access_token == token,
            PlTransport.del_flag == 0,
        )
        .first()
    )
    if not transport:
        raise HTTPException(status_code=404, detail="链接无效或已失效")
    return transport


def _format_plan_date(value) -> Optional[str]:
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.strftime("%Y-%m-%d")
    if isinstance(value, date):
        return value.isoformat()
    return str(value)


def _resolve_plan_flags(plan: Optional[PlPlan], transport: PlTransport) -> tuple[bool, bool]:
    voided = (
        (plan is not None and plan.follow_status == "cancelled")
        or transport.transport_status == "cancelled"
    )
    archived = bool(plan is not None and plan.archive_status == "archived")
    return voided, archived


def _build_public_page(db: Session, transport: PlTransport) -> PublicTransportMultiStopSitePageRead:
    link = (
        db.query(PlPlanTransport)
        .filter(
            PlPlanTransport.transport_id == transport.id,
            PlPlanTransport.del_flag == 0,
        )
        .order_by(PlPlanTransport.create_time.desc())
        .first()
    )
    plan = None
    if link:
        plan = (
            db.query(PlPlan)
            .filter(PlPlan.id == link.plan_id, PlPlan.del_flag == 0)
            .first()
        )

    plan_voided, plan_archived = _resolve_plan_flags(plan, transport)
    stop_reads: list[dict] = []
    if plan:
        segments = load_site_stop_segments(
            db,
            tenant_no=transport.tenant_no,
            plan_id=plan.id,
            transport_id=transport.id,
        )
        stop_reads = build_public_stop_reads_for_segments(
            db,
            tenant_no=transport.tenant_no,
            plan_id=plan.id,
            segments=segments,
        )

    return PublicTransportMultiStopSitePageRead(
        plan_no=plan.plan_no if plan else transport.transport_no,
        plan_date=_format_plan_date(plan.plan_date if plan else None),
        license_plate=transport.license_plate,
        vehicle_no=transport.license_plate,
        plan_voided=plan_voided,
        plan_archived=plan_archived,
        stops=stop_reads,
    )


def _build_site_url(token: str) -> str:
    origin = (settings.PUBLIC_WEB_ORIGIN or "").rstrip("/")
    return f"{origin}/#/public/pl/transport-site/{token}"


def _build_site_history_stop_reads(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    transport: PlTransport,
) -> list[PublicTransportHistoryStopRead]:
    plan_products = (
        db.query(PlPlanProduct)
        .filter(
            PlPlanProduct.plan_id == plan.id,
            PlPlanProduct.tenant_no == tenant_no,
            PlPlanProduct.del_flag == 0,
        )
        .all()
    )
    plan_products_by_id = {row.id: row for row in plan_products}
    crm_product_ids = {
        (row.product_id or "").strip() for row in plan_products if (row.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, {}
    )

    load_row = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.plan_id == plan.id,
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
        )
        .first()
    )
    delivery_rows = (
        db.query(PlPlanDelivery)
        .filter(
            PlPlanDelivery.plan_id == plan.id,
            PlPlanDelivery.tenant_no == tenant_no,
            PlPlanDelivery.del_flag == 0,
        )
        .all()
    )
    delivery_time_by_id = {row.id: row.delivery_time for row in delivery_rows}
    load_time = load_row.load_time if load_row else None

    segments = load_site_stop_segments(
        db,
        tenant_no=tenant_no,
        plan_id=plan.id,
        transport_id=transport.id,
    )
    if not segments:
        return []

    reads: list[PublicTransportHistoryStopRead] = []
    for segment in segments:
        stop_time = load_time if segment.stop_type == "load" else delivery_time_by_id.get(segment.stop_id)
        stop_products: list[PublicTransportHistoryStopProductRead] = []
        for row in segment.products:
            plan_prod = plan_products_by_id.get(row.plan_product_id)
            product_id = (row.product_id or (plan_prod.product_id if plan_prod else "") or "").strip()
            display_name, _alias_name = resolve_product_display_names(
                saved_product_name=row.product_name,
                product_id=product_id,
                standard_by_id=standard_by_id,
                alias_options_by_id=alias_options_by_id,
            )
            stop_products.append(
                PublicTransportHistoryStopProductRead(
                    product_name=display_name or row.product_name or "—",
                    model=row.model,
                    package_spec=row.package_spec,
                    quantity=row.plan_qty,
                    unit=_resolve_unit_label(unit_map, row.unit),
                )
            )
        reads.append(
            PublicTransportHistoryStopRead(
                stop_type=segment.stop_type,
                seq=segment.seq,
                label=segment.label,
                customer_name=segment.customer_name,
                address=segment.address,
                stop_time=stop_time,
                products=stop_products,
            )
        )
    return reads


@router.get("/{token}/history")
def get_public_transport_site_history(
    token: str,
    date: str = Query(..., description="筛选日 YYYY-MM-DD"),
    q: Optional[str] = Query(None, description="车牌号/产品名称模糊"),
    db: Session = Depends(get_db),
):
    transport = _get_transport_by_site_token(db, token)
    day = _parse_history_date(date)
    if day is None:
        return error_response(400, "日期格式无效，请使用 YYYY-MM-DD")

    kw = (q or "").strip()
    query = (
        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 == transport.tenant_no,
            PlPlan.del_flag == 0,
            PlPlan.transport_status == "arranged",
            PlTransport.tenant_no == transport.tenant_no,
            PlTransport.site_access_token.isnot(None),
            PlTransport.site_access_token != "",
            or_(
                PlPlan.plan_date == day,
                and_(PlPlan.pickup_time.isnot(None), func.date(PlPlan.pickup_time) == day),
                and_(PlPlan.delivery_time.isnot(None), func.date(PlPlan.delivery_time) == day),
            ),
        )
    )
    if kw:
        like = f"%{kw}%"
        product_match = (
            db.query(PlPlanProduct.id)
            .filter(
                PlPlanProduct.plan_id == PlPlan.id,
                PlPlanProduct.del_flag == 0,
                PlPlanProduct.product_name.like(like),
            )
            .exists()
        )
        query = query.filter(or_(PlTransport.license_plate.like(like), product_match))

    rows = query.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, list[PublicTransportHistoryStopRead]]] = []
    for plan, tr in rows:
        if plan.id in seen_plan:
            continue
        site_tok = (tr.site_access_token or "").strip()
        if not site_tok:
            continue
        site_stops = _build_site_history_stop_reads(
            db, tenant_no=transport.tenant_no, plan=plan, transport=tr
        )
        if not site_stops:
            continue
        seen_plan.add(plan.id)
        plan_rows.append((plan, tr, site_stops))

    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=transport.tenant_no, product_ids=crm_product_ids
        )
        unit_map = load_dict_label_maps(
            db, [CRM_PRODUCT_UNIT], tenant_no=transport.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, site_stops in plan_rows:
        voided, archived = _resolve_driver_link_flags(plan, tr)
        site_tok = (tr.site_access_token or "").strip()
        records.append(
            PublicTransportHistoryRecordRead(
                plan_id=plan.id,
                plan_no=plan.plan_no or "",
                transport_id=tr.id,
                site_access_token=site_tok or None,
                site_url=_build_site_url(site_tok) if site_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=site_stops,
                plan_voided=voided,
                plan_archived=archived,
                status_label=_history_status_label(voided, archived),
            )
        )

    page = PublicTransportHistoryPageRead(
        license_plate=_normalize_main_plate(transport.license_plate) or None,
        date=day.isoformat(),
        keyword=kw or None,
        records=records,
    )
    return success_response(page.model_dump(by_alias=True))


@router.get("/{token}")
def get_public_transport_site(token: str, db: Session = Depends(get_db)):
    transport = _get_transport_by_site_token(db, token)
    page = _build_public_page(db, transport)
    return success_response(page.model_dump(by_alias=True))


@router.post("/{token}/feedback")
def save_public_site_stop_feedback(
    token: str,
    payload: PublicTransportSiteStopFeedback,
    db: Session = Depends(get_db),
):
    transport = _get_transport_by_site_token(db, token)
    page = _build_public_page(db, transport)
    if page.plan_voided:
        return error_response(400, "运输计划已作废，无法提交回执")
    if page.plan_archived:
        return error_response(400, "运输计划已归档，无法提交回执")

    link = (
        db.query(PlPlanTransport)
        .filter(
            PlPlanTransport.transport_id == transport.id,
            PlPlanTransport.del_flag == 0,
        )
        .order_by(PlPlanTransport.create_time.desc())
        .first()
    )
    if not link or not link.plan_id:
        return error_response(400, "未找到关联计划，无法保存段回传")

    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 error_response(400, "停靠段参数无效")

    segments = load_site_stop_segments(
        db,
        tenant_no=transport.tenant_no,
        plan_id=link.plan_id,
        transport_id=transport.id,
    )
    segment_map = {(item.stop_type, item.stop_id): item for item in segments}
    segment = segment_map.get((stop_type, stop_id))
    if not segment:
        return error_response(400, "该段不在现场回传范围内")

    try:
        save_site_stop_feedback(
            db,
            tenant_no=transport.tenant_no,
            transport_id=transport.id,
            plan_id=link.plan_id,
            stop_type=stop_type,
            stop_id=stop_id,
            business_type=segment.business_type or "入库",
            quantity=payload.quantity,
            attachment_urls=payload.attachment_urls,
            remark=payload.remark,
        )
    except ValueError as exc:
        return error_response(400, str(exc))

    db.commit()
    return success_response({"success": True, "message": "保存成功"})


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"


@router.post("/{token}/attachments")
def upload_public_site_attachment(
    token: str,
    file: UploadFile = File(...),
    folder: Optional[str] = Form(""),
    db: Session = Depends(get_db),
):
    transport = _get_transport_by_site_token(db, token)
    page = _build_public_page(db, transport)
    if page.plan_voided:
        return error_response(400, "运输计划已作废，无法上传附件")
    if page.plan_archived:
        return error_response(400, "运输计划已归档，无法上传附件")
    if not file or not file.filename:
        return error_response(400, "请选择文件")

    try:
        effective_folder = resolve_upload_folder(
            folder=folder or "", biz_type="pl_transport_site_stop_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_site_stop_feedback",
            user_id=None,
            ext1=effective_folder or None,
        )
        db.commit()
        return success_response(
            {
                "id": record.id,
                "fileName": record.original_name,
                "fileUrl": record.url,
                "fileSize": record.file_size,
                "fileSizeDisplay": _format_file_size(record.file_size or 0),
                "fileType": record.file_type,
            }
        )
    except Exception as exc:
        db.rollback()
        return error_response(500, f"上传失败: {str(exc)}")