feat(backend): add sample freight CRUD and link/unlink

Provide freight batch create/update/link/unlink, sample enrichment map, and protocol-save draft grouping without writing legacy transport rows.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/crud/sample_freight.py

@@ -0,0 +1,537 @@
+"""样本运费批次 CRUD（biz_sample_freight / biz_sample_freight_item）"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from decimal import Decimal
+from typing import Any, Dict, List, Optional, Tuple
+
+from sqlalchemy import and_, func, or_
+from sqlalchemy.orm import Session
+
+from app.core.utils import generate_uuid
+from app.models.models import BizSampleFreight, BizSampleFreightItem, SysDictData
+from app.services.sample_transport_mode import DICT_TYPE, resolve_transport_mode_value
+
+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 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 _row_to_dict(
+    obj: BizSampleFreight,
+    *,
+    linked_sample_count: int = 0,
+    transport_mode_label: str = "",
+) -> Dict[str, Any]:
+    amount = obj.freight_amount
+    return {
+        "id": obj.id,
+        "transportMode": obj.transport_mode or "",
+        "transportModeLabel": transport_mode_label or (obj.transport_mode or ""),
+        "freightAmount": float(amount) if amount is not None else 0.0,
+        "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 "",
+        "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 get_freight(db: Session, freight_id: str, tenant_no: str) -> Optional[BizSampleFreight]:
+    return (
+        db.query(BizSampleFreight)
+        .filter(
+            BizSampleFreight.id == freight_id,
+            BizSampleFreight.tenant_no == tenant_no,
+            BizSampleFreight.del_flag == 0,
+        )
+        .first()
+    )
+
+
+def list_freights(
+    db: Session,
+    *,
+    tenant_no: str,
+    page: int = 1,
+    page_size: int = 10,
+    keyword: Optional[str] = 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,
+    )
+    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)]
+        if label_values:
+            conds.append(BizSampleFreight.transport_mode.in_(label_values))
+        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])
+    out = [
+        _row_to_dict(
+            r,
+            linked_sample_count=_linked_count(db, r.id, tenant_no),
+            transport_mode_label=labels.get(r.transport_mode or "", r.transport_mode or ""),
+        )
+        for r in rows
+    ]
+    return out, total
+
+
+def get_freight_detail(db: Session, freight_id: str, tenant_no: str) -> Dict[str, Any]:
+    obj = get_freight(db, freight_id, tenant_no)
+    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])
+    data = _row_to_dict(
+        obj,
+        linked_sample_count=len(items),
+        transport_mode_label=labels.get(obj.transport_mode or "", obj.transport_mode or ""),
+    )
+    data["items"] = [_item_to_dict(i) for i in items]
+    return data
+
+
+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,
+) -> Dict[str, Any]:
+    mode = resolve_transport_mode_value(
+        db, tenant_no=tenant_no, raw=transport_mode_raw, user_id=user_id
+    )
+    amount = _parse_amount(freight_amount)
+    now = _now()
+    obj = BizSampleFreight(
+        id=generate_uuid(),
+        transport_mode=mode,
+        freight_amount=amount,
+        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,
+    )
+    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,
+) -> Dict[str, Any]:
+    obj = get_freight(db, freight_id, tenant_no)
+    if not obj:
+        raise ValueError("运费批次不存在")
+    if transport_mode_raw is not None and str(transport_mode_raw).strip() != "":
+        obj.transport_mode = resolve_transport_mode_value(
+            db, tenant_no=tenant_no, raw=str(transport_mode_raw), user_id=user_id
+        )
+    if freight_amount is not None and freight_amount != "":
+        obj.freight_amount = _parse_amount(freight_amount)
+    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,
+) -> Dict[str, Any]:
+    obj = get_freight(db, freight_id, tenant_no)
+    if not obj:
+        raise ValueError("运费批次不存在")
+    if not items:
+        raise ValueError("请选择要关联的样本")
+    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,
+) -> Dict[str, Any]:
+    obj = get_freight(db, freight_id, tenant_no)
+    if not obj:
+        raise ValueError("运费批次不存在")
+    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 freight_map_for_samples(
+    db: Session,
+    *,
+    tenant_no: str,
+    sample_source: str,
+    sample_ids: List[str],
+) -> 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 ""
+            ),
+            "freightAmount": float(freight.freight_amount)
+            if freight.freight_amount is not None
+            else 0.0,
+            "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,
+) -> None:
+    """协议保存挂钩：按 freightId / freightDraft 关联运费（flush，不单独 commit）。"""
+    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 ""
+        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 "").strip()
+            or str((raw.get("freight") or {}).get("id") or "").strip()
+        )
+        draft = raw.get("freightDraft")
+        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"),
+                    "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"),
+            items=group["items"],
+            commit=False,
+        )
+    db.flush()

backend/tests/test_sample_freight.py

@@ -1,6 +1,7 @@
 """样本运费：运输方式字典解析与关联规则"""
 
 import unittest
+from decimal import Decimal
 from unittest.mock import MagicMock
 
 from app.services.sample_transport_mode import resolve_transport_mode_value
@@ -39,5 +40,26 @@ class TestResolveTransportMode(unittest.TestCase):
         db.add.assert_not_called()
 
 
+class TestFreightAmountValidation(unittest.TestCase):
+    def test_negative_rejected(self):
+        from app.crud.sample_freight import _parse_amount
+
+        with self.assertRaises(ValueError):
+            _parse_amount(-1)
+
+    def test_zero_ok(self):
+        from app.crud.sample_freight import _parse_amount
+
+        self.assertEqual(_parse_amount(0), Decimal("0"))
+
+    def test_empty_rejected(self):
+        from app.crud.sample_freight import _parse_amount
+
+        with self.assertRaises(ValueError):
+            _parse_amount(None)
+        with self.assertRaises(ValueError):
+            _parse_amount("")
+
+
 if __name__ == "__main__":
     unittest.main()