bleeson
/
shandex_edi_2024
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

997 processing flow and small fixes.

master
bleeson 2024-03-14 15:46:57 -07:00
parent 4ad4ad79a5
commit f3782aede4
5 changed files with 141 additions and 607 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
edi_867.py
View File

@ -22,6 +22,7 @@ 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"
SOURCE_867_FILENAME_RE = re.compile(
r"\A 867_YAMAMOTOYAMA_ .* [.]edi \Z", re.X | re.M | re.S
@ -40,7 +41,8 @@ def main():
if SOURCE_867_FILENAME_RE.match(edi_filename.name):
process_file(edi_filename)
#TODO respond with 997?
#shutil.move(edi_filename, THIS_DIRECTORY / "processed_867s" / edi_filename.name)#TODO uncomment
shutil.copy(edi_filename, EDI_997_DIRECTORY / edi_filename.name)
shutil.move(edi_filename, THIS_DIRECTORY / "processed_867s" / edi_filename.name)
combine_zship867s()
@ -145,7 +147,7 @@ def process_file(edi_filename: pathlib.Path):
itmref=itmref,
itmdes=itmdes,
qty=int(qty_str),
netpri=price,
gropri=price,
sau=sau
),
subdetail,
@ -199,7 +201,7 @@ class WarehouseShipmentDetail:
itmdes: str = ""
sau: str = ""
qty: int = 0
netpri: decimal.Decimal = decimal.Decimal()
gropri: decimal.Decimal = decimal.Decimal()
star91: str = ""
star92: str = ""
subdetails: typing.List[WarehouseShipmentSubDetail] = dataclasses.field(
@ -236,7 +238,7 @@ class WarehouseShipmentDetail:
self.itmdes,
self.sau,
self.qty,
self.netpri,
self.gropri,
self.star91,
self.star92,
]

39
edi_944.py
View File

