source_logistics_ecommerce_edi_2024/source_ecommerce_make_shipment.py

#!/usr/bin/env python3
import csv
import pprint
import dataclasses
import datetime
import decimal
import functools
import pathlib
import re
import shutil
import typing

import records  # type: ignore

import yamamotoyama  # type: ignore
import yamamotoyama.x3_imports  # type: ignore

THIS_DIRECTORY = pathlib.Path(__file__).parent
INCOMING_DIRECTORY = THIS_DIRECTORY / "incoming_shipments"
SDH_IMPORT_DIRECTORY = THIS_DIRECTORY / "to_import_SDH"

def main():
    for file in INCOMING_DIRECTORY.iterdir():
        if file.name[-4:] != '.csv':
            continue
        else:
            process_files(file)
            #shutil.move(file, INCOMING_DIRECTORY / "archive" / file.name)
            combine_zship945s()


def combine_zship945s():
    """
    Collect all ZSHPORD imports into a single file for easy import.
    """
    archive_directory = SDH_IMPORT_DIRECTORY / "archive"
    archive_directory.mkdir(exist_ok=True)
    with (SDH_IMPORT_DIRECTORY / "ZSHIP945S.dat").open(
        "w", encoding="utf-8", newline="\n"
    ) as combined_import_file:
        for individual_import_filename in SDH_IMPORT_DIRECTORY.glob(
            "ZSHIP945S_*.dat"
        ):
            with individual_import_filename.open(
                "r", encoding="utf-8", newline="\n"
            ) as individual_import_file:
                for line in individual_import_file:
                    combined_import_file.write(line)
            shutil.move(
                individual_import_filename,
                archive_directory / individual_import_filename.name,
            )

def find_so_from_po(cust_po):
    with yamamotoyama.get_connection('test') as db_connection:#TODO remove 'test'
        return db_connection.query(
            """
            select
                SOHNUM_0
            from FY23TEST.SORDER --TODO change to PROD
            where 
                SOHTYP_0 = 'WEB'
                and BPCORD_0 = 'STSHOPIFY'
                and CUSORDREF_0 = :order
            """,
            order=cust_po,
        ).first()["SOHNUM_0"]

def process_files(file):
    with open(file) as source_file:
        csv_reader = csv.reader(source_file)
        warehouse_shipment = WarehouseShipment()
        previous_order = ''
        previous_item = ''
        for num, row in enumerate(csv_reader):
            if num == 0:
                continue #skip header lines
            if num >= 1: #gather header information
                sohnum = find_so_from_po(row[6])
                if num == 1:
                    previous_order = sohnum
                if previous_order != sohnum:
                    if warehouse_shipment.sohnum != '':
                        time_stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
                        with yamamotoyama.x3_imports.open_import_file(
                            SDH_IMPORT_DIRECTORY / f"ZSHIP945S_{warehouse_shipment.sohnum}_{time_stamp}.dat"
                        ) as import_file:
                            warehouse_shipment.output(import_file)
                        warehouse_shipment = WarehouseShipment()
                        previous_order = sohnum
                order_date = row[9]
                customer_name = row[10]
                # shipadd1 = row[9] # address information is not stored in X3
                # shipadd2 = row[10]
                # shipcity = row[11]
                # shipstate = row[12]
                # shipzip = row[13]
                tracking = row[16]
                weight = row[18]
                ship_charge = row[21]
                taxes = row[22]
                ship_site = row[0]
                discount = row[4]
                warehouse_shipment.sohnum = sohnum
                #warehouse_shipment.header.sohnum = sohnum
                warehouse_shipment.header.shidat = datetime.datetime.strptime(order_date,'%m/%d/%y %I:%M %p')
                warehouse_shipment.header.ylicplate = tracking
                warehouse_shipment.header.growei = weight
                #gather line data
                #TODO how are multiple lots processed?
                line_product = row[1]
                line_qty = row[3]
                line_lot = row[4]
                line_price = row[20]
                subdetail = WarehouseShipmentSubDetail(
                        qtypcu=-1 * int(line_qty),
                        lot=line_lot,
                    )
                if previous_item != line_product:
                    detail = WarehouseShipmentDetail(
                            sohnum=sohnum,
                            itmref=line_product,
                            qty=int(line_qty),
                        )
                    # pprint.pprint(detail)
                    # pprint.pprint(subdetail)
                    warehouse_shipment.append(
                        detail,
                        subdetail,
                    )
                else:
                    warehouse_shipment.append(
                        detail,
                        subdetail,
                    )
        time_stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        with yamamotoyama.x3_imports.open_import_file(
            SDH_IMPORT_DIRECTORY / f"ZSHIP945S_{warehouse_shipment.sohnum}_{time_stamp}.dat"
        ) as import_file:
            warehouse_shipment.output(import_file)
            
