zkxw/backend/app/crud/sample_freight.py

"""样本运费批次 CRUD（biz_sample_freight / biz_sample_freight_item）"""

from __future__ import annotations

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

from sqlalchemy import and_, exists, func, or_
from sqlalchemy.orm import Session

from app.core.data_scope import apply_data_scope, get_user_dept_id
from app.core.utils import generate_uuid
from app.models.models import (
    BizCellSampleType,
    BizCollectionProtocol,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizSampleFreight,
    BizSampleFreightItem,
    SysDictData,
    SysUser,
)
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationParty,
    BizResearchDonationSample,
)
from app.services.sample_transport_mode import DICT_TYPE, resolve_transport_mode_value
from app.services.sf_express_settlement import (
    assert_freight_mutable,
    freight_amount_label,
    resolve_header_for_save,
)

SAMPLE_SOURCE_CELL = "cell_storage"
SAMPLE_SOURCE_DONATION = "research_donation"


def _now() -> datetime:
    return datetime.now()


def _fmt_dt(t: Optional[datetime]) -> str:
    return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""


def _parse_amount(amount: Any) -> Decimal:
    if amount is None or amount == "":
        raise ValueError("运费不能为空")
    try:
        val = Decimal(str(amount))
    except Exception as exc:
        raise ValueError("运费格式无效") from exc
    if not val.is_finite():
        raise ValueError("运费格式无效")
    if val < 0:
        raise ValueError("运费不能为负数")
    return val


def _item_field(raw: Dict[str, Any], camel: str, snake: str, default: str = "") -> str:
    v = raw.get(camel)
    if v is None or v == "":
        v = raw.get(snake)
    if v is None:
        return default
    return str(v).strip()


def _normalize_item(raw: Any) -> Dict[str, Any]:
    if not isinstance(raw, dict):
        raise ValueError("关联样本项无效")
    sample_source = _item_field(raw, "sampleSource", "sample_source")
    sample_id = _item_field(raw, "sampleId", "sample_id")
    if not sample_source or not sample_id:
        raise ValueError("关联样本须包含 sampleSource 与 sampleId")
    if sample_source not in (SAMPLE_SOURCE_CELL, SAMPLE_SOURCE_DONATION):
        raise ValueError("sampleSource 无效")
    protocol_id = _item_field(raw, "protocolId", "protocol_id")
    sample_code = _item_field(raw, "sampleCode", "sample_code")
    return {
        "sample_source": sample_source,
        "sample_id": sample_id,
        "protocol_id": protocol_id or None,
        "sample_code": sample_code or None,
    }


def _label_map(db: Session, tenant_no: str, modes: List[str]) -> Dict[str, str]:
    uniq = [m for m in {str(x).strip() for x in modes if str(x or "").strip()}]
    if not uniq:
        return {}
    rows = (
        db.query(SysDictData.dict_value, SysDictData.dict_label)
        .filter(
            SysDictData.dict_type == DICT_TYPE,
            SysDictData.tenant_no == tenant_no,
            SysDictData.dict_value.in_(uniq),
            SysDictData.del_flag == 0,
        )
        .all()
    )
    return {str(v): (lbl or v or "") for v, lbl in rows}


def _linked_count(db: Session, freight_id: str, tenant_no: str) -> int:
    return (
        db.query(func.count(BizSampleFreightItem.id))
        .filter(
            BizSampleFreightItem.freight_id == freight_id,
            BizSampleFreightItem.tenant_no == tenant_no,
            BizSampleFreightItem.del_flag == 0,
        )
        .scalar()
        or 0
    )


def _user_nick_name_map(db: Session, user_ids: List[str]) -> Dict[str, str]:
    ids = list({x for x in user_ids if x})
    if not ids:
        return {}
    rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(ids)).all()
    return {u.id: (u.nick_name or "").strip() for u in rows}


def _generate_freight_no(
    db: Session, tenant_no: str, when: Optional[datetime] = None
) -> str:
    """生成批次号：YF + YYYYMMDD + 001～999（按租户、按日）。"""
    day = (when or _now()).strftime("%Y%m%d")
    prefix = f"YF{day}"
    existing = (
        db.query(BizSampleFreight.freight_no)
        .filter(
            BizSampleFreight.tenant_no == tenant_no,
            BizSampleFreight.freight_no.like(f"{prefix}%"),
        )
        .order_by(BizSampleFreight.freight_no.desc())
        .first()
    )
    seq = 1
    if existing and existing[0]:
        try:
            seq = int(str(existing[0])[-3:]) + 1
        except ValueError:
            seq = 1
    if seq > 999:
        raise ValueError("当日运费批次号已达上限（999）")
    return f"{prefix}{seq:03d}"


