tradon/modules/automation/booking_report.py

import json

from sql import Literal, Null
from sql.functions import CurrentTimestamp

from trytond.model import ModelSQL, ModelView, fields
from trytond.pool import Pool
from trytond.transaction import Transaction
from trytond.wizard import Button, StateTransition, StateView, Wizard
from trytond.exceptions import UserError


class FreightBookingShipmentReport(ModelSQL, ModelView):
    "Freight Booking Shipment Report"
    __name__ = 'freight.booking.shipment.report'

    automation_document = fields.Many2One(
        'automation.document', "Automation Document")
    bl_number = fields.Function(fields.Char("BL Number"), 'get_bl_number')
    shipment = fields.Function(fields.Many2One(
        'stock.shipment.in', "Shipment"), 'get_shipment')
    status = fields.Function(fields.Selection([
        ('found', "Found"),
        ('error', "Error"),
    ], "Status"), 'get_status')

    @classmethod
    def __setup__(cls):
        super().__setup__()
        cls._order = [
            ('id', 'DESC'),
        ]

    @classmethod
    def table_query(cls):
        AutomationDocument = Pool().get('automation.document')
        document = AutomationDocument.__table__()

        return document.select(
            Literal(None).as_('create_uid'),
            CurrentTimestamp().as_('create_date'),
            Literal(None).as_('write_uid'),
            Literal(None).as_('write_date'),
            document.id.as_('id'),
            document.id.as_('automation_document'),
            where=(
                (document.type == 'weight_report')
                & (document.metadata_json != Null)))

    @classmethod
    def _extract_bl_number(cls, metadata_json):
        if not metadata_json:
            return None
        try:
            metadata = json.loads(str(metadata_json))
        except (TypeError, ValueError):
            return None

        def find_value(value):
            if isinstance(value, dict):
                for key in (
                        'bl_no', 'bl_number', 'bl', 'BL_Number',
                        'BL number', 'B/L No'):
                    bl_number = value.get(key)
                    if bl_number:
                        return str(bl_number).strip()
                for item in value.values():
                    bl_number = find_value(item)
                    if bl_number:
                        return bl_number
            elif isinstance(value, list):
                for item in value:
                    bl_number = find_value(item)
                    if bl_number:
                        return bl_number
            return None

        return find_value(metadata)

    @classmethod
    def _get_report_data(cls, reports):
        ShipmentIn = Pool().get('stock.shipment.in')
        bl_numbers = {}

        for report in reports:
            document = report.automation_document
            bl_number = cls._extract_bl_number(
                document.metadata_json if document else None)
            bl_numbers[report.id] = bl_number

        shipments_by_bl = {}
        for bl_number in {b for b in bl_numbers.values() if b}:
            shipments = ShipmentIn.search([
                ('bl_number', '=', bl_number),
            ], limit=1)
            if not shipments:
                shipments = ShipmentIn.search([
                    ('bl_number', 'ilike', bl_number),
                ], limit=1)
            shipments_by_bl[bl_number] = shipments[0] if shipments else None

        return bl_numbers, shipments_by_bl

    @classmethod
    def get_bl_number(cls, reports, name):
        bl_numbers, _ = cls._get_report_data(reports)
        return bl_numbers

    @classmethod
    def get_shipment(cls, reports, name):
        bl_numbers, shipments_by_bl = cls._get_report_data(reports)
        return {
            report.id: (
                shipments_by_bl[bl_numbers[report.id]].id
                if bl_numbers[report.id]
                and shipments_by_bl.get(bl_numbers[report.id])
                else None)
            for report in reports
        }

    @classmethod
    def get_status(cls, reports, name):
        bl_numbers, shipments_by_bl = cls._get_report_data(reports)
        return {
            report.id: (
                'found'
                if bl_numbers[report.id]
                and shipments_by_bl.get(bl_numbers[report.id])
                else 'error')
            for report in reports
        }


class FreightBookingShipmentReportLinkStart(ModelView):
    "Link Freight Booking Shipment Report"
    __name__ = 'freight.booking.shipment.report.link.start'

    bl_number = fields.Char("BL Number", readonly=True)
    shipment = fields.Many2One(
        'stock.shipment.in', "Shipment", required=True)


class FreightBookingShipmentReportLink(Wizard):
    "Link Freight Booking Shipment Report"
    __name__ = 'freight.booking.shipment.report.link'

    start = StateView(
        'freight.booking.shipment.report.link.start',
        'automation.freight_booking_shipment_report_link_start_form',
        [
            Button('Cancel', 'end', 'tryton-cancel'),
            Button('Link', 'link', 'tryton-ok', default=True),
        ])
    link = StateTransition()

    def default_start(self, fields):
        Report = Pool().get('freight.booking.shipment.report')
        active_ids = Transaction().context.get('active_ids') or []
        reports = Report.browse(active_ids)
        error_reports = [
            report for report in reports
            if report.status == 'error' and report.bl_number]
        if len(error_reports) > 1:
            raise UserError(
                "Link one error line at a time to avoid overwriting the "
                "shipment BL number.")
        return {
            'bl_number': error_reports[0].bl_number if error_reports else None,
        }

    def transition_link(self):
        Report = Pool().get('freight.booking.shipment.report')
        ShipmentIn = Pool().get('stock.shipment.in')
        active_ids = Transaction().context.get('active_ids') or []
        reports = Report.browse(active_ids)
        error_reports = [
            report for report in reports
            if report.status == 'error' and report.bl_number]
        if not error_reports:
            raise UserError(
                "Select at least one error line with a BL number.")
        if len(error_reports) > 1:
            raise UserError(
                "Link one error line at a time to avoid overwriting the "
                "shipment BL number.")

        shipment = self.start.shipment
        shipment.bl_number = error_reports[0].bl_number
        ShipmentIn.save([shipment])
        return 'end'

    def end(self):
        return 'reload'