@ -2,7 +2,7 @@
"""
Consume a generic 944 file from 3PLs, and translate into a Sage X3
readable file - import template ZPTHI.
For Shadex we also need to reply with a 997
For Shandex we also need to reply with a 997
"""
# pylint: disable=too-many-instance-attributes
import dataclasses
@ -13,6 +13,7 @@ import pathlib
import re
import shutil
import typing
import pprint
import records # type: ignore
@ -22,25 +23,21 @@ 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_944_FILENAME_RE = re.compile(
r"\A 944_YAMAMOTOYAMA_ \S+ [.]edi \Z", re.X | re.M | re.S
)
TEST_FILE = THIS_DIRECTORY / "edi-testing" / "944_YAMAMOTOYAMA_765aaebb-06c4-4eea-8d2a-7dddf2fd9ec2.edi"#TODO remove this
TEST_DIR = THIS_DIRECTORY / "edi-testing"
def main():
"""
Do it!
"""
# if SHANDEX_944_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_944_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)
shutil.move(edi_filename, EDI_997_DIRECTORY / edi_filename.name)
def tokens_from_edi_file(
@ -100,9 +97,9 @@ def process_file(edi_filename: pathlib.Path):
subdetail,
)
time_stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
#TODO explode out the stojou lines
with yamamotoyama.x3_imports.open_import_file(
TEST_DIR / f"ZPTHI_{warehouse_receipt.sdhnum}_{time_stamp}.dat" #todo change folder back to IMPORTS_DIRECTORY
IMPORTS_DIRECTORY / f"ZPTHI_{warehouse_receipt.sdhnum}_{time_stamp}.dat"
) as import_file:
warehouse_receipt.output(import_file)
@ -130,13 +127,13 @@ class ReceiptSubDetail:
database.query(
"""
select
'S',
'A',
'S' [Code],
'A' [STA_0],
[STJ].[PCU_0],
cast(cast(-1*[STJ].[QTYSTU_0] as int) as nvarchar),
cast(cast(-1*[STJ].[QTYSTU_0] as int) as nvarchar) [QTYPCU_0],
[STJ].[LOT_0],
'',
''
'' [BPSLOT_0],
'' [SERNUM_0]
from [FY23TEST].[STOJOU] [STJ] --TODO change to PROD
where
[STJ].[VCRNUM_0] = :sdhnum
@ -163,7 +160,6 @@ class ReceiptSubDetail:
self.sta,
self.pcu,
self.qtypcu,
self.loc,
self.lot,
self.bpslot,
self.sernum,
@ -451,8 +447,17 @@ class Receipt:
shipment = detail.sdhnum
item = detail.itmref
for record in subdetail.stojous(shipment, item):
#output(subdetail.convert_to_strings())
output(record)
record_list = [
record['Code'],
record['STA_0'],
record['PCU_0'],
record['QTYPCU_0'],
record['LOT_0'],
record['BPSLOT_0'],
record['SERNUM_0']
]
#pprint.pprint(record_list)
output(record_list)
if __name__ == "__main__":

115
edi_947.py
View File

@ -19,6 +19,7 @@ import re
import shutil
import typing
import smtplib
import pprint
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
@ -31,39 +32,79 @@ 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_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
#TODO remove this and have a single production name? why is this one different
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
"AV" : 'RD',#Damaged in Transit
"55" : 'A',#Product Taken Off Hold
"AA" : 'A',#Physical Count
"05" : 'Q'#Product Put on Hold
}
DAMAGE_CODE_DESCRIPTIONS_MAPPING = {
"07" : "Product Dumped or Destroyed",
"AV" : "Damaged in Transit"
"AV" : "Damaged in Transit",
"55" : "Product Taken Off Hold",
"AA" : "Physical Count",
"05" : "Product Put on Hold"
}
#This transaction can also be used for inventory counts, which we will report on but not process
EMAIL_ONLY_CODES = ['AA']
#When we receive an EDI to change status, it will either be A or Q, the reverse of the earlier code
DAMAGE_CODE_SOURCE_MAPPING = {
"07" : 'A',#Product Dumped or Destroyed
"AV" : 'A',#Damaged in Transit
"55" : 'Q',#Product Taken Off Hold
"AA" : 'A',#Physical Count
"05" : 'A'#Product Put on Hold
}
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)
for edi_filename in X12_DIRECTORY.iterdir():
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)
shutil.move(edi_filename, EDI_997_DIRECTORY / edi_filename.name)
combine_zscs()
def combine_zscs():
"""
Collect all ZSCS imports into a single file for easy import.
"""
archive_directory = IMPORTS_DIRECTORY / "archive"
archive_directory.mkdir(exist_ok=True)
with (IMPORTS_DIRECTORY / "ZSCS.dat").open(
"w", encoding="utf-8", newline="\n"
) as combined_import_file:
for individual_import_filename in IMPORTS_DIRECTORY.glob(
"ZSCS_*.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(
@ -75,10 +116,9 @@ def 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
if fields[0] in {
"ISA",
"GS"
"W15",
"N1",
"N9",
"SE",
@ -99,21 +139,23 @@ def gtin_lookup(gtin):
[ITM].[EANCOD_0],
[ITM].[ZCASEUPC_0]
from [FY23TEST].[ITMMASTER] [ITM]--TODO change back to [PROD]
join [FY23TEST].[ITMFACILIT] [ITF]
on [ITM].[ITMREF_0] = [ITF].[ITMREF_0]
and [ITF].[STOFCY_0] = 'WON'
where
[ITM].[ZCASEUPC_0] = :zcaseupc
""",
zcaseupc=gtin,
).first()["ITMREF_0"]
def stock_movement_alert(gtin, qty, lot, status):
def stock_movement_alert(itmref, 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]}'
emailtext = f'Item: {itmref}\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')
@ -132,35 +174,40 @@ def process_file(edi_filename: pathlib.Path):
warehouse_stockchange.header.iptdat = datetime.datetime.strptime(
iptdat, "%Y%m%d"
).date()
if fields[0] == 'W15':
transaction_number = fields[2]
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)
product = gtin_lookup(gtin)
stock_movement_alert(product, qty, lot, status)
if status in EMAIL_ONLY_CODES:
return
warehouse_stockchange.header.vcrdes = DAMAGE_CODE_DESCRIPTIONS_MAPPING[status]
subdetail = StockChangeSubDetail(
qtypcu=int(qty),
qtystu=int(qty),
sta=DAMAGE_CODE_MAPPING[status],
pcu=uom
)
detail_line = StockChangeDetail(
vcrlin=vcrlin,
itmref=gtin,
itmref=product,
pcu=uom,
sta=DAMAGE_CODE_SOURCE_MAPPING[status],
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
IMPORTS_DIRECTORY / f"ZSCS_{transaction_number}_{time_stamp}.dat"
) 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
@ -211,10 +258,18 @@ class StockChangeSubDetail:
"""
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(
[
"S",
self.pcu,
fix_uom(self.pcu),
self.qtypcu,
self.qtystu,
self.loc,
@ -250,22 +305,25 @@ class StockChangeDetail:
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?
It matters which one we use, best attempt is to get the largest one available
"""
with yamamotoyama.get_connection('test') as db_connection:#todo remove 'test'
return db_connection.query(
"""
select
select top 1
[STO].[STOFCY_0],
[STO].[ITMREF_0],
[STO].[LOT_0],
[STO].[PALNUM_0]
[STO].[PALNUM_0],
[STO].[QTYSTU_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
order by
[STO].[QTYSTU_0] desc
""",
itmref=itmref,
lot=lot,
@ -282,9 +340,11 @@ class StockChangeDetail:
[ITM].[EANCOD_0],
[ITM].[ZCASEUPC_0]
from [FY23TEST].[ITMMASTER] [ITM]--TODO change back to [PROD]
join [FY23TEST].[ITMFACILIT] [ITF]
on [ITM].[ITMREF_0] = [ITF].[ITMREF_0]
and [ITF].[STOFCY_0] = 'WON'
where
[ITM].[ZCASEUPC_0] = :zcaseupc
""",
zcaseupc=gtin,
).first()["ITMREF_0"]
@ -316,12 +376,13 @@ class StockChangeDetail:
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),
self.itmref,
fix_uom(self.pcu),
self.pcustucoe,
self.sta,

548
edi_997_incoming.py
View File

@ -22,7 +22,7 @@ THIS_DIRECTORY = pathlib.Path(__file__).parent
X12_DIRECTORY = THIS_DIRECTORY / "incoming"
SHANDEX_997_FILENAME_RE = re.compile( #TODO FIX REGEX
r"\A 997_YAMAMOTOYAMA_ .* [.]ed \Z", re.X | re.M | re.S
r"\A 997_YAMAMOTOYAMA_ .* [.]edi \Z", re.X | re.M | re.S
)
@ -103,553 +103,7 @@ def process_file(edi_filename: pathlib.Path):
# IMPORTS_DIRECTORY / f"ZSHIP945S_{warehouse_receipt.sohnum}_{time_stamp}.dat"
# ) as import_file:
# warehouse_receipt.output(import_file)
@dataclasses.dataclass
class ReceiptSubDetail:
"""
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 ReceiptDetail:
"""
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[ReceiptSubDetail] = dataclasses.field(
default_factory=list
)
def append(self, subdetail: ReceiptSubDetail):
"""
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
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, ReceiptDetail):
return self.itmref == item.itmref
return False
def fill(self):
"""
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 ReceiptHeader:
"""
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 ReceiptDetailList:
"""
List of shipment details
"""
_details: typing.List[ReceiptDetail]
_item_set: typing.Set[str]
def __init__(self):
self._details = []
self._item_set = set()
def append(
self,
shipment_detail: ReceiptDetail,
shipment_subdetail: ReceiptSubDetail,
):
"""
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 Receipt:
"""
Warehosue shipment, both header & details
"""
header: ReceiptHeader
details: ReceiptDetailList
_sohnum: str
def __init__(self):
self.header = ReceiptHeader()
self._sohnum = ""
self.details = ReceiptDetailList()
def append(
self,
shipment_detail: ReceiptDetail,
shipment_subdetail: ReceiptSubDetail,
):
"""
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() as db_connection:
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 [PROD].[SORDER] as [SOH]
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__":