def _row_to_dict(
    obj: BizSampleFreight,
    *,
    linked_sample_count: int = 0,
    transport_mode_label: str = "",
    create_by_name: str = "",
) -> Dict[str, Any]:
    amount = obj.freight_amount
    create_by = obj.create_by or ""
    return {
        "id": obj.id,
        "freightNo": getattr(obj, "freight_no", None) or "",
        "transportMode": obj.transport_mode or "",
        "transportModeLabel": transport_mode_label or (obj.transport_mode or ""),
        "freightAmount": float(amount) if amount is not None else None,
        "freightAmountLabel": freight_amount_label(obj.transport_mode, amount),
        "waybillNo": getattr(obj, "waybill_no", None) or "",
        "settleStatus": getattr(obj, "settle_status", None) or "",
        "settleTime": _fmt_dt(getattr(obj, "settle_time", None))
        if getattr(obj, "settle_time", None)
        else "",
        "linkedSampleCount": int(linked_sample_count or 0),
        "tenantNo": obj.tenant_no,
        "orderNum": obj.order_num if obj.order_num is not None else 0,
        "remark": obj.remark or "",
        "createBy": create_by,
        "createByName": create_by_name or create_by,
        "createTime": _fmt_dt(obj.create_time),
        "updateTime": _fmt_dt(obj.update_time),
    }


def _item_to_dict(item: BizSampleFreightItem) -> Dict[str, Any]:
    return {
        "id": item.id,
        "freightId": item.freight_id,
        "sampleSource": item.sample_source or "",
        "sampleId": item.sample_id or "",
        "sampleCode": item.sample_code or "",
        "protocolId": item.protocol_id or "",
    }


def _item_to_dict_enriched(
    item: BizSampleFreightItem,
    *,
    protocol_no: str = "",
    sample_type_name: str = "",
    mother_name: str = "",
    newborn_name: str = "",
) -> Dict[str, Any]:
    data = _item_to_dict(item)
    data["protocolNo"] = protocol_no or ""
    data["sampleTypeName"] = sample_type_name or ""
    data["motherName"] = mother_name or ""
    data["newbornName"] = newborn_name or ""
    return data


def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> datetime:
    raw = (value or "").strip()
    if not raw:
        raise ValueError("时间参数无效")
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
        try:
            dt = datetime.strptime(raw, fmt)
            if fmt == "%Y-%m-%d" and end_of_day:
                dt = dt + timedelta(days=1) - timedelta(seconds=1)
            return dt
        except ValueError:
            continue
    try:
        return datetime.fromisoformat(raw.replace("Z", "+00:00")).replace(tzinfo=None)
    except ValueError as exc:
        raise ValueError("时间参数格式无效") from exc


def _enrich_items_maps(
    db: Session,
    items: List[BizSampleFreightItem],
    tenant_no: str,
) -> Dict[str, Dict[str, str]]:
    """批量解析协议号/样本类型/母亲子女姓名，key = sample_source|sample_id。"""
    out: Dict[str, Dict[str, str]] = {}
    if not items:
        return out

    cell_ids = [
        i.sample_id
        for i in items
        if i.sample_source == SAMPLE_SOURCE_CELL and i.sample_id
    ]
    donation_ids = [
        i.sample_id
        for i in items
        if i.sample_source == SAMPLE_SOURCE_DONATION and i.sample_id
    ]

    if cell_ids:
        rows = (
            db.query(
                BizProtocolSample.id,
                BizCollectionProtocol.protocol_no,
                BizCellSampleType.type_name,
                BizProtocolPartyCustomer.mother_name,
                BizProtocolPartyCustomer.newborn_name,
            )
            .outerjoin(
                BizCollectionProtocol,
                and_(
                    BizCollectionProtocol.id == BizProtocolSample.collection_protocol_id,
                    BizCollectionProtocol.del_flag == 0,
                ),
            )
            .outerjoin(
                BizCellSampleType,
                and_(
                    BizCellSampleType.id == BizProtocolSample.sample_type_id,
                    BizCellSampleType.del_flag == 0,
                ),
            )
            .outerjoin(
                BizProtocolPartyCustomer,
                and_(
                    BizProtocolPartyCustomer.collection_protocol_id
                    == BizProtocolSample.collection_protocol_id,
                    BizProtocolPartyCustomer.del_flag == 0,
                ),
            )
            .filter(
                BizProtocolSample.id.in_(cell_ids),
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
            )
            .all()
        )
        for sid, pno, tname, mname, nname in rows:
            out[f"{SAMPLE_SOURCE_CELL}|{sid}"] = {
                "protocolNo": pno or "",
                "sampleTypeName": tname or "",
                "motherName": mname or "",
                "newbornName": nname or "",
            }

    if donation_ids:
        rows = (
            db.query(
                BizResearchDonationSample.id,
                BizResearchDonation.donation_no,
                BizCellSampleType.type_name,
                BizResearchDonationParty.mother_name,
                BizResearchDonationParty.newborn_name,
            )
            .outerjoin(
                BizResearchDonation,
                and_(
                    BizResearchDonation.id
                    == BizResearchDonationSample.research_donation_id,
                    BizResearchDonation.del_flag == 0,
                ),
            )
            .outerjoin(
                BizCellSampleType,
                and_(
                    BizCellSampleType.id == BizResearchDonationSample.sample_type_id,
                    BizCellSampleType.del_flag == 0,
                ),
            )
            .outerjoin(
                BizResearchDonationParty,
                and_(
                    BizResearchDonationParty.research_donation_id
                    == BizResearchDonationSample.research_donation_id,
                    BizResearchDonationParty.del_flag == 0,
                ),
            )
            .filter(
                BizResearchDonationSample.id.in_(donation_ids),
                BizResearchDonationSample.tenant_no == tenant_no,
                BizResearchDonationSample.del_flag == 0,
            )
            .all()
        )
        for sid, dno, tname, mname, nname in rows:
            out[f"{SAMPLE_SOURCE_DONATION}|{sid}"] = {
                "protocolNo": dno or "",
                "sampleTypeName": tname or "",
                "motherName": mname or "",
                "newbornName": nname or "",
            }

    return out


