zkxw/backend/app/crud/cell_product.py

"""细胞产品 CRUD（biz_cell_product）"""
from __future__ import annotations

import math
import re
from datetime import datetime
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.utils.sample_special import sample_special_to_dict
from app.crud import storage_occupancy as occupancy_svc
from app.models.models import BizStorageSlotOccupancy
from app.models.models import (
    BizCellProduct,
    BizCellProductSpec,
    BizProductionProductType,
    BizStorageFacility,
    BizStorageFacilitySlot,
)
from app.services.storage_facility_code import build_full_slot_code

_CELL_PRODUCT_STATUSES = frozenset(
    {"frozen", "in_production", "sold", "used", "voided", "thawed"}
)
_CELL_PRODUCT_RELEASE_STATUSES = frozenset({"sold", "used", "voided"})
_CELL_PRODUCT_STORE_STATUSES = frozenset({"frozen", "thawed", "in_production"})


def _apply_cell_product_status_change(
    db: Session,
    obj: BizCellProduct,
    to_status: str,
    tenant_no: str,
    user_id: Optional[str],
    now: datetime,
) -> None:
    """管理端手动改状态：允许在合法枚举间任意变更，并同步库位占用。"""
    from_status = (obj.product_status or "").strip()
    to_status = (to_status or "").strip()
    if to_status == from_status:
        return
    if to_status not in _CELL_PRODUCT_STATUSES:
        raise ValueError(
            f"状态必须为 {'/'.join(sorted(_CELL_PRODUCT_STATUSES))}"
        )
    obj.product_status = to_status
    slot_id = (obj.storage_slot_id or "").strip()
    if to_status in _CELL_PRODUCT_RELEASE_STATUSES:
        if slot_id:
            try:
                occupancy_svc.remove_by_item(
                    db, "cell_product", obj.id, user_id=user_id,
                )
            except Exception:
                pass
        return
    if to_status in _CELL_PRODUCT_STORE_STATUSES and slot_id:
        existing = db.query(BizStorageSlotOccupancy).filter(
            BizStorageSlotOccupancy.item_type == "cell_product",
            BizStorageSlotOccupancy.item_id == obj.id,
            BizStorageSlotOccupancy.occupancy_status == "stored",
            BizStorageSlotOccupancy.del_flag == 0,
        ).first()
        if existing:
            return
        occupancy_svc.assert_slot_vacant(
            db, slot_id, tenant_no=tenant_no, exclude_product_id=obj.id,
        )
        occupancy_svc._add_stored_occupancy_no_commit(
            db,
            slot_id=slot_id,
            item_type="cell_product",
            item_id=obj.id,
            item_code=obj.product_code or "",
            item_name=obj.product_name or "",
            item_source=obj.source_type or "manual",
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=obj.create_dept_id or "",
            stored_at=now,
        )


def _row_to_dict(c: BizCellProduct) -> Dict[str, Any]:
    return {
        "id": c.id,
        "productCode": c.product_code,
        "productName": c.product_name,
        "productTypeId": c.product_type_id or "",
        "productTypeName": "",
        "specId": c.spec_id or "",
        "specText": c.spec_text or "",
        "specName": c.spec_text or "",
        "batchNo": c.batch_no or "",
        "storageSlotId": c.storage_slot_id or "",
        "storageSlotName": "",
        "passage": c.passage or "",
        "freezeDate": c.freeze_date.strftime("%Y-%m-%d %H:%M:%S") if c.freeze_date else "",
        "expectedCellCount": float(c.expected_cell_count) if c.expected_cell_count is not None else None,
        "viability": float(c.viability) if c.viability is not None else None,
        "productStatus": c.product_status or "",
        "sourceType": c.source_type or "",
        "productionTaskId": c.production_task_id or "",
        "quantity": float(c.quantity) if c.quantity is not None else None,
        "unit": c.unit or "",
        "expiryDate": c.expiry_date.strftime("%Y-%m-%d %H:%M:%S") if c.expiry_date else "",
        "productImages": c.product_images or "",
        "tenantNo": c.tenant_no,
        "orderNum": c.order_num if c.order_num is not None else 0,
        "remark": c.remark or "",
        "createBy": c.create_by or "",
        "createTime": c.create_time.strftime("%Y-%m-%d %H:%M:%S") if c.create_time else "",
        "updateBy": c.update_by or "",
        "updateTime": c.update_time.strftime("%Y-%m-%d %H:%M:%S") if c.update_time else "",
    }


def _build_facility_path_label(node_id: str, by_id: Dict[str, BizStorageFacility]) -> str:
    parts: List[str] = []
    cur_id = (node_id or "").strip()
    visited: set[str] = set()
    while cur_id and cur_id != "0" and cur_id not in visited:
        visited.add(cur_id)
        node = by_id.get(cur_id)
        if not node:
            break
        parts.append(node.node_name or "")
        cur_id = (node.parent_id or "").strip()
    parts.reverse()
    return " / ".join(p for p in parts if p)