34
master_contoller.py
View File

@ -7,6 +7,12 @@ import shutil
import pathlib
import paramiko # type: ignore
import pprint
import edi_997_inbound
import edi_944
import edi_947
import edi_846
import edi_867
import update_shandex_dashboard
#import edi_943multi3pl #TODO remove 940 from this file
THIS_DIRECTORY = pathlib.Path(__file__).parent
@ -18,22 +24,28 @@ def main():
"""
Do it!
"""
#process all EDIs that start with us
#pick up files from Shandex
retrieve_x12_edi_files_shandex()#TODO turn on archiving
#process all EDIs that started with Shandex
edi_997_inbound.main()
edi_944.main()
edi_947.main()
#edi_846.main()
edi_867.main()
# process all EDIs that start with us
#edi_943-multi3pl.main()#TODO make this file Shandex only
#edi_997_outbound
#send them to Shandex
#send_x12_edi_files_shandex()#TODO set this up
#pick up files from Shandex
retrieve_x12_edi_files_shandex()
#process all EDIs that started with Shandex
#edi_997_outbound.main()
#edi_945.main()
# edi_944.main()
# edi_997_inbound.main()
#combine_zship945s()
#update dashboard
# update_shandex_dashboard.main()
# SL_SFTP_HOST = "s-8ade4d252cc44c50b.server.transfer.us-west-1.amazonaws.com"
# SL_SFTP_USERNAME = "yumiddleware2023"
SSH_DIRECTORY = THIS_DIRECTORY / "ssh"
SSH_KNOWN_HOSTS_FILE = str(SSH_DIRECTORY / "known_hosts")
SSH_KEY_FILENAME = str(SSH_DIRECTORY / "id_ed25519")