@dataclasses.dataclass
class WarehouseShipmentSubDetail:
    """
    Information that goes onto a shipment sub-detail line, taken from ZSHIP945 template.
    """

    sta: str = "A"
    pcu: str = ""
    qtypcu: int = 0
    loc: str = ""
    lot: str = ""
    sernum: str = ""

    def convert_to_strings(self) -> typing.List[str]:
        """
        Convert to strings for X3 import writing.
        """
        return yamamotoyama.x3_imports.convert_to_strings(
            [
                "S",
                self.sta,
                self.pcu,
                self.qtypcu,
                self.loc,
                self.lot,
                self.sernum,
            ]
        )


@dataclasses.dataclass
class WarehouseShipmentDetail:
    """
    Information that goes on a shipment detail line, taken from ZSHIP945 template.
    """

    sohnum: str = ""
    soplin: int = 0
    itmref: str = ""
    itmdes: str = ""
    sau: str = ""
    qty: int = 0
    star91: str = ""
    star92: str = ""
    subdetails: typing.List[WarehouseShipmentSubDetail] = dataclasses.field(
        default_factory=list
    )

    def append(self, subdetail: WarehouseShipmentSubDetail):
        """
        Add subdetail
        """
        
        subdetail.pcu = self.sau
        self.subdetails.append(subdetail)

    def check_subdetail_qty(self):
        """
        Check for shortages by totaling up subdetail quantities.
        """
        total_cases = 0
        for subdetail in self.subdetails:
            total_cases += subdetail.qtypcu
            if subdetail.pcu == '':
                subdetail.pcu = self.sau
        return abs(total_cases)

    def convert_to_strings(self) -> typing.List[str]:
        """
        Convert to strings for X3 import writing.
        """
        self.qty = self.check_subdetail_qty()
        return yamamotoyama.x3_imports.convert_to_strings(
            [
                "L",
                self.sohnum,
                self.soplin,
                self.itmref,
                self.itmdes,
                self.sau,
                self.qty,
                self.star91,
                self.star92,
            ]
        )

    def __eq__(self, item: typing.Any) -> bool:
        """
        Test for equality
        """
        if isinstance(item, str):
            return self.itmref == item
        if isinstance(item, WarehouseShipmentDetail):
            return self.itmref == item.itmref
        return False

    def fill(self):
        """
        Set soplin & itmdes from itmref & sohnum
        """

        def get() -> records.Record:
            with yamamotoyama.get_connection('test') as database:#TODO remove test
                how_many = (
                    database.query(
                        """
                    select
                        count(*) as [how_many]
                    from [FY23TEST].[SORDERP] as [SOP] --TODO change to PROD
                    where
                        [SOP].[SOHNUM_0] = :sohnum
                        and [SOP].[ITMREF_0] = :itmref
                    """,
                        sohnum=self.sohnum,
                        itmref=self.itmref,
                    )
                    .first()
                    .how_many
                )
                if how_many == 1:
                    return database.query(
                        """
                        select top 1
                            [SOP].[SOPLIN_0]
                            ,[SOP].[ITMDES1_0]
                            ,[SOP].[SAU_0]
                        from [FY23TEST].[SORDERP] as [SOP] --TODO change to PROD
                        where
                            [SOP].[SOHNUM_0] = :sohnum
                            and [SOP].[ITMREF_0] = :itmref
                        order by
                            [SOP].[SOPLIN_0]
                        """,
                        sohnum=self.sohnum,
                        itmref=self.itmref,
                    ).first()
                else:
                    pprint.pprint(self.sohnum)
                    pprint.pprint(self.itmref)
                    raise NotImplementedError  # TODO
        result = get()
        self.soplin = result.SOPLIN_0
        self.itmdes = result.ITMDES1_0
        self.sau = result.SAU_0


