From cde30287fd494244fba276a17dbff5493a0d9c0f Mon Sep 17 00:00:00 2001 From: wuzilong <704541015@qq.com> Date: Sun, 1 Nov 2026 20:55:59 +0800 Subject: [PATCH] =?UTF-8?q?feat(pl):=20=E6=8C=89=E8=A3=85=E9=80=81?= =?UTF-8?q?=E6=AE=B5=E5=88=86=E5=88=AB=E8=AE=A1=E9=87=8F=E5=8F=8C=E8=AF=81?= =?UTF-8?q?=E9=A2=9D=E5=BA=A6=E5=B9=B6=E4=BF=9D=E7=95=99=E5=8E=86=E5=8F=B2?= =?UTF-8?q?=E5=8D=95=E4=BE=A7=E5=8F=A3=E5=BE=84=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/pl/transport.py | 2 + backend/app/schemas/pl/quota_planning.py | 2 + backend/app/services/pl_plan_service.py | 2 + backend/app/services/pl_quota_business.py | 2 + backend/app/services/pl_quota_planning.py | 60 ++++++++-- backend/app/services/pl_quota_sources.py | 108 ++++++++++-------- .../app/services/pl_quota_usage_service.py | 13 ++- backend/tests/test_pl_quota_business.py | 71 +++++++++++- backend/tests/test_pl_quota_sources.py | 52 +++++++++ 9 files changed, 247 insertions(+), 65 deletions(-) diff --git a/backend/app/api/pl/transport.py b/backend/app/api/pl/transport.py index 509a7f658..b6f1bb105 100644 --- a/backend/app/api/pl/transport.py +++ b/backend/app/api/pl/transport.py @@ -885,6 +885,8 @@ def arrange_transport( selected_escort_id=transport.escort_id, ) apply_resource_snapshot_json(plan=plan, transport=transport, snapshot=snapshot) + from app.services.pl_quota_planning import pin_measurement_version + pin_measurement_version(db, user, plan) plan.transport_status = "arranged" plan.follow_status = "arranged" plan.freight_amount = freight_amount or 0 diff --git a/backend/app/schemas/pl/quota_planning.py b/backend/app/schemas/pl/quota_planning.py index 964bca0ec..b27a23f05 100644 --- a/backend/app/schemas/pl/quota_planning.py +++ b/backend/app/schemas/pl/quota_planning.py @@ -31,6 +31,8 @@ class CertCondition(InputBody): class PlanQuotaConfiguration(InputBody): + # 可回传服务端值；不作为版本选择依据。 + measurement_version: Optional[Literal['legacy_single_side', 'per_stop_v2']] = None source_mapping: Optional[List[SourcePair]] = None purchase_cert_ids: List[str] = Field(default_factory=list) transport_cert_ids: List[str] = Field(default_factory=list) diff --git a/backend/app/services/pl_plan_service.py b/backend/app/services/pl_plan_service.py index eb353c484..1ee2f4f49 100644 --- a/backend/app/services/pl_plan_service.py +++ b/backend/app/services/pl_plan_service.py @@ -1165,6 +1165,8 @@ def create_transport_for_plan( driver_id=driver_id, escort_id=escort_id, ) + from app.services.pl_quota_planning import pin_measurement_version + pin_measurement_version(db, user, plan) plan.transport_status = "arranged" plan.follow_status = "arranged" plan.freight_amount = freight_amount or 0 diff --git a/backend/app/services/pl_quota_business.py b/backend/app/services/pl_quota_business.py index 118b437f2..0854c8b46 100644 --- a/backend/app/services/pl_quota_business.py +++ b/backend/app/services/pl_quota_business.py @@ -128,6 +128,8 @@ def apply_plan_quota(db, user, plan, action, payload=None): [('transport', cid) for cid in config.transport_cert_ids])} validate_conditions(config, certs, plan_fingerprint(request.sources)) plan.cert_configuration = config.model_dump(mode='json', by_alias=True) + plan.cert_configuration = {**(plan.cert_configuration or {}), + 'measurementVersion': config.measurement_version} return result except ValueError as exc: raise HTTPException(409, str(exc)) from exc diff --git a/backend/app/services/pl_quota_planning.py b/backend/app/services/pl_quota_planning.py index 8379fdbde..d6ca420f1 100644 --- a/backend/app/services/pl_quota_planning.py +++ b/backend/app/services/pl_quota_planning.py @@ -45,16 +45,38 @@ def _plan(db, user, plan_id, lock=False): return row +def resolve_measurement_version(db, user, plan): + """只信任持久化的服务端配置与账本，历史无版本记录保持原口径。""" + usages = db.query(PlPlanCertUsage).filter_by(tenant_no=user.tenant_no, plan_id=plan.id, del_flag=0).all() + if usages: + return ('per_stop_v2' if all((u.confirmation_snapshot or {}).get('measurementVersion') == 'per_stop_v2' + for u in usages) else 'legacy_single_side') + saved = (plan.cert_configuration or {}).get('measurementVersion') + if saved in ('per_stop_v2', 'legacy_single_side'): + return saved + if plan.transport_status == 'arranged' or plan.archive_status in ('archived', 'exception'): + return 'legacy_single_side' + return 'per_stop_v2' + + +def pin_measurement_version(db, user, plan): + """安排改变状态前固化版本，与业务及额度在同一事务保存。""" + version = resolve_measurement_version(db, user, plan) + plan.cert_configuration = {**(plan.cert_configuration or {}), 'measurementVersion': version} + return version + + def build_usage_request(db, user, plan_id, operation, action, *, lock=False, check_conditions=True, preview_vehicle=None): """调用者先flush已授权的计划/实际量变更，再组装；不能将客户端UsageRequest直接传账本。""" op = operation if isinstance(operation, PlanQuotaOperation) else PlanQuotaOperation.model_validate(operation) plan = _plan(db, user, plan_id, lock) config = op.configuration or PlanQuotaConfiguration.model_validate(plan.cert_configuration or {}) + config = config.model_copy(update={"measurement_version": resolve_measurement_version(db, user, plan)}) mapping = None if config.source_mapping is None else [r.model_dump(by_alias=True) for r in config.source_mapping] if action == 'release': sources, actuals = [], {} else: - sources, actuals = extract_sources(db, user, plan_id, mapping, actual=action in ('settle', 'actual_adjust'), lock=lock, preview_vehicle=preview_vehicle) + sources, actuals = extract_sources(db, user, plan_id, mapping, actual=action in ('settle', 'actual_adjust'), lock=lock, preview_vehicle=preview_vehicle, measurement_version=config.measurement_version) keys = [('purchase', cid) for cid in config.purchase_cert_ids] + [('transport', cid) for cid in config.transport_cert_ids] if len(set(keys)) != len(keys): raise HTTPException(400, '证件选择重复') @@ -82,8 +104,9 @@ def configuration_context(db, user, plan_id, config, preview_vehicle=None): config = config if isinstance(config, PlanQuotaConfiguration) else PlanQuotaConfiguration.model_validate(config) mapping = None if config.source_mapping is None else [r.model_dump(by_alias=True) for r in config.source_mapping] with db.no_autoflush: - sources, _ = extract_sources(db, user, plan_id, mapping, preview_vehicle=preview_vehicle) - return {'sourceFingerprint': plan_fingerprint(sources), + config = config.model_copy(update={"measurement_version": resolve_measurement_version(db, user, _plan(db, user, plan_id))}) + sources, _ = extract_sources(db, user, plan_id, mapping, preview_vehicle=preview_vehicle, measurement_version=config.measurement_version) + return {'measurementVersion': config.measurement_version, 'sourceFingerprint': plan_fingerprint(sources), 'sources': [s.model_dump(mode='json') for s in sources]} @@ -91,7 +114,7 @@ def preview_plan(db, user, plan_id, config, preview_vehicle=None): from app.services.pl_quota_usage_service import preview_usage config = config if isinstance(config, PlanQuotaConfiguration) else PlanQuotaConfiguration.model_validate(config) with db.no_autoflush: - request, _ = build_usage_request(db, user, plan_id, PlanQuotaOperation(operationId='preview', configuration=config), + request, config = build_usage_request(db, user, plan_id, PlanQuotaOperation(operationId='preview', configuration=config), 'resize', check_conditions=False, preview_vehicle=preview_vehicle) result = preview_usage(db, user, request) fingerprint = plan_fingerprint(request.sources) @@ -102,7 +125,7 @@ def preview_plan(db, user, plan_id, config, preview_vehicle=None): validate_conditions(config, certs, fingerprint) except HTTPException as error: reasons.append(error.detail) - return {**result, 'sourceFingerprint': fingerprint, + return {**result, 'measurementVersion': config.measurement_version, 'sourceFingerprint': fingerprint, 'sources': [s.model_dump(mode='json') for s in request.sources], 'status': 'needs_confirmation' if reasons else ('available' if request.sources else 'not_required'), 'reasons': reasons} @@ -115,8 +138,9 @@ def save_configuration(db, user, plan_id, config): if plan.transport_status == 'arranged' or plan.archive_status == 'archived': raise HTTPException(409, '已安排计划请通过调整运输在同一事务重算用证') # 草稿允许缺少条件确认，但真实来源身份及选证权限必须有效。 + config = config.model_copy(update={"measurement_version": resolve_measurement_version(db, user, plan)}) mapping = None if config.source_mapping is None else [r.model_dump(by_alias=True) for r in config.source_mapping] - extract_sources(db, user, plan_id, mapping, lock=True) + extract_sources(db, user, plan_id, mapping, lock=True, measurement_version=config.measurement_version) for kind, ids in [('purchase', config.purchase_cert_ids), ('transport', config.transport_cert_ids)]: for cid in sorted(ids): _get_row(db, user, kind, cid) @@ -132,18 +156,34 @@ def save_configuration(db, user, plan_id, config): def get_plan_quota(db, user, plan_id): with db.no_autoflush: plan, products, rows, stops = source_context(db, user, plan_id) + from app.services.pl_quota_contract_service import _contract + contract_numbers = {} + for cid in {r.contract_id for values in rows.values() for r in values.values() if r.contract_id}: + try: + contract_numbers[cid] = _contract(db, user, cid).contract_no + except HTTPException as exc: + if exc.status_code not in (403, 404): + raise + contract_numbers[cid] = None options = [] for side, values in rows.items(): for row in values.values(): if products[row.plan_product_id].is_toxic == 1: - options.append({'sourceStopType': side, 'sourceStopProductId': row.id, + stop = stops[side][getattr(row, 'load_id' if side == 'load' else 'delivery_id')] + options.append({'sourceStopId': stop.id, 'stopId': stop.id, + 'stopOrder': getattr(stop, 'seq', None) or stop.order_num or 1, + 'productId': row.product_id, 'productNo': row.product_no, + 'sourceStopType': side, 'sourceStopProductId': row.id, 'planProductId': row.plan_product_id, 'productName': row.product_name, 'model': row.model, 'quantity': row.plan_qty, 'unit': row.unit, - 'contractId': row.contract_id, 'contractProductId': row.contract_product_id, - 'partyName': getattr(stops[side][getattr(row, 'load_id' if side == 'load' else 'delivery_id')], 'customer_name', None)}) + 'contractId': row.contract_id, 'contractNo': contract_numbers.get(row.contract_id), + 'contractProductId': row.contract_product_id, + 'partyName': stop.customer_name}) usages = db.query(PlPlanCertUsage).filter_by(tenant_no=user.tenant_no, plan_id=plan_id, del_flag=0).order_by(PlPlanCertUsage.create_time.desc()).all() + version = resolve_measurement_version(db, user, plan) result = {'planId': plan.id, 'planNo': plan.plan_no, 'licensePlate': plan.license_plate, - 'configuration': plan.cert_configuration or {}, 'sourceOptions': options, + 'measurementVersion': version, + 'configuration': {**(plan.cert_configuration or {}), 'measurementVersion': version}, 'sourceOptions': options, 'canSaveConfiguration': plan.transport_status != 'arranged' and plan.archive_status != 'archived', 'usages': []} for usage in usages: diff --git a/backend/app/services/pl_quota_sources.py b/backend/app/services/pl_quota_sources.py index a0a34602f..3644c1793 100644 --- a/backend/app/services/pl_quota_sources.py +++ b/backend/app/services/pl_quota_sources.py @@ -1,4 +1,4 @@ -"""从当前计划段行提取唯一用证计量侧，绝不按同 SKU 猜测装送配对。""" +"""从真实装送段提取用证来源；保留历史单侧口径。""" import hashlib import json from decimal import Decimal @@ -52,64 +52,72 @@ def _source_context(db, user, plan_id, lock=False): return plan, products, rows, stops -def extract_sources(db, user, plan_id, mapping=None, *, actual=False, lock=False, preview_vehicle=None): +def extract_sources(db, user, plan_id, mapping=None, *, actual=False, lock=False, preview_vehicle=None, measurement_version="legacy_single_side"): """mapping 仅选择真实段行及对侧追溯ID；数量/单位/合同/指纹全部由数据库生成。 - 单侧来源可直接确定；同一计划产品有装送两侧时，必须逐项显式声明计量侧与配对。 - 另一侧可关联多条计量行（例如一装多送），但不再次累计其数量。 + per_stop_v2 自动提取每条装送段行，忽略旧客户端的单侧选择。 + legacy_single_side 保留显式配对和唯一计量侧，避免历史账本静默重计。 """ plan, products, rows, stops = source_context(db, user, plan_id, lock) toxic = {pid for pid, row in products.items() if row.is_toxic == 1} + if measurement_version not in ("legacy_single_side", "per_stop_v2"): + raise HTTPException(400, "未知用证计量版本") if not toxic: - if mapping: + if mapping and measurement_version == "legacy_single_side": raise HTTPException(400, '该计划无需本次两证计量配置') return [], {} groups = {pid: {side: [r for r in values.values() if r.plan_product_id == pid] for side, values in rows.items()} for pid in toxic} - if mapping is None: - mapping = [] - for group in groups.values(): - available = [side for side, values in group.items() if values] - if len(available) != 1: - raise HTTPException(409, '请明确装送产品配对及唯一用证计量侧') - side = available[0] - mapping.extend({'sourceStopType': side, 'sourceStopProductId': r.id, - 'relatedStopProductIds': []} for r in group[side]) - chosen, seen, measured_sides = [], set(), {} - covered_opposite = {pid: set() for pid in toxic} - for entry in mapping: - side, rid = entry.get('sourceStopType'), entry.get('sourceStopProductId') - if side not in SIDE or (side, rid) in seen: - raise HTTPException(400, '计量来源无效或重复') - seen.add((side, rid)) - row = rows[side].get(rid) - if row is None or row.plan_product_id not in toxic: - raise HTTPException(400, '计量来源不属于当前计划易制毒产品') - pid = row.plan_product_id - if pid in measured_sides and measured_sides[pid] != side: - raise HTTPException(400, '同一物理产品不能同时按装送两侧计量') - measured_sides[pid] = side - opposite = 'delivery' if side == 'load' else 'load' - ids = entry.get('relatedStopProductIds', []) - if not isinstance(ids, list) or len(ids) != len(set(ids)): - raise HTTPException(400, '对侧来源关联无效或重复') - related = [] - for other_id in ids: - other = rows[opposite].get(other_id) - if other is None or other.plan_product_id != pid: - raise HTTPException(400, '对侧产品不属于当前计划产品明细') - related.append(other) - if groups[pid][opposite] and not related: - raise HTTPException(400, '存在对侧装送产品，请明确配对') - covered_opposite[pid].update(ids) - chosen.append((side, row, sorted(related, key=lambda r: r.id))) - for pid, group in groups.items(): - side = measured_sides.get(pid) - if side is None or {r.id for r in group[side]} != {r.id for s, r, _ in chosen if s == side and r.plan_product_id == pid}: - raise HTTPException(400, '用证计量来源未覆盖全部计划产品段行') - other_side = 'delivery' if side == 'load' else 'load' - if covered_opposite[pid] != {r.id for r in group[other_side]}: - raise HTTPException(400, '装送来源配对未覆盖全部对侧段行') + if measurement_version == 'per_stop_v2': + chosen = [(side, row, []) for side, values in rows.items() + for row in values.values() if row.plan_product_id in toxic] + if {row.plan_product_id for _, row, _ in chosen} != toxic: + raise HTTPException(409, '用证计量来源未覆盖全部计划产品段行') + else: + if mapping is None: + mapping = [] + for group in groups.values(): + available = [side for side, values in group.items() if values] + if len(available) != 1: + raise HTTPException(409, '请明确装送产品配对及唯一用证计量侧') + side = available[0] + mapping.extend({'sourceStopType': side, 'sourceStopProductId': r.id, + 'relatedStopProductIds': []} for r in group[side]) + chosen, seen, measured_sides = [], set(), {} + covered_opposite = {pid: set() for pid in toxic} + for entry in mapping: + side, rid = entry.get('sourceStopType'), entry.get('sourceStopProductId') + if side not in SIDE or (side, rid) in seen: + raise HTTPException(400, '计量来源无效或重复') + seen.add((side, rid)) + row = rows[side].get(rid) + if row is None or row.plan_product_id not in toxic: + raise HTTPException(400, '计量来源不属于当前计划易制毒产品') + pid = row.plan_product_id + if pid in measured_sides and measured_sides[pid] != side: + raise HTTPException(400, '同一物理产品不能同时按装送两侧计量') + measured_sides[pid] = side + opposite = 'delivery' if side == 'load' else 'load' + ids = entry.get('relatedStopProductIds', []) + if not isinstance(ids, list) or len(ids) != len(set(ids)): + raise HTTPException(400, '对侧来源关联无效或重复') + related = [] + for other_id in ids: + other = rows[opposite].get(other_id) + if other is None or other.plan_product_id != pid: + raise HTTPException(400, '对侧产品不属于当前计划产品明细') + related.append(other) + if groups[pid][opposite] and not related: + raise HTTPException(400, '存在对侧装送产品，请明确配对') + covered_opposite[pid].update(ids) + chosen.append((side, row, sorted(related, key=lambda r: r.id))) + for pid, group in groups.items(): + side = measured_sides.get(pid) + if side is None or {r.id for r in group[side]} != {r.id for s, r, _ in chosen if s == side and r.plan_product_id == pid}: + raise HTTPException(400, '用证计量来源未覆盖全部计划产品段行') + other_side = 'delivery' if side == 'load' else 'load' + if covered_opposite[pid] != {r.id for r in group[other_side]}: + raise HTTPException(400, '装送来源配对未覆盖全部对侧段行') sources, actuals = [], {} for side, row, related in sorted(chosen, key=lambda v: (v[0], v[1].id)): if row.plan_qty is None or row.plan_qty < 0 or not row.unit or not row.contract_id or not row.contract_product_id: @@ -123,6 +131,8 @@ def extract_sources(db, user, plan_id, mapping=None, *, actual=False, lock=False stop_facts.append({key: getattr(parent, key, None) for key in ('id', 'supplier_id', 'supplier_name', 'customer_id', 'customer_name', 'address')}) snapshot = {'source': _facts(row), 'related': [_facts(r) for r in related], 'stops': stop_facts, 'vehicle': plan.license_plate, 'trailer': plan.trailer_no, 'planType': plan.plan_type} + if measurement_version == 'per_stop_v2': + snapshot['measurementVersion'] = measurement_version if preview_vehicle is not None: snapshot['vehicle'] = preview_vehicle.get('licensePlate', plan.license_plate) snapshot['trailer'] = preview_vehicle.get('trailerNo', plan.trailer_no) diff --git a/backend/app/services/pl_quota_usage_service.py b/backend/app/services/pl_quota_usage_service.py index a6fa307aa..7f5ff38c3 100644 --- a/backend/app/services/pl_quota_usage_service.py +++ b/backend/app/services/pl_quota_usage_service.py @@ -156,6 +156,9 @@ def _execute(db,user,r,digest,plan): else: supplement=_settle(db,user,r,certs,usages,items,plan) for k,u in usages.items(): + u.confirmation_snapshot = {**(u.confirmation_snapshot or {}), + 'measurementVersion': (u.confirmation_snapshot or {}).get('measurementVersion', + next((s.snapshot.get('measurementVersion', 'legacy_single_side') for s in r.sources), 'legacy_single_side'))} u.row_version=(u.row_version or 0)+1; u.update_by=user.id if certs[k].reserved_qty<0 or certs[k].used_qty<0 or certs[k].reserved_qty+certs[k].used_qty>certs[k].total_qty: raise ValueError('余额越界') @@ -253,13 +256,14 @@ def _reserve(db,user,r,plan,certs,usages,items,selected,preview=False,lock_previ new_qty=sum((a.permit_quantity for a in allocations),ZERO) c.reserved_qty+=new_qty-u.reserved_qty; u.reserved_qty=new_qty; u.usage_status='reserved' if allocations else 'released' u.cert_revision=c.revision; u.confirmed_by=user.id; u.confirmed_time=datetime.now() - u.confirmation_snapshot={'executed':False,'vehicleNo':r.vehicle_no} + u.confirmation_snapshot={'executed':False,'vehicleNo':r.vehicle_no, + 'measurementVersion': next((s.snapshot.get('measurementVersion', 'legacy_single_side') for s in r.sources), 'legacy_single_side')} for a in allocations: s=sources[a.source_id]; link,scope=scopes[(k,a.source_id)] item=previous.get(a.source_id) if item is None: item=m.PlPlanCertUsageItem(**_audit(user),usage_id=u.id); db.add(item); items[k].append(item) - values=dict(contract_cert_link_id=link.id,contract_cert_product_id=scope.id,contract_scope_revision=link.scope_revision,contract_id=s.contract_id,contract_product_id=s.contract_product_id,plan_product_id=s.plan_product_id,source_stop_type=s.source_stop_type,source_stop_id=s.source_stop_id,source_stop_product_id=s.source_stop_product_id,related_stop_products=s.related_stop_products,allocation_seq=a.allocation_seq,source_unit=s.unit,planned_source_qty=s.quantity,allocated_source_qty=a.source_quantity,planned_cert_qty=a.permit_quantity,reserved_qty=a.permit_quantity,used_qty=ZERO,item_status='reserved',source_fingerprint=s.fingerprint,source_snapshot={**s.snapshot, 'productId': s.product_id},del_flag=0,reserve_input_mode='same_unit' if s.unit==c.quantity_unit else 'manual',reserve_conversion_note=notes.get((k[0],k[1],a.source_id)),reserve_confirmed_by=user.id,reserve_confirmed_time=datetime.now(),update_by=user.id) + values=dict(contract_cert_link_id=link.id,contract_cert_product_id=scope.id,contract_scope_revision=link.scope_revision,contract_id=s.contract_id,contract_product_id=s.contract_product_id,plan_product_id=s.plan_product_id,source_stop_type=s.source_stop_type,source_stop_id=s.source_stop_id,source_stop_product_id=s.source_stop_product_id,related_stop_products=s.related_stop_products,allocation_seq=a.allocation_seq,source_unit=s.unit,planned_source_qty=s.quantity,allocated_source_qty=a.source_quantity,planned_cert_qty=a.permit_quantity,reserved_qty=a.permit_quantity,used_qty=ZERO,item_status='reserved',source_fingerprint=s.fingerprint,source_snapshot={**s.snapshot, 'measurementVersion': s.snapshot.get('measurementVersion', 'legacy_single_side'), 'productId': s.product_id},del_flag=0,reserve_input_mode='same_unit' if s.unit==c.quantity_unit else 'manual',reserve_conversion_note=notes.get((k[0],k[1],a.source_id)),reserve_confirmed_by=user.id,reserve_confirmed_time=datetime.now(),update_by=user.id) for name,value in values.items(): setattr(item,name,value) @@ -425,7 +429,8 @@ def _supplement_items(db,user,r,plan,certs,usages,items,active,item_map,suppleme if usage is None: usage=m.PlPlanCertUsage(**_audit(user),plan_id=plan.id,plan_no_snapshot=plan.plan_no, cert_type=kind,cert_id=pid,cert_no_snapshot=cert.cert_no,cert_revision=cert.revision, - quantity_unit=cert.quantity_unit,reserved_qty=ZERO,used_qty=ZERO,del_flag=0) + quantity_unit=cert.quantity_unit,reserved_qty=ZERO,used_qty=ZERO,del_flag=0, + confirmation_snapshot={'measurementVersion': source.snapshot.get('measurementVersion', 'legacy_single_side')}) db.add(usage); usages[k]=usage; items[k]=[] # 保留曾移除行的稳定ID（唯一键包括软删除行）。 item=next((i for i in items[k] if _key(i)==sid),None) @@ -439,6 +444,6 @@ def _supplement_items(db,user,r,plan,certs,usages,items,active,item_map,suppleme allocation_seq=max((i.allocation_seq for rows in active.values() for i in rows),default=0)+1, source_unit=source.unit,planned_source_qty=source.quantity,allocated_source_qty=ZERO,planned_cert_qty=ZERO, reserved_qty=ZERO,used_qty=ZERO,item_status='settled',source_fingerprint=source.fingerprint, - source_snapshot={**source.snapshot,'productId':source.product_id},del_flag=0) + source_snapshot={**source.snapshot,'measurementVersion':source.snapshot.get('measurementVersion', 'legacy_single_side'),'productId':source.product_id},del_flag=0) for name,value in values.items(): setattr(item,name,value) active.setdefault(k,[]).append(item); item_map[key]=item diff --git a/backend/tests/test_pl_quota_business.py b/backend/tests/test_pl_quota_business.py index 44da35b23..f34cb3d25 100644 --- a/backend/tests/test_pl_quota_business.py +++ b/backend/tests/test_pl_quota_business.py @@ -58,7 +58,7 @@ def business(case, monkeypatch): case.db.add(m.PlPlanLoadProduct(id='sp',load_id='load',plan_id='plan',plan_product_id='pp',product_id='product',product_no='SKU',product_name='酸',plan_qty=29,unit='吨',contract_id='c',contract_product_id='cp',tenant_no='t',inbound_qty=0)) case.db.commit() monkeypatch.setattr(pl_quota_planning,'plan_execution_lock_reason', lambda *a: None) - sources,_=extract_sources(case.db,case.user,'plan') + sources,_=extract_sources(case.db,case.user,'plan', measurement_version='per_stop_v2') fingerprint=plan_fingerprint(sources) payload=TransportArrange(vehicleId='v',operationId='business',quotaConfiguration={'purchaseCertIds':['p'],'transportCertIds':['t'],'conditions':[ {'certType':kind,'certId':cid,'certRevision':1,'sourceFingerprint':fingerprint,'partiesConfirmed':True,'routeConfirmed':True,'note':'已核对'} @@ -349,7 +349,7 @@ def test_single_plan_arrange_cannot_change_shared_transport_or_either_quota(appr c.db.add(c.models.PlPlanLoadProduct(id='spb',load_id='loadb',plan_id='other',plan_product_id='ppb',product_id='product',product_no='SKU',product_name='酸',plan_qty=29,unit='吨',contract_id='c',contract_product_id='cp',tenant_no='t')) c.db.flush() payload_b=c.payload.model_copy(deep=True);payload_b.operation_id='reserve-b' - fingerprint=plan_fingerprint(extract_sources(c.db,c.user,'other')[0]) + fingerprint=plan_fingerprint(extract_sources(c.db,c.user,'other', measurement_version='per_stop_v2')[0]) for condition in payload_b.quota_configuration.conditions: condition.source_fingerprint=fingerprint apply_plan_quota(c.db,c.user,c.plan,'reserve',c.payload) apply_plan_quota(c.db,c.user,other,'reserve',payload_b) @@ -370,3 +370,70 @@ def test_single_plan_arrange_cannot_change_shared_transport_or_either_quota(appr assert error.value.status_code == 409 assert '共享运输' in error.value.detail assert snapshot() == before + + +def test_per_stop_reserves_both_certificates_and_settles_each_actual(business): + from app.services.pl_quota_planning import configuration_context + from app.services.pl_quota_business import apply_plan_quota, settle_plan_quota + c = business + c.db.get(c.models.PlPlanLoadProduct, 'sp').plan_qty = 100 + c.db.get(c.models.PlPlanLoadProduct, 'sp').inbound_qty = 90 + for index, qty in [(1, 40), (2, 60)]: + c.db.add(c.models.PlPlanDelivery(id=f'd{index}', plan_id='plan', tenant_no='t', seq=index)) + c.db.add(c.models.PlPlanDeliveryProduct(id=f'dp{index}', delivery_id=f'd{index}', plan_id='plan', plan_product_id='pp', product_id='product', product_no='SKU', product_name='酸', plan_qty=qty, outbound_qty=qty-5, unit='吨', contract_id='c', contract_product_id='cp', tenant_no='t')) + for model, cid in [(c.m.PlPurchaseCert, 'p'), (c.m.PlTransportCert, 't')]: + c.db.get(model, cid).total_qty = 300 + c.db.flush() + config = c.payload.quota_configuration + fingerprint = configuration_context(c.db, c.user, 'plan', config)['sourceFingerprint'] + for condition in config.conditions: condition.source_fingerprint = fingerprint + apply_plan_quota(c.db, c.user, c.plan, 'reserve', c.payload) + assert c.plan.cert_configuration['measurementVersion'] == 'per_stop_v2' + assert all(u.confirmation_snapshot['measurementVersion'] == 'per_stop_v2' for u in c.db.query(c.m.PlPlanCertUsage)) + assert all(i.source_snapshot['measurementVersion'] == 'per_stop_v2' for i in c.db.query(c.m.PlPlanCertUsageItem)) + for model, cid in [(c.m.PlPurchaseCert, 'p'), (c.m.PlTransportCert, 't')]: + assert c.db.get(model, cid).reserved_qty == 200 + assert settle_plan_quota(c.db, c.user, c.plan, c.payload) + for model, cid in [(c.m.PlPurchaseCert, 'p'), (c.m.PlTransportCert, 't')]: + assert c.db.get(model, cid).used_qty == 180 + + +def test_arrange_pins_v2_before_state_changes(approval_case, monkeypatch): + from app.services import pl_plan_execution_lock, pl_contract_plan_qty + c = approval_case + c.models.PlPlanTransport.__table__.create(c.db.bind) + monkeypatch.setattr(pl_plan_execution_lock, 'plan_execution_lock_reason', lambda *a: None) + monkeypatch.setattr(pl_contract_plan_qty, 'auto_start_execute_approved_contracts', lambda *a, **kw: None) + c.models.PlPlanContract.__table__.create(c.db.bind) + # 检查正式安排路径抵达额度调用时，pending 计划的版本已经固定。 + def check_quota(db, user, plan, action, payload): + assert plan.transport_status == 'arranged' + assert plan.cert_configuration.get('measurementVersion') == 'per_stop_v2' + raise RuntimeError('checked quota boundary') + monkeypatch.setattr(c.api, 'apply_plan_quota', check_quota) + with pytest.raises(RuntimeError, match='checked quota boundary'): + c.api.arrange_transport('plan', c.payload, c.db, c.user) + + +def test_supplement_new_certificate_keeps_server_measurement_version(business): + from datetime import date + from test_pl_quota_usage import _second_pair + from app.services.pl_quota_business import apply_plan_quota, settle_plan_quota + from app.services.pl_quota_planning import configuration_context + from app.schemas.pl.quota_planning import CertCondition + c = business + apply_plan_quota(c.db, c.user, c.plan, 'reserve', c.payload) + _second_pair(c) + for model, cid in [(c.m.PlPurchaseCert, 'p'), (c.m.PlTransportCert, 't')]: + c.db.get(model, cid).total_qty = 100 + c.db.get(model, cid).valid_to = date(2000, 1, 1) + c.db.get(c.models.PlPlanLoadProduct, 'sp').inbound_qty = 35 + c.db.flush() + config = c.payload.quota_configuration.model_copy(deep=True) + config.purchase_cert_ids = ['p2']; config.transport_cert_ids = ['t2'] + fingerprint = configuration_context(c.db, c.user, 'plan', config)['sourceFingerprint'] + config.conditions = [CertCondition(certType=kind, certId=cid, certRevision=1, sourceFingerprint=fingerprint, partiesConfirmed=True, routeConfirmed=True, note='核对补证') for kind, cid in [('purchase', 'p2'), ('transport', 't2')]] + payload = SimpleNamespace(operation_id='supplement', quota_allow_supplement=True, quota_configuration=config, quota_reason='实际超量', quota_source_note='磅单') + assert settle_plan_quota(c.db, c.user, c.plan, payload) + assert all(u.confirmation_snapshot.get('measurementVersion') == 'per_stop_v2' for u in c.db.query(c.m.PlPlanCertUsage)) + assert configuration_context(c.db, c.user, 'plan', config)['measurementVersion'] == 'per_stop_v2' diff --git a/backend/tests/test_pl_quota_sources.py b/backend/tests/test_pl_quota_sources.py index 5fc222546..bff0b3642 100644 --- a/backend/tests/test_pl_quota_sources.py +++ b/backend/tests/test_pl_quota_sources.py @@ -122,6 +122,8 @@ def test_planning_context_and_options_return_readonly_authoritative_facts(case): c = case for model in (PlPlanCertUsage, PlPlanCertUsageItem): model.__table__.create(c.db.bind) + from app.models.crm_models import CrmContract + CrmContract.__table__.create(c.db.bind) info = planning.get_plan_quota(c.db, c.user, 'plan') assert len(info['sourceOptions']) == 2 context = planning.configuration_context(c.db, c.user, 'plan', {'sourceMapping': mapping()}, @@ -155,3 +157,53 @@ def test_overview_filters_real_usage_states_ignores_old_ready_and_other_tenant(c assert list_plans(c.db, other)['total'] == 0 c.db.get(c.m.PlPlanProduct, 'pp').is_toxic = 0; c.db.flush() assert list_plans(c.db, c.user)['list'][0]['quotaStatus'] == 'not_required' + + +def test_per_stop_counts_all_sides_ignoring_old_client_selection(case): + c = case + c.db.get(c.m.PlPlanLoadProduct, 'lp').plan_qty = 100 + c.db.get(c.m.PlPlanDeliveryProduct, 'dp').plan_qty = 40 + c.db.add(c.m.PlPlanDelivery(id='delivery2', plan_id='plan', tenant_no='t', seq=2)) + c.db.add(c.m.PlPlanDeliveryProduct(id='dp2', delivery_id='delivery2', plan_id='plan', + plan_product_id='pp', product_id='sku', product_no='SKU', product_name='硫酸', + plan_qty=60, contract_id='sale2', contract_product_id='cs2', tenant_no='t')) + c.db.flush() + sources, _ = c.svc.extract_sources(c.db, c.user, 'plan', mapping(), measurement_version='per_stop_v2') + assert sum(s.quantity for s in sources) == 200 + assert {(s.contract_id, s.contract_product_id) for s in sources} == {('purchase', 'cp'), ('sale', 'cs'), ('sale2', 'cs2')} + assert all(s.snapshot['measurementVersion'] == 'per_stop_v2' and not s.related_stop_products for s in sources) + + +@pytest.mark.parametrize('history', ['draft', 'arranged', 'archived', 'ledger']) +def test_server_version_cannot_be_overridden_by_client(case, history): + from app.models.pl_quota_models import PlPlanCertUsage + from app.services.pl_quota_planning import configuration_context + c = case + PlPlanCertUsage.__table__.create(c.db.bind) + plan = c.db.get(c.m.PlPlan, 'plan') + if history == 'arranged': plan.transport_status = 'arranged' + if history == 'archived': plan.archive_status = 'archived' + if history == 'ledger': + c.db.add(PlPlanCertUsage(id='u', tenant_no='t', plan_id='plan', plan_no_snapshot='PLAN', cert_type='purchase', cert_id='p', cert_no_snapshot='P', cert_revision=1, quantity_unit='吨')) + c.db.flush() + config = {'sourceMapping': mapping(), 'measurementVersion': 'legacy_single_side' if history == 'draft' else 'per_stop_v2'} + result = configuration_context(c.db, c.user, 'plan', config) + assert result['measurementVersion'] == ('per_stop_v2' if history == 'draft' else 'legacy_single_side') + assert sum(Decimal(s['quantity']) for s in result['sources']) == (60 if history == 'draft' else 30) + + +def test_options_include_stop_contract_and_party_without_foreign_contract_leak(case): + from app.models.crm_models import CrmContract + from app.models.pl_quota_models import PlPlanCertUsage, PlPlanCertUsageItem + from app.services.pl_quota_planning import get_plan_quota + c = case + for model in (CrmContract, PlPlanCertUsage, PlPlanCertUsageItem): model.__table__.create(c.db.bind) + c.db.add(CrmContract(id='purchase', contract_no='PUR-1', contract_name='购买合同', customer_id='customer', contract_type='purchase', tenant_no='t')) + c.db.add(CrmContract(id='sale', contract_no='SECRET', contract_name='其他租户', customer_id='customer', contract_type='sale', tenant_no='other')) + c.db.get(c.m.PlPlanLoad, 'load').customer_name = '装货方' + c.db.flush() + options = {o['sourceStopType']: o for o in get_plan_quota(c.db, c.user, 'plan')['sourceOptions']} + assert options['load']['contractNo'] == 'PUR-1' + assert options['load']['partyName'] == '装货方' + assert options['load']['stopId'] == 'load' and options['load']['stopOrder'] == 1 + assert options['delivery']['contractNo'] is None