diff --git a/backend/app/services/pl_quota_conversion.py b/backend/app/services/pl_quota_conversion.py index 3400febca..785dc8d3a 100644 --- a/backend/app/services/pl_quota_conversion.py +++ b/backend/app/services/pl_quota_conversion.py @@ -1,4 +1,6 @@ """计划段次是否需要把计划单位换算成证面单位。""" +from decimal import Decimal, InvalidOperation + from app.models.pl_models import ( PlPlan, PlPlanDelivery, PlPlanDeliveryProduct, PlPlanLoad, PlPlanLoadProduct, PlPlanProduct, ) @@ -63,12 +65,45 @@ def line_needs_conversion(unit, cert_units): ) -def plan_needs_unit_conversion(plan, lines, cert_units, snapshots): +def _positive_decimal(value): + try: + number = Decimal(str(value).strip()) + except (InvalidOperation, AttributeError, TypeError): + return False + return number > 0 + + +def _saved_manual_keys(configuration): + """已填写完整人工换算的 (sourceId, permitKind, permitId)。""" + rows = (configuration or {}).get('manualAllocations') or (configuration or {}).get('manual_allocations') or [] + keys = set() + for row in rows: + if not isinstance(row, dict): + continue + source_id = row.get('sourceId') or row.get('source_id') + kind = row.get('permitKind') or row.get('permit_kind') + permit_id = row.get('permitId') or row.get('permit_id') + source_qty = row.get('sourceQuantity') if 'sourceQuantity' in row else row.get('source_quantity') + permit_qty = row.get('permitQuantity') if 'permitQuantity' in row else row.get('permit_quantity') + if source_id and kind and permit_id and _positive_decimal(source_qty) and _positive_decimal(permit_qty): + keys.add((str(source_id), str(kind), str(permit_id))) + return keys + + +def plan_needs_unit_conversion(plan, lines, cert_bindings, snapshots): + """证书齐备豁免或异单位段已保存完整人工换算时，不再进入待换算待办。""" + if int(getattr(plan, 'quota_cert_waived', 0) or 0) == 1: + return False version = measurement_version(plan, snapshots) + saved = _saved_manual_keys(plan.cert_configuration) for line in measured_lines(version, lines, plan.cert_configuration): - units = cert_units.get(line['contractProductId'], []) - if line_needs_conversion(line['unit'], units): - return True + bindings = cert_bindings.get(line['contractProductId'], []) + source_id = f"{line['side']}:{line['id']}" + for binding in bindings: + if not line_needs_conversion(line['unit'], [binding['unit']]): + continue + if (source_id, binding['kind'], binding['certId']) not in saved: + return True return False @@ -104,12 +139,14 @@ def _stop_lines(db, tenant_no, plan_ids): return grouped -def _cert_units(db, tenant_no, contract_product_ids): - units = {} +def _cert_bindings(db, tenant_no, contract_product_ids): + bindings = {} if not contract_product_ids: - return units + return bindings for kind, cert_model in (('purchase', PlPurchaseCert), ('transport', PlTransportCert)): - rows = db.query(PlContractCertProduct.contract_product_id, cert_model.quantity_unit).join( + rows = db.query( + PlContractCertProduct.contract_product_id, cert_model.id, cert_model.quantity_unit, + ).join( PlContractCertLink, PlContractCertLink.id == PlContractCertProduct.contract_cert_link_id, ).join(cert_model, cert_model.id == PlContractCertLink.cert_id).filter( PlContractCertLink.tenant_no == tenant_no, PlContractCertLink.del_flag == 0, @@ -119,9 +156,11 @@ def _cert_units(db, tenant_no, contract_product_ids): cert_model.tenant_no == tenant_no, cert_model.del_flag == 0, PlContractCertProduct.contract_product_id.in_(contract_product_ids), ).all() - for product_id, unit in rows: - units.setdefault(product_id, []).append(unit) - return units + for product_id, cert_id, unit in rows: + bindings.setdefault(product_id, []).append({ + 'kind': kind, 'certId': cert_id, 'unit': unit, + }) + return bindings def plans_needing_unit_conversion(db, tenant_no, plan_ids): @@ -137,7 +176,7 @@ def plans_needing_unit_conversion(db, tenant_no, plan_ids): product_ids = { line['contractProductId'] for rows in lines.values() for line in rows if line['contractProductId'] } - cert_units = _cert_units(db, tenant_no, product_ids) + cert_bindings = _cert_bindings(db, tenant_no, product_ids) snapshots = {plan_id: [] for plan_id in ids} usage_rows = db.query(PlPlanCertUsage.plan_id, PlPlanCertUsage.confirmation_snapshot).filter( PlPlanCertUsage.tenant_no == tenant_no, PlPlanCertUsage.del_flag == 0, @@ -146,6 +185,6 @@ def plans_needing_unit_conversion(db, tenant_no, plan_ids): for plan_id, snapshot in usage_rows: snapshots[plan_id].append(snapshot) for plan in plans: - if plan_needs_unit_conversion(plan, lines.get(plan.id, []), cert_units, snapshots.get(plan.id, [])): + if plan_needs_unit_conversion(plan, lines.get(plan.id, []), cert_bindings, snapshots.get(plan.id, [])): needed.add(plan.id) return needed diff --git a/backend/tests/test_pl_quota_conversion.py b/backend/tests/test_pl_quota_conversion.py index 1c9846fa5..fcd5b991b 100644 --- a/backend/tests/test_pl_quota_conversion.py +++ b/backend/tests/test_pl_quota_conversion.py @@ -23,47 +23,77 @@ def line(line_id, side, unit, contract_product_id='cp', customer='客户', toxic } +def bind(unit, kind='purchase', cert_id='buy'): + return {'kind': kind, 'certId': cert_id, 'unit': unit} + + def plan(**values): return SimpleNamespace( cert_configuration=values.get('configuration'), transport_status=values.get('transport_status', 'pending'), archive_status=values.get('archive_status', 'unarchived'), + quota_cert_waived=values.get('quota_cert_waived', 0), ) def test_same_label_and_code_do_not_need_conversion(): current = plan() - assert plan_needs_unit_conversion(current, [line('L', 'load', '吨')], {'cp': ['dun']}, []) is False - assert plan_needs_unit_conversion(current, [line('L', 'load', '吨', toxic=False)], {'cp': ['kg']}, []) is False + assert plan_needs_unit_conversion(current, [line('L', 'load', '吨')], {'cp': [bind('dun')]}, []) is False + assert plan_needs_unit_conversion(current, [line('L', 'load', '吨', toxic=False)], {'cp': [bind('kg')]}, []) is False assert plan_needs_unit_conversion( - current, [line('L', 'load', '吨', customer='安庆市山峡化工有限公司')], {'cp': ['ping']}, [], + current, [line('L', 'load', '吨', customer='安庆市山峡化工有限公司')], {'cp': [bind('ping')]}, [], ) is True - assert plan_needs_unit_conversion(current, [line('L', 'load', '吨')], {'cp': ['kg']}, []) is False - assert plan_needs_unit_conversion(current, [line('L', 'load', '升')], {'cp': ['毫升']}, []) is False + assert plan_needs_unit_conversion(current, [line('L', 'load', '吨')], {'cp': [bind('kg')]}, []) is False + assert plan_needs_unit_conversion(current, [line('L', 'load', '升')], {'cp': [bind('毫升')]}, []) is False def test_unbound_and_unselected_legacy_sides_are_display_only(): both = [line('L', 'load', '吨'), line('D', 'delivery', '吨')] arranged = plan(transport_status='arranged') - assert plan_needs_unit_conversion(arranged, both, {'cp': ['kg']}, []) is False + assert plan_needs_unit_conversion(arranged, both, {'cp': [bind('kg')]}, []) is False chosen = plan(transport_status='arranged', configuration={'sourceMapping': [ {'sourceStopType': 'load', 'sourceStopProductId': 'L', 'relatedStopProductIds': ['D']}, ]}) - assert plan_needs_unit_conversion(chosen, both, {'cp': ['瓶']}, []) is True - assert plan_needs_unit_conversion(chosen, both, {'cp': ['kg']}, []) is False + assert plan_needs_unit_conversion(chosen, both, {'cp': [bind('瓶')]}, []) is True + assert plan_needs_unit_conversion(chosen, both, {'cp': [bind('kg')]}, []) is False def test_new_plan_counts_either_stop_when_a_bound_certificate_uses_another_unit(): current = plan() assert plan_needs_unit_conversion( - current, [line('L', 'load', 'dun'), line('D', 'delivery', 'dun')], {'cp': ['kg']}, [], + current, [line('L', 'load', 'dun'), line('D', 'delivery', 'dun')], {'cp': [bind('kg')]}, [], ) is False assert plan_needs_unit_conversion( - current, [line('L', 'load', 'dun'), line('D', 'delivery', 'dun')], {'cp': ['ping']}, [], + current, [line('L', 'load', 'dun'), line('D', 'delivery', 'dun')], {'cp': [bind('ping')]}, [], ) is True assert plan_needs_unit_conversion(current, [line('L', 'load', 'dun')], {}, []) is False +def test_waived_or_saved_manual_conversion_clears_pending_flag(): + current = plan(quota_cert_waived=1) + assert plan_needs_unit_conversion( + current, [line('L', 'load', 'ping')], {'cp': [bind('sheng', cert_id='buy')]}, [], + ) is False + saved = plan(configuration={ + 'manualAllocations': [{ + 'permitKind': 'purchase', 'sourceId': 'delivery:D', 'permitId': 'buy', + 'sourceQuantity': '10', 'permitQuantity': '2.5', 'note': '1瓶=0.25升', + }], + }) + assert plan_needs_unit_conversion( + saved, [line('D', 'delivery', 'ping')], {'cp': [bind('sheng', cert_id='buy')]}, [], + ) is False + incomplete = plan(configuration={ + 'manualAllocations': [{ + 'permitKind': 'purchase', 'sourceId': 'delivery:D', 'permitId': 'buy', + 'sourceQuantity': '10', 'permitQuantity': '', 'note': '', + }], + }) + assert plan_needs_unit_conversion( + incomplete, [line('D', 'delivery', 'ping')], {'cp': [bind('sheng', cert_id='buy')]}, [], + ) is True + + def test_list_and_badge_count_only_plans_with_a_real_unit_mismatch(monkeypatch): engine = create_engine('sqlite://') models = (