574 lines
16 KiB
Python
574 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Make a 943 in "X12" format without using the Javascript Middleware
|
|
developed by Tech4Biz.
|
|
|
|
A 943 is a replenishment shipping notice: we notify our 3PL that we have
|
|
sent them resupplies of our inventory.
|
|
"""
|
|
import datetime
|
|
import dataclasses
|
|
import typing
|
|
import pathlib
|
|
import pprint
|
|
|
|
import records
|
|
import yamamotoyama # type:ignore
|
|
|
|
|
|
SITE_MAPPING = {
|
|
'WNJ' : 'SOURCELOGISTICS',
|
|
'WCA' : 'SOURCELOGISTICS',
|
|
'WON' : 'SHANDEXTEST ' # TODO CHANGE TO SHANDEX, needs to be 15 characters
|
|
}
|
|
|
|
SHIPPING_CODE_MAPPING = {
|
|
'' : 'LT', #Default to LTL
|
|
'AIR' : 'AP', #Air package carrier
|
|
'DEL' : 'LT', #LTL and the default if mode is not entered
|
|
'GRN' : 'D', #Parcel post
|
|
'OUR' : 'SR', #Supplier truck
|
|
'P/U' : 'CE', #Pickup
|
|
'WCALL' : 'CE', #Pickup
|
|
}
|
|
|
|
THIS_DIRECTORY = pathlib.Path(__file__).parent
|
|
X12_SHANDEX = THIS_DIRECTORY / "outgoing"
|
|
|
|
# X12_SOURCELOGISTICS = THIS_DIRECTORY / "edi-testing" #test directories
|
|
# X12_SHANDEX = THIS_DIRECTORY / "edi-testing"
|
|
|
|
def main():
|
|
"""
|
|
Do it!
|
|
"""
|
|
with yamamotoyama.get_connection('test') as database:
|
|
shipments = list(get_shipments(database))
|
|
for shipment in shipments:
|
|
write_943(database, shipment)
|
|
|
|
|
|
class X12:
|
|
"""
|
|
X12 format parent class.
|
|
"""
|
|
|
|
@staticmethod
|
|
def line(items: typing.List[str]) -> str:
|
|
"""
|
|
Return X12 EDI line with * and ~
|
|
"""
|
|
|
|
def sanitize(thing: str) -> str:
|
|
for bad_character in ("*", "/", "&", "#", ","):
|
|
thing = thing.replace(bad_character, "")
|
|
return thing
|
|
|
|
return "*".join([sanitize(item) for item in items]) + "~"
|
|
|
|
def x12(self) -> str:
|
|
"""
|
|
X12 format.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@staticmethod
|
|
def control_number() -> int:
|
|
"""
|
|
Next EDI serial number
|
|
"""
|
|
filepath = pathlib.Path(__file__).with_suffix(".remember")
|
|
encoding = "utf-8"
|
|
newline = "\n"
|
|
try:
|
|
with filepath.open(
|
|
"r", encoding=encoding, newline=newline
|
|
) as remember_file:
|
|
number = int(remember_file.readline().rstrip("\n"))
|
|
except (OSError, ValueError):
|
|
number = 0
|
|
number += 1
|
|
with filepath.open("w", encoding=encoding, newline=newline) as remember_file:
|
|
remember_file.write(f"{number}\n")
|
|
return number
|
|
|
|
|
|
|
|
def write_943(database: records.Connection, shipment: str):
|
|
"""
|
|
Write out a 943 to a file
|
|
"""
|
|
now = datetime.datetime.now()
|
|
site = str(get_shipment_destination(database, shipment))
|
|
datestamp_string = now.strftime("%Y-%m-%d-%H-%M-%S")
|
|
with (X12_SHANDEX / f"{site}-{shipment}-{datestamp_string}-943.edi").open(
|
|
"w", encoding="utf-8", newline="\n"
|
|
) as x12_file:
|
|
output = x12_file.write
|
|
is_header_output = False
|
|
header = None
|
|
detail_count = 0
|
|
package_count = 0
|
|
for shipment_database_row in get_shipment(database, shipment):
|
|
ship_to_site = shipment_database_row.BPCORD_0
|
|
mdl = shipment_database_row.MDL_0
|
|
x12_ship_to = SITE_MAPPING[ship_to_site]
|
|
x12_mdl = SHIPPING_CODE_MAPPING[mdl]
|
|
header = ShipmentHeader(shipment_database_row)
|
|
detail = ShipmentDetail(shipment_database_row)
|
|
if not is_header_output:
|
|
output(header.x12(x12_ship_to,x12_mdl))
|
|
is_header_output = True
|
|
output(detail.x12())
|
|
detail_count += 1
|
|
package_count += detail.qtystu_0
|
|
if header:
|
|
output(header.footer(package_count, detail_count))
|
|
with database.transaction() as _:
|
|
database.query(
|
|
"""
|
|
update [FY23TEST].[SDELIVERY]
|
|
set [XX4S_943RDY_0] = 1
|
|
where [SOHNUM_0] = :shipment
|
|
""",
|
|
shipment=shipment,
|
|
)
|
|
order = get_order_for_shipment(database, shipment)
|
|
database.query(
|
|
"""
|
|
update [FY23TEST].[SORDER]
|
|
set [XX4S_UDF2_0] = :sent_message
|
|
where [SOHNUM_0] = :order
|
|
""",
|
|
order=order,
|
|
sent_message=f"943 Sent {datetime.date.today().isoformat()}",
|
|
)
|
|
|
|
def get_shipment_destination(database: records.Connection, shipment: str) -> str:
|
|
"""
|
|
Get the destination site
|
|
"""
|
|
return (
|
|
database.query(
|
|
"""
|
|
select
|
|
[SDH].[BPCORD_0]
|
|
from [FY23TEST].[SDELIVERY] as [SDH]
|
|
where
|
|
[SDH].[SDHNUM_0] = :shipment
|
|
""",
|
|
shipment=shipment,
|
|
)
|
|
.first()
|
|
.BPCORD_0
|
|
)
|
|
|
|
def get_order_for_shipment(database: records.Connection, shipment: str) -> str:
|
|
"""
|
|
What is the order for this shipment?
|
|
"""
|
|
return (
|
|
database.query(
|
|
"""
|
|
select
|
|
[SDH].[SOHNUM_0]
|
|
from [FY23TEST].[SDELIVERY] as [SDH]
|
|
where
|
|
[SDH].[SDHNUM_0] = :shipment
|
|
""",
|
|
shipment=shipment,
|
|
)
|
|
.first()
|
|
.SOHNUM_0
|
|
)
|
|
|
|
|
|
def get_shipments(database: records.Connection) -> typing.Iterator[str]:
|
|
"""
|
|
What have we shipped? Fetch from X3.
|
|
"""
|
|
for shipment_result in database.query(
|
|
"""
|
|
select
|
|
[SDH].[SDHNUM_0]
|
|
from [FY23TEST].[SDELIVERY] as [SDH]
|
|
join [FY23TEST].[SORDER] as [SOH]
|
|
on [SOH].[SOHNUM_0] = [SDH].[SOHNUM_0]
|
|
where
|
|
(
|
|
[SDH].[STOFCY_0] in ('PMW','WCA','WNJ')
|
|
and [SDH].[BPCORD_0] in ('WON')
|
|
and nullif([SDH].[MDL_0],'') is not null
|
|
and nullif([SDH].[BPTNUM_0],'') is not null
|
|
)
|
|
and [SDH].[SHIDAT_0] >= {d'2023-10-09'}
|
|
and (
|
|
[SOH].[XX4S_UDF2_0] = ''
|
|
or [SDH].[XX4S_943RDY_0] = 2
|
|
)
|
|
and (
|
|
[SDH].[CFMFLG_0] = 2
|
|
or [SDH].[YLICPLATE_0] <> ''
|
|
or [SDH].[XX4S_943RDY_0] = 2
|
|
)
|
|
"""
|
|
):
|
|
yield shipment_result.SDHNUM_0
|
|
|
|
|
|
def get_shipment(
|
|
database: records.Connection, shipment: str
|
|
) -> typing.Iterator[records.Record]:
|
|
"""
|
|
Get shipment information from X3 database.
|
|
"""
|
|
yield from database.query(
|
|
"""
|
|
select
|
|
[SDHNUM_0]
|
|
,[STOFCY_0]
|
|
,[STOFCY]
|
|
,[SDHCAT_0]
|
|
,[SALFCY_0]
|
|
,[SALFCY]
|
|
,[SOHNUM_0]
|
|
,[CUSORDREF_0]
|
|
,[BPCORD_0]
|
|
,[BPDNAM_0]
|
|
,[BPDADDLIG_0]
|
|
,[BPDADDLIG_1]
|
|
,[BPDADDLIG_2]
|
|
,[CTY_0]
|
|
,[SAT_0]
|
|
,[POSCOD_0]
|
|
,[CRY_0]
|
|
,[CRYNAM_0]
|
|
,[BPCINV_0]
|
|
,[BPINAM_0]
|
|
,[BPTNUM_0]
|
|
,[BPTNAM_0]
|
|
,[MDL_0]
|
|
,[SCAC_0]
|
|
,[YLICPLATE_0]
|
|
,[DLVDAT_0]
|
|
,[CREDAT_0]
|
|
,[UPDDAT_0]
|
|
,[CCE_0]
|
|
,[SOPLIN_0]
|
|
,[ITMREF_0]
|
|
,[ITMDES1_0]
|
|
,[ZCASEUPC_0]
|
|
,[EANCOD_0]
|
|
,[STU_0]
|
|
,[QTYSTU_0]
|
|
,[LotQty]
|
|
,[LOT_0]
|
|
,[NETWEI_0]
|
|
,[GROWEI_0]
|
|
,[CFMFLG_0]
|
|
,[SHIDAT_0]
|
|
from [FY23TEST].[zyumiddleware_shipment] as [SDH]
|
|
where
|
|
[SDH].[SDHNUM_0] = :shipment
|
|
""",
|
|
shipment=shipment,
|
|
)
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class ShipmentHeader(X12):
|
|
"""
|
|
Header
|
|
"""
|
|
|
|
sdhnum: str
|
|
stofcy: str
|
|
stofcy: str
|
|
sdhcat: str
|
|
salfcy: str
|
|
salfcy: str
|
|
sohnum: str
|
|
cusordref: str
|
|
bpcord: str
|
|
bpdnam: str
|
|
bpdaddlig: str
|
|
bpdaddlig_1: str
|
|
bpdaddlig_2: str
|
|
cty: str
|
|
sat: str
|
|
poscod: str
|
|
cry: str
|
|
crynam: str
|
|
bpcinv: str
|
|
bpinam: str
|
|
bptnum: str
|
|
bptnam: str
|
|
scac: str
|
|
ylicplate: str
|
|
dlvdat: datetime.date
|
|
credat: datetime.date
|
|
upddat: datetime.date
|
|
cce: str
|
|
short_control_number: str
|
|
interchange_control_number: str
|
|
header_segments: int
|
|
footer_segments: int
|
|
|
|
def __init__(self, database_row: records.Record):
|
|
self.sdhnum = database_row.SDHNUM_0
|
|
self.stofcy = database_row.STOFCY_0
|
|
self.sdhcat = database_row.SDHCAT_0
|
|
self.salfcy = database_row.SALFCY_0
|
|
self.salfcy = database_row.SALFCY_0
|
|
self.sohnum = database_row.SOHNUM_0
|
|
self.cusordref = database_row.CUSORDREF_0
|
|
self.bpcord = database_row.BPCORD_0
|
|
self.bpdnam = database_row.BPDNAM_0
|
|
self.bpdaddlig = database_row.BPDADDLIG_0
|
|
self.bpdaddlig_1 = database_row.BPDADDLIG_1
|
|
self.bpdaddlig_2 = database_row.BPDADDLIG_2
|
|
self.cty = database_row.CTY_0
|
|
self.sat = database_row.SAT_0
|
|
self.poscod = database_row.POSCOD_0
|
|
self.cry = database_row.CRY_0
|
|
self.crynam = database_row.CRYNAM_0
|
|
self.bpcinv = database_row.BPCINV_0
|
|
self.bpinam = database_row.BPINAM_0
|
|
self.bptnum = database_row.BPTNUM_0
|
|
self.bptnam = database_row.BPTNAM_0
|
|
self.scac = database_row.SCAC_0
|
|
self.ylicplate = database_row.YLICPLATE_0
|
|
self.dlvdat = database_row.DLVDAT_0
|
|
self.credat = database_row.CREDAT_0
|
|
self.upddat = database_row.UPDDAT_0
|
|
self.cce = database_row.CCE_0
|
|
self.shidat = database_row.SHIDAT_0
|
|
raw_control_number = self.control_number()
|
|
self.short_control_number = f"{raw_control_number:04}"
|
|
self.interchange_control_number = (
|
|
f"{raw_control_number:09}" # Format to 9 characters
|
|
)
|
|
self.now = datetime.date.today()
|
|
self.date = self.now.strftime("%y%m%d")
|
|
self.long_date = self.now.strftime("%Y%m%d")
|
|
self.time = self.now.strftime("%H%m")
|
|
self.header_segments = 10
|
|
self.footer_segments = 2
|
|
|
|
def x12(self, receiver_id, mdl) -> str:
|
|
return "".join(
|
|
[
|
|
f"ISA*00* *00* *ZZ*YAMAMOTOYAMA *ZZ*{receiver_id}*",
|
|
self.date,
|
|
"*",
|
|
self.time,
|
|
"*U*00401*",
|
|
self.interchange_control_number,
|
|
"*0*P*>~",
|
|
self.line(
|
|
[
|
|
"GS",
|
|
"OW",#should this be AR? Shandex okayed "OW"
|
|
"YAMAMOTOYAMA",
|
|
f"{receiver_id}",
|
|
self.long_date,
|
|
self.time,
|
|
self.interchange_control_number,
|
|
"X",
|
|
"004010",
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"ST",
|
|
"943",
|
|
self.short_control_number,
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"W06",
|
|
"N",
|
|
self.sohnum,
|
|
self.shidat.strftime("%Y%m%d"),
|
|
self.sdhnum,
|
|
"",
|
|
"",
|
|
"",
|
|
"",
|
|
"",
|
|
"",
|
|
"AS",
|
|
]
|
|
),
|
|
"N1*RE*Pomona Wholesale*1*008930755~",
|
|
"N4*Pomona*CA*91768*US~",
|
|
self.line(
|
|
[
|
|
"N1",
|
|
"ST", # Ship to
|
|
self.bpdnam,
|
|
"53", # Building
|
|
self.bpcord,
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"N3",
|
|
self.bpdaddlig,
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"N4",
|
|
self.cty,
|
|
self.sat,
|
|
self.poscod,
|
|
self.cry,
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"N9",
|
|
"PO",
|
|
self.cusordref,
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"G62",
|
|
"17",
|
|
self.dlvdat.strftime("%Y%m%d"),
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"W27",
|
|
f"{mdl}",
|
|
self.scac,
|
|
]
|
|
),
|
|
]
|
|
)
|
|
|
|
def footer(self, package_count: int, detail_count: int):
|
|
"""
|
|
End footer
|
|
"""
|
|
segment_count = self.header_segments + (detail_count * 3) + self.footer_segments
|
|
return "".join(
|
|
[
|
|
self.line(
|
|
[
|
|
"W03",
|
|
str(int(package_count)),
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"SE",
|
|
str(segment_count),
|
|
self.short_control_number,
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"GE",
|
|
"1",
|
|
self.interchange_control_number,
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"IEA",
|
|
"1",
|
|
self.interchange_control_number,
|
|
]
|
|
),
|
|
]
|
|
)
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class ShipmentDetail(X12):
|
|
"""
|
|
Shipment detail.
|
|
"""
|
|
|
|
soplin_0: str
|
|
itmref_0: str
|
|
itmdes1_0: str
|
|
stu_0: str
|
|
qtystu_0: int
|
|
lot_0: str
|
|
stofcy_0: str
|
|
zcaseupc_0: str
|
|
eancod_0: str
|
|
gtin_or_upc_code: str
|
|
gtin_or_upc_marker: str
|
|
|
|
def __init__(self, db_record: records.Record):
|
|
self.soplin_0 = str(db_record.SOPLIN_0)
|
|
self.itmref_0 = db_record.ITMREF_0
|
|
self.itmdes1_0 = db_record.ITMDES1_0
|
|
self.stu_0 = db_record.STU_0
|
|
self.qtystu_0 = db_record.LotQty
|
|
self.lot_0 = db_record.LOT_0
|
|
self.stofcy_0 = db_record.STOFCY_0
|
|
if self.stofcy_0 == 'WON' and self.stu_0 == 'CS': #Shadex requires CA for cases
|
|
self.stu_0 = 'CA'
|
|
self.zcaseupc_0 = db_record.ZCASEUPC_0
|
|
self.eancod_0 = db_record.EANCOD_0
|
|
self.gtin_or_upc_marker = 'UK'
|
|
if self.stu_0 == 'CS':
|
|
self.gtin_or_upc_code = self.zcaseupc_0
|
|
else:
|
|
self.gtin_or_upc_code = self.eancod_0
|
|
|
|
def x12(self) -> str:
|
|
"""
|
|
Format in X12
|
|
"""
|
|
|
|
return "".join(
|
|
[
|
|
self.line(
|
|
[
|
|
"W04",
|
|
str(int(self.qtystu_0)),
|
|
self.stu_0,
|
|
"",
|
|
"VN", # Vendor's (Seller's) Item Number
|
|
self.itmref_0,
|
|
"LT", # Lot number
|
|
self.lot_0,
|
|
"",
|
|
"",
|
|
"",
|
|
"",
|
|
"",
|
|
"",
|
|
self.gtin_or_upc_marker,
|
|
self.gtin_or_upc_code,#W04-15
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"G69",
|
|
self.itmdes1_0,
|
|
]
|
|
),
|
|
self.line(
|
|
[
|
|
"N9",
|
|
"LI",
|
|
self.soplin_0,
|
|
]
|
|
),
|
|
]
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|