Source code for ferrosoft.apps.emiflow.dataimport.base

from dataclasses import dataclass
from typing import Type, Callable

from django.utils.translation import gettext as _

from ferrosoft.apps.dataimport.models import LogHighlight
from ferrosoft.apps.dataimport.queue import ProcessorArgs, ModelProcessorArgs
from ferrosoft.apps.emiflow.models import (
    Settings,
    VehicleType,
    ImportLine,
    TransportChainDefaults,
)
from ferrosoft.apps.emiflow.services.vehicle import detect_vehicle_type
from ferrosoft.apps.ferrobase.tenant.utils import tenant_db_transaction




@dataclass
class ImportState:
    settings: Settings
    booking_disabled: bool





def load_defaults(pargs: ProcessorArgs):
    settings = Settings.get_instance()
    booking_disabled = settings.report_directory is None

    if booking_disabled:
        pargs.ui_updater.send_log_entry(
            pargs.job,
            _("Please review settings and set defaults."),
            0,
            LogHighlight.DANGER,
        )

    return ImportState(settings, booking_disabled)





def process_transport_chain[T](
    processor_args: ModelProcessorArgs,
    model_class: Type[T],
    creator: Callable[[T, ImportState, ModelProcessorArgs], None],
):
    state = processor_args.processor_state
    model = processor_args.model

    if not isinstance(state, ImportState):
        raise ValueError("State must be %s instance" % ImportState.__name__)
    if not isinstance(model, model_class):
        raise ValueError("Model must be %s instance" % model_class.__name__)

    try:
        with tenant_db_transaction():
            creator(model, state, processor_args)
    except Exception as e:
        processor_args.ui_updater.send_log_entry(
            processor_args.job,
            _("Failed booking transport and calculating emissions"),
            processor_args.line_number,
            LogHighlight.DANGER,
        )
        raise e





def get_transport_chain_defaults(
    model: ImportLine,
    state: ImportState,
    processor_args: ModelProcessorArgs,
) -> TransportChainDefaults | None:
    vehicle_type = (
        VehicleType(model.vehicle_type)
        if model.vehicle_type
        else detect_vehicle_type(model.weight)
    )
    defaults = state.settings.get_transport_chain_defaults(vehicle_type)
    if state.booking_disabled or not defaults.are_defaults_set:
        processor_args.ui_updater.send_log_entry(
            processor_args.job,
            _(
                "Cannot create transport chain: please review settings and set defaults for vehicle type %s."
            )
            % vehicle_type.label,
            processor_args.line_number,
            LogHighlight.DANGER,
        )
        return None
    return defaults