@dataclasses.dataclass
class WarehouseShipmentHeader:
    """
    Information that goes on a shipment header, taken from ZSHIP945 template.
    """

    salfcy: str = "STC"
    stofcy: str = ""
    sdhnum: str = ""
    bpcord: str = ""
    bpaadd: str = "SH001"
    cur: str = "USD"
    shidat: datetime.date = datetime.date(1753, 1, 1)
    cfmflg: int = 1
    pjt: str = ""
    bptnum: str = ""
    ylicplate: str = ""
    invdtaamt_2: decimal.Decimal = decimal.Decimal()
    invdtaamt_3: decimal.Decimal = decimal.Decimal()
    invdtaamt_4: decimal.Decimal = decimal.Decimal()
    invdtaamt_5: decimal.Decimal = decimal.Decimal()
    invdtaamt_6: decimal.Decimal = decimal.Decimal()
    invdtaamt_7: decimal.Decimal = decimal.Decimal()
    invdtaamt_8: decimal.Decimal = decimal.Decimal()
    invdtaamt_9: decimal.Decimal = decimal.Decimal()
    die: str = ""
    die_1: str = ""
    die_2: str = ""
    die_3: str = ""
    die_4: str = ""
    die_5: str = ""
    die_6: str = ""
    die_7: str = ""
    die_8: str = ""
    die_9: str = ""
    die_10: str = ""
    die_11: str = ""
    die_12: str = ""
    die_13: str = ""
    die_14: str = ""
    die_15: str = ""
    die_16: str = ""
    die_17: str = ""
    die_18: str = ""
    die_19: str = ""
    cce: str = ""
    cce_1: str = ""
    cce_2: str = ""
    cce_3: str = ""
    cce_4: str = ""
    cce_5: str = ""
    cce_6: str = ""
    cce_7: str = ""
    cce_8: str = ""
    cce_9: str = ""
    cce_10: str = ""
    cce_11: str = ""
    cce_12: str = ""
    cce_13: str = ""
    cce_14: str = ""
    cce_15: str = ""
    cce_16: str = ""
    cce_17: str = ""
    cce_18: str = ""
    cce_19: str = ""
    bpdnam: str = ""
    bpdaddlig: str = ""
    bpdaddlig_1: str = ""
    bpdaddlig_2: str = ""
    bpdposcod: str = ""
    bpdcty: str = ""
    bpdsat: str = ""
    bpdcry: str = ""
    bpdcrynam: str = ""
    sdhtyp: str = "SDN"
    growei: decimal.Decimal = decimal.Decimal()
    pacnbr: int = 0
    star71: str = ""
    star72: str = ""
    star81: str = ""
    star82: str = ""

    def convert_to_strings(self) -> typing.List[str]:
        """
        Convert to X3 import line
        """
        return yamamotoyama.x3_imports.convert_to_strings(
            [
                "H",
                self.salfcy,
                self.stofcy,
                self.sdhnum,
                self.bpcord,
                self.bpaadd,
                self.cur,
                self.shidat.strftime("%Y%m%d"),
                self.cfmflg,
                self.pjt,
                self.bptnum,
                self.ylicplate,
                self.invdtaamt_2,
                self.invdtaamt_3,
                self.invdtaamt_4,
                self.invdtaamt_5,
                self.invdtaamt_6,
                self.invdtaamt_7,
                self.invdtaamt_8,
                self.invdtaamt_9,
                self.die,
                self.die_1,
                self.die_2,
                self.die_3,
                self.die_4,
                self.die_5,
                self.die_6,
                self.die_7,
                self.die_8,
                self.die_9,
                self.die_10,
                self.die_11,
                self.die_12,
                self.die_13,
                self.die_14,
                self.die_15,
                self.die_16,
                self.die_17,
                self.die_18,
                self.die_19,
                self.cce,
                self.cce_1,
                self.cce_2,
                self.cce_3,
                self.cce_4,
                self.cce_5,
                self.cce_6,
                self.cce_7,
                self.cce_8,
                self.cce_9,
                self.cce_10,
                self.cce_11,
                self.cce_12,
                self.cce_13,
                self.cce_14,
                self.cce_15,
                self.cce_16,
                self.cce_17,
                self.cce_18,
                self.cce_19,
                self.bpdnam,
                self.bpdaddlig,
                self.bpdaddlig_1,
                self.bpdaddlig_2,
                self.bpdposcod,
                self.bpdcty,
                self.bpdsat,
                self.bpdcry,
                self.bpdcrynam,
                self.sdhtyp,
                self.growei,
                self.pacnbr,
                self.star71,
                self.star72,
                self.star81,
                self.star82,
            ]
        )


