zkxw/backend/app/services/market_freight.py

"""移动市场运费批次：按现场范围过滤明细并执行原子写入。"""
from __future__ import annotations

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

from fastapi import HTTPException
from sqlalchemy import and_, exists, func, literal, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, aliased

from app.core.data_scope import apply_data_scope
from app.core.utils import generate_uuid
from app.models.models import (
    BizCellSampleType,
    BizCollectionProtocol,
    BizCollectionRegionStaff,
    BizProtocolCollectionInfo,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizSampleFreight,
    BizSampleFreightItem,
    SysDept,
    SysDictData,
    SysPost,
    SysPostRole,
    SysRole,
    SysRoleDataScopeDetail,
    SysUser,
    SysUserPost,
    SysUserRole,
)
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,
)
from app.services.sample_transport_mode import DICT_TYPE, resolve_transport_mode_value
from app.services.sf_express_settlement import (
    assert_freight_mutable,
    freight_amount_label,
    resolve_header_for_save,
)
from app.services.web_auth_roles import (
    _map_role_name_to_web,
    resolve_web_role_for_user,
)

CELL = 'cell_storage'
DONATION = 'research_donation'
SOURCES = (CELL, DONATION)


def _current_tenant_user_info(
    db: Session, ctx: MarketFieldContext, user_info: dict
) -> dict:
    """用当前租户有效角色覆盖不可信的令牌 role 快照。"""
    if not isinstance(user_info, dict):
        raise PermissionError('用户身份与现场范围上下文不一致')
    user_id = str(user_info.get('user_id') or '').strip()
    tenant_no = str(user_info.get('tenant_no') or '').strip()
    if user_id != ctx.user_id or tenant_no != ctx.tenant_no:
        raise PermissionError('用户身份与现场范围上下文不一致')
    current = dict(user_info)
    current['role'] = resolve_web_role_for_user(
        db, ctx.user_id, tenant_no=ctx.tenant_no
    )
    return current


def _latest_cell_party_id(ctx: MarketFieldContext):
    latest = aliased(BizProtocolPartyCustomer)
    return (
        select(latest.id)
        .where(
            latest.collection_protocol_id == BizCollectionProtocol.id,
            latest.tenant_no == ctx.tenant_no,
            latest.del_flag == 0,
        )
        .order_by(latest.create_time.desc(), latest.id.desc())
        .limit(1)
        .correlate(BizCollectionProtocol)
        .scalar_subquery()
    )


def _latest_donation_party_id(ctx: MarketFieldContext):
    latest = aliased(BizResearchDonationParty)
    return (
        select(latest.id)
        .where(
            latest.research_donation_id == BizResearchDonation.id,
            latest.tenant_no == ctx.tenant_no,
            latest.del_flag == 0,
        )
        .order_by(latest.create_time.desc(), latest.id.desc())
        .limit(1)
        .correlate(BizResearchDonation)
        .scalar_subquery()
    )


def _amount(value: Any) -> Decimal:
    try:
        result = Decimal(str(value))
    except (InvalidOperation, ValueError, TypeError) as exc:
        raise ValueError('运费格式无效') from exc
    if not result.is_finite() or result < 0:
        raise ValueError('运费格式无效')
    return result.quantize(Decimal('0.01'))


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


def _normalize_items(items: List[Dict[str, Any]]) -> List[Tuple[str, str]]:
    if not items:
        raise ValueError('至少选择一个样本')
    result: List[Tuple[str, str]] = []
    seen = set()
    for item in items:
        source = str(item.get('sampleSource') or item.get('sample_source') or '').strip()
        sample_id = str(item.get('sampleId') or item.get('sample_id') or '').strip()
        if source not in SOURCES or not sample_id:
            raise ValueError('关联样本须包含有效的 sampleSource 与 sampleId')
        key = (source, sample_id)
        if key in seen:
            raise ValueError('关联样本不可重复')
        seen.add(key)
        result.append(key)
    return result


