zkxw/backend/app/services/market_sample.py

"""市场移动端统一样本查询与轻量办理。"""
from __future__ import annotations

import uuid
from datetime import datetime
from typing import Any, Dict, Iterable, List, Literal, Optional, Tuple

from fastapi import HTTPException
from sqlalchemy import and_, literal, or_
from sqlalchemy.orm import Session

from app.models.models import (
    BizCellSampleType,
    BizCollectionProtocol,
    BizProtocolCollectionInfo,
    BizCollectionRegionStaff,
    BizProtocolFullmoonTodo,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizProtocolStorageCert,
    BizQcWorkOrder,
    BizSampleReception,
    BizStorageReleaseReview,
    SysUser,
)
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationCollection,
    BizResearchDonationParty,
    BizResearchDonationSample,
)
from app.services.market_field_scope import (
    MarketFieldContext,
    apply_cell_protocol_scope,
    apply_donation_scope,
    leader_region_ids_query,
)
from app.services.market_support import validate_attachment_refs
from app.services.sample_info_progress import (
    apply_handle_payload,
    insurance_handled,
    storage_cert_handled,
)

CELL_SOURCE = 'cell_storage'
RESEARCH_SOURCE = 'research_donation'
VALID_SOURCES = (CELL_SOURCE, RESEARCH_SOURCE)


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


def _fmt_date(value: Any) -> str:
    if not value:
        return ''
    if hasattr(value, 'strftime'):
        return value.strftime('%Y-%m-%d')
    text = str(value)
    return f'{text[:4]}-{text[4:6]}-{text[6:8]}' if len(text) == 8 and text.isdigit() else text


def _parse_dt(value: Any) -> Optional[datetime]:
    if value in (None, ''):
        return None
    if isinstance(value, datetime):
        return value
    text = str(value).strip()
    for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M:%S'):
        try:
            return datetime.strptime(text, fmt)
        except ValueError:
            continue
    raise ValueError('办理时间格式无效')


def _status_meta(code: str, label: str) -> Dict[str, str]:
    normalized = (code or '').strip()
    if normalized in (
        'ycj',
        'collected',
        'js',
        'complete',
        'tg',
        'yrk',
        'handled',
    ):
        tag_type = 'success'
    elif normalized in ('jcz', 'incomplete', 'processing'):
        tag_type = 'warning'
    elif normalized in ('yzf', 'voided', 'bj', 'btg', 'rejected'):
        tag_type = 'danger'
    else:
        tag_type = 'info'
    return {
        'code': normalized,
        'label': label,
        'statusTone': tag_type,
        'tagType': tag_type,
    }


def _sample_status(status: str) -> Dict[str, str]:
    status = (status or '').strip()
    label = {
        'ycj': '已采集',
        'jcz': '检测中',
        'yrk': '已入库',
        'yzf': '已作废',
    }.get(status, '未采集' if not status else status)
    return _status_meta(status or 'none', label)


def _collection_progress(status: str) -> Dict[str, str]:
    status = (status or '').strip()
    if status == 'yzf':
        return _status_meta('voided', '已作废')
    if status in ('ycj', 'jcz', 'yrk'):
        return _status_meta('collected', '已采集')
    return _status_meta('uncollected', '未采集')


def _apply_joined_collection_scope(query, model, db: Session, ctx: MarketFieldContext):
    if ctx.scope_kind == 'specialist':
        return query.filter(model.sales_user_id == ctx.user_id)
    return query.filter(
        or_(
            model.region_id.in_(leader_region_ids_query(db, ctx)),
            model.region_manager_user_id == ctx.user_id,
        )
    )


def _stable_id(source: str, sample_code: str, row_id: str) -> str:
    code = (sample_code or '').strip()
    return f'{source}:{code}' if code else f'{source}:@id:{row_id}'


def _parse_stable_id(sample_id: str) -> Tuple[str, str, bool]:
    source, separator, token = (sample_id or '').partition(':')
    token = token.strip()
    if not separator or source not in VALID_SOURCES or not token:
        raise HTTPException(status_code=404, detail='数据不存在或无权访问')
    if token.startswith('@id:'):
        row_id = token[4:].strip()
        if not row_id:
            raise HTTPException(status_code=404, detail='数据不存在或无权访问')
        return source, row_id, True
    return source, token, False


def _materialization_order(query, source: str):
    if source == CELL_SOURCE:
        return query.order_by(
            BizProtocolCollectionInfo.create_time.desc(),
            BizProtocolCollectionInfo.id.desc(),
            BizProtocolPartyCustomer.create_time.desc(),
            BizProtocolPartyCustomer.id.desc(),
        )
    return query.order_by(
        BizResearchDonationCollection.create_time.desc(),
        BizResearchDonationCollection.id.desc(),
        BizResearchDonationParty.create_time.desc(),
        BizResearchDonationParty.id.desc(),
    )


