311 lines
8.5 KiB
Python
311 lines
8.5 KiB
Python
#!/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()
|