def _enrich_product_dicts(
    db: Session,
    tenant_no: str,
    items: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
    if not items:
        return items

    type_ids = {i.get("productTypeId") for i in items if i.get("productTypeId")}
    spec_ids = {i.get("specId") for i in items if i.get("specId")}
    slot_ids = {i.get("storageSlotId") for i in items if i.get("storageSlotId")}

    type_map: Dict[str, str] = {}
    if type_ids:
        rows = (
            db.query(BizProductionProductType)
            .filter(
                BizProductionProductType.id.in_(type_ids),
                BizProductionProductType.del_flag == 0,
            )
            .all()
        )
        type_map = {r.id: r.product_type_name or "" for r in rows}

    spec_map: Dict[str, str] = {}
    if spec_ids:
        rows = (
            db.query(BizCellProductSpec)
            .filter(
                BizCellProductSpec.id.in_(spec_ids),
                BizCellProductSpec.del_flag == 0,
            )
            .all()
        )
        spec_map = {r.id: r.spec_name or "" for r in rows}

    slot_map: Dict[str, str] = {}
    if slot_ids:
        slots = (
            db.query(BizStorageFacilitySlot)
            .filter(
                BizStorageFacilitySlot.id.in_(slot_ids),
                BizStorageFacilitySlot.del_flag == 0,
            )
            .all()
        )
        facilities = (
            db.query(BizStorageFacility)
            .filter(
                BizStorageFacility.tenant_no == tenant_no,
                BizStorageFacility.del_flag == 0,
            )
            .all()
        )
        facility_by_id = {f.id: f for f in facilities}
        for slot in slots:
            slot_map[slot.id] = build_full_slot_code(
                slot.facility_node_id,
                slot.slot_code,
                facility_by_id,
            )

    # --- 来源信息富化 ---
    from app.models.models import BizProductionTask, BizProductionWorkOrder

    task_ids = {i.get("productionTaskId") for i in items if i.get("productionTaskId")}
    task_map: Dict[str, Any] = {}
    task_to_wo: Dict[str, str] = {}
    wo_map: Dict[str, Any] = {}

    if task_ids:
        tasks = (
            db.query(BizProductionTask)
            .filter(
                BizProductionTask.id.in_(task_ids),
                BizProductionTask.del_flag == 0,
            )
            .all()
        )
        for t in tasks:
            task_map[t.id] = t
            if t.work_order_id:
                task_to_wo[t.id] = t.work_order_id

        wo_ids = set(task_to_wo.values())
        if wo_ids:
            wos = (
                db.query(BizProductionWorkOrder)
                .filter(
                    BizProductionWorkOrder.id.in_(wo_ids),
                    BizProductionWorkOrder.del_flag == 0,
                )
                .all()
            )
            wo_map = {w.id: w for w in wos}

    source_type_labels = {"production": "生产入库", "manual": "手动录入"}

    for item in items:
        tid = item.get("productTypeId") or ""
        sid = item.get("specId") or ""
        slot_id = item.get("storageSlotId") or ""
        item["productTypeName"] = type_map.get(tid, "") if tid else ""
        stored_spec = (item.get("specText") or item.get("spec_text") or "").strip()
        if stored_spec:
            item["specName"] = stored_spec
        elif sid:
            item["specName"] = spec_map.get(sid, "")
        else:
            item["specName"] = (item.get("specName") or "").strip()
        item["storageSlotName"] = slot_map.get(slot_id, "") if slot_id else ""
        # 来源信息
        src_type = item.get("sourceType", "")
        item["sourceTypeLabel"] = source_type_labels.get(src_type, src_type or "")
        tid = item.get("productionTaskId") or ""
        if tid and tid in task_map:
            t_obj = task_map[tid]
            item["sourceTaskNo"] = t_obj.task_no or ""
            wo_id = task_to_wo.get(tid)
            if wo_id and wo_id in wo_map:
                wo_obj = wo_map[wo_id]
                item["sourceWorkOrderNo"] = wo_obj.work_order_no or ""
                item["sourceProtocolNo"] = wo_obj.protocol_no or ""
                item["sourceReceptionSource"] = wo_obj.reception_source or ""
            else:
                item["sourceWorkOrderNo"] = ""
                item["sourceProtocolNo"] = ""
                item["sourceReceptionSource"] = ""
        else:
            item["sourceTaskNo"] = ""
            item["sourceWorkOrderNo"] = ""
            item["sourceProtocolNo"] = ""
            item["sourceReceptionSource"] = ""
    return items


def product_to_response(db: Session, tenant_no: str, obj: BizCellProduct) -> Dict[str, Any]:
    enriched = _enrich_product_dicts(db, tenant_no, [_row_to_dict(obj)])
    return enriched[0] if enriched else _row_to_dict(obj)


def list_cell_products(
    db: Session,
    *,
    tenant_no: str,
    page: int = 1,
    page_size: int = 20,
    keyword: Optional[str] = None,
    product_code: Optional[str] = None,
    product_name: Optional[str] = None,
    product_type_id: Optional[str] = None,
    product_status: Optional[str] = None,
    passage: Optional[str] = None,
    source_type: Optional[str] = None,
    user_info=None,
) -> Dict[str, Any]:
    def _filtered():
        q0 = db.query(BizCellProduct).filter(
            BizCellProduct.del_flag == 0,
            BizCellProduct.tenant_no == tenant_no,
        )
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizCellProduct.product_code.ilike(like),
                    BizCellProduct.product_name.ilike(like),
                )
            )
        pc = (product_code or "").strip()
        if pc:
            q0 = q0.filter(BizCellProduct.product_code.like(f"%{pc}%"))
        pn = (product_name or "").strip()
        if pn:
            q0 = q0.filter(BizCellProduct.product_name.like(f"%{pn}%"))
        if product_type_id:
            q0 = q0.filter(BizCellProduct.product_type_id == product_type_id)
        if product_status:
            q0 = q0.filter(BizCellProduct.product_status == product_status)
        if passage:
            q0 = q0.filter(BizCellProduct.passage == passage)
        if source_type:
            q0 = q0.filter(BizCellProduct.source_type == source_type)
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizCellProduct, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizCellProduct.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(
            BizCellProduct.create_time.desc(),
            BizCellProduct.product_code.asc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    items = _enrich_product_dicts(
        db, tenant_no, [_row_to_dict(x) for x in rows]
    )
    return {"items": items, "total": int(total)}


def get_cell_product(db: Session, id: str, tenant_no: str) -> Optional[BizCellProduct]:
    return (
        db.query(BizCellProduct)
        .filter(
            and_(
                BizCellProduct.id == id,
                BizCellProduct.tenant_no == tenant_no,
                BizCellProduct.del_flag == 0,
            )
        )
        .first()
    )


def create_cell_product(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: Optional[str] = None,
    data: Any = None,
) -> BizCellProduct:
    code = (data.product_code or "").strip()
    if not code:
        raise ValueError("产品编码不能为空")

    existing = db.query(BizCellProduct).filter(
        BizCellProduct.tenant_no == tenant_no,
        BizCellProduct.product_code == code,
        BizCellProduct.del_flag == 0,
    ).first()
    if existing:
        raise ValueError(f"产品编码 {code} 已存在")

    obj = BizCellProduct(
        id=generate_uuid(),
        product_code=code,
        product_name=(data.product_name or "").strip(),
        product_type_id=(data.product_type_id or "").strip() or None,
        spec_id=(data.spec_id or "").strip() or None,
        spec_text=(getattr(data, "spec_name", None) or "").strip() or None,
        batch_no=(data.batch_no or "").strip() or None,
        storage_slot_id=(data.storage_slot_id or "").strip() or None,
        passage=((data.passage or "").strip() or None) if getattr(data, "passage", None) else None,
        freeze_date=data.freeze_date if hasattr(data, "freeze_date") else None,
        expected_cell_count=data.expected_cell_count if hasattr(data, "expected_cell_count") else None,
        viability=data.viability if hasattr(data, "viability") else None,
        product_status=data.product_status or "frozen",
        source_type=data.source_type or "manual",
        quantity=data.quantity,
        unit=(data.unit or "").strip() or None,
        expiry_date=data.expiry_date,
        product_images=(data.product_images or "").strip() or None,
        tenant_no=tenant_no,
        order_num=0,
        remark=(data.remark or "").strip() or None,
        create_by=user_id,
        create_dept_id=create_dept_id or "",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(obj)
    db.flush()

    slot_id = (obj.storage_slot_id or "").strip()
    if slot_id and (obj.product_status or "") in _CELL_PRODUCT_STORE_STATUSES:
        occupancy_svc.assert_slot_vacant(db, slot_id, tenant_no=tenant_no, exclude_product_id=obj.id)
        occupancy_svc._add_stored_occupancy_no_commit(
            db,
            slot_id=slot_id,
            item_type="cell_product",
            item_id=obj.id,
            item_code=obj.product_code or "",
            item_name=obj.product_name or "",
            item_source=obj.source_type or "manual",
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=create_dept_id or "",
        )

    db.commit()
    db.refresh(obj)
    return obj


def update_cell_product(
    db: Session,
    id: str,
    tenant_no: str,
    user_id: Optional[str] = None,
    data: Any = None,
) -> BizCellProduct:
    obj = get_cell_product(db, id, tenant_no)
    if not obj:
        raise ValueError("细胞产品不存在")

    old_slot_id = (obj.storage_slot_id or "").strip()
    slot_change_requested = data.storage_slot_id is not None

    if getattr(data, "product_code", None) is not None:
        code = (data.product_code or "").strip()
        if code and code != (obj.product_code or "").strip():
            if product_code_exists(db, tenant_no, code, exclude_id=obj.id):
                raise ValueError(f"产品编码 {code} 已存在")
            obj.product_code = code

    if data.product_name is not None:
        obj.product_name = (data.product_name or "").strip()
    if data.product_type_id is not None:
        obj.product_type_id = (data.product_type_id or "").strip() or None
    if data.spec_id is not None:
        obj.spec_id = (data.spec_id or "").strip() or None
    if getattr(data, "spec_name", None) is not None:
        obj.spec_text = (data.spec_name or "").strip() or None
    if data.batch_no is not None:
        obj.batch_no = (data.batch_no or "").strip() or None
    if data.storage_slot_id is not None:
        obj.storage_slot_id = (data.storage_slot_id or "").strip() or None
    if hasattr(data, "passage") and data.passage is not None:
        obj.passage = (data.passage or "").strip() or None
    if hasattr(data, "freeze_date") and data.freeze_date is not None:
        obj.freeze_date = data.freeze_date
    if hasattr(data, "expected_cell_count") and data.expected_cell_count is not None:
        obj.expected_cell_count = data.expected_cell_count
    if hasattr(data, "viability") and data.viability is not None:
        obj.viability = data.viability
    if data.product_status is not None and data.product_status != obj.product_status:
        _apply_cell_product_status_change(
            db,
            obj,
            data.product_status,
            tenant_no,
            user_id,
            datetime.now(),
        )
    if data.quantity is not None:
        obj.quantity = data.quantity
    if data.unit is not None:
        obj.unit = (data.unit or "").strip() or None
    if data.expiry_date is not None:
        obj.expiry_date = data.expiry_date
    if data.product_images is not None:
        obj.product_images = (data.product_images or "").strip() or None
    if data.remark is not None:
        obj.remark = (data.remark or "").strip() or None

    new_slot_id = (obj.storage_slot_id or "").strip()
    in_store = (obj.product_status or "") in _CELL_PRODUCT_STORE_STATUSES
    if slot_change_requested and old_slot_id != new_slot_id:
        stored_occ = db.query(BizStorageSlotOccupancy).filter(
            BizStorageSlotOccupancy.item_type == "cell_product",
            BizStorageSlotOccupancy.item_id == obj.id,
            BizStorageSlotOccupancy.occupancy_status == "stored",
            BizStorageSlotOccupancy.del_flag == 0,
        ).first()
        if stored_occ and (stored_occ.slot_id or "").strip() != new_slot_id:
            occupancy_svc.remove_by_item_no_commit(db, "cell_product", obj.id, user_id=user_id)
            stored_occ = None
        if new_slot_id and in_store and not stored_occ:
            occupancy_svc.assert_slot_vacant(
                db, new_slot_id, tenant_no=tenant_no, exclude_product_id=obj.id,
            )
            occupancy_svc._add_stored_occupancy_no_commit(
                db,
                slot_id=new_slot_id,
                item_type="cell_product",
                item_id=obj.id,
                item_code=obj.product_code or "",
                item_name=obj.product_name or "",
                item_source=obj.source_type or "manual",
                tenant_no=tenant_no,
                user_id=user_id,
                create_dept_id=obj.create_dept_id or "",
            )
        elif not new_slot_id and stored_occ:
            occupancy_svc.remove_by_item_no_commit(db, "cell_product", obj.id, user_id=user_id)

    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def delete_cell_product(
    db: Session,
    id: str,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> None:
    obj = get_cell_product(db, id, tenant_no)
    if not obj:
        raise ValueError("细胞产品不存在")
    occupancy_svc.remove_by_item_no_commit(db, "cell_product", obj.id, user_id=user_id)
    obj.del_flag = 1
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


def _next_code_start_seq(db: Session, tenant_no: str, prefix: str) -> int:
    last = (
        db.query(BizCellProduct)
        .filter(
            BizCellProduct.tenant_no == tenant_no,
            BizCellProduct.product_code.like(f"{prefix}%"),
            BizCellProduct.del_flag == 0,
        )
        .order_by(BizCellProduct.product_code.desc())
        .first()
    )
    seq = 1
    if last and last.product_code:
        try:
            base = normalize_product_code_base(last.product_code)
            seq = int(base[-6:]) + 1
        except (ValueError, IndexError):
            seq = 1
    return seq


def get_next_code(db: Session, tenant_no: str) -> str:
    today_str = datetime.now().strftime("%Y%m%d")
    prefix = f"CP{today_str}"
    seq = _next_code_start_seq(db, tenant_no, prefix)
    return f"{prefix}{seq:06d}"


_PRODUCT_CODE_SUFFIX_RE = re.compile(r"^(.+)-(\d{2})$")
_PASSAGE_PRODUCT_CODE_RE = re.compile(r"^(.+)-(P\d+)-(\d{2})$", re.IGNORECASE)


def normalize_passage_product_base(code: str) -> str:
    """去掉末尾 -P{n}-XX 或 -XX，得到代次产品编码的基础前缀。"""
    c = (code or "").strip()
    if not c:
        return ""
    m = _PASSAGE_PRODUCT_CODE_RE.match(c)
    if m:
        return m.group(1).strip()
    return normalize_product_code_base(c)


def get_max_passage_product_seq(
    db: Session,
    tenant_no: str,
    base_code: str,
    passage: str,
) -> int:
    """查询 `{base_code}-{passage}-XX` 形式产品编码在库中的最大两位序号（无匹配则 0）。"""
    base = normalize_passage_product_base(base_code)
    passage = (passage or "").strip()
    if not base or not passage:
        return 0
    prefix = f"{base}-{passage}-"
    suffix_re = re.compile(rf"^{re.escape(prefix)}(\d{{2}})$")
    rows = (
        db.query(BizCellProduct.product_code)
        .filter(
            BizCellProduct.tenant_no == tenant_no,
            BizCellProduct.product_code.like(f"{prefix}%"),
            BizCellProduct.del_flag == 0,
        )
        .all()
    )
    max_seq = 0
    for (code,) in rows:
        m = suffix_re.match((code or "").strip())
        if not m:
            continue
        try:
            max_seq = max(max_seq, int(m.group(1)))
        except ValueError:
            continue
    return max_seq


def get_max_passage_product_seq_map(
    db: Session,
    tenant_no: str,
    base_code: str,
    passages: List[str],
) -> Dict[str, int]:
    """批量查询各代次在 `{base_code}-{passage}-XX` 规则下的最大序号。"""
    result: Dict[str, int] = {}
    seen: set[str] = set()
    for raw in passages or []:
        passage = (raw or "").strip()
        if not passage or passage in seen:
            continue
        seen.add(passage)
        result[passage] = get_max_passage_product_seq(db, tenant_no, base_code, passage)
    return result


def get_max_product_suffix_seq(
    db: Session,
    tenant_no: str,
    base_code: str,
) -> int:
    """查询 `{base_code}-XX` 形式产品编码在库中的最大两位序号（无匹配则 0）。"""
    base = normalize_product_code_base((base_code or "").strip())
    if not base:
        return 0
    prefix = f"{base}-"
    suffix_re = re.compile(rf"^{re.escape(base)}-(\d{{2}})$")
    rows = (
        db.query(BizCellProduct.product_code)
        .filter(
            BizCellProduct.tenant_no == tenant_no,
            BizCellProduct.product_code.like(f"{prefix}%"),
            BizCellProduct.del_flag == 0,
        )
        .all()
    )
    max_seq = 0
    for (code,) in rows:
        m = suffix_re.match((code or "").strip())
        if not m:
            continue
        try:
            max_seq = max(max_seq, int(m.group(1)))
        except ValueError:
            continue
    return max_seq


def product_code_exists(
    db: Session,
    tenant_no: str,
    product_code: str,
    exclude_id: Optional[str] = None,
) -> bool:
    code = (product_code or "").strip()
    if not code:
        return False
    q = db.query(BizCellProduct.id).filter(
        BizCellProduct.tenant_no == tenant_no,
        BizCellProduct.product_code == code,
        BizCellProduct.del_flag == 0,
    )
    if exclude_id:
        q = q.filter(BizCellProduct.id != exclude_id)
    return q.first() is not None


def normalize_product_code_base(code: str) -> str:
    """去掉末尾 -01/-02 等序号，得到多份入库时的基础编码。"""
    c = (code or "").strip()
    if not c:
        return ""
    m = _PRODUCT_CODE_SUFFIX_RE.match(c)
    if m:
        return m.group(1).strip()
    return c


def build_split_product_codes(base: str, count: int) -> List[str]:
    """按基础编码生成 基础-01、基础-02 …"""
    if count < 1:
        return []
    return [f"{base}-{i:02d}" for i in range(1, count + 1)]


def reserve_next_codes(db: Session, tenant_no: str, count: int) -> List[str]:
    if count < 1:
        return []
    count = min(count, 50)
    today_str = datetime.now().strftime("%Y%m%d")
    prefix = f"CP{today_str}"
    start = _next_code_start_seq(db, tenant_no, prefix)
    return [f"{prefix}{start + i:06d}" for i in range(count)]


def _validate_batch_items(
    db: Session,
    tenant_no: str,
    items: List[Any],
    *,
    require_storage_slot: bool = True,
) -> Tuple[List[str], List[Optional[str]]]:
    if not items:
        raise ValueError("至少需要一份产品明细")
    if len(items) > 50:
        raise ValueError("单次入库份数不能超过 50")

    codes: List[str] = []
    slot_ids: List[Optional[str]] = []
    for it in items:
        code = (getattr(it, "product_code", None) or "").strip()
        if not code:
            raise ValueError("产品编码不能为空")
        if code in codes:
            raise ValueError(f"产品编码 {code} 重复")
        codes.append(code)

        sid = (getattr(it, "storage_slot_id", None) or "").strip() or None
        if require_storage_slot and not sid:
            raise ValueError("每份产品须选择储物格位置")
        slot_ids.append(sid)

    occupancy_svc.assert_batch_slots_unique(slot_ids)

    existing_codes = (
        db.query(BizCellProduct.product_code)
        .filter(
            BizCellProduct.tenant_no == tenant_no,
            BizCellProduct.product_code.in_(codes),
            BizCellProduct.del_flag == 0,
        )
        .all()
    )
    if existing_codes:
        dup = existing_codes[0][0]
        raise ValueError(f"产品编码 {dup} 已存在")

    return codes, slot_ids


def create_cell_products_by_quantity(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: Optional[str] = None,
    data: Any = None,
    count: int,
) -> List[BizCellProduct]:
    """按入库份数拆分：基础编码-01、基础编码-02 …（须通过 items 为每份指定独立储物格）"""
    count = max(1, min(int(count), 50))
    if count > 1:
        raise ValueError("入库多份时请通过入库明细为每份产品分别选择储物格")
    base = normalize_product_code_base(getattr(data, "product_code", None) or "")
    if not base:
        raise ValueError("产品编码不能为空")
    codes = build_split_product_codes(base, count)
    slot_id = (getattr(data, "storage_slot_id", None) or "").strip() or None
    from types import SimpleNamespace

    items = [
        SimpleNamespace(product_code=c, storage_slot_id=slot_id) for c in codes
    ]
    return create_cell_products_batch(
        db,
        tenant_no=tenant_no,
        user_id=user_id,
        create_dept_id=create_dept_id,
        data=data,
        items=items,
        require_storage_slot=False,
    )


def create_cell_products_batch(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: Optional[str] = None,
    data: Any = None,
    items: List[Any],
    require_storage_slot: bool = True,
) -> List[BizCellProduct]:
    codes, slot_ids = _validate_batch_items(
        db,
        tenant_no,
        items,
        require_storage_slot=require_storage_slot,
    )
    for slot_id in slot_ids:
        if slot_id:
            occupancy_svc.assert_slot_vacant(db, slot_id, tenant_no=tenant_no)
    now = datetime.now()
    created: List[BizCellProduct] = []
    for it, slot_id in zip(items, slot_ids):
        code = (getattr(it, "product_code", None) or "").strip()
        item_passage = (getattr(it, "passage", None) or "").strip() or None
        if not item_passage and getattr(data, "passage", None):
            item_passage = (data.passage or "").strip() or None
        item_spec = (getattr(it, "spec_name", None) or "").strip() or None
        if not item_spec and getattr(data, "spec_name", None):
            item_spec = (data.spec_name or "").strip() or None
        item_expected = getattr(it, "expected_cell_count", None)
        if item_expected is None:
            item_expected = data.expected_cell_count if hasattr(data, "expected_cell_count") else None
        item_viability = getattr(it, "viability", None)
        if item_viability is None:
            item_viability = data.viability if hasattr(data, "viability") else None
        obj = BizCellProduct(
            id=generate_uuid(),
            product_code=code,
            product_name=(data.product_name or "").strip(),
            product_type_id=(data.product_type_id or "").strip() or None,
            spec_id=(data.spec_id or "").strip() or None,
            spec_text=item_spec,
            batch_no=(data.batch_no or "").strip() or None,
            storage_slot_id=slot_id,
            passage=item_passage,
            freeze_date=data.freeze_date if hasattr(data, "freeze_date") else None,
            expected_cell_count=item_expected,
            viability=item_viability,
            product_status=data.product_status or "frozen",
            source_type=data.source_type or "manual",
            quantity=1,
            unit=(data.unit or "").strip() or None,
            expiry_date=data.expiry_date,
            product_images=(data.product_images or "").strip() or None,
            tenant_no=tenant_no,
            order_num=0,
            remark=(data.remark or "").strip() or None,
            create_by=user_id,
            create_dept_id=create_dept_id or "",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(obj)
        created.append(obj)
    db.flush()
    for obj in created:
        slot_id = (obj.storage_slot_id or "").strip()
        if slot_id and (obj.product_status or "") in _CELL_PRODUCT_STORE_STATUSES:
            occupancy_svc._add_stored_occupancy_no_commit(
                db,
                slot_id=slot_id,
                item_type="cell_product",
                item_id=obj.id,
                item_code=obj.product_code,
                item_name=obj.product_name,
                item_source=obj.source_type or "manual",
                tenant_no=tenant_no,
                user_id=user_id,
                create_dept_id=create_dept_id or "",
                stored_at=now,
            )

    db.commit()
    for obj in created:
        db.refresh(obj)
    return created


def total_pages(total: int, page_size: int) -> int:
    if total <= 0:
        return 0
    return int(math.ceil(total / float(page_size)))


# ---------------------------------------------------------------------------
# 来源链路 API
# ---------------------------------------------------------------------------

_MAX_CHAIN_DEPTH = 10

_SOURCE_TYPE_LABELS = {"production": "生产入库", "manual": "手动录入"}


def _product_node(obj: BizCellProduct, is_current: bool = False) -> Dict[str, Any]:
    return {
        "id": obj.id,
        "productCode": obj.product_code or "",
        "productName": obj.product_name or "",
        "nodeType": "cell_product",
        "productStatus": obj.product_status or "",
        "passage": obj.passage or "",
        "isCurrent": is_current,
    }


def _nest_ancestor_parent(chain: List[Dict[str, Any]]) -> Any:
    """多进一时 parent 为数组；单链时为对象；无上游为 null。"""
    if not chain:
        return None
    if len(chain) == 1:
        return chain[0]
    return chain


_WORK_ORDER_TYPE_LABELS = {
    "inbound": "入库生产",
    "temporary": "临时生产",
    "outbound": "出库生产",
}


def _edge_meta(task: Any, wo: Any = None) -> Dict[str, Any]:
    edge: Dict[str, Any] = {
        "workOrderNo": "",
        "workOrderType": "",
        "workOrderTypeLabel": "",
        "taskNo": "",
        "taskName": "",
        "taskId": "",
    }
    if task:
        edge["taskNo"] = task.task_no or ""
        edge["taskName"] = task.task_name or ""
        edge["taskId"] = task.id or ""
    if wo:
        edge["workOrderNo"] = wo.work_order_no or ""
        wo_type = getattr(wo, "work_order_type", None) or "inbound"
        edge["workOrderType"] = wo_type
        edge["workOrderTypeLabel"] = _WORK_ORDER_TYPE_LABELS.get(wo_type, wo_type)
    return edge


_SOURCE_LABELS = {"cell_storage": "细胞存储", "research_donation": "科研捐赠"}


def _sample_customer_dict(db: Session, protocol_id: str, tenant_no: str) -> Dict[str, Any]:
    if not protocol_id:
        return {}
    from app.models.models import BizProtocolPartyCustomer

    pc = (
        db.query(BizProtocolPartyCustomer)
        .filter(
            BizProtocolPartyCustomer.collection_protocol_id == protocol_id,
            BizProtocolPartyCustomer.tenant_no == tenant_no,
            BizProtocolPartyCustomer.del_flag == 0,
        )
        .first()
    )
    if not pc:
        return {}
    return {
        "motherName": pc.mother_name or "",
        "motherPhone": pc.mother_phone or "",
        "motherIdCard": pc.mother_id_card or "",
        "newbornName": pc.newborn_name or "",
    }


def _donation_sample_customer_dict(db: Session, donation_id: str, tenant_no: str) -> Dict[str, Any]:
    if not donation_id:
        return {}
    from app.models.research_donation import BizResearchDonationParty

    party = (
        db.query(BizResearchDonationParty)
        .filter(
            BizResearchDonationParty.research_donation_id == donation_id,
            BizResearchDonationParty.tenant_no == tenant_no,
            BizResearchDonationParty.del_flag == 0,
        )
        .first()
    )
    if not party:
        return {}
    return {
        "motherName": party.mother_name or "",
        "motherPhone": party.mother_phone or "",
        "motherIdCard": party.mother_id_card or "",
        "newbornName": party.newborn_name or "",
    }


def _sample_type_name_from_id(db: Session, sample_type_id: Optional[str]) -> str:
    if not sample_type_id:
        return ""
    from app.models.models import BizCellSampleType

    row = (
        db.query(BizCellSampleType.type_name)
        .filter(BizCellSampleType.id == sample_type_id, BizCellSampleType.del_flag == 0)
        .first()
    )
    return (row[0] if row else "") or ""


def _find_protocol_sample(
    db: Session,
    tenant_no: str,
    input_id: str,
    input_code: str,
) -> Optional[Any]:
    from app.models.models import BizProtocolSample

    sample = None
    if input_id:
        sample = (
            db.query(BizProtocolSample)
            .filter(
                BizProtocolSample.id == input_id,
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
            )
            .first()
        )
    if not sample and input_code:
        sample = (
            db.query(BizProtocolSample)
            .filter(
                BizProtocolSample.sample_code == input_code,
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
            )
            .first()
        )
    return sample


def _find_donation_sample(
    db: Session,
    tenant_no: str,
    input_id: str,
    input_code: str,
    wo: Any = None,
) -> Optional[Any]:
    from app.models.research_donation import BizResearchDonationSample
    from app.crud.sample_reception import _donation_sample_id_col

    sample = None
    if input_id:
        sample = (
            db.query(BizResearchDonationSample)
            .filter(
                _donation_sample_id_col(BizResearchDonationSample.id) == input_id,
                BizResearchDonationSample.tenant_no == tenant_no,
                BizResearchDonationSample.del_flag == 0,
            )
            .first()
        )
    if not sample and input_code:
        sample = (
            db.query(BizResearchDonationSample)
            .filter(
                BizResearchDonationSample.sample_code == input_code,
                BizResearchDonationSample.tenant_no == tenant_no,
                BizResearchDonationSample.del_flag == 0,
            )
            .first()
        )
    if not sample and wo and (getattr(wo, "reception_source", None) or "") == "research_donation":
        wo_sample_id = (getattr(wo, "protocol_sample_id", None) or "").strip()
        if wo_sample_id:
            sample = (
                db.query(BizResearchDonationSample)
                .filter(
                    _donation_sample_id_col(BizResearchDonationSample.id) == wo_sample_id,
                    BizResearchDonationSample.tenant_no == tenant_no,
                    BizResearchDonationSample.del_flag == 0,
                )
                .first()
            )
    return sample


def _protocol_sample_chain_node(
    db: Session,
    tenant_no: str,
    sample: Any,
    *,
    wo: Any,
    edge: Dict[str, Any],
    sample_type_name: str,
) -> Dict[str, Any]:
    reception_source = getattr(wo, "reception_source", None) if wo else ""
    reception_source = reception_source or "cell_storage"
    protocol_id = sample.collection_protocol_id or ""
    return {
        "node": {
            "id": sample.id,
            "productCode": sample.sample_code or "",
            "productName": "",
            "nodeType": "sample",
            "sampleCode": sample.sample_code or "",
            "sampleTypeName": sample_type_name,
            "protocolNo": wo.protocol_no if wo else "",
            "collectionProtocolId": protocol_id,
            "receptionSource": reception_source,
            "receptionSourceLabel": _SOURCE_LABELS.get(reception_source, reception_source),
            "customer": _sample_customer_dict(db, protocol_id, tenant_no),
            "isCurrent": False,
            **sample_special_to_dict(sample),
        },
        "edge": edge,
        "parent": None,
    }


def _donation_sample_chain_node(
    db: Session,
    tenant_no: str,
    sample: Any,
    *,
    wo: Any,
    edge: Dict[str, Any],
    sample_type_name: str,
) -> Dict[str, Any]:
    donation_id = sample.research_donation_id or ""
    type_name = sample_type_name or _sample_type_name_from_id(db, sample.sample_type_id)
    return {
        "node": {
            "id": sample.id,
            "productCode": sample.sample_code or "",
            "productName": "",
            "nodeType": "sample",
            "sampleCode": sample.sample_code or "",
            "sampleTypeName": type_name,
            "protocolNo": wo.protocol_no if wo else "",
            "collectionProtocolId": donation_id,
            "receptionSource": "research_donation",
            "receptionSourceLabel": _SOURCE_LABELS["research_donation"],
            "customer": _donation_sample_customer_dict(db, donation_id, tenant_no),
            "isCurrent": False,
            **sample_special_to_dict(sample),
        },
        "edge": edge,
        "parent": None,
    }


def _ancestor_sample_chain_link(
    db: Session,
    tenant_no: str,
    sample_input: Any,
    wo: Any,
    edge: Dict[str, Any],
) -> Optional[Dict[str, Any]]:
    """解析任务样本投入物，返回祖先链单段（样本终止节点）。"""
    input_id = (sample_input.input_id or "").strip()
    input_code = (sample_input.input_code or "").strip()
    if not input_id and not input_code:
        if wo and (getattr(wo, "reception_source", None) or "") == "research_donation":
            input_code = (getattr(wo, "sample_code", None) or "").strip()
            input_id = (getattr(wo, "protocol_sample_id", None) or "").strip()
        if not input_id and not input_code:
            return None

    input_name = (sample_input.input_name or "").strip()
    reception_source = (getattr(wo, "reception_source", None) or "") if wo else ""

    donation_sample = None
    if reception_source == "research_donation":
        donation_sample = _find_donation_sample(db, tenant_no, input_id, input_code, wo)
    if donation_sample:
        return _donation_sample_chain_node(
            db,
            tenant_no,
            donation_sample,
            wo=wo,
            edge=edge,
            sample_type_name=input_name,
        )

    protocol_sample = _find_protocol_sample(db, tenant_no, input_id, input_code)
    if protocol_sample:
        return _protocol_sample_chain_node(
            db,
            tenant_no,
            protocol_sample,
            wo=wo,
            edge=edge,
            sample_type_name=input_name
            or _sample_type_name_from_id(db, protocol_sample.sample_type_id),
        )

    if reception_source != "research_donation":
        donation_sample = _find_donation_sample(db, tenant_no, input_id, input_code, wo)
        if donation_sample:
            return _donation_sample_chain_node(
                db,
                tenant_no,
                donation_sample,
                wo=wo,
                edge=edge,
                sample_type_name=input_name,
            )

    return None


def _find_cell_product_by_input(db: Session, inp: Any) -> Optional[BizCellProduct]:
    if not (inp.input_id or inp.input_code):
        return None
    if inp.input_id:
        product = (
            db.query(BizCellProduct)
            .filter(BizCellProduct.id == inp.input_id, BizCellProduct.del_flag == 0)
            .first()
        )
        if product:
            return product
    if inp.input_code:
        return (
            db.query(BizCellProduct)
            .filter(BizCellProduct.product_code == inp.input_code, BizCellProduct.del_flag == 0)
            .first()
        )
    return None


def _ancestor_link_from_cell_product_input(
    db: Session,
    tenant_no: str,
    inp: Any,
    edge: Dict[str, Any],
    depth: int,
) -> Optional[Dict[str, Any]]:
    parent_product = _find_cell_product_by_input(db, inp)
    if not parent_product:
        return None
    parent_chain = _build_ancestor_chain(db, parent_product, depth + 1, tenant_no)
    return {
        "node": _product_node(parent_product),
        "edge": edge,
        "parent": _nest_ancestor_parent(parent_chain),
    }


def _build_ancestor_chain(
    db: Session, product: BizCellProduct, depth: int, tenant_no: str
) -> List[Dict[str, Any]]:
    if depth >= _MAX_CHAIN_DEPTH:
        return []
    task_id = product.production_task_id
    if not task_id:
        return []

    from app.models.models import BizProductionTask, BizProductionWorkOrder
    from app.models.models import BizProductionTaskInput

    task = (
        db.query(BizProductionTask)
        .filter(BizProductionTask.id == task_id, BizProductionTask.del_flag == 0)
        .first()
    )
    if not task:
        return []

    wo = None
    if task.work_order_id:
        wo = (
            db.query(BizProductionWorkOrder)
            .filter(BizProductionWorkOrder.id == task.work_order_id, BizProductionWorkOrder.del_flag == 0)
            .first()
        )
    edge = _edge_meta(task, wo)

    inputs = (
        db.query(BizProductionTaskInput)
        .filter(
            BizProductionTaskInput.task_id == task_id,
            BizProductionTaskInput.input_type.in_(("cell_product", "sample")),
            BizProductionTaskInput.del_flag == 0,
        )
        .order_by(
            BizProductionTaskInput.order_num.asc(),
            BizProductionTaskInput.create_time.asc(),
        )
        .all()
    )
    if not inputs:
        return []

    links: List[Dict[str, Any]] = []
    seen_node_ids: set = set()

    for inp in inputs:
        link: Optional[Dict[str, Any]] = None
        if inp.input_type == "cell_product":
            link = _ancestor_link_from_cell_product_input(
                db, tenant_no, inp, edge, depth + 1
            )
        elif inp.input_type == "sample":
            link = _ancestor_sample_chain_link(db, tenant_no, inp, wo, edge)

        if not link:
            continue
        node_id = (link.get("node") or {}).get("id") or ""
        if node_id and node_id in seen_node_ids:
            continue
        if node_id:
            seen_node_ids.add(node_id)
        links.append(link)

    return links


def _build_descendant_chain(
    db: Session, product_id: str, product_code: str, depth: int, tenant_no: str
) -> List[Dict[str, Any]]:
    if depth >= _MAX_CHAIN_DEPTH:
        return []

    from app.models.models import BizProductionTaskInput, BizProductionTaskOutput, BizProductionTask, BizProductionWorkOrder

    # 找到引用此产品作为投入物的所有任务（id 或编码）
    input_match = [BizProductionTaskInput.input_id == product_id]
    if product_code:
        input_match.append(BizProductionTaskInput.input_code == product_code)
    input_rows = (
        db.query(BizProductionTaskInput)
        .filter(
            BizProductionTaskInput.input_type == "cell_product",
            BizProductionTaskInput.del_flag == 0,
            or_(*input_match),
        )
        .all()
    )
    if not input_rows:
        return []

    task_ids = list({r.task_id for r in input_rows})
    tasks = (
        db.query(BizProductionTask)
        .filter(BizProductionTask.id.in_(task_ids), BizProductionTask.del_flag == 0)
        .all()
    )
    task_by_id = {t.id: t for t in tasks}

    wo_ids = list({t.work_order_id for t in tasks if t.work_order_id})
    wo_by_id: Dict[str, Any] = {}
    if wo_ids:
        wos = (
            db.query(BizProductionWorkOrder)
            .filter(BizProductionWorkOrder.id.in_(wo_ids), BizProductionWorkOrder.del_flag == 0)
            .all()
        )
        wo_by_id = {w.id: w for w in wos}

    # 对每个任务，找产出
    result: List[Dict[str, Any]] = []
    output_rows = (
        db.query(BizProductionTaskOutput)
        .filter(
            BizProductionTaskOutput.task_id.in_(task_ids),
            BizProductionTaskOutput.del_flag == 0,
        )
        .all()
    )

    child_product_ids = list({o.cell_product_id for o in output_rows if o.cell_product_id})
    child_products = {}
    if child_product_ids:
        cps = (
            db.query(BizCellProduct)
            .filter(BizCellProduct.id.in_(child_product_ids), BizCellProduct.del_flag == 0)
            .all()
        )
        child_products = {cp.id: cp for cp in cps}

    seen_child_ids: set = set()
    for o in output_rows:
        cp = child_products.get(o.cell_product_id)
        if not cp or cp.id in seen_child_ids:
            continue
        seen_child_ids.add(cp.id)
        t = task_by_id.get(o.task_id)
        wo_obj = wo_by_id.get(t.work_order_id) if t and t.work_order_id else None
        edge = _edge_meta(t, wo_obj)

        result.append({
            "node": _product_node(cp),
            "edge": edge,
            "children": _build_descendant_chain(
                db, cp.id, cp.product_code or "", depth + 1, tenant_no
            ),
        })

    return result


def _lineage_node_key(node: Dict[str, Any]) -> str:
    nt = (node.get("nodeType") or "").strip()
    nid = (node.get("id") or "").strip()
    if nt == "sample":
        code = (node.get("sampleCode") or "").strip()
        return f"sample:{(nid or code)}"
    return f"{nt}:{nid}"


def _walk_ancestor_link(
    link: Any,
    out: List[Dict[str, Any]],
    seen: set,
) -> None:
    if not link:
        return
    if isinstance(link, list):
        for item in link:
            _walk_ancestor_link(item, out, seen)
        return
    if not isinstance(link, dict):
        return
    parent = link.get("parent")
    _walk_ancestor_link(parent, out, seen)
    node = link.get("node")
    if not isinstance(node, dict):
        return
    key = _lineage_node_key(node)
    if not key or key in seen:
        return
    seen.add(key)
    out.append(dict(node))


def _flatten_lineage_nodes(chain: Dict[str, Any]) -> List[Dict[str, Any]]:
    """将 get_product_chain 结果展平为上游→下游节点列表（含 current，不含 descendants）。"""
    out: List[Dict[str, Any]] = []
    seen: set = set()
    for link in chain.get("ancestors") or []:
        _walk_ancestor_link(link, out, seen)
    current = chain.get("current")
    if isinstance(current, dict):
        key = _lineage_node_key(current)
        if key and key not in seen:
            seen.add(key)
            node = dict(current)
            node["isCurrent"] = True
            out.append(node)
        elif key:
            for n in out:
                if _lineage_node_key(n) == key:
                    n["isCurrent"] = True
                    break
    return out


def get_product_chain(db: Session, product_id: str, tenant_no: str) -> Optional[Dict[str, Any]]:
    product = get_cell_product(db, product_id, tenant_no)
    if not product:
        return None

    return {
        "current": _product_node(product, is_current=True),
        "ancestors": _build_ancestor_chain(db, product, 0, tenant_no),
        "descendants": _build_descendant_chain(
            db, product.id, product.product_code or "", 0, tenant_no
        ),
    }


def _wo_status_label(status: str) -> str:
    from app.crud.qc_work_order import _WORK_ORDER_STATUS_LABELS

    return _WORK_ORDER_STATUS_LABELS.get(status or "", status or "")


def _slim_task_for_lineage(
    task: Dict[str, Any], results: List[Dict[str, Any]]
) -> Dict[str, Any]:
    return {
        "id": task.get("id") or "",
        "taskNo": task.get("taskNo") or "",
        "itemName": task.get("itemName") or "",
        "itemCode": task.get("itemCode") or "",
        "taskType": task.get("taskType") or "",
        "taskStatus": task.get("taskStatus") or "",
        "taskStatusLabel": task.get("taskStatusLabel") or "",
        "conclusion": task.get("conclusion") or "",
        "conclusionLabel": task.get("conclusionLabel") or "",
        "remark": task.get("remark") or "",
        "completedTime": task.get("completedTime") or "",
        "companyName": task.get("companyName") or "",
        "results": [
            {
                "paramName": r.get("paramName") or "",
                "paramValue": r.get("paramValue") or "",
                "paramUnit": r.get("paramUnit") or "",
            }
            for r in (results or [])
        ],
        "attachmentFiles": task.get("attachmentFiles") or [],
    }


def _load_lineage_qc_for_nodes(
    db: Session, nodes: List[Dict[str, Any]], tenant_no: str
) -> List[Dict[str, Any]]:
    from app.models.models import BizQcWorkOrder
    from app.crud import qc_task as task_svc
    from app.crud import qc_task_result as result_svc
    from app.crud.qc_work_order import list_work_orders_by_sample

    product_ids = [
        (n.get("id") or "").strip()
        for n in nodes
        if (n.get("nodeType") or "") == "cell_product" and (n.get("id") or "").strip()
    ]
    wo_by_product: Dict[str, List[BizQcWorkOrder]] = {pid: [] for pid in product_ids}
    if product_ids:
        rows = (
            db.query(BizQcWorkOrder)
            .filter(
                BizQcWorkOrder.cell_product_id.in_(product_ids),
                BizQcWorkOrder.tenant_no == tenant_no,
                BizQcWorkOrder.del_flag == 0,
            )
            .order_by(BizQcWorkOrder.create_time.desc())
            .all()
        )
        for wo in rows:
            pid = (wo.cell_product_id or "").strip()
            if pid in wo_by_product:
                wo_by_product[pid].append(wo)

    sample_wo_cache: Dict[Tuple[str, str], List[Dict[str, Any]]] = {}

    node_work_orders_orm: List[List[BizQcWorkOrder]] = []
    for n in nodes:
        if (n.get("nodeType") or "") == "cell_product":
            node_work_orders_orm.append(wo_by_product.get((n.get("id") or "").strip(), []))
        else:
            node_work_orders_orm.append([])

    sample_wo_dicts: Dict[int, List[Dict[str, Any]]] = {}
    for idx, n in enumerate(nodes):
        if (n.get("nodeType") or "") != "sample":
            continue
        code = (n.get("sampleCode") or "").strip()
        rs = (n.get("receptionSource") or "").strip()
        if not code or not rs:
            sample_wo_dicts[idx] = []
            continue
        key = (code, rs)
        if key not in sample_wo_cache:
            sample_wo_cache[key] = list_work_orders_by_sample(
                db,
                sample_code=code,
                reception_source=rs,
                tenant_no=tenant_no,
                user_info=None,
            )
        sample_wo_dicts[idx] = sample_wo_cache[key]

    pending_wo_ids: List[str] = []
    for wos in node_work_orders_orm:
        for wo in wos:
            pending_wo_ids.append(wo.id)
    for wos in sample_wo_dicts.values():
        for wo in wos:
            wid = (wo.get("id") or "").strip()
            if wid:
                pending_wo_ids.append(wid)

    unique_wo_ids = list({(x or "").strip() for x in pending_wo_ids if (x or "").strip()})
    tasks_by_wo: Dict[str, List[Dict[str, Any]]] = {}
    for i in range(0, len(unique_wo_ids), 100):
        chunk = unique_wo_ids[i : i + 100]
        chunk_tasks = task_svc.list_tasks_by_work_order_ids(db, chunk, tenant_no)
        for wo_id, tasks in chunk_tasks.items():
            tasks_by_wo.setdefault(wo_id, []).extend(tasks)

    all_task_ids = [
        (t.get("id") or "").strip()
        for tasks in tasks_by_wo.values()
        for t in tasks
        if (t.get("id") or "").strip()
    ]
    results_by_task = result_svc.list_saved_results_by_task_ids(db, all_task_ids, tenant_no)

    def pack_orm_wos(wos: List[BizQcWorkOrder]) -> List[Dict[str, Any]]:
        packed = []
        for wo in wos:
            tasks = tasks_by_wo.get(wo.id, [])
            packed.append(
                {
                    "id": wo.id,
                    "workOrderNo": wo.work_order_no or "",
                    "workOrderType": wo.work_order_type or "",
                    "workOrderStatus": wo.work_order_status or "",
                    "workOrderStatusLabel": _wo_status_label(wo.work_order_status or ""),
                    "tasks": [
                        _slim_task_for_lineage(
                            t, results_by_task.get(t.get("id") or "", [])
                        )
                        for t in tasks
                    ],
                }
            )
        return packed

    def pack_dict_wos(wos: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        packed = []
        for wo in wos:
            wid = (wo.get("id") or "").strip()
            tasks = tasks_by_wo.get(wid, [])
            packed.append(
                {
                    "id": wid,
                    "workOrderNo": wo.get("workOrderNo") or "",
                    "workOrderType": wo.get("workOrderType") or "",
                    "workOrderStatus": wo.get("workOrderStatus") or "",
                    "workOrderStatusLabel": wo.get("workOrderStatusLabel")
                    or _wo_status_label(wo.get("workOrderStatus") or ""),
                    "tasks": [
                        _slim_task_for_lineage(
                            t, results_by_task.get(t.get("id") or "", [])
                        )
                        for t in tasks
                    ],
                }
            )
        return packed

    out_nodes: List[Dict[str, Any]] = []
    for idx, n in enumerate(nodes):
        nt = n.get("nodeType") or ""
        if nt == "sample":
            label = (n.get("sampleCode") or n.get("id") or "").strip()
            work_orders = pack_dict_wos(sample_wo_dicts.get(idx, []))
        else:
            label = (n.get("productCode") or n.get("id") or "").strip()
            work_orders = pack_orm_wos(node_work_orders_orm[idx])
        out_nodes.append(
            {
                "nodeType": nt,
                "id": n.get("id") or "",
                "label": label,
                "productName": n.get("productName") or "",
                "passage": n.get("passage") or None,
                "receptionSource": n.get("receptionSource") or None,
                "receptionSourceLabel": n.get("receptionSourceLabel") or None,
                "isCurrent": bool(n.get("isCurrent")),
                "workOrders": work_orders,
            }
        )
    return out_nodes


def get_product_lineage_qc(
    db: Session, product_id: str, tenant_no: str
) -> Optional[Dict[str, Any]]:
    chain = get_product_chain(db, product_id, tenant_no)
    if not chain:
        return None
    current = chain.get("current") or {}
    flat = _flatten_lineage_nodes(chain)
    nodes = _load_lineage_qc_for_nodes(db, flat, tenant_no)
    return {
        "productId": current.get("id") or product_id,
        "productCode": current.get("productCode") or "",
        "productName": current.get("productName") or "",
        "nodes": nodes,
    }