shandex_edi_2024/edi_850.py

#!/usr/bin/env python3
"""
Consume a 850 file from Shandex, and translate into a Sage X3
readable file - import template ZPOH.
For Shandex we also need to reply with a 997
"""
# pylint: disable=too-many-instance-attributes
import dataclasses
import datetime
import decimal
import functools
import pathlib
import re
import shutil
import typing
import pprint
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

import records  # type: ignore

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

THIS_DIRECTORY = pathlib.Path(__file__).parent
X12_DIRECTORY = THIS_DIRECTORY / "incoming"
IMPORTS_DIRECTORY = THIS_DIRECTORY / "x3_imports"
EDI_997_DIRECTORY = THIS_DIRECTORY / "997_processing"

SHANDEX_850_FILENAME_RE = re.compile(
    r"\A 850_STASH-YAMAMOTOYAMA_ \S+ [.]edi \Z", re.X | re.M | re.S
)

def main():
    """
    Do it!
    """
    for edi_filename in X12_DIRECTORY.iterdir():
        if SHANDEX_850_FILENAME_RE.match(edi_filename.name):
            process_file(edi_filename)
            # file moved to 997 processing folder to be sent later
            shutil.move(edi_filename, EDI_997_DIRECTORY / edi_filename.name)
    combine_zpohs()


def new_850_alert(ordref, orddat):
    msg = MIMEMultipart()
    msg['Subject'] = 'New PO from Shandex'
    msg['Precedence'] = 'bulk'
    msg['From'] = 'x3report@stashtea.com'
    msg['To'] = 'icortes@yamamotoyama.com'
    msg['CC'] = 'bleeson@stashtea.com'
    emailtext = f'Ref: {ordref}\nDate: {orddat}'
    msg.attach(MIMEText(emailtext, 'plain'))
    with smtplib.SMTP_SSL("smtp.gmail.com", 465) as smtp:
        smtp.login(user='x3reportmk2@yamamotoyama.com', password=r'n</W<7fr"VD~\2&[pZc5')
        smtp.send_message(msg)

def combine_zpohs():
    """
    Collect all ZPOH imports into a single file for easy import.
    """
    archive_directory = IMPORTS_DIRECTORY / "archive"
    archive_directory.mkdir(exist_ok=True)
    with (IMPORTS_DIRECTORY / "ZPOH.dat").open(
        "w", encoding="utf-8", newline="\n"
    ) as combined_import_file:
        for individual_import_filename in IMPORTS_DIRECTORY.glob(
            "ZPOH_*.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 tokens_from_edi_file(
    edi_filename: pathlib.Path,
) -> typing.Iterator[typing.List[str]]:
    """
    Read tokens from EDI file
    """
    with edi_filename.open(encoding="utf-8", newline="") as edi_file:
        for record in edi_file.read().split("~"):
            fields = record.split("*")
            if fields[0] in {
                "ISA",
                "ST",
                "N2",
                "N3",
                "N4",
            }:
                continue
            yield fields


def process_file(edi_filename: pathlib.Path):
    """
    Convert a specific EDI file into an import file.
    """
    with yamamotoyama.get_connection() as database:
        purchase_order = PO()
        pohnum = ''
        for fields in tokens_from_edi_file(edi_filename):
            #BEG*00*SA*PO0040865**20241209
            if fields[0] == "BEG":
                _, _, _, ordref, _, orddat = fields[:6]
                purchase_order.header.ordref = ordref
                purchase_order.header.orddat = datetime.datetime.strptime(
                    orddat, "%Y%m%d"
                ).date()  # 20230922
                purchase_order.header.pohfcy='WON'
                purchase_order.header.bpsnum='PMW'
                purchase_order.header.cur='USD'
            #DTM*010*20250109
            if fields[0] == "DTM" and fields[1] == '010':
                expected_ship_date = fields[2]
            #PO1*1*2688*CA*15.02**VP*C08225*IN*10077652082255
            if fields[0] == "PO1":
                _, lineno, qty_str, uom, pricestr, _, _, itmref, gtin = fields[:9]
                detail = PODetail(
                    itmref=itmref,
                    itmdes=get_itmdes(itmref, database),
                    prhfcy='WON',
                    uom=uom,
                    qtyuom=int(qty_str),
                    extrcpdat=expected_ship_date,
                    gropri=pricestr,
                )
                purchase_order.append(detail)
        time_stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        new_850_alert(ordref, purchase_order.header.orddat)#TODO new alert type, after it hits X3?
        with yamamotoyama.x3_imports.open_import_file(
            IMPORTS_DIRECTORY / f"ZPOH_{purchase_order.header.ordref}_{time_stamp}.dat"
        ) as import_file:
            purchase_order.output(import_file)


