556 lines
17 KiB
Python
556 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Consume a generic 947 file from 3PLs, and translate into a Sage X3
|
|
readable file - import template ZSCS.
|
|
For Shadex we also need to reply with a 997
|
|
947 is warehouse advice, to alert us of damages or amount changes from something
|
|
like a count
|
|
"""
|
|
#what about serial numbers?
|
|
#status on L line?
|
|
#remove negative line, needs to look up stocou? not sure.
|
|
# pylint: disable=too-many-instance-attributes
|
|
import dataclasses
|
|
import datetime
|
|
import decimal
|
|
import functools
|
|
import pathlib
|
|
import re
|
|
import shutil
|
|
import typing
|
|
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"
|
|
|
|
SHANDEX_947_FILENAME_RE = re.compile(
|
|
r"\A 947_YAMAMOTOYAMA_ \S+ [.]edi \Z", re.X | re.M | re.S
|
|
)
|
|
#TODO remove this and have a single production name
|
|
SHANDEX_947_FILENAME_RE = re.compile(
|
|
r"\A STASH_TEST_947 \S+ [.]edi \Z", re.X | re.M | re.S
|
|
)
|
|
|
|
TEST_FILE = THIS_DIRECTORY / "edi-testing" / "STASH_TEST_947_e19f1f58-d77c-47da-a1f1-2ba449eba6ae.edi"#TODO remove this
|
|
TEST_DIR = THIS_DIRECTORY / "edi-testing"
|
|
|
|
DAMAGE_CODE_MAPPING = {
|
|
"07" : 'RD',#Product Dumped or Destroyed
|
|
"AV" : 'RD'#Damaged in Transit
|
|
}
|
|
|
|
DAMAGE_CODE_DESCRIPTIONS_MAPPING = {
|
|
"07" : "Product Dumped or Destroyed",
|
|
"AV" : "Damaged in Transit"
|
|
}
|
|
|
|
def main():
|
|
"""
|
|
Do it!
|
|
"""
|
|
if SHANDEX_947_FILENAME_RE.match(TEST_FILE.name):#TODO remove these 2 lines
|
|
process_file(TEST_FILE)
|
|
# for edi_filename in X12_DIRECTORY.iterdir(): #TODO uncomment and review
|
|
# if SHANDEX_947_FILENAME_RE.match(edi_filename.name):
|
|
# process_file(edi_filename)
|
|
# file moved to 997 processing folder to be sent later
|
|
# shutil.move(edi_filename, X12_DIRECTORY / "997_processing" / edi_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 { #TODO see if there are more fields used in vendor EDI
|
|
"ISA",
|
|
"GS"
|
|
"W15",
|
|
"N1",
|
|
"N9",
|
|
"SE",
|
|
"GE",
|
|
"IEA"
|
|
}:
|
|
continue
|
|
yield fields
|
|
|
|
|
|
def gtin_lookup(gtin):
|
|
with yamamotoyama.get_connection('test') as db_connection:#todo remove 'test'
|
|
return db_connection.query(
|
|
"""
|
|
select
|
|
[ITM].[ITMREF_0],
|
|
[ITM].[ITMDES1_0],
|
|
[ITM].[EANCOD_0],
|
|
[ITM].[ZCASEUPC_0]
|
|
from [FY23TEST].[ITMMASTER] [ITM]--TODO change back to [PROD]
|
|
where
|
|
[ITM].[ZCASEUPC_0] = :zcaseupc
|
|
|
|
""",
|
|
zcaseupc=gtin,
|
|
).first()["ITMREF_0"]
|
|
|
|
|
|
def stock_movement_alert(gtin, qty, lot, status):
|
|
msg = MIMEMultipart()
|
|
msg['Subject'] = 'New Stock Change from Shandex'
|
|
msg['Precedence'] = 'bulk'
|
|
msg['From'] = 'x3report@stashtea.com'
|
|
msg['To'] = 'bleeson@stashtea.com'#TODO correct receipientscares
|
|
emailtext = f'Item: {gtin_lookup(gtin)}\nQty: {qty}\nLot: {lot}\nStatus: {DAMAGE_CODE_MAPPING[status]}\nReason: {DAMAGE_CODE_DESCRIPTIONS_MAPPING[status]}'
|
|
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 process_file(edi_filename: pathlib.Path):
|
|
"""
|
|
Convert a specific EDI file into an import file.
|
|
"""
|
|
warehouse_stockchange = StockChange()
|
|
vcrlin = 0
|
|
for fields in tokens_from_edi_file(edi_filename):
|
|
if fields[0] == "G62":
|
|
iptdat = fields[2]
|
|
warehouse_stockchange.header.iptdat = datetime.datetime.strptime(
|
|
iptdat, "%Y%m%d"
|
|
).date()
|
|
if fields[0] == "W19":
|
|
vcrlin += 1000
|
|
# W19*AV*35*CA**UK*10077652082651***03022026C
|
|
_, status, qty, uom, _, _, gtin, _, _, lot = fields[:10]
|
|
stock_movement_alert(gtin, qty, lot, status)
|
|
warehouse_stockchange.header.vcrdes = DAMAGE_CODE_DESCRIPTIONS_MAPPING[status]
|
|
subdetail = StockChangeSubDetail(
|
|
qtypcu=int(qty),
|
|
sta=DAMAGE_CODE_MAPPING[status],
|
|
pcu=uom
|
|
)
|
|
detail_line = StockChangeDetail(
|
|
vcrlin=vcrlin,
|
|
itmref=gtin,
|
|
pcu=uom,
|
|
lot=lot
|
|
)
|
|
warehouse_stockchange.append(
|
|
detail_line,
|
|
subdetail,
|
|
)
|
|
|
|
time_stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
with yamamotoyama.x3_imports.open_import_file(
|
|
TEST_DIR / f"ZSCS_{warehouse_stockchange.sdhnum}_{time_stamp}.dat" #todo change folder back to IMPORTS_DIRECTORY
|
|
) as import_file:
|
|
warehouse_stockchange.output(import_file)
|
|
#TODO email notification about stock change record creation and reason code to Ilora, anyone else too?
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class StockChangeSubDetail:
|
|
"""
|
|
Information that goes onto a stockchange sub-detail line, taken from ZPTHI template.
|
|
"""
|
|
|
|
pcu: str = ""
|
|
qtypcu: int = 0
|
|
qtystu: int = 0
|
|
loc: str = ""
|
|
sta: str = "A"
|
|
|
|
# def stojous(self, shipment, item) -> typing.List[str]:
|
|
# """
|
|
# Convert grouped lot quantities into individual STOJOU records to fit on stockchange
|
|
# """
|
|
# with yamamotoyama.get_connection('test') as database: #todo remove 'test'
|
|
# details = (
|
|
# database.query(
|
|
# """
|
|
# select
|
|
# 'S',
|
|
# 'A',
|
|
# [STJ].[PCU_0],
|
|
# cast(cast(-1*[STJ].[QTYSTU_0] as int) as nvarchar),
|
|
# [STJ].[LOT_0],
|
|
# '',
|
|
# ''
|
|
# from [FY23TEST].[STOJOU] [STJ] --TODO change to PROD
|
|
# where
|
|
# [STJ].[VCRNUM_0] = :sdhnum
|
|
# and [STJ].[ITMREF_0] = :itmref
|
|
# and [STJ].[LOT_0] = :lot
|
|
# and [STJ].[TRSTYP_0] = 4
|
|
# """,
|
|
# sdhnum=shipment,
|
|
# itmref=item,
|
|
# lot=self.lot,
|
|
# )
|
|
# .all()
|
|
# )
|
|
# return details
|
|
|
|
|
|
def convert_to_strings(self) -> typing.List[str]:
|
|
"""
|
|
Convert to strings for X3 import writing.
|
|
"""
|
|
return yamamotoyama.x3_imports.convert_to_strings(
|
|
[
|
|
"S",
|
|
self.pcu,
|
|
self.qtypcu,
|
|
self.qtystu,
|
|
self.loc,
|
|
self.sta,
|
|
]
|
|
)
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class StockChangeDetail:
|
|
"""
|
|
Information that goes on a stockchange detail line, taken from ZPTHI template.
|
|
"""
|
|
|
|
vcrlin: int = 0
|
|
itmref: str = ""
|
|
pcu: str = ""
|
|
pcustucoe: int = 1 #does this need a lookup?
|
|
sta: str = "A" #todo this needs to flip based on the transaction A > R, A > Q, what about Q > A?
|
|
loctyp: str = ""
|
|
loc: str = ""
|
|
lot: str = ""
|
|
slo: str = ""
|
|
sernum: str = ""
|
|
palnum: str = ""
|
|
ctrnum: str = ""
|
|
qlyctldem: str= ""
|
|
owner: str = "WON"
|
|
subdetails: typing.List[StockChangeSubDetail] = dataclasses.field(
|
|
default_factory=list
|
|
)
|
|
|
|
def palnum_lookup(self, itmref, lot, status):
|
|
"""
|
|
Pick a palnum from X3 using the lot, location, and status
|
|
It doesn't matter which one we choose?
|
|
"""
|
|
with yamamotoyama.get_connection('test') as db_connection:#todo remove 'test'
|
|
return db_connection.query(
|
|
"""
|
|
select
|
|
[STO].[STOFCY_0],
|
|
[STO].[ITMREF_0],
|
|
[STO].[LOT_0],
|
|
[STO].[PALNUM_0]
|
|
from [FY23TEST].[STOCK] [STO] --TODO change to PROD
|
|
where
|
|
[STO].[ITMREF_0] = :itmref
|
|
and [STO].[STOFCY_0] = 'WON'
|
|
and [STO].[LOT_0] = :lot
|
|
and [STO].[STA_0] = :status
|
|
""",
|
|
itmref=itmref,
|
|
lot=lot,
|
|
status=status
|
|
).first()["PALNUM_0"]
|
|
|
|
def gtin_lookup(self, gtin):
|
|
with yamamotoyama.get_connection('test') as db_connection:#todo remove 'test'
|
|
return db_connection.query(
|
|
"""
|
|
select
|
|
[ITM].[ITMREF_0],
|
|
[ITM].[ITMDES1_0],
|
|
[ITM].[EANCOD_0],
|
|
[ITM].[ZCASEUPC_0]
|
|
from [FY23TEST].[ITMMASTER] [ITM]--TODO change back to [PROD]
|
|
where
|
|
[ITM].[ZCASEUPC_0] = :zcaseupc
|
|
|
|
""",
|
|
zcaseupc=gtin,
|
|
).first()["ITMREF_0"]
|
|
|
|
def append(self, subdetail: StockChangeSubDetail):
|
|
"""
|
|
Add subdetail
|
|
"""
|
|
subdetail.pcu = self.pcu
|
|
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
|
|
return abs(total_cases)
|
|
|
|
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
|
|
self.qty = self.check_subdetail_qty()
|
|
return yamamotoyama.x3_imports.convert_to_strings(
|
|
[
|
|
"L",
|
|
self.vcrlin,
|
|
self.gtin_lookup(self.itmref),
|
|
fix_uom(self.pcu),
|
|
self.pcustucoe,
|
|
self.sta,
|
|
self.loctyp,
|
|
self.loc,
|
|
self.lot,
|
|
self.slo,
|
|
self.sernum,
|
|
self.palnum_lookup(self.itmref, self.lot, self.sta),
|
|
self.ctrnum,
|
|
self.qlyctldem,
|
|
self.owner
|
|
]
|
|
)
|
|
|
|
def __eq__(self, item: typing.Any) -> bool:
|
|
"""
|
|
Test for equality
|
|
"""
|
|
if isinstance(item, str):
|
|
return self.itmref == item
|
|
if isinstance(item, StockChangeDetail):
|
|
return self.itmref == item.itmref
|
|
return False
|
|
|
|
# def fill(self):#not needed for stockchanges
|
|
# """
|
|
# Set soplin & itmdes from itmref & sohnum
|
|
# """
|
|
|
|
# def get() -> records.Record:
|
|
# with yamamotoyama.get_connection() as database:
|
|
# how_many = (
|
|
# database.query(
|
|
# """
|
|
# select
|
|
# count(*) as [how_many]
|
|
# from [PROD].[SORDERP] as [SOP]
|
|
# 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 [PROD].[SORDERP] as [SOP]
|
|
# where
|
|
# [SOP].[SOHNUM_0] = :sohnum
|
|
# and [SOP].[ITMREF_0] = :itmref
|
|
# order by
|
|
# [SOP].[SOPLIN_0]
|
|
# """,
|
|
# sohnum=self.sohnum,
|
|
# itmref=self.itmref,
|
|
# ).first()
|
|
|
|
|
|
# result = get()
|
|
# self.soplin = result.SOPLIN_0
|
|
# self.itmdes = result.ITMDES1_0
|
|
# self.sau = result.SAU_0
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class StockChangeHeader:
|
|
"""
|
|
Information that goes on a stockchange header, taken from ZSCS template.
|
|
"""
|
|
vcrnum: str = ""
|
|
stofcy: str = "WON"
|
|
iptdat: datetime.date = datetime.date(1753, 1, 1)
|
|
vcrdes: str = ""
|
|
pjt: str = ""
|
|
trsfam: str = "CHX"
|
|
|
|
|
|
def convert_to_strings(self) -> typing.List[str]:
|
|
"""
|
|
Convert to X3 import line
|
|
"""
|
|
return yamamotoyama.x3_imports.convert_to_strings(
|
|
[
|
|
"E",
|
|
self.vcrnum,
|
|
self.stofcy,
|
|
self.iptdat.strftime("%Y%m%d"),
|
|
self.vcrdes,
|
|
self.pjt,
|
|
self.trsfam,
|
|
]
|
|
)
|
|
|
|
|
|
class StockChangeDetailList:
|
|
"""
|
|
List of stockchange details
|
|
"""
|
|
|
|
_details: typing.List[StockChangeDetail]
|
|
_item_set: typing.Set[str]
|
|
|
|
def __init__(self):
|
|
self._details = []
|
|
self._item_set = set()
|
|
|
|
def append(
|
|
self,
|
|
stockchange_detail: StockChangeDetail,
|
|
stockchange_subdetail: StockChangeSubDetail,
|
|
):
|
|
"""
|
|
Append
|
|
"""
|
|
itmref = stockchange_detail.itmref
|
|
if itmref in self._item_set:
|
|
for detail in self._details:
|
|
if detail == itmref:
|
|
detail.subdetails.append(stockchange_subdetail)
|
|
return
|
|
self._item_set.add(itmref)
|
|
#stockchange_detail.fill()
|
|
stockchange_detail.append(stockchange_subdetail)
|
|
self._details.append(stockchange_detail)
|
|
|
|
def __iter__(self):
|
|
return iter(self._details)
|
|
|
|
|
|
class StockChange:
|
|
"""
|
|
Warehouse stockchange, both header & details
|
|
"""
|
|
|
|
header: StockChangeHeader
|
|
details: StockChangeDetailList
|
|
_sdhnum: str
|
|
|
|
def __init__(self):
|
|
self.header = StockChangeHeader()
|
|
self._sdhnum = ""
|
|
self.details = StockChangeDetailList()
|
|
|
|
def append(
|
|
self,
|
|
stockchange_detail: StockChangeDetail,
|
|
stockchange_subdetail: StockChangeSubDetail,
|
|
):
|
|
"""
|
|
Add detail information.
|
|
"""
|
|
self.details.append(stockchange_detail, stockchange_subdetail)
|
|
|
|
@property
|
|
def sdhnum(self):
|
|
"""
|
|
shipment number
|
|
"""
|
|
return self._sdhnum
|
|
|
|
@sdhnum.setter
|
|
def sdhnum(self, value: str):
|
|
if self._sdhnum != value:
|
|
self._sdhnum = value
|
|
if value:
|
|
self._fill_info_from_shipment()
|
|
|
|
# def _get_shipment_from_x3(self) -> records.Record:
|
|
# """
|
|
# Fetch shipment from X3 database.
|
|
# """
|
|
# with yamamotoyama.get_connection('test') as db_connection:#todo remove 'test'
|
|
# return db_connection.query(
|
|
# """
|
|
# select
|
|
# [SDH].[STOFCY_0],
|
|
# [SDH].[SDHNUM_0],
|
|
# [SDH].[SALFCY_0],
|
|
# [SDH].[BPCORD_0],
|
|
# [SDH].[CUR_0],
|
|
# [SDH].[SOHNUM_0]
|
|
# from [FY23TEST].[SDELIVERY] [SDH]--TODO change back to [PROD]
|
|
# where
|
|
# [SDH].[SDHNUM_0] = :shipment
|
|
# """,
|
|
# shipment=self.sdhnum,
|
|
# ).first()
|
|
|
|
# def _fill_info_from_shipment(self):
|
|
# """
|
|
# When we learn the SOHNUM, we can copy information from the sales order.
|
|
# """
|
|
# result = self._get_shipment_from_x3()
|
|
# self.header.stofcy = result.STOFCY_0
|
|
# self.header.sdhnum = result.SDHNUM_0
|
|
# self.header.salfcy = result.SALFCY_0
|
|
# self.header.bpcord = result.BPCORD_0
|
|
# self.header.cur = result.CUR_0
|
|
# self.header.sohnum = result.SOHNUM_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:
|
|
#shipment = detail.sdhnum
|
|
#item = detail.itmref
|
|
output(subdetail.convert_to_strings())
|
|
# for record in subdetail.stojous(shipment, item):
|
|
# output(subdetail.convert_to_strings())
|
|
# output(record)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|