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

"""运费结算业务服务"""
from __future__ import annotations

import json
from datetime import date, datetime
from decimal import Decimal
from typing import Any, Dict, List, Optional, Tuple

from fastapi import HTTPException
from sqlalchemy import exists, false, func, or_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.services.wechat_approval_notices import sync_approval_notifications
from app.services.wechat_completion_notices import sync_completion_notification
from app.core.delete_guard import assert_row_deletable_by_creator, user_info_from_context
from app.models.models import SysUser
from app.services.pl_data_scope import apply_pl_scope, assert_pl_row_in_scope
from app.services.pl.freight_party_customers import load_plan_party_customers_map
from app.models.pl_models import (
    PlFreightSettlement,
    PlFreightSettlementApproval,
    PlFreightSettlementTransport,
    PlPlan,
    PlPlanDelivery,
    PlPlanLoad,
    PlRemittanceVoucher,
    PlTransport,
)
from app.schemas.pl.freight import (
    FreightSettlementDisburseBody,
    FreightSettlementSubmitApprovalBody,
    FreightSettlementUpdate,
)

_FMT = "%Y-%m-%d %H:%M:%S"

_DRAFT_SETTLEMENT_STATUSES = ("pending", "rejected")

TAB_STATUS_MAP = {
    "bills": ("pending", "rejected", "approving", "approved", "confirmed", "cancelled"),
    "approval_pending": ("approving",),
    "approval_done": ("approved", "confirmed"),
    "disburse": ("approved",),
    "settled": ("confirmed",),
}


def _resolve_user_display_name(db: Session, user_id: str) -> str:
    uid = (user_id or "").strip()
    if not uid:
        return ""
    u = db.query(SysUser).filter(SysUser.id == uid, SysUser.del_flag == 0).first()
    if not u:
        return ""
    return (u.nick_name or u.user_no or "").strip()


def _occupied_transport_ids_subquery(db: Session, tenant_no: str):
    return (
        db.query(PlFreightSettlementTransport.transport_id)
        .join(
            PlFreightSettlement,
            PlFreightSettlement.id == PlFreightSettlementTransport.settlement_id,
        )
        .filter(
            PlFreightSettlementTransport.tenant_no == tenant_no,
            PlFreightSettlementTransport.del_flag == 0,
            PlFreightSettlement.tenant_no == tenant_no,
            PlFreightSettlement.del_flag == 0,
            PlFreightSettlement.settlement_status != "cancelled",
        )
    )


def get_settlement_row(
    db: Session,
    tenant_no: str,
    settlement_id: str,
    *,
    user=None,
    skip_scope: bool = False,
    for_update: bool = False,
) -> PlFreightSettlement:
    query = (
        db.query(PlFreightSettlement)
        .filter(
            PlFreightSettlement.id == settlement_id,
            PlFreightSettlement.tenant_no == tenant_no,
            PlFreightSettlement.del_flag == 0,
        )
    )
    if for_update:
        query = query.populate_existing().with_for_update()
    row = query.first()
    if not row:
        raise HTTPException(status_code=404, detail="结算单不存在")
    if user and not skip_scope:
        assert_pl_row_in_scope(db, row, PlFreightSettlement, user, tenant_no)
    return row


def get_settlement_by_transport_id(
    db: Session,
    *,
    tenant_no: str,
    transport_id: str,
    user=None,
) -> Optional[Dict[str, Any]]:
    if not transport_id:
        return None
    row = (
        db.query(PlFreightSettlement)
        .join(
            PlFreightSettlementTransport,
            PlFreightSettlementTransport.settlement_id == PlFreightSettlement.id,
        )
        .filter(
            PlFreightSettlementTransport.transport_id == transport_id,
            PlFreightSettlementTransport.tenant_no == tenant_no,
            PlFreightSettlementTransport.del_flag == 0,
            PlFreightSettlement.tenant_no == tenant_no,
            PlFreightSettlement.del_flag == 0,
        )
        .order_by(PlFreightSettlement.create_time.desc())
        .first()
    )
    if not row:
        return None
    if user:
        assert_pl_row_in_scope(db, row, PlFreightSettlement, user, tenant_no)
    return {
        "id": row.id,
        "settlementNo": row.settlement_no,
        "invoiceAmount": row.invoice_amount,
        "invoiceNo": row.invoice_no,
        "invoiceDate": row.invoice_date,
        "invoiceAttachmentUrls": row.invoice_attachment_urls,
        "settlementStatus": row.settlement_status,
    }


def unsettled_plans_q_filter(q: str):
    """简易搜索：计划编号 / 装货送货客户 / 车牌 / 承运商名称。"""
    load_match = exists().where(
        PlPlanLoad.plan_id == PlPlan.id,
        PlPlanLoad.tenant_no == PlPlan.tenant_no,
        PlPlanLoad.del_flag == 0,
        PlPlanLoad.customer_name.contains(q),
    )
    delivery_match = exists().where(
        PlPlanDelivery.plan_id == PlPlan.id,
        PlPlanDelivery.tenant_no == PlPlan.tenant_no,
        PlPlanDelivery.del_flag == 0,
        PlPlanDelivery.customer_name.contains(q),
    )
    return or_(
        PlPlan.plan_no.contains(q),
        PlPlan.customer_name.contains(q),
        PlPlan.supplier_name.contains(q),
        load_match,
        delivery_match,
        PlPlan.license_plate.contains(q),
        PlTransport.license_plate.contains(q),
        PlTransport.carrier_name.contains(q),
    )