def get_freight(
    db: Session, freight_id: str, tenant_no: str, user_info: Optional[dict] = None
) -> Optional[BizSampleFreight]:
    query = db.query(BizSampleFreight).filter(
        BizSampleFreight.id == freight_id,
        BizSampleFreight.tenant_no == tenant_no,
        BizSampleFreight.del_flag == 0,
    )
    return apply_data_scope(query, BizSampleFreight, user_info, db, tenant_no).first()


def _validated_items(
    db: Session, items: List[Dict[str, Any]], tenant_no: str, user_info: dict
) -> List[Dict[str, Any]]:
    """写入前校验全部样本的租户和数据范围，快照只从数据库派生。"""
    validated = []
    for raw in items:
        item = _normalize_item(raw)
        if item["sample_source"] == SAMPLE_SOURCE_CELL:
            sample_model, parent_model = BizProtocolSample, BizCollectionProtocol
            parent_key = BizProtocolSample.collection_protocol_id
        else:
            sample_model, parent_model = BizResearchDonationSample, BizResearchDonation
            parent_key = BizResearchDonationSample.research_donation_id
        query = db.query(sample_model, parent_model).join(
            parent_model, parent_model.id == parent_key
        ).filter(
            sample_model.id == item["sample_id"],
            sample_model.tenant_no == tenant_no,
            sample_model.del_flag == 0,
            parent_model.tenant_no == tenant_no,
            parent_model.del_flag == 0,
        )
        row = apply_data_scope(query, parent_model, user_info, db, tenant_no).first()
        if not row:
            raise PermissionError("样本不存在或无权访问")
        existing_links = db.query(BizSampleFreightItem).filter(
            BizSampleFreightItem.tenant_no == tenant_no,
            BizSampleFreightItem.sample_source == item["sample_source"],
            BizSampleFreightItem.sample_id == item["sample_id"],
            BizSampleFreightItem.del_flag == 0,
        ).all()
        if any(
            not get_freight(db, link.freight_id, tenant_no, user_info)
            for link in existing_links
        ):
            raise PermissionError("无权修改样本当前关联的运费批次")
        sample, parent = row
        item["sample_code"] = sample.sample_code
        item["protocol_id"] = parent.id
        validated.append(item)
    return validated


