"""移动市场运费批次：按现场范围过滤明细并执行原子写入。""" 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