def list_unsettled_plans(
    db: Session,
    *,
    tenant_no: str,
    user=None,
    q: Optional[str] = None,
    carrier_id: Optional[str] = None,
    plan_no: Optional[str] = None,
    plan_type: Optional[str] = None,
    plan_type_in: Optional[str] = None,
    license_plate: Optional[str] = None,
    license_plate_in: Optional[str] = None,
    transport_company: Optional[str] = None,
    transport_company_in: Optional[str] = None,
    freight_amount_op: Optional[str] = None,
    freight_amount: Optional[str] = None,
    freight_amount_min: Optional[str] = None,
    freight_amount_max: Optional[str] = None,
    freight_type: Optional[str] = None,
    freight_type_in: Optional[str] = None,
    transport_qty_op: Optional[str] = None,
    transport_qty: Optional[str] = None,
    transport_qty_min: Optional[str] = None,
    transport_qty_max: Optional[str] = None,
    has_invoice: Optional[str] = None,
    has_invoice_in: Optional[str] = None,
    freight_remark: Optional[str] = None,
    freight_remark_in: Optional[str] = None,
    plan_no_in: Optional[str] = None,
    plan_date_from=None,
    plan_date_to=None,
    freight_price_op: Optional[str] = None,
    freight_price: Optional[str] = None,
    freight_price_min: Optional[str] = None,
    freight_price_max: Optional[str] = None,
    archive_time_from=None,
    archive_time_to=None,
    load_customers: Optional[str] = None,
    delivery_customers: Optional[str] = None,
    ids: Optional[List[str]] = None,
    page: int = 1,
    page_size: int = 20,
    max_page_size: Optional[int] = None,
) -> Tuple[int, List[Dict[str, Any]]]:
    occupied = _occupied_transport_ids_subquery(db, tenant_no).subquery()

    query = (
        db.query(PlPlan, PlTransport)
        .join(PlTransport, PlTransport.id == PlPlan.transport_id)
        .filter(
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
            PlPlan.archive_status == "archived",
            PlPlan.has_freight == 1,
            PlPlan.follow_status != "cancelled",
            PlPlan.transport_id.isnot(None),
            PlTransport.del_flag == 0,
            PlTransport.freight_amount > 0,
            ~PlPlan.transport_id.in_(occupied),
        )
    )

    if q:
        query = query.filter(unsettled_plans_q_filter(q))
    if carrier_id:
        query = query.filter(PlTransport.carrier_id == carrier_id)

    from app.services.pl.freight_list_filters import build_unsettled_plan_column_filters

    column_filters = build_unsettled_plan_column_filters(
        plan_no=plan_no,
        plan_no_in=plan_no_in,
        plan_type=plan_type,
        plan_type_in=plan_type_in,
        license_plate=license_plate,
        license_plate_in=license_plate_in,
        transport_company=transport_company,
        transport_company_in=transport_company_in,
        freight_amount_op=freight_amount_op,
        freight_amount=freight_amount,
        freight_amount_min=freight_amount_min,
        freight_amount_max=freight_amount_max,
        freight_type=freight_type,
        freight_type_in=freight_type_in,
        transport_qty_op=transport_qty_op,
        transport_qty=transport_qty,
        transport_qty_min=transport_qty_min,
        transport_qty_max=transport_qty_max,
        has_invoice=has_invoice,
        has_invoice_in=has_invoice_in,
        freight_remark=freight_remark,
        freight_remark_in=freight_remark_in,
        plan_date_from=plan_date_from,
        plan_date_to=plan_date_to,
        freight_price_op=freight_price_op,
        freight_price=freight_price,
        freight_price_min=freight_price_min,
        freight_price_max=freight_price_max,
        archive_time_from=archive_time_from,
        archive_time_to=archive_time_to,
        load_customers=load_customers,
        delivery_customers=delivery_customers,
    )
    if column_filters:
        query = query.filter(*column_filters)
    if ids:
        query = query.filter(PlPlan.id.in_(ids))

    if user:
        query = apply_pl_scope(query, PlPlan, user, db, tenant_no)

    total = query.count()
    page = max(1, page)
    cap = int(max_page_size) if max_page_size else 500
    page_size = min(max(1, page_size), cap)
    rows = (
        query.order_by(PlPlan.update_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )

    party_map = load_plan_party_customers_map(db, tenant_no, [plan for plan, _ in rows])
    items: List[Dict[str, Any]] = []
    for plan, transport in rows:
        load_customers, delivery_customers = party_map.get(plan.id, ([], []))
        items.append(
            {
                "id": plan.id,
                "planNo": plan.plan_no,
                "planDate": plan.plan_date,
                "planType": plan.plan_type,
                "customerName": plan.customer_name,
                "supplierName": plan.supplier_name,
                "loadCustomers": load_customers,
                "deliveryCustomers": delivery_customers,
                "licensePlate": plan.license_plate or transport.license_plate,
                "transportCompany": transport.carrier_name,
                "freightAmount": plan.freight_amount or transport.freight_amount,
                "freightType": transport.freight_type,
                "transportQty": transport.transport_qty,
                "freightPrice": transport.freight_price,
                "hasInvoice": int(transport.has_invoice or 0),
                "freightRemark": plan.freight_remark,
                "archiveTime": plan.update_time,
                "transportId": transport.id,
                "transportNo": transport.transport_no,
                "driverName": transport.driver_name,
                "transportMedium": transport.transport_medium,
                "startPoint": transport.start_point,
                "endPoint": transport.end_point,
            }
        )
    return total, items


def _resolve_transports_for_create(
    db: Session, tenant_no: str, *, plan_ids: Optional[List[str]], transport_ids: Optional[List[str]]
) -> List[PlTransport]:
    ids: List[str] = []
    if plan_ids:
        plans = (
            db.query(PlPlan)
            .filter(
                PlPlan.id.in_(plan_ids),
                PlPlan.tenant_no == tenant_no,
                PlPlan.del_flag == 0,
            )
            .all()
        )
        if len(plans) != len(set(plan_ids)):
            raise HTTPException(status_code=400, detail="存在无效的计划")
        for p in plans:
            if p.archive_status != "archived":
                raise HTTPException(status_code=400, detail=f"计划 {p.plan_no} 未归档，不能结算")
            if int(p.has_freight or 0) != 1:
                raise HTTPException(status_code=400, detail=f"计划 {p.plan_no} 无运输费用")
            if not p.transport_id:
                raise HTTPException(status_code=400, detail=f"计划 {p.plan_no} 未安排运输")
            ids.append(p.transport_id)
    if transport_ids:
        ids.extend(transport_ids)

    ids = list(dict.fromkeys(ids))
    if not ids:
        raise HTTPException(status_code=400, detail="请选择有效的运输单")

    occupied = {r[0] for r in _occupied_transport_ids_subquery(db, tenant_no).all()}
    conflict = [tid for tid in ids if tid in occupied]
    if conflict:
        raise HTTPException(status_code=400, detail="所选运输单已纳入其他结算单")

    transports = (
        db.query(PlTransport)
        .filter(
            PlTransport.id.in_(ids),
            PlTransport.tenant_no == tenant_no,
            PlTransport.del_flag == 0,
        )
        .all()
    )
    if len(transports) != len(ids):
        raise HTTPException(status_code=400, detail="请选择有效的运输单")
    return transports


def create_settlement(
    db: Session,
    *,
    tenant_no: str,
    plan_ids: Optional[List[str]],
    transport_ids: Optional[List[str]],
    user_id: str,
    dept_id: str,
    settlement_no: str,
) -> PlFreightSettlement:
    transports = _resolve_transports_for_create(
        db, tenant_no, plan_ids=plan_ids, transport_ids=transport_ids
    )
    total_freight = sum(Decimal(str(t.freight_amount or 0)) for t in transports)

    settlement = PlFreightSettlement(
        id=generate_uuid(),
        tenant_no=tenant_no,
        settlement_no=settlement_no,
        total_freight=total_freight,
        settlement_status="pending",
        create_by=user_id,
        create_dept_id=dept_id or "",
    )
    db.add(settlement)

    for t in transports:
        db.add(
            PlFreightSettlementTransport(
                id=generate_uuid(),
                tenant_no=tenant_no,
                settlement_id=settlement.id,
                transport_id=t.id,
                freight_amount=t.freight_amount,
                create_by=user_id,
                create_dept_id=dept_id or "",
            )
        )

    db.commit()
    db.refresh(settlement)
    return settlement


def _recalc_settlement_total_freight_from_links(
    db: Session,
    settlement_id: str,
    *,
    user_id: str,
    now: Optional[datetime] = None,
) -> None:
    # SessionLocal 关闭了 autoflush，须先 flush 再聚合查询，否则 SUM 仍读到旧明细金额
    db.flush()
    total = (
        db.query(func.coalesce(func.sum(PlFreightSettlementTransport.freight_amount), 0))
        .filter(
            PlFreightSettlementTransport.settlement_id == settlement_id,
            PlFreightSettlementTransport.del_flag == 0,
        )
        .scalar()
    )
    settlement = (
        db.query(PlFreightSettlement)
        .filter(
            PlFreightSettlement.id == settlement_id,
            PlFreightSettlement.del_flag == 0,
        )
        .first()
    )
    if not settlement:
        return
    ts = now or datetime.now()
    settlement.total_freight = Decimal(str(total or 0))
    settlement.update_by = user_id
    settlement.update_time = ts


def sync_draft_settlement_freight_for_transport(
    db: Session,
    *,
    tenant_no: str,
    transport_id: str,
    freight_amount: Decimal,
    user_id: str,
) -> None:
    """运输单运费变更后，同步关联草稿结算单的明细快照与合计。"""
    tid = (transport_id or "").strip()
    if not tid:
        return

    links = (
        db.query(PlFreightSettlementTransport)
        .join(
            PlFreightSettlement,
            PlFreightSettlement.id == PlFreightSettlementTransport.settlement_id,
        )
        .filter(
            PlFreightSettlementTransport.transport_id == tid,
            PlFreightSettlementTransport.tenant_no == tenant_no,
            PlFreightSettlementTransport.del_flag == 0,
            PlFreightSettlement.tenant_no == tenant_no,
            PlFreightSettlement.del_flag == 0,
            PlFreightSettlement.settlement_status.in_(_DRAFT_SETTLEMENT_STATUSES),
        )
        .all()
    )
    if not links:
        return

    now = datetime.now()
    amount = Decimal(str(freight_amount or 0))
    settlement_ids = set()
    for link in links:
        link.freight_amount = amount
        link.update_by = user_id
        link.update_time = now
        settlement_ids.add(link.settlement_id)

    for settlement_id in settlement_ids:
        _recalc_settlement_total_freight_from_links(
            db, settlement_id, user_id=user_id, now=now
        )


def refresh_draft_settlement_freight(
    db: Session,
    settlement: PlFreightSettlement,
    *,
    user_id: str = "",
) -> bool:
    """草稿/已驳回结算单按当前运输单运费刷新明细快照与合计。"""
    if settlement.settlement_status not in _DRAFT_SETTLEMENT_STATUSES:
        return False

    links = (
        db.query(PlFreightSettlementTransport)
        .filter(
            PlFreightSettlementTransport.settlement_id == settlement.id,
            PlFreightSettlementTransport.del_flag == 0,
        )
        .all()
    )
    if not links:
        return False

    transport_ids = [link.transport_id for link in links]
    transports = {
        t.id: t
        for t in db.query(PlTransport)
        .filter(PlTransport.id.in_(transport_ids), PlTransport.del_flag == 0)
        .all()
    }

    old_total = Decimal(str(settlement.total_freight or 0))
    now = datetime.now()
    changed = False
    for link in links:
        transport = transports.get(link.transport_id)
        if not transport:
            continue
        new_amount = Decimal(str(transport.freight_amount or 0))
        if Decimal(str(link.freight_amount or 0)) != new_amount:
            link.freight_amount = new_amount
            link.update_time = now
            if user_id:
                link.update_by = user_id
            changed = True

    _recalc_settlement_total_freight_from_links(
        db, settlement.id, user_id=user_id, now=now
    )
    new_total = Decimal(str(settlement.total_freight or 0))
    return changed or new_total != old_total


def delete_draft_settlement(
    db: Session, *, tenant_no: str, settlement_id: str, user_id: str, user=None
) -> None:
    settlement = get_settlement_row(db, tenant_no, settlement_id, user=user)
    if settlement.settlement_status not in _DRAFT_SETTLEMENT_STATUSES:
        raise HTTPException(status_code=400, detail="仅草稿或已驳回结算单可删除")

    if user:
        assert_row_deletable_by_creator(settlement, user_info_from_context(user))
    elif user_id:
        assert_row_deletable_by_creator(settlement, {"user_id": user_id})

    now = datetime.now()
    settlement.del_flag = 1
    settlement.update_by = user_id
    settlement.update_time = now

    links = (
        db.query(PlFreightSettlementTransport)
        .filter(
            PlFreightSettlementTransport.settlement_id == settlement_id,
            PlFreightSettlementTransport.del_flag == 0,
        )
        .all()
    )
    for link in links:
        link.del_flag = 1
        link.update_time = now

    db.flush()
    sync_approval_notifications(
        db, "pl_freight", settlement_id, tenant_no,
        actor_id=user_id or "",
    )
    db.commit()


def cancel_draft_settlement(
    db: Session,
    *,
    tenant_no: str,
    settlement_id: str,
    user_id: str,
    user=None,
    cancel_reason: Optional[str] = None,
) -> PlFreightSettlement:
    from app.schemas.pl.cancel_reason import normalize_optional_reason

    settlement = get_settlement_row(db, tenant_no, settlement_id, user=user)
    if settlement.settlement_status not in _DRAFT_SETTLEMENT_STATUSES:
        raise HTTPException(status_code=400, detail="仅草稿或已驳回结算单可作废")

    if user:
        assert_row_deletable_by_creator(settlement, user_info_from_context(user))
    elif user_id:
        assert_row_deletable_by_creator(settlement, {"user_id": user_id})

    now = datetime.now()
    settlement.settlement_status = "cancelled"
    settlement.cancel_reason = normalize_optional_reason(cancel_reason)
    settlement.cancel_time = now
    settlement.cancel_by = user_id
    settlement.update_by = user_id
    settlement.update_time = now

    links = (
        db.query(PlFreightSettlementTransport)
        .filter(
            PlFreightSettlementTransport.settlement_id == settlement_id,
            PlFreightSettlementTransport.del_flag == 0,
        )
        .all()
    )
    for link in links:
        link.del_flag = 1
        link.update_time = now

    db.flush()
    sync_approval_notifications(
        db, "pl_freight", settlement_id, tenant_no,
        actor_id=user_id or "",
    )
    db.commit()
    db.refresh(settlement)
    return settlement


def _get_transport_ids(db: Session, settlement_id: str) -> List[str]:
    return [
        r.transport_id
        for r in db.query(PlFreightSettlementTransport)
        .filter(
            PlFreightSettlementTransport.settlement_id == settlement_id,
            PlFreightSettlementTransport.del_flag == 0,
        )
        .all()
    ]


def _get_plans_for_transports(db: Session, tenant_no: str, transport_ids: List[str]) -> List[PlPlan]:
    if not transport_ids:
        return []
    return (
        db.query(PlPlan)
        .filter(
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
            PlPlan.transport_id.in_(transport_ids),
        )
        .all()
    )


def _approval_to_dict(row: PlFreightSettlementApproval) -> Dict[str, Any]:
    return {
        "id": row.id,
        "step": row.step,
        "stepName": row.step_name,
        "approverId": row.approver_id,
        "approverName": row.approver_name,
        "approvalType": row.approval_type,
        "approvalResult": row.approval_result,
        "approvalOpinion": row.approval_opinion,
        "approvalTime": row.approval_time,
        "isCurrent": row.is_current,
    }


def create_settlement_approval_records(
    db: Session,
    *,
    tenant_no: str,
    settlement_id: str,
    approval_type: str,
    approver_ids: List[str],
    user_id: str,
    dept_id: str,
) -> None:
    approval_type = (approval_type or "sequential").strip() or "sequential"
    cleaned = [aid.strip() for aid in approver_ids if (aid or "").strip()]
    if not cleaned:
        raise HTTPException(status_code=400, detail="请选择审批人")

    if approval_type in ("parallel", "or"):
        step = 1
        for idx, aid in enumerate(cleaned):
            nick = _resolve_user_display_name(db, aid) or f"用户{aid}"
            db.add(
                PlFreightSettlementApproval(
                    id=generate_uuid(),
                    settlement_id=settlement_id,
                    step=step,
                    step_name="会签审批" if approval_type == "parallel" else "或签审批",
                    approver_id=aid,
                    approver_name=nick,
                    approval_type=approval_type,
                    approval_result="pending",
                    is_current=1,
                    tenant_no=tenant_no,
                    create_by=user_id,
                    create_dept_id=dept_id or "",
                )
            )
    else:
        step = 1
        for idx, aid in enumerate(cleaned):
            nick = _resolve_user_display_name(db, aid) or f"用户{aid}"
            db.add(
                PlFreightSettlementApproval(
                    id=generate_uuid(),
                    settlement_id=settlement_id,
                    step=step,
                    step_name=f"第{step}步审批",
                    approver_id=aid,
                    approver_name=nick,
                    approval_type="sequential",
                    approval_result="pending",
                    is_current=1 if idx == 0 else 0,
                    tenant_no=tenant_no,
                    create_by=user_id,
                    create_dept_id=dept_id or "",
                )
            )
            step += 1
    db.flush()


def _clear_settlement_approval_records(db: Session, settlement_id: str) -> None:
    """软删除结算单历史审批记录（重新提交审批前清理）"""
    now = datetime.now()
    rows = (
        db.query(PlFreightSettlementApproval)
        .filter(
            PlFreightSettlementApproval.settlement_id == settlement_id,
            PlFreightSettlementApproval.del_flag == 0,
        )
        .all()
    )
    for row in rows:
        row.del_flag = 1
        row.update_time = now
    if rows:
        db.flush()


def _sync_settlement_status_if_approval_complete(
    db: Session, settlement: PlFreightSettlement
) -> bool:
    """审批记录已全部完成但结算单仍停留在 approving 时自动纠正（兼容历史脏数据）"""
    if settlement.settlement_status != "approving":
        return False
    if not _settlement_all_approved(db, settlement.id):
        return False
    settlement.settlement_status = "approved"
    settlement.update_time = datetime.now()
    return True


def _sync_settlement_status_if_disbursed(
    db: Session, settlement: PlFreightSettlement
) -> bool:
    """已有汇款凭证但结算单仍停留在 approved 时自动纠正（兼容历史脏数据）"""
    if settlement.settlement_status != "approved":
        return False
    voucher = (
        db.query(PlRemittanceVoucher)
        .filter(
            PlRemittanceVoucher.settlement_id == settlement.id,
            PlRemittanceVoucher.del_flag == 0,
        )
        .first()
    )
    if not voucher:
        return False
    settlement.settlement_status = "confirmed"
    settlement.update_time = datetime.now()
    return True


def _sync_disbursed_settlements(db: Session, tenant_no: str) -> None:
    """批量纠正已有汇款凭证但仍为 approved 的结算单"""
    rows = (
        db.query(PlFreightSettlement)
        .filter(
            PlFreightSettlement.tenant_no == tenant_no,
            PlFreightSettlement.del_flag == 0,
            PlFreightSettlement.settlement_status == "approved",
            db.query(PlRemittanceVoucher.id)
            .filter(
                PlRemittanceVoucher.settlement_id == PlFreightSettlement.id,
                PlRemittanceVoucher.del_flag == 0,
            )
            .exists(),
        )
        .all()
    )
    if not rows:
        return
    now = datetime.now()
    for row in rows:
        row.settlement_status = "confirmed"
        row.update_time = now
    db.commit()


def can_user_approve_settlement(db: Session, settlement_id: str, user_id: str) -> bool:
    if not settlement_id or not user_id:
        return False

    approvals = (
        db.query(PlFreightSettlementApproval)
        .filter(
            PlFreightSettlementApproval.settlement_id == settlement_id,
            PlFreightSettlementApproval.del_flag == 0,
        )
        .order_by(PlFreightSettlementApproval.step.asc())
        .all()
    )
    if not approvals:
        return False

    user_approval = next((a for a in approvals if a.approver_id == user_id), None)
    if not user_approval or user_approval.approval_result != "pending" or user_approval.is_current != 1:
        return False

    approval_type = user_approval.approval_type or "sequential"
    if approval_type == "sequential":
        for approval in approvals:
            if approval.step == user_approval.step:
                break
            if approval.approval_result != "approved":
                return False
    return True


def _settlement_all_approved(db: Session, settlement_id: str) -> bool:
    pending = (
        db.query(PlFreightSettlementApproval)
        .filter(
            PlFreightSettlementApproval.settlement_id == settlement_id,
            PlFreightSettlementApproval.approval_result == "pending",
            PlFreightSettlementApproval.del_flag == 0,
        )
        .count()
    )
    return pending == 0


def _advance_after_approval(
    db: Session, settlement: PlFreightSettlement, current: PlFreightSettlementApproval, result: str
) -> None:
    now = datetime.now()
    if result == "rejected":
        settlement.settlement_status = "rejected"
        settlement.update_time = now
        others = (
            db.query(PlFreightSettlementApproval)
            .filter(
                PlFreightSettlementApproval.settlement_id == settlement.id,
                PlFreightSettlementApproval.id != current.id,
                PlFreightSettlementApproval.approval_result == "pending",
                PlFreightSettlementApproval.del_flag == 0,
            )
            .all()
        )
        for row in others:
            row.approval_result = "skipped"
            row.is_current = 0
            row.update_time = now
        return

    approval_type = current.approval_type or "sequential"
    step = current.step

    if approval_type == "or":
        others = (
            db.query(PlFreightSettlementApproval)
            .filter(
                PlFreightSettlementApproval.settlement_id == settlement.id,
                PlFreightSettlementApproval.step == step,
                PlFreightSettlementApproval.id != current.id,
                PlFreightSettlementApproval.del_flag == 0,
            )
            .all()
        )
        for row in others:
            if row.approval_result == "pending":
                row.approval_result = "skipped"
                row.is_current = 0
                row.update_time = now
    elif approval_type == "parallel":
        step_pending = (
            db.query(PlFreightSettlementApproval)
            .filter(
                PlFreightSettlementApproval.settlement_id == settlement.id,
                PlFreightSettlementApproval.step == step,
                PlFreightSettlementApproval.approval_result == "pending",
                PlFreightSettlementApproval.del_flag == 0,
            )
            .count()
        )
        if step_pending > 0:
            return
    else:
        nxt = (
            db.query(PlFreightSettlementApproval)
            .filter(
                PlFreightSettlementApproval.settlement_id == settlement.id,
                PlFreightSettlementApproval.approval_result == "pending",
                PlFreightSettlementApproval.step > step,
                PlFreightSettlementApproval.del_flag == 0,
            )
            .order_by(PlFreightSettlementApproval.step.asc())
            .first()
        )
        if nxt:
            nxt.is_current = 1
            nxt.update_time = now
            return

    # 或签跳过其他节点后先落库；Session autoflush=False，否则仍会统计到 pending。
    db.flush()
    if _settlement_all_approved(db, settlement.id):
        settlement.settlement_status = "approved"
        settlement.update_time = now


def submit_settlement_approval(
    db: Session,
    *,
    tenant_no: str,
    settlement_id: str,
    body: FreightSettlementSubmitApprovalBody,
    user_id: str,
    dept_id: str,
    user=None,
) -> PlFreightSettlement:
    settlement = get_settlement_row(db, tenant_no, settlement_id, user=user)
    if settlement.settlement_status not in _DRAFT_SETTLEMENT_STATUSES:
        raise HTTPException(status_code=400, detail="仅草稿或已驳回状态可提交审批")

    settlement.settlement_status = "approving"
    if body.remark:
        settlement.remark = body.remark
    settlement.update_by = user_id
    db.flush()

    _clear_settlement_approval_records(db, settlement_id)

    create_settlement_approval_records(
        db,
        tenant_no=tenant_no,
        settlement_id=settlement_id,
        approval_type=body.approval_type,
        approver_ids=body.approver_ids,
        user_id=user_id,
        dept_id=dept_id,
    )
    db.flush()
    sync_approval_notifications(
        db, "pl_freight", settlement_id, tenant_no,
        new_cycle=True, actor_id=user_id or "",
    )
    db.commit()
    db.refresh(settlement)
    return settlement


def process_settlement_approval(
    db: Session,
    *,
    tenant_no: str,
    approval_id: str,
    action: str,
    comment: Optional[str],
    user_id: str,
) -> PlFreightSettlement:
    db.query(PlFreightSettlement).join(
        PlFreightSettlementApproval,
        PlFreightSettlementApproval.settlement_id == PlFreightSettlement.id,
    ).filter(
        PlFreightSettlementApproval.id == approval_id,
        PlFreightSettlementApproval.tenant_no == tenant_no,
        PlFreightSettlement.tenant_no == tenant_no,
        PlFreightSettlement.del_flag == 0,
    ).populate_existing().with_for_update().first()
    approval = (
        db.query(PlFreightSettlementApproval)
        .filter(
            PlFreightSettlementApproval.id == approval_id,
            PlFreightSettlementApproval.tenant_no == tenant_no,
            PlFreightSettlementApproval.del_flag == 0,
        )
        .first()
    )
    if not approval:
        raise HTTPException(status_code=404, detail="审批记录不存在")
    db.refresh(approval, with_for_update=True)

    settlement = get_settlement_row(db, tenant_no, approval.settlement_id, skip_scope=True)
    if settlement.settlement_status != "approving":
        raise HTTPException(status_code=400, detail="结算单不在审批中")

    if approval.approver_id != user_id:
        raise HTTPException(status_code=403, detail="只有审批人才能审批")
    if not can_user_approve_settlement(db, settlement.id, user_id):
        raise HTTPException(status_code=400, detail="当前不可审批")

    result = "approved" if action == "approve" else "rejected"
    now = datetime.now()
    approval.approval_result = result
    approval.approval_opinion = comment or ""
    approval.approval_time = now
    approval.is_current = 0
    approval.update_by = user_id
    approval.update_time = now

    # SessionLocal 关闭了 autoflush，须先 flush 再查询 pending 计数，否则当前审批仍会被算作 pending
    db.flush()
    _advance_after_approval(db, settlement, approval, result)
    settlement.update_by = user_id
    db.flush()
    sync_approval_notifications(
        db, "pl_freight", settlement.id, tenant_no,
        actor_id=user_id or "",
    )
    db.commit()
    db.refresh(settlement)
    return settlement


def withdraw_settlement_approval(
    db: Session, *, tenant_no: str, settlement_id: str, user_id: str, user=None
) -> PlFreightSettlement:
    settlement = get_settlement_row(db, tenant_no, settlement_id, user=user)
    if settlement.settlement_status != "approving":
        raise HTTPException(status_code=400, detail="仅审批中可撤回")
    if settlement.create_by != user_id:
        raise HTTPException(status_code=403, detail="仅提交人可撤回")

    now = datetime.now()
    rows = (
        db.query(PlFreightSettlementApproval)
        .filter(
            PlFreightSettlementApproval.settlement_id == settlement_id,
            PlFreightSettlementApproval.del_flag == 0,
        )
        .all()
    )
    for row in rows:
        row.del_flag = 1
        row.update_time = now

    settlement.settlement_status = "pending"
    settlement.update_by = user_id
    settlement.update_time = now
    db.flush()
    sync_approval_notifications(
        db, "pl_freight", settlement_id, tenant_no,
        actor_id=user_id or "",
    )
    db.commit()
    db.refresh(settlement)
    return settlement


def disburse_settlement(
    db: Session,
    *,
    tenant_no: str,
    settlement_id: str,
    body: FreightSettlementDisburseBody,
    user_id: str,
    dept_id: str,
    user=None,
) -> PlFreightSettlement:
    # 锁住结算单，避免并发发放创建两份凭证和两条完成通知。
    settlement = get_settlement_row(
        db, tenant_no, settlement_id, user=user, for_update=True
    )
    if settlement.settlement_status != "approved":
        raise HTTPException(status_code=400, detail="仅待发放状态可登记")

    total = Decimal(str(settlement.total_freight or 0))
    amount = Decimal(str(body.remittance_amount))
    if amount != total and not (body.remark or "").strip():
        raise HTTPException(status_code=400, detail="汇款金额与运费合计不一致时备注必填")

    voucher = PlRemittanceVoucher(
        id=generate_uuid(),
        tenant_no=tenant_no,
        settlement_id=settlement.id,
        remittance_date=body.remittance_date,
        remittance_amount=body.remittance_amount,
        voucher_urls=json.dumps(body.voucher_urls),
        remark=body.remark,
        create_by=user_id,
        create_dept_id=dept_id or "",
    )
    db.add(voucher)
    settlement.settlement_status = "confirmed"
    settlement.update_by = user_id
    db.flush()
    sync_completion_notification(
        db, "pl_freight", settlement_id, tenant_no, actor_id=user_id, event_id=voucher.id
    )
    db.commit()
    db.refresh(settlement)
    return settlement


def get_approval_flow(db: Session, settlement_id: str) -> List[Dict[str, Any]]:
    rows = (
        db.query(PlFreightSettlementApproval)
        .filter(
            PlFreightSettlementApproval.settlement_id == settlement_id,
            PlFreightSettlementApproval.del_flag == 0,
        )
        .order_by(PlFreightSettlementApproval.step.asc(), PlFreightSettlementApproval.create_time.asc())
        .all()
    )
    return [_approval_to_dict(r) for r in rows]


def get_approval_history(db: Session, settlement_id: str) -> List[Dict[str, Any]]:
    """返回全部已完成的审批动作（含撤回/重提前的历史轮次），按时间正序。"""
    rows = (
        db.query(PlFreightSettlementApproval)
        .filter(
            PlFreightSettlementApproval.settlement_id == settlement_id,
            PlFreightSettlementApproval.approval_result.in_(["approved", "rejected"]),
        )
        .order_by(
            PlFreightSettlementApproval.approval_time.asc(),
            PlFreightSettlementApproval.create_time.asc(),
        )
        .all()
    )
    return [_approval_to_dict(r) for r in rows]


def _current_approver_name(db: Session, settlement_id: str) -> Optional[str]:
    row = (
        db.query(PlFreightSettlementApproval)
        .filter(
            PlFreightSettlementApproval.settlement_id == settlement_id,
            PlFreightSettlementApproval.approval_result == "pending",
            PlFreightSettlementApproval.is_current == 1,
            PlFreightSettlementApproval.del_flag == 0,
        )
        .order_by(PlFreightSettlementApproval.step.asc())
        .first()
    )
    return row.approver_name if row else None


def _current_approval_id(db: Session, settlement_id: str, user_id: str) -> Optional[str]:
    if not can_user_approve_settlement(db, settlement_id, user_id):
        return None
    row = (
        db.query(PlFreightSettlementApproval)
        .filter(
            PlFreightSettlementApproval.settlement_id == settlement_id,
            PlFreightSettlementApproval.approver_id == user_id,
            PlFreightSettlementApproval.approval_result == "pending",
            PlFreightSettlementApproval.is_current == 1,
            PlFreightSettlementApproval.del_flag == 0,
        )
        .first()
    )
    return row.id if row else None


def _last_approval_time(db: Session, settlement_id: str) -> Optional[datetime]:
    row = (
        db.query(PlFreightSettlementApproval)
        .filter(
            PlFreightSettlementApproval.settlement_id == settlement_id,
            PlFreightSettlementApproval.approval_result == "approved",
            PlFreightSettlementApproval.del_flag == 0,
        )
        .order_by(PlFreightSettlementApproval.approval_time.desc())
        .first()
    )
    return row.approval_time if row else None


def enrich_settlement_list_item(
    db: Session, settlement: PlFreightSettlement, user_id: str
) -> Dict[str, Any]:
    transport_ids = _get_transport_ids(db, settlement.id)
    plans = _get_plans_for_transports(db, settlement.tenant_no, transport_ids)
    plates = [p.license_plate for p in plans if p.license_plate]
    if not plates:
        transports = (
            db.query(PlTransport)
            .filter(PlTransport.id.in_(transport_ids), PlTransport.del_flag == 0)
            .all()
            if transport_ids
            else []
        )
        plates = [t.license_plate for t in transports if t.license_plate]

    voucher = (
        db.query(PlRemittanceVoucher)
        .filter(
            PlRemittanceVoucher.settlement_id == settlement.id,
            PlRemittanceVoucher.del_flag == 0,
        )
        .order_by(PlRemittanceVoucher.create_time.desc())
        .first()
    )

    submitter = _resolve_user_display_name(db, settlement.create_by or "")

    item = {
        "id": settlement.id,
        "settlementNo": settlement.settlement_no,
        "totalFreight": settlement.total_freight,
        "invoiceAmount": settlement.invoice_amount,
        "invoiceNo": settlement.invoice_no,
        "invoiceDate": settlement.invoice_date,
        "invoiceAttachmentUrls": settlement.invoice_attachment_urls,
        "settlementStatus": settlement.settlement_status,
        "remark": settlement.remark,
        "tenantNo": settlement.tenant_no,
        "orderNum": settlement.order_num,
        "createBy": settlement.create_by,
        "createDeptId": settlement.create_dept_id,
        "createTime": settlement.create_time,
        "updateBy": settlement.update_by,
        "updateTime": settlement.update_time,
        "delFlag": settlement.del_flag,
        "planCount": len(plans),
        "licensePlateSummary": "、".join(plates[:3]) + ("…" if len(plates) > 3 else ""),
        "currentApproverName": _current_approver_name(db, settlement.id),
        "canApprove": can_user_approve_settlement(db, settlement.id, user_id),
        "canWithdraw": settlement.settlement_status == "approving" and settlement.create_by == user_id,
        "currentApprovalId": _current_approval_id(db, settlement.id, user_id),
        "approvalFinishedTime": _last_approval_time(db, settlement.id),
        "disburseTime": voucher.create_time if voucher else None,
        "disburseAmount": voucher.remittance_amount if voucher else None,
        "submitterName": submitter or None,
    }
    return item


def _split_status_csv(raw: Optional[str]) -> List[str]:
    if not raw or not str(raw).strip():
        return []
    return [s.strip() for s in str(raw).split(",") if s.strip()]


def list_settlements_enriched(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    user=None,
    q: Optional[str] = None,
    carrier_id: Optional[str] = None,
    settlement_no: Optional[str] = None,
    settlement_status: Optional[str] = None,
    settlement_status_in: Optional[str] = None,
    tab: Optional[str] = None,
    ids: Optional[List[str]] = None,
    page: int = 1,
    page_size: int = 20,
    max_page_size: Optional[int] = None,
    total_freight_op: Optional[str] = None,
    total_freight: Optional[str] = None,
    total_freight_min: Optional[str] = None,
    total_freight_max: Optional[str] = None,
    create_time_from=None,
    create_time_to=None,
    submitter_name: Optional[str] = None,
    current_approver_name: Optional[str] = None,
    approval_finished_time_from=None,
    approval_finished_time_to=None,
    disburse_amount_op: Optional[str] = None,
    disburse_amount: Optional[str] = None,
    disburse_amount_min: Optional[str] = None,
    disburse_amount_max: Optional[str] = None,
    disburse_time_from=None,
    disburse_time_to=None,
) -> Tuple[int, List[Dict[str, Any]]]:
    _sync_disbursed_settlements(db, tenant_no)
    query = db.query(PlFreightSettlement).filter(
        PlFreightSettlement.tenant_no == tenant_no,
        PlFreightSettlement.del_flag == 0,
    )

    selected_statuses = _split_status_csv(settlement_status_in)
    if not selected_statuses and settlement_status and str(settlement_status).strip():
        selected_statuses = [str(settlement_status).strip()]

    if tab and tab in TAB_STATUS_MAP:
        tab_statuses = list(TAB_STATUS_MAP[tab])
        if selected_statuses:
            intersected = [s for s in selected_statuses if s in tab_statuses]
            if not intersected:
                query = query.filter(false())
            else:
                query = query.filter(PlFreightSettlement.settlement_status.in_(intersected))
        else:
            query = query.filter(PlFreightSettlement.settlement_status.in_(tab_statuses))
    elif selected_statuses:
        query = query.filter(PlFreightSettlement.settlement_status.in_(selected_statuses))

    from app.services.pl.freight_list_filters import build_settlement_column_filters

    column_filters = build_settlement_column_filters(
        settlement_no=settlement_no,
        total_freight_op=total_freight_op,
        total_freight=total_freight,
        total_freight_min=total_freight_min,
        total_freight_max=total_freight_max,
        create_time_from=create_time_from,
        create_time_to=create_time_to,
        submitter_name=submitter_name,
        current_approver_name=current_approver_name,
        approval_finished_time_from=approval_finished_time_from,
        approval_finished_time_to=approval_finished_time_to,
        disburse_amount_op=disburse_amount_op,
        disburse_amount=disburse_amount,
        disburse_amount_min=disburse_amount_min,
        disburse_amount_max=disburse_amount_max,
        disburse_time_from=disburse_time_from,
        disburse_time_to=disburse_time_to,
    )
    if column_filters:
        query = query.filter(*column_filters)

    if q:
        query = query.filter(PlFreightSettlement.settlement_no.contains(q))

    if carrier_id:
        carrier_settlement_ids = (
            db.query(PlFreightSettlementTransport.settlement_id)
            .join(PlTransport, PlTransport.id == PlFreightSettlementTransport.transport_id)
            .filter(
                PlFreightSettlementTransport.tenant_no == tenant_no,
                PlFreightSettlementTransport.del_flag == 0,
                PlTransport.del_flag == 0,
                PlTransport.carrier_id == carrier_id,
            )
            .distinct()
        )
        query = query.filter(PlFreightSettlement.id.in_(carrier_settlement_ids))
    if ids:
        query = query.filter(PlFreightSettlement.id.in_(ids))

    if user and tab != "approval_pending":
        query = apply_pl_scope(query, PlFreightSettlement, user, db, tenant_no)

    total = query.count()
    page = max(1, page)
    cap = int(max_page_size) if max_page_size else 500
    page_size = min(max(1, page_size), cap)
    rows = (
        query.order_by(PlFreightSettlement.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    status_fixed = False
    freight_fixed = False
    for row in rows:
        if _sync_settlement_status_if_approval_complete(db, row):
            status_fixed = True
        if _sync_settlement_status_if_disbursed(db, row):
            status_fixed = True
        if refresh_draft_settlement_freight(db, row, user_id=user_id):
            freight_fixed = True
    if status_fixed or freight_fixed:
        db.commit()
        for row in rows:
            db.refresh(row)

    if tab == "bills":
        bills_statuses = set(TAB_STATUS_MAP["bills"])
        rows = [r for r in rows if r.settlement_status in bills_statuses]

    return total, [enrich_settlement_list_item(db, r, user_id) for r in rows]


def get_settlement_detail(
    db: Session, *, tenant_no: str, settlement_id: str, user_id: str, user=None
) -> Dict[str, Any]:
    settlement = get_settlement_row(db, tenant_no, settlement_id, user=user)
    if _sync_settlement_status_if_approval_complete(db, settlement):
        db.commit()
        db.refresh(settlement)
    if _sync_settlement_status_if_disbursed(db, settlement):
        db.commit()
        db.refresh(settlement)
    if refresh_draft_settlement_freight(db, settlement, user_id=user_id):
        db.commit()
        db.refresh(settlement)
    base = enrich_settlement_list_item(db, settlement, user_id)

    transport_ids = _get_transport_ids(db, settlement_id)
    transports = (
        db.query(PlTransport)
        .filter(PlTransport.id.in_(transport_ids), PlTransport.del_flag == 0)
        .all()
        if transport_ids
        else []
    )
    plans = _get_plans_for_transports(db, tenant_no, transport_ids)
    plan_summaries = [
        {
            "id": p.id,
            "planNo": p.plan_no,
            "planDate": p.plan_date,
            "licensePlate": p.license_plate,
            "freightAmount": p.freight_amount,
            "freightRemark": p.freight_remark,
        }
        for p in plans
    ]

    vouchers = (
        db.query(PlRemittanceVoucher)
        .filter(
            PlRemittanceVoucher.settlement_id == settlement_id,
            PlRemittanceVoucher.del_flag == 0,
        )
        .order_by(PlRemittanceVoucher.create_time.desc())
        .all()
    )

    transport_payload = []
    for t in transports:
        transport_payload.append(
            {
                "id": t.id,
                "transportNo": t.transport_no,
                "licensePlate": t.license_plate,
                "driverName": t.driver_name,
                "carrierName": t.carrier_name,
                "transportMedium": t.transport_medium,
                "startPoint": t.start_point,
                "endPoint": t.end_point,
                "freightAmount": t.freight_amount,
                "transportStatus": t.transport_status,
            }
        )

    voucher_payload = []
    for v in vouchers:
        voucher_payload.append(
            {
                "id": v.id,
                "settlementId": v.settlement_id,
                "remittanceDate": v.remittance_date,
                "remittanceAmount": v.remittance_amount,
                "voucherUrls": v.voucher_urls,
                "remark": v.remark,
                "createBy": v.create_by,
                "createTime": v.create_time,
            }
        )

    base["transports"] = transport_payload
    base["plans"] = plan_summaries
    base["approvalFlow"] = get_approval_flow(db, settlement_id)
    base["approvalHistory"] = get_approval_history(db, settlement_id)
    base["vouchers"] = voucher_payload
    return base


def _validate_transports_for_settlement_update(
    db: Session,
    *,
    tenant_no: str,
    settlement_id: str,
    transport_ids: List[str],
) -> List[PlTransport]:
    cleaned = list(dict.fromkeys([tid.strip() for tid in transport_ids if (tid or "").strip()]))
    if not cleaned:
        raise HTTPException(status_code=400, detail="至少保留一条关联运输")

    transports = (
        db.query(PlTransport)
        .filter(
            PlTransport.id.in_(cleaned),
            PlTransport.tenant_no == tenant_no,
            PlTransport.del_flag == 0,
        )
        .all()
    )
    if len(transports) != len(cleaned):
        raise HTTPException(status_code=400, detail="存在无效的运输单")

    transport_map = {t.id: t for t in transports}
    current_ids = set(_get_transport_ids(db, settlement_id))
    added_ids = [tid for tid in cleaned if tid not in current_ids]

    if added_ids:
        occupied = {r[0] for r in _occupied_transport_ids_subquery(db, tenant_no).all()}
        conflict = [tid for tid in added_ids if tid in occupied]
        if conflict:
            raise HTTPException(status_code=400, detail="所选运输单已纳入其他结算单")

        plans = (
            db.query(PlPlan)
            .filter(
                PlPlan.tenant_no == tenant_no,
                PlPlan.del_flag == 0,
                PlPlan.transport_id.in_(added_ids),
            )
            .all()
        )
        plan_by_transport = {p.transport_id: p for p in plans if p.transport_id}
        for tid in added_ids:
            plan = plan_by_transport.get(tid)
            if not plan:
                raise HTTPException(status_code=400, detail="存在未关联计划的运输单")
            if plan.archive_status != "archived":
                raise HTTPException(status_code=400, detail=f"计划 {plan.plan_no} 未归档，不能结算")
            if int(plan.has_freight or 0) != 1:
                raise HTTPException(status_code=400, detail=f"计划 {plan.plan_no} 无运输费用")
            if plan.follow_status == "cancelled":
                raise HTTPException(status_code=400, detail=f"计划 {plan.plan_no} 已作废")
            amount = Decimal(str(transport_map[tid].freight_amount or 0))
            if amount <= 0:
                raise HTTPException(status_code=400, detail=f"运输单 {transport_map[tid].transport_no} 运费无效")

    return [transport_map[tid] for tid in cleaned]


def _sync_settlement_transport_links(
    db: Session,
    *,
    tenant_no: str,
    settlement_id: str,
    transport_ids: List[str],
    transports: List[PlTransport],
    user_id: str,
    dept_id: str,
) -> None:
    cleaned = list(dict.fromkeys([tid.strip() for tid in transport_ids if (tid or "").strip()]))
    transport_map = {t.id: t for t in transports}
    now = datetime.now()

    current_links = (
        db.query(PlFreightSettlementTransport)
        .filter(
            PlFreightSettlementTransport.settlement_id == settlement_id,
            PlFreightSettlementTransport.del_flag == 0,
        )
        .all()
    )
    current_map = {link.transport_id: link for link in current_links}
    new_set = set(cleaned)

    for tid, link in current_map.items():
        if tid not in new_set:
            link.del_flag = 1
            link.update_by = user_id
            link.update_time = now

    for tid in cleaned:
        transport = transport_map[tid]
        amount = transport.freight_amount
        if tid in current_map:
            link = current_map[tid]
            if Decimal(str(link.freight_amount or 0)) != Decimal(str(amount or 0)):
                link.freight_amount = amount
                link.update_by = user_id
                link.update_time = now
            continue
        db.add(
            PlFreightSettlementTransport(
                id=generate_uuid(),
                tenant_no=tenant_no,
                settlement_id=settlement_id,
                transport_id=tid,
                freight_amount=amount,
                create_by=user_id,
                create_dept_id=dept_id or "",
            )
        )

    _recalc_settlement_total_freight_from_links(db, settlement_id, user_id=user_id, now=now)


def update_settlement_draft(
    db: Session,
    *,
    tenant_no: str,
    settlement_id: str,
    payload: FreightSettlementUpdate,
    user_id: str,
    user=None,
) -> PlFreightSettlement:
    settlement = get_settlement_row(db, tenant_no, settlement_id, user=user)
    status = settlement.settlement_status
    if status not in (*_DRAFT_SETTLEMENT_STATUSES, "approved"):
        raise HTTPException(status_code=400, detail="当前状态不可编辑")

    update_data = payload.model_dump(exclude_unset=True, by_alias=False)
    transport_ids = update_data.pop("transport_ids", None)

    if status == "approved":
        if transport_ids is not None:
            raise HTTPException(status_code=400, detail="当前状态不可修改关联运输")
        allowed = {"invoice_amount", "invoice_no", "invoice_date", "invoice_attachment_urls", "remark"}
        update_data = {k: v for k, v in update_data.items() if k in allowed}
        if not update_data:
            raise HTTPException(status_code=400, detail="无有效更新字段")
    elif status in _DRAFT_SETTLEMENT_STATUSES and transport_ids is not None:
        dept_id = getattr(user, "dept_id", "") if user else ""
        transports = _validate_transports_for_settlement_update(
            db,
            tenant_no=tenant_no,
            settlement_id=settlement_id,
            transport_ids=transport_ids,
        )
        _sync_settlement_transport_links(
            db,
            tenant_no=tenant_no,
            settlement_id=settlement_id,
            transport_ids=transport_ids,
            transports=transports,
            user_id=user_id,
            dept_id=dept_id,
        )

    for key, value in update_data.items():
        setattr(settlement, key, value)
    settlement.update_by = user_id
    db.commit()
    db.refresh(settlement)
    return settlement