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

"""运输计划 · 现场按段页（history / page / feedback / upload）共享逻辑"""
from __future__ import annotations

import os
from datetime import date, 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 (
    PlPlan,
    PlPlanDelivery,
    PlPlanLoad,
    PlPlanProduct,
    PlPlanTransport,
    PlTransport,
)
from app.schemas.pl.transport import (
    PublicTransportHistoryPageRead,
    PublicTransportHistoryProductRead,
    PublicTransportHistoryRecordRead,
    PublicTransportHistoryStopProductRead,
    PublicTransportHistoryStopRead,
    PublicTransportMultiStopSitePageRead,
)
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,
    _normalize_main_plate,
    _parse_calendar_month,
    _parse_history_date,
    _resolve_driver_link_flags,
    _resolve_unit_label,
)
from app.services.pl_transport_stop_feedback_service import (
    build_public_stop_reads_for_segments,
    build_transport_site_url,
    load_site_stop_segments,
    save_site_stop_feedback,
)
from app.utils.excel_export import load_dict_label_maps


def plan_matches_history_day(plan: PlPlan, day: date) -> bool:
    """Mirrors SQL day filters in build_site_history_page for unit tests."""
    if plan.plan_date == day:
        return True
    if plan.pickup_time is not None:
        pt = plan.pickup_time
        if (pt.date() if isinstance(pt, datetime) else pt) == day:
            return True
    if plan.delivery_time is not None:
        dt = plan.delivery_time
        if (dt.date() if isinstance(dt, datetime) else dt) == day:
            return True
    return False


def resolve_transport_for_plan(
    db: Session, *, tenant_no: str, plan_id: str
) -> Tuple[Optional[PlPlan], Optional[PlTransport]]:
    """本租户计划 + 关联运输（须有有效 site_access_token）。找不到返回 (None, None)。"""
    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,
            PlTransport.site_access_token.isnot(None),
            PlTransport.site_access_token != "",
        )
        .first()
    )
    return plan, 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 _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_site_page_for_transport(
    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_driver_link_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_page_for_plan(
    db: Session, *, tenant_no: str, plan_id: str
) -> tuple[Optional[PublicTransportMultiStopSitePageRead], Optional[str]]:
    plan, transport = resolve_transport_for_plan(db, tenant_no=tenant_no, plan_id=plan_id)
    if not plan:
        return None, "计划不存在"
    if not transport:
        return None, "未找到关联运输或现场链接无效"
    return build_site_page_for_transport(db, transport), None


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


def build_site_history_page(
    db: Session,
    *,
    tenant_no: str,
    date_str: str,
    keyword: Optional[str] = None,
    license_plate_hint: Optional[str] = None,
) -> tuple[Optional[PublicTransportHistoryPageRead], Optional[str]]:
    day = _parse_history_date(date_str)
    if day is None:
        return None, "日期格式无效，请使用 YYYY-MM-DD"

    kw = (keyword 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 == tenant_no,
            PlPlan.del_flag == 0,
            PlPlan.transport_status == "arranged",
            PlTransport.tenant_no == 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=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=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, 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_transport_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(license_plate_hint) or None,
        date=day.isoformat(),
        keyword=kw or None,
        records=records,
    )
    return page, None


def build_site_calendar_counts(
    db: Session,
    *,
    tenant_no: 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

    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,
            PlTransport.site_access_token.isnot(None),
            PlTransport.site_access_token != "",
            or_(
                and_(
                    PlPlan.plan_date.isnot(None),
                    PlPlan.plan_date >= month_start,
                    PlPlan.plan_date <= month_end,
                ),
                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, tr in rows:
        if plan.id in seen:
            continue
        site_tok = (tr.site_access_token or "").strip()
        if not site_tok:
            continue
        # 与 history 一致：无现场段则不计入
        if not load_site_stop_segments(
            db, tenant_no=tenant_no, plan_id=plan.id, transport_id=tr.id
        ):
            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=True,
        ),
        None,
    )


def save_site_feedback_for_transport(
    db: Session,
    transport: PlTransport,
    *,
    stop_type: str,
    stop_id: str,
    quantity: Optional[Decimal] = None,
    attachment_urls: Optional[list[str]] = None,
    remark: Optional[str] = None,
) -> tuple[bool, Optional[str]]:
    """保存段回传。成功返回 (True, None)；失败返回 (False, msg)。不 commit。"""
    page = build_site_page_for_transport(db, transport)
    if page.plan_voided:
        return False, "运输计划已作废，无法提交回执"
    if page.plan_archived:
        return False, "运输计划已归档，无法提交回执"

    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 False, "未找到关联计划，无法保存段回传"

    stop_type_n = (stop_type or "").strip()
    stop_id_n = (stop_id or "").strip()
    if stop_type_n not in ("load", "delivery") or not stop_id_n:
        return False, "停靠段参数无效"

    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_n, stop_id_n))
    if not segment:
        return False, "该段不在现场回传范围内"

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

    return True, None


def save_site_feedback_for_plan(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    stop_type: str,
    stop_id: str,
    quantity: Optional[Decimal] = None,
    attachment_urls: Optional[list[str]] = None,
    remark: Optional[str] = None,
) -> tuple[bool, Optional[str]]:
    plan, transport = resolve_transport_for_plan(db, tenant_no=tenant_no, plan_id=plan_id)
    if not plan:
        return False, "计划不存在"
    if not transport:
        return False, "未找到关联运输或现场链接无效"
    return save_site_feedback_for_transport(
        db,
        transport,
        stop_type=stop_type,
        stop_id=stop_id,
        quantity=quantity,
        attachment_urls=attachment_urls,
        remark=remark,
    )


def upload_site_attachment_for_transport(
    db: Session,
    transport: PlTransport,
    *,
    file: UploadFile,
    folder: Optional[str] = "",
) -> tuple[Optional[dict[str, Any]], Optional[tuple[int, str]]]:
    """上传附件。成功返回 (data, None) 并 commit；失败返回 (None, (code, msg))。"""
    page = build_site_page_for_transport(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_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 (
            {
                "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,
            },
            None,
        )
    except Exception as exc:
        db.rollback()
        return None, (500, f"上传失败: {str(exc)}")


def upload_site_attachment_for_plan(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
    file: UploadFile,
    folder: Optional[str] = "",
) -> tuple[Optional[dict[str, Any]], Optional[tuple[int, str]]]:
    plan, transport = resolve_transport_for_plan(db, tenant_no=tenant_no, plan_id=plan_id)
    if not plan:
        return None, (400, "计划不存在")
    if not transport:
        return None, (400, "未找到关联运输或现场链接无效")
    return upload_site_attachment_for_transport(
        db, transport, file=file, folder=folder
    )