shanxia2.0/backend/app/services/pl_quota_conversion.py

"""计划段次是否需要把计划单位换算成证面单位。"""
from decimal import Decimal, InvalidOperation

from app.models.pl_models import (
    PlPlan, PlPlanDelivery, PlPlanDeliveryProduct, PlPlanLoad, PlPlanLoadProduct, PlPlanProduct,
)
from app.models.pl_quota_models import (
    PlContractCertLink, PlContractCertProduct, PlPlanCertUsage, PlPurchaseCert, PlTransportCert,
)
from app.services.pl_quota_units import same_quantity_unit, standard_conversion_ratio

_VERSIONS = ('per_stop_v2', 'legacy_single_side')


def measurement_version(plan, snapshots):
    """与 resolve_measurement_version 同一口径，供列表和角标批量判断。"""
    if snapshots:
        return 'per_stop_v2' if all((snap or {}).get('measurementVersion') == 'per_stop_v2' for snap in snapshots) else 'legacy_single_side'
    saved = ((plan.cert_configuration or {}).get('measurementVersion'))
    if saved in _VERSIONS:
        return saved
    if plan.transport_status == 'arranged' or plan.archive_status in ('archived', 'exception'):
        return 'legacy_single_side'
    return 'per_stop_v2'


def _mapping_ids(configuration):
    mapping = (configuration or {}).get('sourceMapping') or (configuration or {}).get('source_mapping') or []
    ids = set()
    for item in mapping:
        if isinstance(item, dict):
            value = item.get('sourceStopProductId') or item.get('source_stop_product_id')
            if value:
                ids.add(value)
    return ids


def measured_lines(version, lines, configuration):
    """只保留查看用证里会要求换算的易制毒段。旧口径两侧都在、尚未选定计量侧时不算。"""
    toxic = [line for line in lines if line['isToxic']]
    if version == 'per_stop_v2':
        return toxic
    chosen = _mapping_ids(configuration)
    if chosen:
        return [line for line in toxic if line['id'] in chosen]
    by_product = {}
    for line in toxic:
        by_product.setdefault(line['planProductId'], {'load': [], 'delivery': []})[line['side']].append(line)
    result = []
    for sides in by_product.values():
        available = [side for side, rows in sides.items() if rows]
        if len(available) == 1:
            result.extend(sides[available[0]])
    return result


def line_needs_conversion(unit, cert_units):
    left = str(unit or '').strip()
    if not left:
        return False
    return any(
        str(value or '').strip() and not same_quantity_unit(left, value)
        and standard_conversion_ratio(left, value) is None
        for value in cert_units
    )


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):
        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


def _chunks(values, size=400):
    values = list(values)
    return [values[index:index + size] for index in range(0, len(values), size)] or []


def _stop_lines(db, tenant_no, plan_ids):
    grouped = {plan_id: [] for plan_id in plan_ids}
    sides = (
        ('load', PlPlanLoadProduct, PlPlanLoad, PlPlanLoadProduct.load_id),
        ('delivery', PlPlanDeliveryProduct, PlPlanDelivery, PlPlanDeliveryProduct.delivery_id),
    )
    for side, product_model, stop_model, parent_id in sides:
        rows = db.query(
            product_model.plan_id, product_model.id, product_model.plan_product_id, product_model.unit,
            product_model.contract_product_id, stop_model.customer_name, PlPlanProduct.is_toxic,
        ).join(stop_model, stop_model.id == parent_id).join(
            PlPlanProduct, PlPlanProduct.id == product_model.plan_product_id,
        ).filter(
            product_model.tenant_no == tenant_no, product_model.del_flag == 0,
            stop_model.tenant_no == tenant_no, stop_model.del_flag == 0,
            PlPlanProduct.tenant_no == tenant_no, PlPlanProduct.del_flag == 0,
            product_model.plan_id.in_(plan_ids),
        ).all()
        for plan_id, line_id, plan_product_id, unit, contract_product_id, customer_name, is_toxic in rows:
            grouped[plan_id].append({
                'id': line_id, 'planProductId': plan_product_id, 'side': side, 'unit': unit,
                'contractProductId': contract_product_id, 'customerName': customer_name,
                'isToxic': is_toxic == 1,
            })
    return grouped


def _cert_bindings(db, tenant_no, contract_product_ids):
    bindings = {}
    if not contract_product_ids:
        return bindings
    for kind, cert_model in (('purchase', PlPurchaseCert), ('transport', PlTransportCert)):
        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,
            PlContractCertLink.link_status == 'active', PlContractCertLink.cert_type == kind,
            PlContractCertProduct.tenant_no == tenant_no, PlContractCertProduct.del_flag == 0,
            PlContractCertProduct.scope_status == 'active',
            cert_model.tenant_no == tenant_no, cert_model.del_flag == 0,
            PlContractCertProduct.contract_product_id.in_(contract_product_ids),
        ).all()
        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):
    needed = set()
    for chunk in _chunks(plan_ids):
        plans = db.query(PlPlan).filter(
            PlPlan.tenant_no == tenant_no, PlPlan.del_flag == 0, PlPlan.id.in_(chunk),
        ).all()
        if not plans:
            continue
        ids = [plan.id for plan in plans]
        lines = _stop_lines(db, tenant_no, ids)
        product_ids = {
            line['contractProductId'] for rows in lines.values() for line in rows if line['contractProductId']
        }
        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,
            PlPlanCertUsage.plan_id.in_(ids),
        ).all()
        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_bindings, snapshots.get(plan.id, [])):
                needed.add(plan.id)
    return needed