class WarehouseShipmentDetailList:
    """
    List of shipment details
    """

    _details: typing.List[WarehouseShipmentDetail]
    _item_set: typing.Set[str]

    def __init__(self):
        self._details = []
        self._item_set = set()

    def append(
        self,
        shipment_detail: WarehouseShipmentDetail,
        shipment_subdetail: WarehouseShipmentSubDetail,
    ):
        """
        Append
        """
        itmref = shipment_detail.itmref
        if itmref in self._item_set:
            for detail in self._details:
                if detail == itmref:
                    detail.subdetails.append(shipment_subdetail)
                    return
        self._item_set.add(itmref)
        shipment_detail.fill()
        shipment_detail.append(shipment_subdetail)
        self._details.append(shipment_detail)

    def __iter__(self):
        return iter(self._details)


class WarehouseShipment:
    """
    Warehosue shipment, both header & details
    """

    header: WarehouseShipmentHeader
    details: WarehouseShipmentDetailList
    _sohnum: str

    def __init__(self):
        self.header = WarehouseShipmentHeader()
        self._sohnum = ""
        self.details = WarehouseShipmentDetailList()

    def append(
        self,
        shipment_detail: WarehouseShipmentDetail,
        shipment_subdetail: WarehouseShipmentSubDetail,
    ):
        """
        Add detail information.
        """
        self.details.append(shipment_detail, shipment_subdetail)

    @property
    def sohnum(self):
        """
        Sales order number
        """
        return self._sohnum

    @sohnum.setter
    def sohnum(self, value: str):
        if self._sohnum != value:
            self._sohnum = value
            if value:
                self._fill_info_from_so()

    def _get_so_from_x3(self) -> records.Record:
        """
        Fetch sales order from X3 database.
        """
        with yamamotoyama.get_connection('test') as db_connection:#TODO remove 'test'
            return db_connection.query(
                """
                select
                     [SOH].[SALFCY_0]
                    ,[SOH].[STOFCY_0]
                    ,[SOH].[BPCORD_0]
                    ,[SOH].[BPAADD_0]
                    ,[SOH].[CUR_0]
                    ,[SOH].[INVDTAAMT_2]
                    ,[SOH].[INVDTAAMT_3]
                    ,[SOH].[INVDTAAMT_4]
                    ,[SOH].[INVDTAAMT_5]
                    ,[SOH].[INVDTAAMT_6]
                    ,[SOH].[INVDTAAMT_7]
                    ,[SOH].[INVDTAAMT_8]
                    ,[SOH].[INVDTAAMT_9]
                    ,[SOH].[DIE_0]
                    ,[SOH].[DIE_1]
                    ,[SOH].[DIE_2]
                    ,[SOH].[DIE_3]
                    ,[SOH].[DIE_4]
                    ,[SOH].[DIE_5]
                    ,[SOH].[DIE_6]
                    ,[SOH].[DIE_7]
                    ,[SOH].[DIE_8]
                    ,[SOH].[DIE_9]
                    ,[SOH].[DIE_10]
                    ,[SOH].[DIE_11]
                    ,[SOH].[DIE_12]
                    ,[SOH].[DIE_13]
                    ,[SOH].[DIE_14]
                    ,[SOH].[DIE_15]
                    ,[SOH].[DIE_16]
                    ,[SOH].[DIE_17]
                    ,[SOH].[DIE_18]
                    ,[SOH].[DIE_19]
                    ,[SOH].[CCE_0]
                    ,[SOH].[CCE_1]
                    ,[SOH].[CCE_2]
                    ,[SOH].[CCE_3]
                    ,[SOH].[CCE_4]
                    ,[SOH].[CCE_5]
                    ,[SOH].[CCE_6]
                    ,[SOH].[CCE_7]
                    ,[SOH].[CCE_8]
                    ,[SOH].[CCE_9]
                    ,[SOH].[CCE_10]
                    ,[SOH].[CCE_11]
                    ,[SOH].[CCE_12]
                    ,[SOH].[CCE_13]
                    ,[SOH].[CCE_14]
                    ,[SOH].[CCE_15]
                    ,[SOH].[CCE_16]
                    ,[SOH].[CCE_17]
                    ,[SOH].[CCE_18]
                    ,[SOH].[CCE_19]
                    ,[SOH].[BPDNAM_0]
                    ,[SOH].[BPDADDLIG_0]
                    ,[SOH].[BPDADDLIG_1]
                    ,[SOH].[BPDADDLIG_2]
                    ,[SOH].[BPDPOSCOD_0]
                    ,[SOH].[BPDCTY_0]
                    ,[SOH].[BPDSAT_0]
                    ,[SOH].[BPDCRY_0]
                    ,[SOH].[BPDCRYNAM_0]
                from [FY23TEST].[SORDER] as [SOH]--TODO change back to PROD
                where
                    [SOH].[SOHNUM_0] = :order
                """,
                order=self.sohnum,
            ).first()

    def _copy_accounting_codes(self, result: records.Record):
        """
        Fill in all the accounting codes
        """
        self.header.die = result.DIE_0
        self.header.die_1 = result.DIE_1
        self.header.die_2 = result.DIE_2
        self.header.die_3 = result.DIE_3
        self.header.die_4 = result.DIE_4
        self.header.die_5 = result.DIE_5
        self.header.die_6 = result.DIE_6
        self.header.die_7 = result.DIE_7
        self.header.die_8 = result.DIE_8
        self.header.die_9 = result.DIE_9
        self.header.die_10 = result.DIE_10
        self.header.die_11 = result.DIE_11
        self.header.die_12 = result.DIE_12
        self.header.die_13 = result.DIE_13
        self.header.die_14 = result.DIE_14
        self.header.die_15 = result.DIE_15
        self.header.die_16 = result.DIE_16
        self.header.die_17 = result.DIE_17
        self.header.die_18 = result.DIE_18
        self.header.die_19 = result.DIE_19
        self.header.cce = result.CCE_0
        self.header.cce_1 = result.CCE_1
        self.header.cce_2 = result.CCE_2
        self.header.cce_3 = result.CCE_3
        self.header.cce_4 = result.CCE_4
        self.header.cce_5 = result.CCE_5
        self.header.cce_6 = result.CCE_6
        self.header.cce_7 = result.CCE_7
        self.header.cce_8 = result.CCE_8
        self.header.cce_9 = result.CCE_9
        self.header.cce_10 = result.CCE_10
        self.header.cce_11 = result.CCE_11
        self.header.cce_12 = result.CCE_12
        self.header.cce_13 = result.CCE_13
        self.header.cce_14 = result.CCE_14
        self.header.cce_15 = result.CCE_15
        self.header.cce_16 = result.CCE_16
        self.header.cce_17 = result.CCE_17
        self.header.cce_18 = result.CCE_18
        self.header.cce_19 = result.CCE_19

    def _fill_info_from_so(self):
        """
        When we learn the SOHNUM, we can copy information from the sales order.
        """
        result = self._get_so_from_x3()
        self.header.salfcy = result.SALFCY_0
        self.header.stofcy = result.STOFCY_0
        self.header.bpcord = result.BPCORD_0
        self.header.bpaadd = result.BPAADD_0
        self.header.cur = result.CUR_0
        self.header.invdtaamt_2 = result.INVDTAAMT_2
        self.header.invdtaamt_3 = result.INVDTAAMT_3
        self.header.invdtaamt_4 = result.INVDTAAMT_4
        self.header.invdtaamt_5 = result.INVDTAAMT_5
        self.header.invdtaamt_6 = result.INVDTAAMT_6
        self.header.invdtaamt_7 = result.INVDTAAMT_7
        self.header.invdtaamt_8 = result.INVDTAAMT_8
        self.header.invdtaamt_9 = result.INVDTAAMT_9
        self._copy_accounting_codes(result)
        self.header.bpdnam = result.BPDNAM_0
        self.header.bpdaddlig = result.BPDADDLIG_0
        self.header.bpdaddlig_1 = result.BPDADDLIG_1
        self.header.bpdaddlig_2 = result.BPDADDLIG_2
        self.header.bpdposcod = result.BPDPOSCOD_0
        self.header.bpdcty = result.BPDCTY_0
        self.header.bpdsat = result.BPDSAT_0
        self.header.bpdcry = result.BPDCRY_0
        self.header.bpdcrynam = result.BPDCRYNAM_0

    def output(self, import_file: typing.TextIO):
        """
        Output entire order to import_file.
        """
        output = functools.partial(
            yamamotoyama.x3_imports.output_with_file, import_file
        )
        output(self.header.convert_to_strings())
        for detail in self.details:
            output(detail.convert_to_strings())
            for subdetail in detail.subdetails:
                output(subdetail.convert_to_strings())


if __name__ == "__main__":
    main()