def list_freights(
    db: Session,
    *,
    tenant_no: str,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    transport_mode: Optional[str] = None,
    sample_source: Optional[str] = None,
    create_time_from: Optional[str] = None,
    create_time_to: Optional[str] = None,
    include_items: bool = False,
    user_info: Optional[dict] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    page = max(1, int(page or 1))
    page_size = max(1, min(100, int(page_size or 10)))
    q = db.query(BizSampleFreight).filter(
        BizSampleFreight.tenant_no == tenant_no,
        BizSampleFreight.del_flag == 0,
    )

    q = apply_data_scope(q, BizSampleFreight, user_info, db, tenant_no)

    mode = (transport_mode or "").strip()
    if mode:
        q = q.filter(BizSampleFreight.transport_mode == mode)

    src = (sample_source or "").strip()
    if src:
        if src not in (SAMPLE_SOURCE_CELL, SAMPLE_SOURCE_DONATION):
            raise ValueError("sampleSource 无效")
        item_exists = exists().where(
            and_(
                BizSampleFreightItem.freight_id == BizSampleFreight.id,
                BizSampleFreightItem.tenant_no == tenant_no,
                BizSampleFreightItem.del_flag == 0,
                BizSampleFreightItem.sample_source == src,
            )
        )
        q = q.filter(item_exists)

    if create_time_from and str(create_time_from).strip():
        q = q.filter(
            BizSampleFreight.create_time
            >= _normalize_date_bound(str(create_time_from).strip())
        )
    if create_time_to and str(create_time_to).strip():
        q = q.filter(
            BizSampleFreight.create_time
            <= _normalize_date_bound(str(create_time_to).strip(), end_of_day=True)
        )

    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        label_values = [
            r[0]
            for r in db.query(SysDictData.dict_value)
            .filter(
                SysDictData.dict_type == DICT_TYPE,
                SysDictData.tenant_no == tenant_no,
                SysDictData.del_flag == 0,
                SysDictData.dict_label.like(like),
            )
            .all()
        ]
        conds = [
            BizSampleFreight.transport_mode.like(like),
            BizSampleFreight.freight_no.like(like),
        ]
        if label_values:
            conds.append(BizSampleFreight.transport_mode.in_(label_values))

        sample_code_exists = exists().where(
            and_(
                BizSampleFreightItem.freight_id == BizSampleFreight.id,
                BizSampleFreightItem.tenant_no == tenant_no,
                BizSampleFreightItem.del_flag == 0,
                BizSampleFreightItem.sample_code.like(like),
            )
        )
        conds.append(sample_code_exists)

        q = q.filter(or_(*conds))

    total = q.count()
    rows = (
        q.order_by(BizSampleFreight.create_time.desc(), BizSampleFreight.id.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    labels = _label_map(db, tenant_no, [r.transport_mode for r in rows])
    name_map = _user_nick_name_map(db, [r.create_by for r in rows if r.create_by])

    items_by_freight: Dict[str, List[BizSampleFreightItem]] = {}
    if include_items and rows:
        freight_ids = [r.id for r in rows]
        all_items = (
            db.query(BizSampleFreightItem)
            .filter(
                BizSampleFreightItem.freight_id.in_(freight_ids),
                BizSampleFreightItem.tenant_no == tenant_no,
                BizSampleFreightItem.del_flag == 0,
            )
            .order_by(
                BizSampleFreightItem.create_time.asc(),
                BizSampleFreightItem.id.asc(),
            )
            .all()
        )
        for it in all_items:
            items_by_freight.setdefault(it.freight_id, []).append(it)
        enrich_map = _enrich_items_maps(db, all_items, tenant_no)
    else:
        enrich_map = {}

    out: List[Dict[str, Any]] = []
    for r in rows:
        linked = (
            len(items_by_freight.get(r.id, []))
            if include_items
            else _linked_count(db, r.id, tenant_no)
        )
        data = _row_to_dict(
            r,
            linked_sample_count=linked,
            transport_mode_label=labels.get(r.transport_mode or "", r.transport_mode or ""),
            create_by_name=name_map.get(r.create_by or "", ""),
        )
        if include_items:
            enriched = []
            for it in items_by_freight.get(r.id, []):
                key = f"{it.sample_source}|{it.sample_id}"
                info = enrich_map.get(key, {})
                enriched.append(
                    _item_to_dict_enriched(
                        it,
                        protocol_no=info.get("protocolNo", ""),
                        sample_type_name=info.get("sampleTypeName", ""),
                        mother_name=info.get("motherName", ""),
                        newborn_name=info.get("newbornName", ""),
                    )
                )
            data["items"] = enriched
        out.append(data)
    return out, total


def get_freight_detail(
    db: Session, freight_id: str, tenant_no: str, user_info: Optional[dict] = None
) -> Dict[str, Any]:
    obj = get_freight(db, freight_id, tenant_no, user_info)
    if not obj:
        raise ValueError("运费批次不存在")
    items = (
        db.query(BizSampleFreightItem)
        .filter(
            BizSampleFreightItem.freight_id == freight_id,
            BizSampleFreightItem.tenant_no == tenant_no,
            BizSampleFreightItem.del_flag == 0,
        )
        .order_by(BizSampleFreightItem.create_time.asc())
        .all()
    )
    labels = _label_map(db, tenant_no, [obj.transport_mode])
    name_map = _user_nick_name_map(db, [obj.create_by] if obj.create_by else [])
    data = _row_to_dict(
        obj,
        linked_sample_count=len(items),
        transport_mode_label=labels.get(obj.transport_mode or "", obj.transport_mode or ""),
        create_by_name=name_map.get(obj.create_by or "", ""),
    )
    enrich_map = _enrich_items_maps(db, items, tenant_no)
    enriched = []
    for i in items:
        key = f"{i.sample_source}|{i.sample_id}"
        info = enrich_map.get(key, {})
        enriched.append(
            _item_to_dict_enriched(
                i,
                protocol_no=info.get("protocolNo", ""),
                sample_type_name=info.get("sampleTypeName", ""),
                mother_name=info.get("motherName", ""),
                newborn_name=info.get("newbornName", ""),
            )
        )
    data["items"] = enriched
    return data


def list_sample_options(
    db: Session,
    *,
    tenant_no: str,
    page: int = 1,
    page_size: int = 20,
    keyword: Optional[str] = None,
    sample_source: Optional[str] = None,
    user_info: Optional[dict] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    """可选样本列表（配置运费弹窗），支持细胞存储 / 科研捐赠。"""
    page = max(1, int(page or 1))
    page_size = max(1, min(100, int(page_size or 20)))
    src = (sample_source or "").strip() or None
    if src and src not in (SAMPLE_SOURCE_CELL, SAMPLE_SOURCE_DONATION):
        raise ValueError("sampleSource 无效")
    kw = (keyword or "").strip()
    like = f"%{kw}%" if kw else None

    rows: List[Dict[str, Any]] = []

    if src in (None, SAMPLE_SOURCE_CELL):
        cq = (
            db.query(
                BizProtocolSample.id,
                BizProtocolSample.sample_code,
                BizProtocolSample.collection_protocol_id,
                BizProtocolSample.create_time,
                BizCollectionProtocol.protocol_no,
                BizCellSampleType.type_name,
                BizProtocolPartyCustomer.mother_name,
                BizProtocolPartyCustomer.newborn_name,
            )
            .outerjoin(
                BizCollectionProtocol,
                and_(
                    BizCollectionProtocol.id
                    == BizProtocolSample.collection_protocol_id,
                    BizCollectionProtocol.del_flag == 0,
                ),
            )
            .outerjoin(
                BizCellSampleType,
                and_(
                    BizCellSampleType.id == BizProtocolSample.sample_type_id,
                    BizCellSampleType.del_flag == 0,
                ),
            )
            .outerjoin(
                BizProtocolPartyCustomer,
                and_(
                    BizProtocolPartyCustomer.collection_protocol_id
                    == BizProtocolSample.collection_protocol_id,
                    BizProtocolPartyCustomer.del_flag == 0,
                ),
            )
            .filter(
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
            )
        )
        cq = cq.filter(
            BizCollectionProtocol.tenant_no == tenant_no,
            BizCollectionProtocol.del_flag == 0,
        )
        cq = apply_data_scope(cq, BizCollectionProtocol, user_info, db, tenant_no)
        if like:
            cq = cq.filter(
                or_(
                    BizProtocolSample.sample_code.like(like),
                    BizProtocolPartyCustomer.mother_name.like(like),
                    BizProtocolPartyCustomer.newborn_name.like(like),
                )
            )
        for sid, scode, pid, ctime, pno, tname, mname, nname in cq.all():
            rows.append(
                {
                    "sampleSource": SAMPLE_SOURCE_CELL,
                    "sampleId": sid,
                    "sampleCode": scode or "",
                    "protocolId": pid or "",
                    "protocolNo": pno or "",
                    "sampleTypeName": tname or "",
                    "motherName": mname or "",
                    "newbornName": nname or "",
                    "createTime": ctime,
                    "freightId": "",
                }
            )

    if src in (None, SAMPLE_SOURCE_DONATION):
        dq = (
            db.query(
                BizResearchDonationSample.id,
                BizResearchDonationSample.sample_code,
                BizResearchDonationSample.research_donation_id,
                BizResearchDonationSample.create_time,
                BizResearchDonation.donation_no,
                BizCellSampleType.type_name,
                BizResearchDonationParty.mother_name,
                BizResearchDonationParty.newborn_name,
            )
            .outerjoin(
                BizResearchDonation,
                and_(
                    BizResearchDonation.id
                    == BizResearchDonationSample.research_donation_id,
                    BizResearchDonation.del_flag == 0,
                ),
            )
            .outerjoin(
                BizCellSampleType,
                and_(
                    BizCellSampleType.id == BizResearchDonationSample.sample_type_id,
                    BizCellSampleType.del_flag == 0,
                ),
            )
            .outerjoin(
                BizResearchDonationParty,
                and_(
                    BizResearchDonationParty.research_donation_id
                    == BizResearchDonationSample.research_donation_id,
                    BizResearchDonationParty.del_flag == 0,
                ),
            )
            .filter(
                BizResearchDonationSample.tenant_no == tenant_no,
                BizResearchDonationSample.del_flag == 0,
            )
        )
        dq = dq.filter(
            BizResearchDonation.tenant_no == tenant_no,
            BizResearchDonation.del_flag == 0,
        )
        dq = apply_data_scope(dq, BizResearchDonation, user_info, db, tenant_no)
        if like:
            dq = dq.filter(
                or_(
                    BizResearchDonationSample.sample_code.like(like),
                    BizResearchDonationParty.mother_name.like(like),
                    BizResearchDonationParty.newborn_name.like(like),
                )
            )
        for sid, scode, pid, ctime, dno, tname, mname, nname in dq.all():
            rows.append(
                {
                    "sampleSource": SAMPLE_SOURCE_DONATION,
                    "sampleId": sid,
                    "sampleCode": scode or "",
                    "protocolId": pid or "",
                    "protocolNo": dno or "",
                    "sampleTypeName": tname or "",
                    "motherName": mname or "",
                    "newbornName": nname or "",
                    "createTime": ctime,
                    "freightId": "",
                }
            )

    # 左连当前运费关联
    if rows:
        sample_ids = [r["sampleId"] for r in rows]
        link_rows = (
            db.query(
                BizSampleFreightItem.sample_source,
                BizSampleFreightItem.sample_id,
                BizSampleFreightItem.freight_id,
            )
            .filter(
                BizSampleFreightItem.tenant_no == tenant_no,
                BizSampleFreightItem.del_flag == 0,
                BizSampleFreightItem.sample_id.in_(sample_ids),
            )
            .all()
        )
        freight_map = {
            f"{src_}|{sid}": fid for src_, sid, fid in link_rows if src_ and sid and fid
        }
        for r in rows:
            r["freightId"] = freight_map.get(f"{r['sampleSource']}|{r['sampleId']}", "") or ""

    rows.sort(
        key=lambda r: (r["createTime"] or datetime.min, r["sampleId"]),
        reverse=True,
    )
    total = len(rows)
    start = (page - 1) * page_size
    page_rows = rows[start : start + page_size]
    for r in page_rows:
        r["createTime"] = _fmt_dt(r.get("createTime"))
    return page_rows, total


def _upsert_item(
    db: Session,
    *,
    freight_id: str,
    tenant_no: str,
    user_id: Optional[str],
    item: Dict[str, Any],
) -> None:
    norm = _normalize_item(item)
    existing = (
        db.query(BizSampleFreightItem)
        .filter(
            BizSampleFreightItem.tenant_no == tenant_no,
            BizSampleFreightItem.sample_source == norm["sample_source"],
            BizSampleFreightItem.sample_id == norm["sample_id"],
        )
        .order_by(BizSampleFreightItem.del_flag.asc(), BizSampleFreightItem.create_time.desc())
        .first()
    )
    now = _now()
    if existing:
        existing.freight_id = freight_id
        existing.protocol_id = norm["protocol_id"] or existing.protocol_id
        existing.sample_code = norm["sample_code"] or existing.sample_code
        existing.del_flag = 0
        existing.update_by = user_id
        existing.update_time = now
        db.add(existing)
        return
    db.add(
        BizSampleFreightItem(
            id=generate_uuid(),
            freight_id=freight_id,
            sample_source=norm["sample_source"],
            sample_id=norm["sample_id"],
            sample_code=norm["sample_code"],
            protocol_id=norm["protocol_id"],
            tenant_no=tenant_no,
            order_num=0,
            remark=None,
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
    )


def create_freight(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str],
    transport_mode_raw: str,
    freight_amount: Any,
    items: Optional[List[Dict[str, Any]]] = None,
    commit: bool = True,
    user_info: Optional[dict] = None,
    waybill_no: Any = None,
    waybill_supplied: bool = True,
    amount_supplied: bool = True,
) -> Dict[str, Any]:
    if user_info is not None:
        items = _validated_items(db, items or [], tenant_no, user_info)
    mode = resolve_transport_mode_value(
        db, tenant_no=tenant_no, raw=transport_mode_raw, user_id=user_id
    )
    plan = resolve_header_for_save(
        transport_mode=mode,
        existing_waybill=None,
        existing_amount=None,
        existing_settle_status=None,
        waybill_supplied=waybill_supplied,
        waybill_no=waybill_no,
        amount_supplied=amount_supplied,
        freight_amount=freight_amount,
    )
    now = _now()
    obj = BizSampleFreight(
        id=generate_uuid(),
        freight_no=_generate_freight_no(db, tenant_no, when=now),
        transport_mode=plan.transport_mode,
        freight_amount=plan.freight_amount,
        waybill_no=plan.waybill_no,
        settle_status=plan.settle_status,
        settle_time=None,
        tenant_no=tenant_no,
        order_num=0,
        remark=None,
        create_by=user_id,
        create_dept_id=(get_user_dept_id(db, user_id) or "") if user_info is not None else "",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    db.flush()
    for it in items or []:
        _upsert_item(db, freight_id=obj.id, tenant_no=tenant_no, user_id=user_id, item=it)
    if commit:
        db.commit()
    else:
        db.flush()
    return get_freight_detail(db, obj.id, tenant_no)


def update_freight(
    db: Session,
    freight_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str],
    transport_mode_raw: Optional[str] = None,
    freight_amount: Any = None,
    commit: bool = True,
    user_info: Optional[dict] = None,
    waybill_no: Any = None,
    waybill_supplied: bool = False,
    amount_supplied: bool = False,
) -> Dict[str, Any]:
    obj = get_freight(db, freight_id, tenant_no, user_info)
    if not obj:
        raise ValueError("运费批次不存在")
    mode = obj.transport_mode or ""
    if transport_mode_raw is not None and str(transport_mode_raw).strip() != "":
        mode = resolve_transport_mode_value(
            db, tenant_no=tenant_no, raw=str(transport_mode_raw), user_id=user_id
        )
    plan = resolve_header_for_save(
        transport_mode=mode,
        existing_waybill=obj.waybill_no,
        existing_amount=obj.freight_amount,
        existing_settle_status=obj.settle_status,
        waybill_supplied=waybill_supplied,
        waybill_no=waybill_no,
        amount_supplied=amount_supplied,
        freight_amount=freight_amount,
    )
    obj.transport_mode = plan.transport_mode
    obj.waybill_no = plan.waybill_no
    obj.freight_amount = plan.freight_amount
    obj.settle_status = plan.settle_status
    if plan.clear_settle_time:
        obj.settle_time = None
    obj.update_by = user_id
    obj.update_time = _now()
    db.add(obj)
    if commit:
        db.commit()
    else:
        db.flush()
    return get_freight_detail(db, freight_id, tenant_no)


def link_samples(
    db: Session,
    freight_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str],
    items: List[Dict[str, Any]],
    commit: bool = True,
    user_info: Optional[dict] = None,
) -> Dict[str, Any]:
    header = get_freight(db, freight_id, tenant_no, user_info)
    if not header:
        raise ValueError("运费批次不存在")
    assert_freight_mutable(header.settle_status, "items")
    if not items:
        raise ValueError("请选择要关联的样本")
    if user_info is not None:
        items = _validated_items(db, items, tenant_no, user_info)
    for it in items:
        _upsert_item(db, freight_id=freight_id, tenant_no=tenant_no, user_id=user_id, item=it)
    if commit:
        db.commit()
    else:
        db.flush()
    return get_freight_detail(db, freight_id, tenant_no)


def unlink_samples(
    db: Session,
    freight_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str],
    items: List[Dict[str, Any]],
    commit: bool = True,
    user_info: Optional[dict] = None,
) -> Dict[str, Any]:
    header = get_freight(db, freight_id, tenant_no, user_info)
    if not header:
        raise ValueError("运费批次不存在")
    assert_freight_mutable(header.settle_status, "items")
    obj = header
    now = _now()
    for it in items or []:
        norm = _normalize_item(it)
        row = (
            db.query(BizSampleFreightItem)
            .filter(
                BizSampleFreightItem.freight_id == freight_id,
                BizSampleFreightItem.tenant_no == tenant_no,
                BizSampleFreightItem.sample_source == norm["sample_source"],
                BizSampleFreightItem.sample_id == norm["sample_id"],
                BizSampleFreightItem.del_flag == 0,
            )
            .first()
        )
        if not row:
            continue
        row.del_flag = 1
        row.update_by = user_id
        row.update_time = now
        db.add(row)
    db.flush()
    remaining = _linked_count(db, freight_id, tenant_no)
    if remaining == 0:
        obj.del_flag = 1
        obj.update_by = user_id
        obj.update_time = now
        db.add(obj)
    if commit:
        db.commit()
    else:
        db.flush()
    if remaining == 0:
        return {
            "id": freight_id,
            "deleted": True,
            "linkedSampleCount": 0,
        }
    return get_freight_detail(db, freight_id, tenant_no)


def delete_freight(
    db: Session,
    freight_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str],
    commit: bool = True,
    user_info: Optional[dict] = None,
) -> Dict[str, Any]:
    """级联软删运费批次及其全部有效关联样本。"""
    header = get_freight(db, freight_id, tenant_no, user_info)
    if not header:
        raise ValueError("运费批次不存在")
    assert_freight_mutable(header.settle_status, "delete")
    obj = (
        db.query(BizSampleFreight)
        .filter(
            BizSampleFreight.id == freight_id,
            BizSampleFreight.tenant_no == tenant_no,
            BizSampleFreight.del_flag == 0,
        )
        .with_for_update()
        .first()
    )
    if not obj:
        raise ValueError("运费批次不存在")
    now = _now()
    items = (
        db.query(BizSampleFreightItem)
        .filter(
            BizSampleFreightItem.freight_id == freight_id,
            BizSampleFreightItem.tenant_no == tenant_no,
            BizSampleFreightItem.del_flag == 0,
        )
        .with_for_update()
        .all()
    )
    for row in items:
        row.del_flag = 1
        row.update_by = user_id
        row.update_time = now
        db.add(row)
    obj.del_flag = 1
    obj.update_by = user_id
    obj.update_time = now
    db.add(obj)
    if commit:
        db.commit()
    else:
        db.flush()
    return {
        "id": freight_id,
        "deleted": True,
        "linkedSampleCount": 0,
    }


def freight_amount_payload(
    amount: Any, transport_mode: Optional[str], money_as_string: bool
) -> Dict[str, Any]:
    if amount is None:
        freight_amount: Any = None
    elif money_as_string:
        freight_amount = format(Decimal(str(amount)).quantize(Decimal("0.01")), "f")
    else:
        freight_amount = float(amount)
    return {
        "freightAmount": freight_amount,
        "freightAmountLabel": freight_amount_label(transport_mode, amount),
    }


def freight_map_for_samples(
    db: Session,
    *,
    tenant_no: str,
    sample_source: str,
    sample_ids: List[str],
    money_as_string: bool = False,
) -> Dict[str, Dict[str, Any]]:
    ids = [str(x).strip() for x in sample_ids if str(x or "").strip()]
    if not ids:
        return {}
    rows = (
        db.query(BizSampleFreightItem, BizSampleFreight)
        .join(
            BizSampleFreight,
            and_(
                BizSampleFreight.id == BizSampleFreightItem.freight_id,
                BizSampleFreight.tenant_no == tenant_no,
                BizSampleFreight.del_flag == 0,
            ),
        )
        .filter(
            BizSampleFreightItem.tenant_no == tenant_no,
            BizSampleFreightItem.sample_source == sample_source,
            BizSampleFreightItem.sample_id.in_(ids),
            BizSampleFreightItem.del_flag == 0,
        )
        .all()
    )
    if not rows:
        return {}
    freight_ids = list({f.id for _, f in rows})
    count_rows = (
        db.query(BizSampleFreightItem.freight_id, func.count(BizSampleFreightItem.id))
        .filter(
            BizSampleFreightItem.freight_id.in_(freight_ids),
            BizSampleFreightItem.tenant_no == tenant_no,
            BizSampleFreightItem.del_flag == 0,
        )
        .group_by(BizSampleFreightItem.freight_id)
        .all()
    )
    count_map = {fid: int(c) for fid, c in count_rows}
    labels = _label_map(db, tenant_no, [f.transport_mode for _, f in rows])
    out: Dict[str, Dict[str, Any]] = {}
    for item, freight in rows:
        out[item.sample_id] = {
            "id": freight.id,
            "transportMode": freight.transport_mode or "",
            "transportModeLabel": labels.get(
                freight.transport_mode or "", freight.transport_mode or ""
            ),
            **freight_amount_payload(
                freight.freight_amount, freight.transport_mode, money_as_string
            ),
            "waybillNo": getattr(freight, "waybill_no", None) or "",
            "settleStatus": getattr(freight, "settle_status", None) or "",
            "linkedSampleCount": count_map.get(freight.id, 0),
        }
    return out


def apply_freight_on_protocol_save(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str],
    sample_source: str,
    protocol_id: str,
    saved_samples: list,
    input_samples: list,
    replaced_sample_ids: Optional[List[str]] = None,
) -> None:
    """协议保存挂钩：按 freightId / freightDraft 关联运费（flush，不单独 commit）。"""
    old_ids = [
        str(sample_id).strip()
        for sample_id in (replaced_sample_ids or [])
        if str(sample_id or "").strip()
    ]
    stale_filters = [BizSampleFreightItem.protocol_id == protocol_id]
    if old_ids:
        stale_filters.append(BizSampleFreightItem.sample_id.in_(old_ids))
    stale_items = (
        db.query(BizSampleFreightItem)
        .filter(
            BizSampleFreightItem.tenant_no == tenant_no,
            BizSampleFreightItem.sample_source == sample_source,
            BizSampleFreightItem.del_flag == 0,
            or_(*stale_filters),
        )
        .order_by(BizSampleFreightItem.id.asc())
        .with_for_update()
        .all()
    )
    now = _now()
    affected_freight_ids = {item.freight_id for item in stale_items}
    for stale in stale_items:
        stale.del_flag = 1
        stale.update_by = user_id
        stale.update_time = now
        db.add(stale)
    if stale_items:
        db.flush()

    draft_groups: Dict[str, Dict[str, Any]] = {}
    for idx, sample_obj in enumerate(saved_samples or []):
        if idx >= len(input_samples or []):
            break
        raw = input_samples[idx] if isinstance(input_samples[idx], dict) else {}
        sample_id = getattr(sample_obj, "id", None) or ""
        sample_code = (
            getattr(sample_obj, "sample_code", None)
            or raw.get("sampleCode")
            or raw.get("sample_code")
            or ""
        )
        if not sample_id:
            continue
        item = {
            "sampleSource": sample_source,
            "sampleId": sample_id,
            "protocolId": protocol_id,
            "sampleCode": sample_code,
        }
        freight_id = (
            str(raw.get("freightId") or raw.get("freight_id") or "").strip()
            or str((raw.get("freight") or {}).get("id") or "").strip()
        )
        draft = raw.get("freightDraft")
        if draft is None:
            draft = raw.get("freight_draft")
        if freight_id:
            link_samples(
                db,
                freight_id,
                tenant_no=tenant_no,
                user_id=user_id,
                items=[item],
                commit=False,
            )
            continue
        if isinstance(draft, dict):
            client_key = str(draft.get("clientKey") or "").strip()
            if not client_key:
                client_key = f"__idx_{idx}"
            group = draft_groups.get(client_key)
            if not group:
                draft_groups[client_key] = {
                    "transportMode": draft.get("transportMode") or draft.get("transport_mode"),
                    "freightAmount": draft.get("freightAmount")
                    if draft.get("freightAmount") is not None
                    else draft.get("freight_amount"),
                    "waybillNo": draft.get("waybillNo")
                    if draft.get("waybillNo") is not None
                    else draft.get("waybill_no"),
                    "items": [item],
                }
            else:
                group["items"].append(item)

    for group in draft_groups.values():
        create_freight(
            db,
            tenant_no=tenant_no,
            user_id=user_id,
            transport_mode_raw=str(group.get("transportMode") or ""),
            freight_amount=group.get("freightAmount"),
            waybill_no=group.get("waybillNo"),
            waybill_supplied=True,
            amount_supplied=True,
            items=group["items"],
            commit=False,
        )
    for freight_id in sorted(affected_freight_ids):
        if _linked_count(db, freight_id, tenant_no) != 0:
            continue
        freight = (
            db.query(BizSampleFreight)
            .filter(
                BizSampleFreight.id == freight_id,
                BizSampleFreight.tenant_no == tenant_no,
                BizSampleFreight.del_flag == 0,
            )
            .with_for_update()
            .first()
        )
        if freight is not None:
            freight.del_flag = 1
            freight.update_by = user_id
            freight.update_time = now
            db.add(freight)
    db.flush()