def get_itmdes(itmref, database):
    result  = database.query(
                """
            select
                ITMDES1_0
            from PROD.ITMMASTER
            where 
            ITMREF_0 = :itmref
            """,
                itmref=itmref,
            ).first()['ITMDES1_0']
    return result

@dataclasses.dataclass
class PODetail:
    """
    Information that goes on a PO detail line, taken from ZPOH template.
    """
    
    itmref: str = ""
    itmdes: str = ""
    prhfcy: str = ""
    uom: str = ""
    qtyuom: int = 0
    extrcpdat: datetime.date = datetime.date(1753, 1, 1)
    gropri: decimal.Decimal = 0
    discrgval1: decimal.Decimal = 0
    discrgval2: decimal.Decimal = 0
    discrgval3: decimal.Decimal = 0
    pjt: str = ""
    vat: str = ''
    star91: str = ""
    star92: str = ""

    def convert_to_strings(self) -> typing.List[str]:
        """
        Convert to strings for X3 import writing.
        """
        def fix_uom(uom):
            x3_uom = ''
            if uom == 'CA':
                x3_uom = 'CS'
            else:
                x3_uom = uom
            return x3_uom

        return yamamotoyama.x3_imports.convert_to_strings(
            [
                "L",
                self.itmref,
                self.itmdes,
                self.prhfcy,
                fix_uom(self.uom),
                self.qtyuom,
                self.extrcpdat,
                self.gropri,
                self.discrgval1,
                self.discrgval2,
                self.discrgval3,
                self.pjt,
                self.vat,
                self.star91,
                self.star92,
            ]
        )

@dataclasses.dataclass
class POHeader:
    """
    Information that goes on a po header, taken from ZPOH template.
    """

    pohfcy: str = ""
    pohnum: str = ""
    orddat: datetime.date = datetime.date(1753, 1, 1)
    bpsnum: str = ""
    ordref: str = ""
    bpsnum: str = ""
    cur: str = "USD"
    star71 = ""
    star72 = ""
    star81 = ""
    star82 = ""
    
    
    def convert_to_strings(self) -> typing.List[str]:
        """
        Convert to X3 import line
        """
        return yamamotoyama.x3_imports.convert_to_strings(
            [
                "E",
                self.pohfcy,
                '',#self.pohnum,
                self.orddat.strftime("%Y%m%d"),
                self.bpsnum,
                self.ordref,
                self.cur,
                self.star71,
                self.star72,
                self.star81,
                self.star82,
            ]
        )


class PODetailList:
    """
    List of PO details
    """

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

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

    def append(
        self,
        po_detail: PODetail
    ):
        """
        Append
        """
        self._details.append(po_detail)

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


class PO:
    """
    Warehouse po, both header & details
    """

    header: POHeader
    details: PODetailList

    def __init__(self):
        self.header = POHeader()
        self.details = PODetailList()

    def append(
        self,
        po_detail: PODetail,
    ):
        """
        Add detail information.
        """
        self.details.append(po_detail)

    def output(self, import_file: typing.TextIO):
        """
        Output entire po 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())


if __name__ == "__main__":
    main()