def _cell_query(db: Session, ctx: MarketFieldContext, user_info: dict):
    query = (
        db.query(
            BizProtocolSample,
            BizCollectionProtocol,
            BizProtocolCollectionInfo,
            BizProtocolPartyCustomer,
        )
        .join(
            BizCollectionProtocol,
            and_(
                BizCollectionProtocol.id == BizProtocolSample.collection_protocol_id,
                BizCollectionProtocol.tenant_no == ctx.tenant_no,
                BizCollectionProtocol.del_flag == 0,
            ),
        )
        .join(
            BizProtocolCollectionInfo,
            and_(
                BizProtocolCollectionInfo.collection_protocol_id == BizCollectionProtocol.id,
                BizProtocolCollectionInfo.tenant_no == ctx.tenant_no,
                BizProtocolCollectionInfo.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolPartyCustomer,
            and_(
                BizProtocolPartyCustomer.collection_protocol_id == BizCollectionProtocol.id,
                BizProtocolPartyCustomer.tenant_no == ctx.tenant_no,
                BizProtocolPartyCustomer.del_flag == 0,
            ),
        )
        .filter(
            BizProtocolSample.tenant_no == ctx.tenant_no,
            BizProtocolSample.del_flag == 0,
        )
    )
    query = apply_cell_protocol_scope(query, db, ctx, user_info=user_info)
    return _apply_joined_collection_scope(
        query, BizProtocolCollectionInfo, db, ctx
    )


def _research_query(db: Session, ctx: MarketFieldContext, user_info: dict):
    query = (
        db.query(
            BizResearchDonationSample,
            BizResearchDonation,
            BizResearchDonationCollection,
            BizResearchDonationParty,
        )
        .join(
            BizResearchDonation,
            and_(
                BizResearchDonation.id == BizResearchDonationSample.research_donation_id,
                BizResearchDonation.tenant_no == ctx.tenant_no,
                BizResearchDonation.del_flag == 0,
            ),
        )
        .join(
            BizResearchDonationCollection,
            and_(
                BizResearchDonationCollection.research_donation_id == BizResearchDonation.id,
                BizResearchDonationCollection.tenant_no == ctx.tenant_no,
                BizResearchDonationCollection.del_flag == 0,
            ),
        )
        .outerjoin(
            BizResearchDonationParty,
            and_(
                BizResearchDonationParty.research_donation_id == BizResearchDonation.id,
                BizResearchDonationParty.tenant_no == ctx.tenant_no,
                BizResearchDonationParty.del_flag == 0,
            ),
        )
        .filter(
            BizResearchDonationSample.tenant_no == ctx.tenant_no,
            BizResearchDonationSample.del_flag == 0,
        )
    )
    query = apply_donation_scope(query, db, ctx, user_info=user_info)
    return _apply_joined_collection_scope(
        query, BizResearchDonationCollection, db, ctx
    )


def _apply_filters(query, source: str, keyword: Optional[str], status: Optional[str]):
    if source == CELL_SOURCE:
        sample = BizProtocolSample
        parent = BizCollectionProtocol
        party = BizProtocolPartyCustomer
    else:
        sample = BizResearchDonationSample
        parent = BizResearchDonation
        party = BizResearchDonationParty
    text = (keyword or '').strip()
    if text:
        like = f'%{text}%'
        number_col = (
            BizCollectionProtocol.protocol_no
            if source == CELL_SOURCE
            else BizResearchDonation.donation_no
        )
        query = query.filter(
            or_(sample.sample_code.like(like), number_col.like(like), party.mother_name.like(like))
        )
    normalized_status = (status or '').strip()
    if normalized_status and normalized_status not in ('all', 'pending', 'processing', 'completed'):
        query = query.filter(sample.sample_status == normalized_status)
    elif normalized_status == 'pending':
        query = query.filter(or_(sample.sample_status.is_(None), sample.sample_status == ''))
    elif normalized_status == 'processing':
        query = query.filter(sample.sample_status.in_(('ycj', 'jcz')))
    elif normalized_status == 'completed':
        query = query.filter(sample.sample_status == 'yrk')
    return query


def _identity_query(query, source: str):
    sample = BizProtocolSample if source == CELL_SOURCE else BizResearchDonationSample
    return query.with_entities(
        literal(source).label('source'),
        sample.id.label('row_id'),
        sample.create_time.label('create_time'),
    ).distinct()


def _load_rows(
    db: Session,
    identities: Iterable[Any],
    ctx: MarketFieldContext,
    user_info: dict,
) -> List[Dict[str, Any]]:
    ordered = [(row.source, row.row_id) for row in identities]
    cell_ids = [row_id for source, row_id in ordered if source == CELL_SOURCE]
    research_ids = [row_id for source, row_id in ordered if source == RESEARCH_SOURCE]
    rows_by_key: Dict[Tuple[str, str], Dict[str, Any]] = {}
    if cell_ids:
        query = _cell_query(db, ctx, user_info).filter(
            BizProtocolSample.id.in_(cell_ids)
        )
        for sample, parent, collection, party in _materialization_order(
            query, CELL_SOURCE
        ).all():
            rows_by_key.setdefault((CELL_SOURCE, sample.id), _raw_item(
                CELL_SOURCE, sample, parent, collection, party
            ))
    if research_ids:
        query = _research_query(db, ctx, user_info).filter(
            BizResearchDonationSample.id.in_(research_ids)
        )
        for sample, parent, collection, party in _materialization_order(
            query, RESEARCH_SOURCE
        ).all():
            rows_by_key.setdefault((RESEARCH_SOURCE, sample.id), _raw_item(
                RESEARCH_SOURCE, sample, parent, collection, party
            ))
    return [rows_by_key[key] for key in ordered if key in rows_by_key]


def _raw_item(source: str, sample, parent, collection, party) -> Dict[str, Any]:
    number = parent.protocol_no if source == CELL_SOURCE else parent.donation_no
    return {
        'id': _stable_id(source, sample.sample_code or '', sample.id),
        'source': source,
        'sourceLabel': '细胞存储' if source == CELL_SOURCE else '科研捐赠',
        'sampleRowId': sample.id,
        'parentId': parent.id,
        'protocolNo': number or '',
        'protocolStatus': (
            parent.protocol_status
            if source == CELL_SOURCE
            else parent.donation_status
        ) or '',
        'sampleCode': sample.sample_code or '',
        'sampleTypeId': sample.sample_type_id or '',
        'sampleTypeName': '',
        'sampleStatus': sample.sample_status or '',
        'customerName': party.mother_name if party else '',
        'motherName': party.mother_name if party else '',
        'newbornName': party.newborn_name if party else '',
        'expectedDeliveryDate': _fmt_date(
            party.expected_delivery_date if party else None
        ),
        'regionId': collection.region_id or '',
        'hospitalId': collection.hospital_id or '',
        'salesUserId': collection.sales_user_id or '',
        'regionManagerUserId': collection.region_manager_user_id or '',
        'sampleCollectTime': _fmt_dt(collection.sample_collect_time),
        'createTime': _fmt_dt(sample.create_time),
    }


def _latest_maps(
    db: Session, tenant_no: str, items: List[Dict[str, Any]]
) -> Tuple[dict, dict, dict]:
    keys = {
        (item['source'], (item.get('sampleCode') or '').strip())
        for item in items
        if (item.get('sampleCode') or '').strip()
    }
    if not keys:
        return {}, {}, {}
    clauses = [
        and_(
            BizSampleReception.reception_source == source,
            BizSampleReception.sample_code.in_(
                [code for key_source, code in keys if key_source == source]
            ),
        )
        for source in VALID_SOURCES
        if any(key_source == source for key_source, _ in keys)
    ]
    receptions = (
        db.query(BizSampleReception)
        .filter(
            BizSampleReception.tenant_no == tenant_no,
            BizSampleReception.del_flag == 0,
            or_(*clauses),
        )
        .order_by(
            BizSampleReception.reception_time.desc(),
            BizSampleReception.create_time.desc(),
            BizSampleReception.id.desc(),
        )
        .all()
    )
    qc_clauses = [
        and_(
            BizQcWorkOrder.reception_source == source,
            BizQcWorkOrder.sample_code.in_(
                [code for key_source, code in keys if key_source == source]
            ),
        )
        for source in VALID_SOURCES
        if any(key_source == source for key_source, _ in keys)
    ]
    qc_rows = (
        db.query(BizQcWorkOrder)
        .filter(
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.del_flag == 0,
            BizQcWorkOrder.trigger_source == 'reception',
            or_(*qc_clauses),
        )
        .order_by(
            BizQcWorkOrder.update_time.desc(),
            BizQcWorkOrder.create_time.desc(),
            BizQcWorkOrder.id.desc(),
        )
        .all()
    )
    storage_clauses = [
        and_(
            BizStorageReleaseReview.reception_source == source,
            BizStorageReleaseReview.sample_code.in_(
                [code for key_source, code in keys if key_source == source]
            ),
        )
        for source in VALID_SOURCES
        if any(key_source == source for key_source, _ in keys)
    ]
    storage_rows = (
        db.query(BizStorageReleaseReview)
        .filter(
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
            or_(*storage_clauses),
        )
        .order_by(
            BizStorageReleaseReview.storage_time.desc(),
            BizStorageReleaseReview.update_time.desc(),
            BizStorageReleaseReview.create_time.desc(),
            BizStorageReleaseReview.id.desc(),
        )
        .all()
    )
    reception_map = {}
    qc_map = {}
    storage_map = {}
    for row in receptions:
        reception_map.setdefault((row.reception_source, (row.sample_code or '').strip()), row)
    for row in qc_rows:
        qc_map.setdefault((row.reception_source, (row.sample_code or '').strip()), row)
    for row in storage_rows:
        storage_map.setdefault((row.reception_source, (row.sample_code or '').strip()), row)
    return reception_map, qc_map, storage_map


def _handle_maps(db: Session, tenant_no: str, items: List[Dict[str, Any]]) -> Tuple[dict, dict]:
    parent_ids = {
        item['parentId'] for item in items if item['source'] == CELL_SOURCE
    }
    if not parent_ids:
        return {}, {}
    cert_rows = (
        db.query(BizProtocolStorageCert)
        .filter(
            BizProtocolStorageCert.tenant_no == tenant_no,
            BizProtocolStorageCert.del_flag == 0,
            BizProtocolStorageCert.collection_protocol_id.in_(parent_ids),
        )
        .order_by(
            BizProtocolStorageCert.update_time.desc(),
            BizProtocolStorageCert.create_time.desc(),
            BizProtocolStorageCert.id.desc(),
        )
        .all()
    )
    todo_rows = (
        db.query(BizProtocolFullmoonTodo)
        .filter(
            BizProtocolFullmoonTodo.tenant_no == tenant_no,
            BizProtocolFullmoonTodo.del_flag == 0,
            BizProtocolFullmoonTodo.collection_protocol_id.in_(parent_ids),
        )
        .order_by(
            BizProtocolFullmoonTodo.update_time.desc(),
            BizProtocolFullmoonTodo.create_time.desc(),
            BizProtocolFullmoonTodo.id.desc(),
        )
        .all()
    )
    cert_map = {}
    todo_map = {}
    for row in cert_rows:
        cert_map.setdefault(row.collection_protocol_id, row)
    for row in todo_rows:
        todo_map.setdefault(row.collection_protocol_id, row)
    return cert_map, todo_map


def _handle_summary(
    row, *, insurance: bool = False, handler_user_name: str = ''
) -> Dict[str, Any]:
    if row is None:
        return {
            **_status_meta('pending', '未办理'),
            'handleTime': '',
            'handlerUserId': '',
            'handlerUserName': '',
            'attachments': '',
            'remark': '',
        }
    attachments = row.insurance_attachments if insurance else row.cert_image_urls
    remark = row.handle_remark if insurance else row.remark
    handled = insurance_handled(row) if insurance else storage_cert_handled(row)
    return {
        **_status_meta('handled' if handled else 'pending', '已办理' if handled else '未办理'),
        'handleTime': _fmt_dt(row.handle_time),
        'handlerUserId': row.handler_user_id or '',
        'handlerUserName': handler_user_name,
        'attachments': attachments or '',
        'remark': remark or '',
    }


def _referenced_user_name_map(
    db: Session, tenant_no: str, user_ids: List[str]
) -> Dict[str, str]:
    wanted = {str(user_id or '').strip() for user_id in user_ids if user_id}
    if not wanted:
        return {}
    return {
        str(user_id): str(nick_name or '').strip()
        for user_id, nick_name in db.query(SysUser.id, SysUser.nick_name)
        .filter(
            SysUser.id.in_(wanted),
            SysUser.tenant_no == tenant_no,
        )
        .all()
    }


def _enrich(db: Session, tenant_no: str, items: List[Dict[str, Any]]) -> None:
    type_ids = {item['sampleTypeId'] for item in items if item['sampleTypeId']}
    type_map = {}
    if type_ids:
        type_map = {
            row.id: row.type_name or ''
            for row in db.query(BizCellSampleType).filter(
                BizCellSampleType.id.in_(type_ids),
                BizCellSampleType.del_flag == 0,
                or_(BizCellSampleType.tenant_no == tenant_no, BizCellSampleType.tenant_no == ''),
            )
        }
    reception_map, qc_map, storage_map = _latest_maps(db, tenant_no, items)
    cert_map, todo_map = _handle_maps(db, tenant_no, items)
    handler_names = _referenced_user_name_map(
        db,
        tenant_no,
        [
            row.handler_user_id
            for row in (*cert_map.values(), *todo_map.values())
            if row.handler_user_id
        ],
    )
    for item in items:
        item['sampleTypeName'] = type_map.get(item['sampleTypeId'], '')
        sample_status = _sample_status(item['sampleStatus'])
        collection_progress = _collection_progress(item['sampleStatus'])
        item.update(
            statusLabel=sample_status['label'],
            statusTone=sample_status['statusTone'],
            tagType=sample_status['tagType'],
        )
        code = (item['sampleCode'] or '').strip()
        key = (item['source'], code)
        reception = reception_map.get(key) if code else None
        qc = qc_map.get(key) if code else None
        storage = storage_map.get(key) if code else None
        reception_code = (reception.accept_conclusion or '').strip() if reception else 'none'
        reception_label = {'js': '已接收', 'bj': '拒收'}.get(reception_code, '未接收')
        qc_code = (qc.conclusion or '').strip() if qc else 'incomplete'
        if qc and qc.work_order_status == 'yzf':
            qc_code = 'yzf'
        qc_label = {'tg': '通过', 'btg': '不通过', 'yzf': '已作废'}.get(qc_code, '未完成')
        if storage and storage.storage_status == 'yrk':
            storage_code, storage_label = 'yrk', '已入库'
        elif storage and storage.review_conclusion == 'btg':
            storage_code, storage_label = 'rejected', '拒绝入库'
        else:
            storage_code, storage_label = 'none', '未入库'
        item['progress'] = {
            'collection': collection_progress,
            'reception': {
                **_status_meta(reception_code, reception_label),
                'recordId': reception.id if reception else '',
                'time': _fmt_dt(reception.reception_time) if reception else '',
            },
            'qc': {
                **_status_meta(qc_code, qc_label),
                'recordId': qc.id if qc else '',
                'workOrderNo': qc.work_order_no if qc else '',
            },
            'storage': {
                **_status_meta(storage_code, storage_label),
                'recordId': storage.id if storage else '',
                'time': _fmt_dt(storage.storage_time) if storage else '',
            },
        }
        item['labStorageCode'] = (
            getattr(reception, 'lab_storage_code', None) or ''
        )
        if item['source'] == CELL_SOURCE:
            cert = cert_map.get(item['parentId'])
            todo = todo_map.get(item['parentId'])
            item['storageCert'] = _handle_summary(
                cert,
                handler_user_name=handler_names.get(
                    getattr(cert, 'handler_user_id', '') or '', ''
                ),
            )
            item['insurance'] = _handle_summary(
                todo,
                insurance=True,
                handler_user_name=handler_names.get(
                    getattr(todo, 'handler_user_id', '') or '', ''
                ),
            )


def list_samples(
    db: Session,
    *,
    ctx: MarketFieldContext,
    user_info: dict,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    status: Optional[str] = None,
    source: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    page = max(1, int(page or 1))
    page_size = min(max(1, int(page_size or 10)), 100)
    identities = []
    if source in (None, '', CELL_SOURCE):
        cell = _apply_filters(_cell_query(db, ctx, user_info), CELL_SOURCE, keyword, status)
        identities.append(_identity_query(cell, CELL_SOURCE))
    if source in (None, '', RESEARCH_SOURCE):
        research = _apply_filters(
            _research_query(db, ctx, user_info), RESEARCH_SOURCE, keyword, status
        )
        identities.append(_identity_query(research, RESEARCH_SOURCE))
    if not identities:
        return [], 0
    combined = identities[0].union_all(*identities[1:]).subquery()
    total = db.query(combined).count()
    page_rows = (
        db.query(combined)
        .order_by(combined.c.create_time.desc(), combined.c.row_id.desc(), combined.c.source.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    items = _load_rows(db, page_rows, ctx, user_info)
    _enrich(db, ctx.tenant_no, items)
    return items, total


def _cell_sample_lookup_query(
    db: Session,
    *,
    token: str,
    by_row_id: bool,
    ctx: MarketFieldContext,
    user_info: dict,
    for_update: bool = False,
):
    query = _cell_query(db, ctx, user_info)
    if by_row_id:
        query = query.filter(
            BizProtocolSample.id == token,
            or_(
                BizProtocolSample.sample_code.is_(None),
                BizProtocolSample.sample_code == '',
            ),
        )
    else:
        query = query.filter(BizProtocolSample.sample_code == token)
    query = query.order_by(
        BizProtocolSample.create_time.desc(),
        BizProtocolSample.id.desc(),
        BizProtocolCollectionInfo.create_time.desc(),
        BizProtocolCollectionInfo.id.desc(),
        BizProtocolPartyCustomer.create_time.desc(),
        BizProtocolPartyCustomer.id.desc(),
    )
    if for_update:
        query = query.populate_existing().with_for_update()
    return query


def _find_sample(
    db: Session,
    sample_id: str,
    ctx: MarketFieldContext,
    user_info: dict,
    *,
    for_update: bool = False,
) -> Dict[str, Any]:
    source, token, by_row_id = _parse_stable_id(sample_id)
    if source == CELL_SOURCE:
        row = _cell_sample_lookup_query(
            db,
            token=token,
            by_row_id=by_row_id,
            ctx=ctx,
            user_info=user_info,
            for_update=for_update,
        ).first()
    else:
        query = _research_query(db, ctx, user_info)
        if by_row_id:
            query = query.filter(
                BizResearchDonationSample.id == token,
                or_(
                    BizResearchDonationSample.sample_code.is_(None),
                    BizResearchDonationSample.sample_code == '',
                ),
            )
        else:
            query = query.filter(BizResearchDonationSample.sample_code == token)
        query = query.order_by(
            BizResearchDonationSample.create_time.desc(),
            BizResearchDonationSample.id.desc(),
            BizResearchDonationCollection.create_time.desc(),
            BizResearchDonationCollection.id.desc(),
            BizResearchDonationParty.create_time.desc(),
            BizResearchDonationParty.id.desc(),
        )
        if for_update:
            query = query.populate_existing().with_for_update()
        row = query.first()
    if row is None:
        raise HTTPException(status_code=404, detail='数据不存在或无权访问')
    return _raw_item(source, *row)


def get_sample_detail(
    db: Session,
    *,
    sample_id: str,
    ctx: MarketFieldContext,
    user_info: dict,
) -> Dict[str, Any]:
    item = _find_sample(db, sample_id, ctx, user_info)
    _enrich(db, ctx.tenant_no, [item])
    progress = item.pop('progress')
    storage_cert = item.pop('storageCert', None)
    insurance = item.pop('insurance', None)
    sections = [
        {
            'key': 'basic',
            'title': '基本信息',
            'fields': [
                {'label': '业务来源', 'value': item['sourceLabel']},
                {'label': '协议编号', 'value': item['protocolNo']},
                {'label': '样本编码', 'value': item['sampleCode']},
                {'label': '样本类型', 'value': item['sampleTypeName']},
                {'label': '实验室入库编码', 'value': item['labStorageCode']},
            ],
        },
        {
            'key': 'progress',
            'title': '办理进度',
            'fields': [
                {'label': '采集', 'value': progress['collection']['label']},
                {'label': '接收', 'value': progress['reception']['label']},
                {'label': '质检', 'value': progress['qc']['label']},
                {'label': '入库', 'value': progress['storage']['label']},
            ],
        },
    ]
    if item['source'] == CELL_SOURCE:
        sections.append(
            {
                'key': 'market-handles',
                'title': '市场办理',
                'storageCert': storage_cert,
                'insurance': insurance,
            }
        )
    actions = []
    if item['source'] == CELL_SOURCE and item['sampleCode']:
        actions = [
            {'key': 'storageCert', 'label': '办理储存证'},
            {'key': 'insurance', 'label': '办理保险'},
        ]
    return {'id': item['id'], 'card': item, 'progress': progress, 'sections': sections, 'actions': actions}


def _eligible_region_staff_query(
    db: Session, tenant_no: str, region_id: str
):
    return (
        db.query(SysUser)
        .join(
            BizCollectionRegionStaff,
            BizCollectionRegionStaff.user_id == SysUser.id,
        )
        .filter(
            BizCollectionRegionStaff.region_id == region_id,
            BizCollectionRegionStaff.tenant_no == tenant_no,
            BizCollectionRegionStaff.del_flag == 0,
            SysUser.tenant_no == tenant_no,
            SysUser.status == '1',
            SysUser.del_flag == 0,
        )
    )


def get_sample_form_options(
    db: Session,
    *,
    sample_id: str,
    action: Optional[Literal['storage-cert', 'insurance']] = None,
    ctx: MarketFieldContext,
    user_info: dict,
) -> Dict[str, List[Dict[str, str]]]:
    item = _find_sample(db, sample_id, ctx, user_info)
    if item['source'] != CELL_SOURCE or not item['sampleCode']:
        raise ValueError('该样本不支持办理人员选择')
    users = (
        _eligible_region_staff_query(db, ctx.tenant_no, item['regionId'])
        .distinct()
        .order_by(SysUser.nick_name.asc(), SysUser.id.asc())
        .all()
    )
    referenced_ids = []
    if action:
        model = (
            BizProtocolStorageCert
            if action == 'storage-cert'
            else BizProtocolFullmoonTodo
        )
        historical = (
            db.query(model)
            .filter(
                model.collection_protocol_id == item['parentId'],
                model.tenant_no == ctx.tenant_no,
                model.del_flag == 0,
            )
            .order_by(
                model.update_time.desc(),
                model.create_time.desc(),
                model.id.desc(),
            )
            .first()
        )
        if historical is not None and historical.handler_user_id:
            referenced_ids.append(historical.handler_user_id)
    referenced_names = _referenced_user_name_map(
        db, ctx.tenant_no, referenced_ids
    )
    staff = [
        {'value': str(user.id), 'label': str(user.nick_name or '')}
        for user in users
    ]
    included = {option['value'] for option in staff}
    for user_id in referenced_ids:
        value = str(user_id)
        if value not in included and value in referenced_names:
            staff.append({'value': value, 'label': referenced_names[value]})
            included.add(value)
    return {
        'staff': staff
    }


def _validate_handler(
    db: Session,
    tenant_no: str,
    region_id: str,
    handler_user_id: str,
) -> SysUser:
    user = (
        _eligible_region_staff_query(db, tenant_no, region_id)
        .filter(SysUser.id == handler_user_id)
        .first()
    )
    if user is None:
        raise ValueError('办理人员无效')
    return user


def _handle_identity_changed(row, normalized: Dict[str, Any]) -> bool:
    """全量表单会回传原办理人/时间；只有语义变化时重新校验资格。"""
    return (
        str(normalized['handler_user_id'] or '').strip()
        != str(row.handler_user_id or '').strip()
        or normalized['handle_time'] != row.handle_time
    )


def _cell_parent_lock_query(
    db: Session, parent_id: str, ctx: MarketFieldContext
):
    return (
        db.query(BizCollectionProtocol)
        .populate_existing()
        .filter(
            BizCollectionProtocol.id == parent_id,
            BizCollectionProtocol.tenant_no == ctx.tenant_no,
            BizCollectionProtocol.del_flag == 0,
        )
        .with_for_update()
    )


def _lock_cell_parent(
    db: Session, parent_id: str, ctx: MarketFieldContext
) -> BizCollectionProtocol:
    parent = _cell_parent_lock_query(db, parent_id, ctx).first()
    if parent is None:
        raise HTTPException(status_code=404, detail='数据不存在或无权访问')
    return parent


def _handle_child_current_query(
    db: Session,
    model,
    parent_id: str,
    ctx: MarketFieldContext,
):
    return (
        db.query(model)
        .populate_existing()
        .filter(
            model.collection_protocol_id == parent_id,
            model.tenant_no == ctx.tenant_no,
            model.del_flag == 0,
        )
        .order_by(
            model.update_time.desc(),
            model.create_time.desc(),
            model.id.desc(),
        )
        .with_for_update()
    )


def _normalize_partial_handle(
    row,
    fields: Dict[str, Any],
    *,
    insurance: bool,
    handle_time: Optional[datetime],
    handler_user_id: str,
    attachments: str,
    remark: str,
) -> Dict[str, Any]:
    if 'handleTime' in fields:
        return apply_handle_payload(
            handle_time=handle_time,
            handler_user_id=handler_user_id,
            attachments=attachments,
            remark=remark,
        )
    if 'handlerUserId' in fields:
        if not handler_user_id:
            raise ValueError('办理人员不能为空')
        handled = (
            insurance_handled(row) if insurance else storage_cert_handled(row)
        )
        if not handled:
            raise ValueError('记录尚未办理，不能单独设置办理人员')
        return {
            'handle_time': row.handle_time,
            'handler_user_id': handler_user_id,
            'attachments': attachments,
            'remark': remark,
            'make_or_todo_status': '1',
            'legacy_date': '',
        }
    return {
        'handle_time': row.handle_time,
        'handler_user_id': row.handler_user_id,
        'attachments': attachments,
        'remark': remark,
        'make_or_todo_status': '0',
        'legacy_date': '',
    }


def _write_handle(
    db: Session,
    *,
    sample_id: str,
    fields: Dict[str, Any],
    ctx: MarketFieldContext,
    user_info: dict,
    kind: str,
) -> Dict[str, Any]:
    source, _, by_row_id = _parse_stable_id(sample_id)
    if source != CELL_SOURCE or by_row_id:
        raise ValueError('该样本不支持此办理操作')
    item = _find_sample(db, sample_id, ctx, user_info)
    if not item['sampleCode']:
        raise ValueError('该样本不支持此办理操作')
    initial_parent_id = item['parentId']
    try:
        _lock_cell_parent(db, initial_parent_id, ctx)
        item = _find_sample(
            db,
            sample_id,
            ctx,
            user_info,
            for_update=True,
        )
        if item['parentId'] != initial_parent_id:
            raise HTTPException(
                status_code=404, detail='数据不存在或无权访问'
            )
    except Exception:
        db.rollback()
        raise
    now = datetime.now()
    create_dept_id = str(
        user_info.get('dept_id') or user_info.get('deptId') or user_info.get('region_id') or ''
    )
    try:
        if kind == 'storage_cert':
            row = _handle_child_current_query(
                db, BizProtocolStorageCert, item['parentId'], ctx
            ).first()
            creating_handle = row is None
            if row is None:
                row = BizProtocolStorageCert(
                    id=str(uuid.uuid4()),
                    collection_protocol_id=item['parentId'],
                    tenant_no=ctx.tenant_no,
                    create_by=ctx.user_id,
                    create_dept_id=create_dept_id,
                    create_time=now,
                    del_flag=0,
                )
                db.add(row)
            handle_time = (
                _parse_dt(fields.get('handleTime'))
                if 'handleTime' in fields
                else row.handle_time
            )
            handler_user_id = (
                str(fields.get('handlerUserId') or '').strip()
                if 'handlerUserId' in fields
                else str(row.handler_user_id or '').strip()
            )
            attachments = (
                str(fields.get('attachments') or '').strip()
                if 'attachments' in fields
                else str(row.cert_image_urls or '').strip()
            )
            if 'attachments' in fields:
                attachments = validate_attachment_refs(
                    db,
                    attachments,
                    module='sample',
                    ctx=ctx,
                    creating=creating_handle,
                    existing_refs=row.cert_image_urls,
                    business_id=row.id,
                )
            remark = (
                str(fields.get('remark') or '').strip()
                if 'remark' in fields
                else str(row.remark or '').strip()
            )
            normalized = _normalize_partial_handle(
                row,
                fields,
                insurance=False,
                handle_time=handle_time,
                handler_user_id=handler_user_id,
                attachments=attachments,
                remark=remark,
            )
            should_validate_handler = creating_handle or _handle_identity_changed(
                row, normalized
            )
            handler = (
                _validate_handler(
                    db,
                    ctx.tenant_no,
                    item['regionId'],
                    normalized['handler_user_id'],
                )
                if normalized['handler_user_id'] and should_validate_handler
                else None
            )
            row.handle_time = normalized['handle_time']
            row.handler_user_id = normalized['handler_user_id']
            row.cert_image_urls = normalized['attachments']
            row.remark = normalized['remark']
            if 'handleTime' in fields:
                row.make_status = normalized['make_or_todo_status']
                row.make_date = normalized['legacy_date'] or None
        else:
            row = _handle_child_current_query(
                db, BizProtocolFullmoonTodo, item['parentId'], ctx
            ).first()
            creating_handle = row is None
            if row is None:
                row = BizProtocolFullmoonTodo(
                    id=str(uuid.uuid4()),
                    collection_protocol_id=item['parentId'],
                    tenant_no=ctx.tenant_no,
                    create_by=ctx.user_id,
                    create_dept_id=create_dept_id,
                    create_time=now,
                    del_flag=0,
                )
                db.add(row)
            handle_time = (
                _parse_dt(fields.get('handleTime'))
                if 'handleTime' in fields
                else row.handle_time
            )
            handler_user_id = (
                str(fields.get('handlerUserId') or '').strip()
                if 'handlerUserId' in fields
                else str(row.handler_user_id or '').strip()
            )
            attachments = (
                str(fields.get('attachments') or '').strip()
                if 'attachments' in fields
                else str(row.insurance_attachments or '').strip()
            )
            if 'attachments' in fields:
                attachments = validate_attachment_refs(
                    db,
                    attachments,
                    module='sample',
                    ctx=ctx,
                    creating=creating_handle,
                    existing_refs=row.insurance_attachments,
                    business_id=row.id,
                )
            remark = (
                str(fields.get('remark') or '').strip()
                if 'remark' in fields
                else str(row.handle_remark or '').strip()
            )
            normalized = _normalize_partial_handle(
                row,
                fields,
                insurance=True,
                handle_time=handle_time,
                handler_user_id=handler_user_id,
                attachments=attachments,
                remark=remark,
            )
            should_validate_handler = creating_handle or _handle_identity_changed(
                row, normalized
            )
            handler = (
                _validate_handler(
                    db,
                    ctx.tenant_no,
                    item['regionId'],
                    normalized['handler_user_id'],
                )
                if normalized['handler_user_id'] and should_validate_handler
                else None
            )
            row.handle_time = normalized['handle_time']
            row.handler_user_id = normalized['handler_user_id']
            row.insurance_attachments = normalized['attachments']
            row.handle_remark = normalized['remark']
            if 'handleTime' in fields:
                row.todo_status = normalized['make_or_todo_status']
                row.handle_date = normalized['legacy_date'] or None
        row.update_by = ctx.user_id
        row.update_time = now
        db.flush()
        db.commit()
        db.refresh(row)
    except Exception:
        db.rollback()
        raise
    data = _handle_summary(row, insurance=kind == 'insurance')
    data['handlerName'] = handler.nick_name if handler else ''
    return data


def update_storage_cert(db: Session, **kwargs) -> Dict[str, Any]:
    return _write_handle(db, kind='storage_cert', **kwargs)


def update_insurance(db: Session, **kwargs) -> Dict[str, Any]:
    return _write_handle(db, kind='insurance', **kwargs)