def _cell_item_query(db: Session, ctx: MarketFieldContext, user_info: dict):
    query = (
        db.query(
            BizSampleFreightItem,
            BizProtocolSample,
            BizCollectionProtocol,
            BizProtocolPartyCustomer,
            BizCellSampleType,
        )
        .join(
            BizProtocolSample,
            and_(
                BizProtocolSample.id == BizSampleFreightItem.sample_id,
                BizProtocolSample.tenant_no == ctx.tenant_no,
                BizProtocolSample.del_flag == 0,
            ),
        )
        .join(
            BizCollectionProtocol,
            and_(
                BizCollectionProtocol.id == BizProtocolSample.collection_protocol_id,
                BizCollectionProtocol.tenant_no == ctx.tenant_no,
                BizCollectionProtocol.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolPartyCustomer,
            and_(
                BizProtocolPartyCustomer.id == _latest_cell_party_id(ctx),
                BizProtocolPartyCustomer.tenant_no == ctx.tenant_no,
                BizProtocolPartyCustomer.del_flag == 0,
            ),
        )
        .outerjoin(
            BizCellSampleType,
            and_(
                BizCellSampleType.id == BizProtocolSample.sample_type_id,
                BizCellSampleType.tenant_no == ctx.tenant_no,
                BizCellSampleType.del_flag == 0,
            ),
        )
        .filter(
            BizSampleFreightItem.tenant_no == ctx.tenant_no,
            BizSampleFreightItem.del_flag == 0,
            BizSampleFreightItem.sample_source == CELL,
        )
    )
    return apply_cell_protocol_scope(query, db, ctx, user_info=user_info)


def _donation_item_query(db: Session, ctx: MarketFieldContext, user_info: dict):
    query = (
        db.query(
            BizSampleFreightItem,
            BizResearchDonationSample,
            BizResearchDonation,
            BizResearchDonationParty,
            BizCellSampleType,
        )
        .join(
            BizResearchDonationSample,
            and_(
                BizResearchDonationSample.id == BizSampleFreightItem.sample_id,
                BizResearchDonationSample.tenant_no == ctx.tenant_no,
                BizResearchDonationSample.del_flag == 0,
            ),
        )
        .join(
            BizResearchDonation,
            and_(
                BizResearchDonation.id == BizResearchDonationSample.research_donation_id,
                BizResearchDonation.tenant_no == ctx.tenant_no,
                BizResearchDonation.del_flag == 0,
            ),
        )
        .outerjoin(
            BizResearchDonationParty,
            and_(
                BizResearchDonationParty.id == _latest_donation_party_id(ctx),
                BizResearchDonationParty.tenant_no == ctx.tenant_no,
                BizResearchDonationParty.del_flag == 0,
            ),
        )
        .outerjoin(
            BizCellSampleType,
            and_(
                BizCellSampleType.id == BizResearchDonationSample.sample_type_id,
                BizCellSampleType.tenant_no == ctx.tenant_no,
                BizCellSampleType.del_flag == 0,
            ),
        )
        .filter(
            BizSampleFreightItem.tenant_no == ctx.tenant_no,
            BizSampleFreightItem.del_flag == 0,
            BizSampleFreightItem.sample_source == DONATION,
        )
    )
    return apply_donation_scope(query, db, ctx, user_info=user_info)


def _visible_item_query(db: Session, source: str, ctx: MarketFieldContext, user_info: dict):
    return _cell_item_query(db, ctx, user_info) if source == CELL else _donation_item_query(db, ctx, user_info)


def _visible_rows(db: Session, freight_ids: List[str], ctx: MarketFieldContext, user_info: dict):
    rows = []
    if not freight_ids:
        return rows
    for source in SOURCES:
        rows.extend(
            _visible_item_query(db, source, ctx, user_info)
            .filter(BizSampleFreightItem.freight_id.in_(freight_ids))
            .order_by(BizSampleFreightItem.create_time.asc(), BizSampleFreightItem.id.asc())
            .all()
        )
    rows.sort(key=lambda row: (row[0].create_time or datetime.min, row[0].id))
    # 业务库历史上可能存在多条有效客户方信息；明细身份始终以关联行 ID 为准。
    unique = {}
    for row in rows:
        unique.setdefault(row[0].id, row)
    return list(unique.values())


def _item_dict(row) -> Dict[str, Any]:
    item, sample, parent, party, sample_type = row
    source = item.sample_source
    return {
        'id': item.id,
        'freightId': item.freight_id,
        'sampleSource': source,
        'sampleId': sample.id,
        'sampleCode': sample.sample_code or '',
        'protocolId': parent.id,
        'protocolNo': (parent.protocol_no if source == CELL else parent.donation_no) or '',
        'sampleTypeName': sample_type.type_name if sample_type else '',
        'motherName': party.mother_name if party else '',
        'newbornName': party.newborn_name if party else '',
    }


def _label_map(db: Session, tenant_no: str, values: List[str]) -> Dict[str, str]:
    wanted = list({value for value in values if value})
    if not wanted:
        return {}
    rows = db.query(SysDictData.dict_value, SysDictData.dict_label).filter(
        SysDictData.tenant_no == tenant_no,
        SysDictData.dict_type == DICT_TYPE,
        SysDictData.dict_value.in_(wanted),
        SysDictData.del_flag == 0,
    ).all()
    return {value: label or value for value, label in rows}


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


def _batch_dict(
    obj: BizSampleFreight, count: int, label: str, creator_name: str = ''
) -> Dict[str, Any]:
    amount = obj.freight_amount
    settle_time = getattr(obj, 'settle_time', None)
    return {
        'id': obj.id,
        'freightNo': obj.freight_no or '',
        'transportMode': obj.transport_mode or '',
        'transportModeLabel': label or obj.transport_mode or '',
        'freightAmount': None
        if amount is None
        else str(Decimal(amount).quantize(Decimal('0.01'))),
        'freightAmountLabel': freight_amount_label(obj.transport_mode, amount),
        'waybillNo': getattr(obj, 'waybill_no', None) or '',
        'settleStatus': getattr(obj, 'settle_status', None) or '',
        'settleTime': _fmt_dt(settle_time) if settle_time else '',
        'visibleItemCount': int(count),
        'remark': obj.remark or '',
        'createBy': obj.create_by or '',
        'createByName': creator_name,
        'createTime': _fmt_dt(obj.create_time),
        'updateTime': _fmt_dt(obj.update_time),
    }


def _freight_accessible(
    db: Session, freight_id: str, *, ctx: MarketFieldContext, user_info: dict
) -> Optional[BizSampleFreight]:
    """按 Web 运费单数据权限解析批次；不再要求现场可见明细。"""
    q = db.query(BizSampleFreight).filter(
        BizSampleFreight.id == freight_id,
        BizSampleFreight.tenant_no == ctx.tenant_no,
        BizSampleFreight.del_flag == 0,
    )
    return apply_data_scope(q, BizSampleFreight, user_info, db, ctx.tenant_no).first()


def list_freights(
    db: Session, *, ctx: MarketFieldContext, user_info: dict, page: int = 1,
    page_size: int = 10, keyword: Optional[str] = None,
    transport_mode: Optional[str] = None, sample_source: Optional[str] = None,
    include_items: bool = False,
    create_time_start: Optional[str] = None,
    create_time_end: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    """列表与 Web 一致：租户 + 通用数据权限；明细仍按现场范围脱敏。"""
    user_info = _current_tenant_user_info(db, ctx, user_info)
    page, page_size = max(1, int(page)), max(1, min(100, int(page_size)))
    source = (sample_source or '').strip()
    if source and source not in SOURCES:
        raise ValueError('sampleSource 无效')
    q = db.query(BizSampleFreight).filter(
        BizSampleFreight.tenant_no == ctx.tenant_no,
        BizSampleFreight.del_flag == 0,
    )
    q = apply_data_scope(q, BizSampleFreight, user_info, db, ctx.tenant_no)

    def date_value(raw: Optional[str], *, end: bool = False) -> Optional[datetime]:
        text = str(raw or '').strip()
        if not text:
            return None
        try:
            parsed = datetime.strptime(text, '%Y-%m-%d')
        except ValueError as exc:
            raise ValueError('日期格式无效，应为 YYYY-MM-DD') from exc
        return parsed + timedelta(days=1) if end else parsed

    created_from = date_value(create_time_start)
    created_until = date_value(create_time_end, end=True)
    if created_from:
        q = q.filter(BizSampleFreight.create_time >= created_from)
    if created_until:
        q = q.filter(BizSampleFreight.create_time < created_until)
    if source:
        item_exists = exists().where(
            and_(
                BizSampleFreightItem.freight_id == BizSampleFreight.id,
                BizSampleFreightItem.tenant_no == ctx.tenant_no,
                BizSampleFreightItem.del_flag == 0,
                BizSampleFreightItem.sample_source == source,
            )
        )
        q = q.filter(item_exists)
    if transport_mode:
        q = q.filter(BizSampleFreight.transport_mode == transport_mode.strip())
    text = (keyword or '').strip()
    if text:
        like = f'%{text}%'
        label_values = [
            value
            for value, in db.query(SysDictData.dict_value)
            .filter(
                SysDictData.dict_type == DICT_TYPE,
                SysDictData.tenant_no == ctx.tenant_no,
                SysDictData.del_flag == 0,
                SysDictData.dict_label.like(like),
            )
            .all()
        ]
        sample_code_exists = exists().where(
            and_(
                BizSampleFreightItem.freight_id == BizSampleFreight.id,
                BizSampleFreightItem.tenant_no == ctx.tenant_no,
                BizSampleFreightItem.del_flag == 0,
                BizSampleFreightItem.sample_code.like(like),
            )
        )
        conds = [
            BizSampleFreight.freight_no.like(like),
            BizSampleFreight.transport_mode.like(like),
            sample_code_exists,
        ]
        if label_values:
            conds.append(BizSampleFreight.transport_mode.in_(label_values))
        q = q.filter(or_(*conds))
    total = q.count()
    batches = q.order_by(BizSampleFreight.create_time.desc(), BizSampleFreight.id.desc()).offset((page - 1) * page_size).limit(page_size).all()
    ids = [row.id for row in batches]
    visible_rows = _visible_rows(db, ids, ctx, user_info)
    active_counts = {
        freight_id: int(count)
        for freight_id, count in db.query(
            BizSampleFreightItem.freight_id,
            func.count(BizSampleFreightItem.id),
        ).filter(
            BizSampleFreightItem.freight_id.in_(ids or ['']),
            BizSampleFreightItem.tenant_no == ctx.tenant_no,
            BizSampleFreightItem.del_flag == 0,
        ).group_by(BizSampleFreightItem.freight_id).all()
    }
    by_batch: Dict[str, list] = {}
    for row in visible_rows:
        by_batch.setdefault(row[0].freight_id, []).append(row)
    labels = _label_map(db, ctx.tenant_no, [row.transport_mode for row in batches])
    creator_names = _creator_name_map(
        db, ctx.tenant_no, [row.create_by for row in batches]
    )
    result = []
    for batch in batches:
        rows = by_batch.get(batch.id, [])
        data = _batch_dict(
            batch,
            len(rows),
            labels.get(batch.transport_mode or '', ''),
            creator_names.get(batch.create_by or '', ''),
        )
        has_outside_scope_item = active_counts.get(batch.id, 0) > len(rows)
        data['allowedActions'] = (
            ['link', 'unlink']
            if has_outside_scope_item
            else ['edit', 'delete', 'link', 'unlink']
        )
        if include_items:
            data['items'] = [_item_dict(row) for row in rows]
        result.append(data)
    return result, total


def get_freight_detail(db: Session, freight_id: str, *, ctx: MarketFieldContext, user_info: dict) -> Dict[str, Any]:
    user_info = _current_tenant_user_info(db, ctx, user_info)
    obj = _freight_accessible(db, freight_id, ctx=ctx, user_info=user_info)
    if not obj:
        raise HTTPException(status_code=404, detail='数据不存在或无权访问')
    rows = _visible_rows(db, [freight_id], ctx, user_info)
    labels = _label_map(db, ctx.tenant_no, [obj.transport_mode])
    creator_names = _creator_name_map(db, ctx.tenant_no, [obj.create_by])
    data = _batch_dict(
        obj,
        len(rows),
        labels.get(obj.transport_mode or '', ''),
        creator_names.get(obj.create_by or '', ''),
    )
    data['items'] = [_item_dict(row) for row in rows]
    visible_item_ids = {row[0].id for row in rows}
    has_outside_scope_item = (
        db.query(BizSampleFreightItem.id)
        .filter(
            BizSampleFreightItem.freight_id == freight_id,
            BizSampleFreightItem.tenant_no == ctx.tenant_no,
            BizSampleFreightItem.del_flag == 0,
            *([BizSampleFreightItem.id.notin_(visible_item_ids)] if visible_item_ids else []),
        )
        .first()
        is not None
    )
    data['allowedActions'] = (
        ['link', 'unlink']
        if has_outside_scope_item or not rows
        else ['edit', 'delete', 'link', 'unlink']
    )
    return data


def _option_query(db: Session, source: str, ctx: MarketFieldContext, user_info: dict, keyword: str):
    if source == CELL:
        q = db.query(
            literal(CELL).label('source'), BizProtocolSample.id.label('sample_id'),
            BizProtocolSample.sample_code.label('sample_code'), BizCollectionProtocol.id.label('protocol_id'),
            BizCollectionProtocol.protocol_no.label('protocol_no'), BizProtocolPartyCustomer.mother_name.label('customer_name'),
            BizProtocolPartyCustomer.newborn_name.label('newborn_name'),
            BizCellSampleType.type_name.label('sample_type_name'),
            BizProtocolSample.create_time.label('create_time'),
        ).join(BizCollectionProtocol, and_(BizCollectionProtocol.id == BizProtocolSample.collection_protocol_id, BizCollectionProtocol.tenant_no == ctx.tenant_no, BizCollectionProtocol.del_flag == 0)).outerjoin(BizProtocolPartyCustomer, and_(BizProtocolPartyCustomer.id == _latest_cell_party_id(ctx), BizProtocolPartyCustomer.tenant_no == ctx.tenant_no, BizProtocolPartyCustomer.del_flag == 0)).outerjoin(BizCellSampleType, and_(BizCellSampleType.id == BizProtocolSample.sample_type_id, BizCellSampleType.tenant_no == ctx.tenant_no, BizCellSampleType.del_flag == 0)).filter(BizProtocolSample.tenant_no == ctx.tenant_no, BizProtocolSample.del_flag == 0)
        q = apply_cell_protocol_scope(q, db, ctx, user_info=user_info)
        if keyword:
            like = f'%{keyword}%'
            q = q.filter(or_(BizProtocolSample.sample_code.like(like), BizCollectionProtocol.protocol_no.like(like), BizProtocolPartyCustomer.mother_name.like(like)))
        return q.distinct()
    q = db.query(
        literal(DONATION).label('source'), BizResearchDonationSample.id.label('sample_id'),
        BizResearchDonationSample.sample_code.label('sample_code'), BizResearchDonation.id.label('protocol_id'),
        BizResearchDonation.donation_no.label('protocol_no'), BizResearchDonationParty.mother_name.label('customer_name'),
        BizResearchDonationParty.newborn_name.label('newborn_name'),
        BizCellSampleType.type_name.label('sample_type_name'),
        BizResearchDonationSample.create_time.label('create_time'),
    ).join(BizResearchDonation, and_(BizResearchDonation.id == BizResearchDonationSample.research_donation_id, BizResearchDonation.tenant_no == ctx.tenant_no, BizResearchDonation.del_flag == 0)).outerjoin(BizResearchDonationParty, and_(BizResearchDonationParty.id == _latest_donation_party_id(ctx), BizResearchDonationParty.tenant_no == ctx.tenant_no, BizResearchDonationParty.del_flag == 0)).outerjoin(BizCellSampleType, and_(BizCellSampleType.id == BizResearchDonationSample.sample_type_id, BizCellSampleType.tenant_no == ctx.tenant_no, BizCellSampleType.del_flag == 0)).filter(BizResearchDonationSample.tenant_no == ctx.tenant_no, BizResearchDonationSample.del_flag == 0)
    q = apply_donation_scope(q, db, ctx, user_info=user_info)
    if keyword:
        like = f'%{keyword}%'
        q = q.filter(or_(BizResearchDonationSample.sample_code.like(like), BizResearchDonation.donation_no.like(like), BizResearchDonationParty.mother_name.like(like)))
    return q.distinct()


def list_sample_options(db: Session, *, ctx: MarketFieldContext, user_info: dict, page: int = 1, page_size: int = 20, keyword: Optional[str] = None, sample_source: Optional[str] = None):
    user_info = _current_tenant_user_info(db, ctx, user_info)
    page, page_size = max(1, int(page)), max(1, min(100, int(page_size)))
    source = (sample_source or '').strip()
    if source and source not in SOURCES:
        raise ValueError('sampleSource 无效')
    sources = (source,) if source else SOURCES
    queries = [_option_query(db, src, ctx, user_info, (keyword or '').strip()) for src in sources]
    combined = queries[0] if len(queries) == 1 else queries[0].union_all(*queries[1:])
    sub = combined.subquery()
    total = db.query(func.count()).select_from(sub).scalar() or 0
    rows = db.query(sub).order_by(sub.c.create_time.desc(), sub.c.source.desc(), sub.c.sample_id.desc()).offset((page - 1) * page_size).limit(page_size).all()
    keys = [(row.source, row.sample_id) for row in rows]
    links = {}
    if keys:
        clauses = [and_(BizSampleFreightItem.sample_source == src, BizSampleFreightItem.sample_id == sid) for src, sid in keys]
        for item, freight_no in db.query(
            BizSampleFreightItem, BizSampleFreight.freight_no
        ).join(
            BizSampleFreight,
            and_(
                BizSampleFreight.id == BizSampleFreightItem.freight_id,
                BizSampleFreight.tenant_no == ctx.tenant_no,
                BizSampleFreight.del_flag == 0,
            ),
        ).filter(
            BizSampleFreightItem.tenant_no == ctx.tenant_no,
            BizSampleFreightItem.del_flag == 0,
            or_(*clauses),
        ).all():
            links[(item.sample_source, item.sample_id)] = (
                item.freight_id, freight_no or ''
            )
    return [{
        'sampleSource': row.source, 'sampleId': row.sample_id,
        'sampleCode': row.sample_code or '', 'protocolId': row.protocol_id,
        'protocolNo': row.protocol_no or '', 'customerName': row.customer_name or '',
        'newbornName': row.newborn_name or '',
        'sampleTypeName': row.sample_type_name or '',
        'createTime': _fmt_dt(row.create_time),
        'freightId': (links.get((row.source, row.sample_id)) or ('', ''))[0],
        'freightNo': (links.get((row.source, row.sample_id)) or ('', ''))[1],
    } for row in rows], int(total)


def _sample_parent_lock_query(
    db: Session,
    source: str,
    sample_ids: List[str],
    ctx: MarketFieldContext,
):
    ids = sorted(set(sample_ids))
    if source == CELL:
        query = (
            db.query(BizProtocolSample, BizCollectionProtocol)
            .join(
                BizCollectionProtocol,
                and_(
                    BizCollectionProtocol.id
                    == BizProtocolSample.collection_protocol_id,
                    BizCollectionProtocol.tenant_no == ctx.tenant_no,
                    BizCollectionProtocol.del_flag == 0,
                ),
            )
            .filter(
                BizProtocolSample.id.in_(ids),
                BizProtocolSample.tenant_no == ctx.tenant_no,
                BizProtocolSample.del_flag == 0,
            )
        )
        return query.order_by(
            BizProtocolSample.id.asc(), BizCollectionProtocol.id.asc()
        ).populate_existing().with_for_update()
    query = (
        db.query(BizResearchDonationSample, BizResearchDonation)
        .join(
            BizResearchDonation,
            and_(
                BizResearchDonation.id
                == BizResearchDonationSample.research_donation_id,
                BizResearchDonation.tenant_no == ctx.tenant_no,
                BizResearchDonation.del_flag == 0,
            ),
        )
        .filter(
            BizResearchDonationSample.id.in_(ids),
            BizResearchDonationSample.tenant_no == ctx.tenant_no,
            BizResearchDonationSample.del_flag == 0,
        )
    )
    return query.order_by(
        BizResearchDonationSample.id.asc(), BizResearchDonation.id.asc()
    ).populate_existing().with_for_update()


def _collection_lock_query(
    db: Session,
    source: str,
    parent_ids: List[str],
    ctx: MarketFieldContext,
):
    ids = sorted(set(parent_ids))
    if source == CELL:
        return (
            db.query(BizProtocolCollectionInfo)
            .filter(
                BizProtocolCollectionInfo.collection_protocol_id.in_(ids),
                BizProtocolCollectionInfo.tenant_no == ctx.tenant_no,
                BizProtocolCollectionInfo.del_flag == 0,
            )
            .order_by(
                BizProtocolCollectionInfo.collection_protocol_id.asc(),
                BizProtocolCollectionInfo.id.asc(),
            )
            .populate_existing()
            .with_for_update()
        )
    return (
        db.query(BizResearchDonationCollection)
        .filter(
            BizResearchDonationCollection.research_donation_id.in_(ids),
            BizResearchDonationCollection.tenant_no == ctx.tenant_no,
            BizResearchDonationCollection.del_flag == 0,
        )
        .order_by(
            BizResearchDonationCollection.research_donation_id.asc(),
            BizResearchDonationCollection.id.asc(),
        )
        .populate_existing()
        .with_for_update()
    )


def _leader_lock_query(db: Session, ctx: MarketFieldContext):
    """锁定经理当前有效的地区负责人关系，供字段范围直接计算。"""
    return (
        db.query(BizCollectionRegionStaff)
        .filter(
            BizCollectionRegionStaff.tenant_no == ctx.tenant_no,
            BizCollectionRegionStaff.del_flag == 0,
            BizCollectionRegionStaff.user_id == ctx.user_id,
            BizCollectionRegionStaff.is_leader == 1,
        )
        .order_by(
            BizCollectionRegionStaff.region_id.asc(),
            BizCollectionRegionStaff.id.asc(),
        )
        .populate_existing()
        .with_for_update()
    )


def _current_field_visible_parent_ids(
    source: str,
    collection_rows: List[Any],
    leader_rows: List[BizCollectionRegionStaff],
    ctx: MarketFieldContext,
) -> set[str]:
    """只使用已加锁的采集分配/负责人当前值计算现场字段范围。"""
    parent_attr = (
        'collection_protocol_id'
        if source == CELL
        else 'research_donation_id'
    )
    leader_regions = {row.region_id for row in leader_rows}
    visible: set[str] = set()
    for row in collection_rows:
        if ctx.scope_kind == 'specialist':
            allowed = row.sales_user_id == ctx.user_id
        else:
            allowed = (
                row.region_id in leader_regions
                or row.region_manager_user_id == ctx.user_id
            )
        if allowed:
            visible.add(getattr(row, parent_attr))
    return visible


def _current_data_scope_parent_ids(
    db: Session,
    parents: List[Any],
    ctx: MarketFieldContext,
    user_info: dict,
) -> set[str]:
    """锁定数据权限依赖并按锁后当前值计算多角色并集范围。"""
    if not isinstance(user_info, dict) or str(
        user_info.get('user_id') or ''
    ).strip() != ctx.user_id:
        raise PermissionError('用户身份与现场范围上下文不一致')
    parent_ids = {parent.id for parent in parents}

    current_users = (
        db.query(SysUser)
        .filter(
            SysUser.id == ctx.user_id,
            SysUser.tenant_no == ctx.tenant_no,
            SysUser.del_flag == 0,
        )
        .order_by(SysUser.id.asc())
        .populate_existing()
        .with_for_update()
        .all()
    )
    if not current_users:
        return set()
    current_user = current_users[0]

    direct_links = (
        db.query(SysUserRole)
        .filter(
            SysUserRole.user_id == ctx.user_id,
            SysUserRole.tenant_no == ctx.tenant_no,
            SysUserRole.del_flag == 0,
        )
        .order_by(SysUserRole.role_id.asc(), SysUserRole.id.asc())
        .populate_existing()
        .with_for_update()
        .all()
    )
    user_posts = (
        db.query(SysUserPost)
        .filter(
            SysUserPost.user_id == ctx.user_id,
            SysUserPost.tenant_no == ctx.tenant_no,
            SysUserPost.del_flag == 0,
        )
        .order_by(SysUserPost.post_id.asc(), SysUserPost.id.asc())
        .populate_existing()
        .with_for_update()
        .all()
    )
    post_ids = sorted({row.post_id for row in user_posts})
    posts = []
    if post_ids:
        posts = (
            db.query(SysPost)
            .filter(
                SysPost.id.in_(post_ids),
                SysPost.tenant_no == ctx.tenant_no,
                SysPost.del_flag == 0,
                SysPost.status == '1',
            )
            .order_by(SysPost.id.asc())
            .populate_existing()
            .with_for_update()
            .all()
        )
    enabled_post_ids = sorted({row.id for row in posts})
    post_role_links = []
    if enabled_post_ids:
        post_role_links = (
            db.query(SysPostRole)
            .filter(
                SysPostRole.post_id.in_(enabled_post_ids),
                SysPostRole.tenant_no == ctx.tenant_no,
                SysPostRole.del_flag == 0,
            )
            .order_by(
                SysPostRole.post_id.asc(),
                SysPostRole.role_id.asc(),
                SysPostRole.id.asc(),
            )
            .populate_existing()
            .with_for_update()
            .all()
        )
    role_ids = sorted(
        {row.role_id for row in direct_links}
        | {row.role_id for row in post_role_links}
    )
    roles = []
    if role_ids:
        roles = (
            db.query(SysRole)
            .filter(
                SysRole.id.in_(role_ids),
                SysRole.tenant_no == ctx.tenant_no,
                SysRole.del_flag == 0,
                SysRole.status == '1',
            )
            .order_by(SysRole.id.asc())
            .populate_existing()
            .with_for_update()
            .all()
        )
    if not roles:
        return set()
    if any(
        _map_role_name_to_web(role.role_name) == 'super_admin'
        for role in roles
    ):
        return parent_ids
    if any((role.data_scope or '1') == '1' for role in roles):
        return parent_ids

    active_role_ids = sorted({role.id for role in roles})
    details = (
        db.query(SysRoleDataScopeDetail)
        .filter(
            SysRoleDataScopeDetail.role_id.in_(active_role_ids),
            SysRoleDataScopeDetail.tenant_no == ctx.tenant_no,
            SysRoleDataScopeDetail.del_flag == 0,
        )
        .order_by(
            SysRoleDataScopeDetail.role_id.asc(),
            SysRoleDataScopeDetail.scope_type.asc(),
            SysRoleDataScopeDetail.scope_value.asc(),
            SysRoleDataScopeDetail.id.asc(),
        )
        .populate_existing()
        .with_for_update()
        .all()
    )
    details_by_role: Dict[str, List[SysRoleDataScopeDetail]] = {}
    for detail in details:
        details_by_role.setdefault(detail.role_id, []).append(detail)

    dept_ids: set[str] = set()
    user_ids: set[str] = set()
    current_dept_id = (current_user.dept_id or '').strip()
    needs_descendants = any(
        (role.data_scope or '1') == '3' for role in roles
    )
    dept_rows = []
    if needs_descendants and current_dept_id:
        dept_rows = (
            db.query(SysDept)
            .filter(
                SysDept.tenant_no == ctx.tenant_no,
                SysDept.del_flag == 0,
            )
            .order_by(SysDept.id.asc())
            .populate_existing()
            .with_for_update()
            .all()
        )
    descendants = {current_dept_id} if current_dept_id else set()
    if dept_rows and current_dept_id:
        pending = [current_dept_id]
        while pending:
            parent_id = pending.pop(0)
            for dept in dept_rows:
                if dept.parent_id == parent_id and dept.id not in descendants:
                    descendants.add(dept.id)
                    pending.append(dept.id)

    for role in roles:
        scope = role.data_scope or '1'
        if scope == '2' and current_dept_id:
            dept_ids.add(current_dept_id)
        elif scope == '3':
            dept_ids.update(descendants)
        elif scope == '4':
            user_ids.add(ctx.user_id)
        elif scope == '5':
            for detail in details_by_role.get(role.id, []):
                value = (detail.scope_value or '').strip()
                if not value:
                    continue
                if detail.scope_type == 'dept':
                    dept_ids.add(value)
                elif detail.scope_type == 'user':
                    user_ids.add(value)

    return {
        parent.id
        for parent in parents
        if (
            (parent.create_dept_id in dept_ids if dept_ids else False)
            or (parent.create_by in user_ids if user_ids else False)
        )
    }


def _active_item_lock_query(
    db: Session,
    ctx: MarketFieldContext,
    *,
    freight_id: Optional[str] = None,
    keys: Optional[List[Tuple[str, str]]] = None,
):
    conditions = []
    if freight_id:
        conditions.append(BizSampleFreightItem.freight_id == freight_id)
    for source, sample_id in sorted(set(keys or [])):
        conditions.append(
            and_(
                BizSampleFreightItem.sample_source == source,
                BizSampleFreightItem.sample_id == sample_id,
            )
        )
    query = db.query(BizSampleFreightItem).filter(
        BizSampleFreightItem.tenant_no == ctx.tenant_no,
        BizSampleFreightItem.del_flag == 0,
    )
    if conditions:
        query = query.filter(or_(*conditions))
    else:
        query = query.filter(False)
    return query.order_by(
        BizSampleFreightItem.sample_source.asc(),
        BizSampleFreightItem.sample_id.asc(),
        BizSampleFreightItem.id.asc(),
    ).populate_existing().with_for_update()


def _lock_batch_header(
    db: Session, freight_id: str, ctx: MarketFieldContext
) -> BizSampleFreight:
    obj = (
        db.query(BizSampleFreight)
        .filter(
            BizSampleFreight.id == freight_id,
            BizSampleFreight.tenant_no == ctx.tenant_no,
            BizSampleFreight.del_flag == 0,
        )
        .populate_existing()
        .with_for_update()
        .first()
    )
    if not obj:
        raise HTTPException(status_code=404, detail='数据不存在或无权访问')
    return obj


def _discover_batch_keys(
    db: Session, freight_id: str, ctx: MarketFieldContext
) -> List[Tuple[str, str]]:
    """批次头已锁后发现参与者；最终关联锁会再次验证该快照。"""
    rows = (
        db.query(
            BizSampleFreightItem.sample_source,
            BizSampleFreightItem.sample_id,
        )
        .filter(
            BizSampleFreightItem.freight_id == freight_id,
            BizSampleFreightItem.tenant_no == ctx.tenant_no,
            BizSampleFreightItem.del_flag == 0,
        )
        .order_by(
            BizSampleFreightItem.sample_source.asc(),
            BizSampleFreightItem.sample_id.asc(),
            BizSampleFreightItem.id.asc(),
        )
        .all()
    )
    return [(source, sample_id) for source, sample_id in rows]


def _lock_sample_graph(
    db: Session,
    keys: List[Tuple[str, str]],
    ctx: MarketFieldContext,
    user_info: dict,
) -> Dict[Tuple[str, str], Tuple[Any, Any]]:
    """按来源和 ID 锁样本/父单/采集分配，再用锁定当前读重算范围。"""
    normalized = sorted(set(keys))
    locked_rows: Dict[str, List[Tuple[Any, Any]]] = {}
    locked_collections: Dict[str, List[Any]] = {}
    for source in SOURCES:
        ids = [sample_id for item_source, sample_id in normalized if item_source == source]
        if not ids:
            continue
        rows = _sample_parent_lock_query(db, source, ids, ctx).all()
        locked_rows[source] = rows
        parent_ids = [parent.id for _, parent in rows]
        locked_collections[source] = _collection_lock_query(
            db, source, parent_ids, ctx
        ).all()

    leader_rows = (
        _leader_lock_query(db, ctx).all()
        if ctx.scope_kind == 'manager'
        else []
    )
    all_parents = [
        parent
        for rows in locked_rows.values()
        for _, parent in rows
    ]
    data_scope_ids = _current_data_scope_parent_ids(
        db, all_parents, ctx, user_info
    )
    visible: Dict[Tuple[str, str], Tuple[Any, Any]] = {}
    for source, rows in locked_rows.items():
        field_scope_ids = _current_field_visible_parent_ids(
            source, locked_collections[source], leader_rows, ctx
        )
        for sample, parent in rows:
            if parent.id in field_scope_ids and parent.id in data_scope_ids:
                visible[(source, sample.id)] = (sample, parent)
    return visible


def _lock_active_items(
    db: Session,
    ctx: MarketFieldContext,
    *,
    freight_id: Optional[str] = None,
    keys: Optional[List[Tuple[str, str]]] = None,
    expected_batch_keys: Optional[List[Tuple[str, str]]] = None,
) -> List[BizSampleFreightItem]:
    rows = _active_item_lock_query(
        db, ctx, freight_id=freight_id, keys=keys
    ).all()
    if freight_id is not None and expected_batch_keys is not None:
        current = sorted(
            (row.sample_source, row.sample_id)
            for row in rows
            if row.freight_id == freight_id
        )
        if current != sorted(expected_batch_keys):
            raise PermissionError('批次明细已发生变化，请重试')
    return rows


def _next_freight_no(db: Session, tenant_no: str, now: datetime, offset: int = 0) -> str:
    prefix = f"YF{now.strftime('%Y%m%d')}"
    latest = db.query(BizSampleFreight.freight_no).filter(BizSampleFreight.tenant_no == tenant_no, BizSampleFreight.freight_no.like(f'{prefix}%')).order_by(BizSampleFreight.freight_no.desc()).with_for_update().first()
    seq = int(str(latest[0])[-3:]) + 1 if latest and str(latest[0])[-3:].isdigit() else 1
    seq += offset
    if seq > 999:
        raise ValueError('当日运费批次号已达上限（999）')
    return f'{prefix}{seq:03d}'


def create_freight(db: Session, *, ctx: MarketFieldContext, user_info: dict, transport_mode: str, freight_amount: Any, items: List[Dict[str, Any]], waybill_no: Any = None):
    keys = _normalize_items(items)
    try:
        samples = _lock_sample_graph(db, keys, ctx, user_info)
        if len(samples) != len(set(keys)):
            raise HTTPException(status_code=404, detail='样本不存在或无权操作')
        conflicts = _lock_active_items(db, ctx, keys=keys)
        if conflicts:
            raise ValueError('样本已关联其他运费批次')
        mode = resolve_transport_mode_value(db, tenant_no=ctx.tenant_no, raw=transport_mode, user_id=ctx.user_id)
        plan = resolve_header_for_save(
            transport_mode=mode,
            existing_waybill=None,
            existing_amount=None,
            existing_settle_status=None,
            waybill_supplied=True,
            waybill_no=waybill_no,
            amount_supplied=True,
            freight_amount=freight_amount,
        )
        now = datetime.now()
        obj = None
        for attempt in range(3):
            candidate = BizSampleFreight(id=generate_uuid(), freight_no=_next_freight_no(db, ctx.tenant_no, now, attempt), transport_mode=plan.transport_mode, freight_amount=plan.freight_amount, waybill_no=plan.waybill_no, settle_status=plan.settle_status, settle_time=None, tenant_no=ctx.tenant_no, order_num=0, remark=None, create_by=ctx.user_id, create_dept_id=str(user_info.get('dept_id') or user_info.get('region_id') or ''), create_time=now, update_by=ctx.user_id, update_time=now, del_flag=0)
            try:
                with db.begin_nested():
                    db.add(candidate)
                    db.flush()
                obj = candidate
                break
            except IntegrityError:
                if attempt == 2:
                    raise ValueError('运费批次号生成冲突，请重试')
        assert obj is not None
        for source, sample_id in keys:
            sample, parent = samples[(source, sample_id)]
            db.add(BizSampleFreightItem(id=generate_uuid(), freight_id=obj.id, sample_source=source, sample_id=sample.id, sample_code=sample.sample_code, protocol_id=parent.id, tenant_no=ctx.tenant_no, order_num=0, remark=None, create_by=ctx.user_id, create_dept_id=str(user_info.get('dept_id') or user_info.get('region_id') or ''), create_time=now, update_by=ctx.user_id, update_time=now, del_flag=0))
        db.flush()
        db.commit()
        return get_freight_detail(db, obj.id, ctx=ctx, user_info=user_info)
    except IntegrityError as exc:
        db.rollback()
        raise ValueError('样本已关联运费批次，不允许移动') from exc
    except Exception:
        db.rollback()
        raise


def update_freight(db: Session, freight_id: str, *, ctx: MarketFieldContext, user_info: dict, transport_mode: Optional[str] = None, freight_amount: Any = None, waybill_no: Any = None, waybill_supplied: bool = False, amount_supplied: bool = False):
    try:
        obj = _lock_batch_header(db, freight_id, ctx)
        expected_keys = _discover_batch_keys(db, freight_id, ctx)
        visible = _lock_sample_graph(db, expected_keys, ctx, user_info)
        active = _lock_active_items(
            db,
            ctx,
            freight_id=freight_id,
            expected_batch_keys=expected_keys,
        )
        if not active or not visible:
            raise HTTPException(status_code=404, detail='数据不存在或无权访问')
        if any(
            (row.sample_source, row.sample_id) not in visible for row in active
        ):
            raise PermissionError('仅当批次全部有效样本均在当前范围内时可修改批次信息')
        mode = obj.transport_mode or ''
        if transport_mode is not None and str(transport_mode).strip() != '':
            mode = resolve_transport_mode_value(db, tenant_no=ctx.tenant_no, raw=str(transport_mode), user_id=ctx.user_id)
        plan = resolve_header_for_save(
            transport_mode=mode,
            existing_waybill=obj.waybill_no,
            existing_amount=obj.freight_amount,
            existing_settle_status=obj.settle_status,
            waybill_supplied=waybill_supplied,
            waybill_no=waybill_no,
            amount_supplied=amount_supplied,
            freight_amount=freight_amount,
        )
        obj.transport_mode = plan.transport_mode
        obj.waybill_no = plan.waybill_no
        obj.freight_amount = plan.freight_amount
        obj.settle_status = plan.settle_status
        if plan.clear_settle_time:
            obj.settle_time = None
        obj.update_by, obj.update_time = ctx.user_id, datetime.now()
        db.flush()
        db.commit()
        return get_freight_detail(db, freight_id, ctx=ctx, user_info=user_info)
    except Exception:
        db.rollback()
        raise


def assert_link_target_access(
    db: Session,
    freight_id: str,
    *,
    ctx: MarketFieldContext,
    user_info: dict,
) -> None:
    """确认目标批次在 Web 同款数据权限内可关联（不要求已有现场可见明细）。"""
    user_info = _current_tenant_user_info(db, ctx, user_info)
    if not _freight_accessible(db, freight_id, ctx=ctx, user_info=user_info):
        raise HTTPException(status_code=404, detail='数据不存在或无权访问')


def link_samples(db: Session, freight_id: str, *, ctx: MarketFieldContext, user_info: dict, items: List[Dict[str, Any]]):
    keys = _normalize_items(items)
    try:
        user_info = _current_tenant_user_info(db, ctx, user_info)
        if not _freight_accessible(db, freight_id, ctx=ctx, user_info=user_info):
            raise HTTPException(status_code=404, detail='数据不存在或无权访问')
        obj = _lock_batch_header(db, freight_id, ctx)
        assert_freight_mutable(obj.settle_status, 'items')
        expected_keys = _discover_batch_keys(db, freight_id, ctx)
        samples = _lock_sample_graph(
            db, sorted(set(expected_keys + keys)), ctx, user_info
        )
        locked_items = _lock_active_items(
            db,
            ctx,
            freight_id=freight_id,
            keys=keys,
            expected_batch_keys=expected_keys,
        )
        if any(key not in samples for key in keys):
            raise HTTPException(status_code=404, detail='样本不存在或无权操作')
        requested = set(keys)
        conflicts = [
            row
            for row in locked_items
            if (row.sample_source, row.sample_id) in requested
        ]
        if conflicts:
            raise ValueError('样本已关联运费批次，不允许移动')
        now = datetime.now()
        for source, sample_id in keys:
            sample, parent = samples[(source, sample_id)]
            db.add(BizSampleFreightItem(id=generate_uuid(), freight_id=obj.id, sample_source=source, sample_id=sample.id, sample_code=sample.sample_code, protocol_id=parent.id, tenant_no=ctx.tenant_no, order_num=0, remark=None, create_by=ctx.user_id, create_dept_id=str(user_info.get('dept_id') or user_info.get('region_id') or ''), create_time=now, update_by=ctx.user_id, update_time=now, del_flag=0))
        db.flush()
        db.commit()
        return get_freight_detail(db, freight_id, ctx=ctx, user_info=user_info)
    except IntegrityError as exc:
        db.rollback()
        raise ValueError('样本已关联运费批次，不允许移动') from exc
    except Exception:
        db.rollback()
        raise


def unlink_samples(db: Session, freight_id: str, *, ctx: MarketFieldContext, user_info: dict, items: List[Dict[str, Any]]):
    keys = _normalize_items(items)
    try:
        obj = _lock_batch_header(db, freight_id, ctx)
        assert_freight_mutable(obj.settle_status, 'items')
        expected_keys = _discover_batch_keys(db, freight_id, ctx)
        visible = _lock_sample_graph(db, expected_keys, ctx, user_info)
        active = _lock_active_items(
            db,
            ctx,
            freight_id=freight_id,
            expected_batch_keys=expected_keys,
        )
        if not any(
            (row.sample_source, row.sample_id) in visible for row in active
        ):
            raise HTTPException(status_code=404, detail='数据不存在或无权访问')
        requested = set(keys)
        matched = [
            row
            for row in active
            if (row.sample_source, row.sample_id) in visible
            and (row.sample_source, row.sample_id) in requested
        ]
        if len(matched) != len(keys):
            raise ValueError('关联明细不存在或无权操作')
        now = datetime.now()
        for row in matched:
            row.del_flag, row.update_by, row.update_time = 1, ctx.user_id, now
        db.flush()
        matched_ids = {row.id for row in matched}
        remaining = [row for row in active if row.id not in matched_ids]
        remaining_visible = [
            row
            for row in remaining
            if (row.sample_source, row.sample_id) in visible
        ]
        if not remaining_visible:
            if not remaining:
                obj.del_flag, obj.update_by, obj.update_time = 1, ctx.user_id, now
            db.commit()
            return {'id': freight_id, 'visibleItemCount': 0, 'disappeared': True}
        db.commit()
        return get_freight_detail(db, freight_id, ctx=ctx, user_info=user_info)
    except Exception:
        db.rollback()
        raise


def delete_freight(db: Session, freight_id: str, *, ctx: MarketFieldContext, user_info: dict):
    """级联软删运费单；仅当全部有效关联样本均在当前范围内时允许。"""
    try:
        obj = _lock_batch_header(db, freight_id, ctx)
        assert_freight_mutable(obj.settle_status, 'delete')
        expected_keys = _discover_batch_keys(db, freight_id, ctx)
        visible = _lock_sample_graph(db, expected_keys, ctx, user_info)
        active = _lock_active_items(
            db,
            ctx,
            freight_id=freight_id,
            expected_batch_keys=expected_keys,
        )
        if not active or not visible:
            raise HTTPException(status_code=404, detail='数据不存在或无权访问')
        if any(
            (row.sample_source, row.sample_id) not in visible for row in active
        ):
            raise PermissionError('仅当批次全部有效样本均在当前范围内时可删除运费单')
        now = datetime.now()
        for row in active:
            row.del_flag, row.update_by, row.update_time = 1, ctx.user_id, now
        obj.del_flag, obj.update_by, obj.update_time = 1, ctx.user_id, now
        db.flush()
        db.commit()
        return {'id': freight_id, 'visibleItemCount': 0, 'disappeared': True}
    except Exception:
        db.rollback()
        raise