546 lines
16 KiB
Python
546 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Convert an 856 file from Source Logistics into an ASN file to upload to TrueCommerce.
|
|
Note: Tare is not described in Souece's file, they send Pallets as Packages, and Packages as Items
|
|
"""
|
|
import sys
|
|
import csv
|
|
import dataclasses
|
|
import datetime
|
|
import decimal
|
|
import functools
|
|
import pathlib
|
|
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
|
|
|
|
OVERRIDE_FLAG = True
|
|
|
|
CUSTOMER_MAP = {
|
|
" Amazon - Golden State - Moreno Vly": "AMAZ0026",
|
|
" Amazon Canada - Bolton YYZ7": "AMAZ0100",
|
|
" Amazon Canada - Brampton YYZ4": "AMAZ0049",
|
|
" Amazon Canada - Calgary YYC1": "AMAZ0090",
|
|
" Amazon Canada - Delta YVR2": "AMAZ0014",
|
|
" Amazon Canada - Mount Hope YHM1": "AMAZ0169",
|
|
" Amazon Canada - Nepean": "AMAZ0176",
|
|
" Amazon Canada - Richmond YXX2": "AMAZ0150",
|
|
" Amazon Canada - Tsawwassen YVR4": "AMAZ0099",
|
|
" Amazon.com - Bloomington SBD1": "AMAZ0143",
|
|
" Amazon.com - Fontana LAX9": "AMAZ0092",
|
|
" Amazon.com - Madison AL HSV1": "AMAZ0173",
|
|
" Amazon.com - Rialto CA": "AMAZ0072",
|
|
" Amazon.com - Somerset NJ TEB9": "AMAZ0156",
|
|
" Amazon.com - Stockton CA": "AMAZ0084",
|
|
" Amazon.com - Stockton SCK1": "AMAZ0104",
|
|
" Amazon.com - Stockton SCK4": "AMAZ0154",
|
|
" Amazon.com Serv VGT2 Las Vegas": "AMAZ0180",
|
|
}
|
|
|
|
RECORD_SEPARATOR = "\n"
|
|
UNIT_SEPARATOR = "\t"
|
|
|
|
THIS_DIRECTORY = pathlib.Path(__file__).parent
|
|
ASN_DIRECTORY = THIS_DIRECTORY
|
|
SOURCE_FILENAME_GLOB = "YAMAMOTOYAMA*.edi"
|
|
|
|
DECIMAL_FORMAT = decimal.Decimal("0.01")
|
|
|
|
UOM_MAP = {
|
|
"BX": "Box",
|
|
"CS": "Case",
|
|
"EA": "Each",
|
|
"GM": "Gram",
|
|
"RL": "Roll",
|
|
}
|
|
|
|
SSCC_ERRORS = []
|
|
|
|
INSERT_STATEMENT = """\
|
|
execute [PROD].[insert_sl_856_data]
|
|
:customer,
|
|
:shipment_id,
|
|
:ship_date,
|
|
:sdd,
|
|
:scac,
|
|
:bol,
|
|
:arn,
|
|
:pro,
|
|
:pallets,
|
|
:growei,
|
|
:po,
|
|
:cty,
|
|
:sat,
|
|
:poscod,
|
|
:sscc,
|
|
:wei,
|
|
:weu,
|
|
:itmrefbpc,
|
|
:lot,
|
|
:itmdes1,
|
|
:credat;
|
|
"""
|
|
|
|
|
|
STASH_GS1 = '0776520'
|
|
|
|
|
|
def sscc_check(sscc):#TODO recurse through pallets and packages
|
|
stashtea_gs1 = '0776520'
|
|
application_id = sscc[:2]
|
|
extension = sscc[2]
|
|
gs1_prefix = sscc[3:10]
|
|
serial = sscc[10:19]
|
|
check_digit = sscc[19]
|
|
if gs1_prefix != stashtea_gs1:
|
|
pprint.pprint(f'{sscc} does not match the Stash GS1 code.')
|
|
|
|
|
|
class Pallet:
|
|
def __init__(self):
|
|
self.pallet_sscc = ''
|
|
self.package_list = []
|
|
|
|
|
|
class Package:
|
|
def __init__(self):
|
|
self.package_sscc = ''
|
|
self.sku = ''
|
|
self.des = ''
|
|
self.asin = ''
|
|
self.weight = decimal.Decimal()
|
|
self.weight_unit = ''
|
|
self.qty = 1
|
|
self.lot = ''
|
|
self.gtin = ''
|
|
self.upc = ''
|
|
self.uom: str
|
|
self.shelf_life = 0
|
|
self.shelf_life_uom = ''
|
|
self.expiration_date_str = ''
|
|
self.manufacture_date = ''
|
|
|
|
@dataclasses.dataclass
|
|
class ShipmentInformationRecord:
|
|
# pylint: disable=too-many-instance-attributes
|
|
"""
|
|
One ASN record.
|
|
"""
|
|
customer: str
|
|
shipment_id: str
|
|
ship_date: datetime.date
|
|
scheduled_delivery_date: datetime.date
|
|
scheduled_delivery_date_str: str
|
|
scac: str
|
|
bill_of_lading: str
|
|
arn: str
|
|
carrier_pro_number: str
|
|
number_of_cartons_shipped: int
|
|
to_code: str
|
|
p_o: str
|
|
sohnum: str
|
|
unit_count: str
|
|
weight: decimal.Decimal
|
|
weight_unit: str
|
|
description: str
|
|
quantity: int
|
|
total_gross_weight: decimal.Decimal
|
|
to_name: str
|
|
to_address_1: str
|
|
to_address_2: str
|
|
to_city: str
|
|
to_state: str
|
|
to_zip: str
|
|
from_code: str
|
|
from_address_1: str
|
|
from_address_2: str
|
|
from_city: str
|
|
from_state: str
|
|
from_zip: str
|
|
number_of_cartons_shipped_str: str
|
|
pallet_list = []
|
|
|
|
|
|
def fill_information(self, database):
|
|
try:
|
|
(
|
|
sales_order,
|
|
self.to_code,
|
|
self.to_name,
|
|
self.to_address_1,
|
|
self.to_address_2,
|
|
self.to_city,
|
|
self.to_state,
|
|
self.to_zip,
|
|
self.customer,
|
|
) = get_order_information(database, self.sohnum).values()
|
|
except AttributeError:
|
|
print(f"Difficulty with {self.sohnum}")
|
|
raise
|
|
if self.scheduled_delivery_date_str is None or self.scheduled_delivery_date_str == '':
|
|
self.scheduled_delivery_date_str = self.ship_date
|
|
|
|
|
|
def __init__(self, database: records.Connection):
|
|
customer = '',
|
|
self.shipment_id = ''
|
|
ship_date_str = ''
|
|
self.scheduled_delivery_date_str = ''
|
|
self.scac = ''
|
|
self.bill_of_lading = ''
|
|
self.arn = ''
|
|
self.carrier_pro_number = ''
|
|
number_of_cartons_shipped_str = '0'
|
|
total_gross_weight_str = '0'
|
|
self.p_o = ''
|
|
self.sohnum = ''
|
|
self.from_city = ''
|
|
self.from_state = ''
|
|
self.from_zip = ''
|
|
weight_str = '0'
|
|
self.weight_unit = ''
|
|
self.asin = ''
|
|
self.lot = ''
|
|
self.description = ''
|
|
self.customer = CUSTOMER_MAP.get(customer, customer)
|
|
self.from_code = ''
|
|
self.number_of_cartons_shipped = int(number_of_cartons_shipped_str)
|
|
self.total_gross_weight = decimal.Decimal(total_gross_weight_str).quantize(
|
|
DECIMAL_FORMAT
|
|
)
|
|
self.pallet_list = []
|
|
|
|
def __bool__(self):
|
|
return bool(self.customer)
|
|
|
|
|
|
class ASNFile:
|
|
"""
|
|
Create an ASN flat-file for TrueCommerce.
|
|
"""
|
|
|
|
def __init__(self, file_path: pathlib.Path):
|
|
# pylint: disable=consider-using-with
|
|
self._file = file_path.open("xt", encoding="us-ascii", newline=RECORD_SEPARATOR)
|
|
|
|
def print_row(self, *output_row: str):
|
|
"""
|
|
Print one row, simply.
|
|
"""
|
|
for i, item in enumerate(output_row):
|
|
assert isinstance(item, str), f"not str {i} “{item}”"
|
|
print(UNIT_SEPARATOR.join(output_row), end=RECORD_SEPARATOR, file=self._file)
|
|
|
|
def close(self):
|
|
"""
|
|
Close resources.
|
|
"""
|
|
self._file.close()
|
|
|
|
|
|
def main():
|
|
"""
|
|
Do it!
|
|
"""
|
|
ASN_DIRECTORY.mkdir(parents=True, exist_ok=True)
|
|
with yamamotoyama.get_connection() as database:
|
|
for file_path in ASN_DIRECTORY.glob(SOURCE_FILENAME_GLOB):
|
|
convert(file_path, database)
|
|
|
|
|
|
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",
|
|
"GS",
|
|
"ST",
|
|
"BSN",
|
|
|
|
}:
|
|
continue
|
|
yield fields
|
|
|
|
|
|
def convert(file_path: pathlib.Path, database):
|
|
"""
|
|
Process a file
|
|
"""
|
|
prev_shipment_id = None
|
|
asn_file = None
|
|
#TODO where does the PRO come from?
|
|
record = None
|
|
record = ShipmentInformationRecord(database)
|
|
|
|
#The x12 format does not state which address an N3 or N4 belongs to
|
|
last_seen_address_type = ''
|
|
last_seen_hierarchical_level = ''
|
|
|
|
pallet = None
|
|
package = None
|
|
for fields in tokens_from_edi_file(file_path):
|
|
#TD1*PLT*2****G*1605.50999999997*LB
|
|
if fields[0] == 'TD1':
|
|
record.growei = fields[7]
|
|
record.weight_unit = fields[8]
|
|
if fields[1] == 'PLT':
|
|
record.number_of_cartons_shipped_str = fields[2]
|
|
#TD5**2*NRCL**TL
|
|
elif fields[0] == 'TD5':
|
|
record.scac = fields[3]
|
|
elif fields[0] == 'REF' and fields[1] == '2I':
|
|
record.carrier_pro_number = fields[2]
|
|
elif fields[0] == 'DTM' and fields[1] == '011':
|
|
record.ship_date = fields[2]
|
|
elif fields[0] == 'N1' and fields[1] == 'ST':
|
|
last_seen_address_type = 'ST'
|
|
elif fields[0] == 'N1' and fields[1] == 'SF':
|
|
last_seen_address_type = 'SF'
|
|
elif fields[0] == 'N3':
|
|
if last_seen_address_type == 'ST':
|
|
record.to_address_1 = fields[1]
|
|
record.to_address_2 = ''
|
|
elif last_seen_address_type == 'SF':
|
|
record.from_address_1 = fields[1]
|
|
record.from_address_2 = ''
|
|
elif fields[0] == 'N4':
|
|
if last_seen_address_type == 'ST':
|
|
record.to_city = fields[1]
|
|
record.to_state = fields[2]
|
|
record.to_zip = fields[3]
|
|
elif last_seen_address_type == 'SF':
|
|
record.from_city = fields[1]
|
|
record.from_state = fields[2]
|
|
record.from_zip = fields[3]
|
|
elif fields[0] == 'PRF':
|
|
record.p_o = fields[1]
|
|
elif fields[0] == 'REF' and fields[1] == 'OI':
|
|
record.shipment_id = fields[2]
|
|
elif fields[0] == 'REF' and fields[1] == 'ZZ':#SL claimed this is a BOL, what is it?
|
|
record.sohnum = fields[2]
|
|
elif fields[0] == 'REF' and fields[1] == 'BM':#SL claimed this is a BOL, what is it?
|
|
record.bill_of_lading = fields[2]
|
|
elif fields[0] == 'HL':#create new hierarchical object and apppend
|
|
last_seen_hierarchical_level = fields[3]
|
|
if last_seen_hierarchical_level == 'P':#Source does not send this as a Tare
|
|
if pallet:
|
|
record.pallet_list.append(pallet)
|
|
pallet = Pallet()
|
|
elif last_seen_hierarchical_level == 'I':#Source does not send this as a Package
|
|
if package:
|
|
pallet.package_list.append(package)
|
|
package = Package()
|
|
elif fields[0] == 'MAN':
|
|
sscc = fields[2]
|
|
if last_seen_hierarchical_level == 'P':
|
|
pallet.pallet_sscc = sscc
|
|
elif last_seen_hierarchical_level == 'I':
|
|
package.package_sscc = sscc
|
|
sscc_check(sscc)
|
|
elif fields[0] == 'LIN':
|
|
package.sku = fields[3]
|
|
package.asin = fields[3]
|
|
elif fields[0] == 'SN1':
|
|
package.qty = fields[2]
|
|
package.uom = fields[3]
|
|
elif fields[0] == 'PID':
|
|
package.des = fields[5]
|
|
elif fields[0] == 'REF' and fields[1] == 'LT':
|
|
package.lot = fields[2]
|
|
elif fields[0] == 'DTM' and fields[1] == '036':
|
|
package.expiration_date_str = fields[2][:4]+'-'+fields[2][4:6]+'-'+fields[2][-2:]
|
|
elif fields[0] == 'PO4':
|
|
package.weight = fields[6]
|
|
package.weight_unit = fields[7]
|
|
record.pallet_list.append(pallet)#add the last pallet onto record
|
|
record.fill_information(database)
|
|
|
|
asn_file = ASNFile(asn_filename(record.shipment_id))
|
|
asn_file.print_row(
|
|
"SHP-tar", record.customer, "Original", record.shipment_id
|
|
)
|
|
asn_file.print_row(
|
|
"SHP-shipping",
|
|
record.ship_date,#record.ship_date.isoformat(),
|
|
record.scheduled_delivery_date_str,#.isoformat(),
|
|
record.scac,
|
|
record.bill_of_lading,
|
|
record.carrier_pro_number,#TODO
|
|
str(record.number_of_cartons_shipped_str),#this is actually the number of pallets shipped
|
|
str(record.growei),
|
|
record.arn,#this must still be filled outby hand
|
|
)
|
|
asn_file.print_row(
|
|
"SHP-address-from",
|
|
record.from_code,
|
|
"Stash Tea Company",
|
|
record.from_address_1,
|
|
record.from_address_2,
|
|
record.from_city,
|
|
record.from_state,
|
|
record.from_zip,
|
|
"US",
|
|
"SAXR9",
|
|
)
|
|
asn_file.print_row(
|
|
"SHP-address-to",
|
|
record.to_code,
|
|
record.to_name,
|
|
record.to_address_1,
|
|
record.to_address_2,
|
|
record.to_city,
|
|
record.to_state,
|
|
record.to_zip,
|
|
)
|
|
asn_file.print_row("SHP-address-vendor", "SAXR9")
|
|
asn_file.print_row("SHP-address-dc", record.to_code)
|
|
asn_file.print_row("ORD", record.p_o, record.to_code,'','')
|
|
|
|
|
|
for pallet in record.pallet_list:
|
|
asn_file.print_row('TAR',pallet.pallet_sscc)
|
|
for package in pallet.package_list:
|
|
line_info = get_line_information(database, record.sohnum, package.sku, package.lot)
|
|
asn_file.print_row(
|
|
"PKG",
|
|
package.package_sscc,
|
|
package.qty,
|
|
str(package.weight),
|
|
package.weight_unit,
|
|
"",
|
|
)
|
|
asn_file.print_row(
|
|
"ITM",
|
|
line_info['XX4S_LUDF3_0'],#package.asin,
|
|
line_info['UPC'],#package.upc,
|
|
line_info['ISBN'],# package.isbn,
|
|
line_info['GTIN'],# package.gtin,
|
|
package.sku,
|
|
line_info['UPC'],#package.upc,
|
|
package.lot,
|
|
line_info['ITMDES1_0'],#package.description,
|
|
str(package.qty),
|
|
line_info['SAU_0'],#package.uom,#TODO spell out Case from CS
|
|
str(line_info['SHL_0']),#str(package.shelf_life),
|
|
line_info['SHLUOM'],#package.shelf_life_uom,
|
|
package.expiration_date_str,
|
|
line_info['LOTCREDAT_0'].strftime('%Y-%m-%d'),#package.manufacture_date.isoformat(),
|
|
)
|
|
if asn_file is not None:
|
|
asn_file.close()
|
|
|
|
|
|
def asn_filename(shipment: str) -> pathlib.Path:
|
|
"""
|
|
Filename for ASN flat file, where TM will be able to find it.
|
|
"""
|
|
# return pathlib.Path(rf"E:\edi\TransactionManager\Import\{shipment}.txt")
|
|
for_truecommerce = ASN_DIRECTORY / "for_truecommerce"
|
|
for_truecommerce.mkdir(parents=True, exist_ok=True)
|
|
return for_truecommerce / f"{shipment}.txt"
|
|
|
|
|
|
@functools.lru_cache(5)
|
|
def get_line_information(
|
|
database: records.Connection, sales_order: str, itmref: str, lot: str
|
|
) -> records.Record:
|
|
"""
|
|
Get detail information about this shipment line:
|
|
- Stash SKU
|
|
- Pack-size
|
|
- UPC, ISBN, GTIN
|
|
- Unit of measure
|
|
- Shelf date
|
|
- Manufacture date
|
|
"""
|
|
results = database.query(
|
|
"""
|
|
select
|
|
[SOP].[ITMREF_0]
|
|
,[SOP].[ITMDES1_0]
|
|
,[SOP].[XX4S_LUDF3_0]
|
|
,[ITM].[YPACKSIZECS_0]
|
|
,'' as [UPC]
|
|
,'' as [ISBN]
|
|
,'' as [GTIN]
|
|
,[ITM].[SAU_0]
|
|
,coalesce([STL].[SHL_0], 100) as [SHL_0]
|
|
,case coalesce([STL].[SHLUOM_0], 2)
|
|
when 2 then 'Months'
|
|
else 'Days'
|
|
end as [SHLUOM] -- Shelf life unit of measure
|
|
,coalesce([STL].[LOTCREDAT_0], {d'1753-01-01'}) as [LOTCREDAT_0]
|
|
from [PROD].[SORDERP] as [SOP]
|
|
join [PROD].[ITMMASTER] as [ITM]
|
|
on [ITM].[ITMREF_0] = [SOP].[ITMREF_0]
|
|
left join [PROD].[STOLOT] as [STL]
|
|
on [STL].[ITMREF_0] = [SOP].[ITMREF_0]
|
|
and [STL].[LOT_0] = :lot
|
|
and [STL].[SLO_0] = ''
|
|
where
|
|
[SOP].[SOHNUM_0] = :sales_order
|
|
and [SOP].[ITMREF_0] = :itmref
|
|
""",
|
|
sales_order=sales_order,
|
|
itmref=itmref,
|
|
lot=lot,
|
|
)
|
|
return results.first()
|
|
|
|
|
|
@functools.lru_cache(5)
|
|
def get_order_information(
|
|
database: records.Connection, sohnum: str
|
|
) -> records.Record:
|
|
"""
|
|
Get order information:
|
|
- Stash Sales Order number
|
|
- Amazon EDI ship-to-code
|
|
- Delivery name & address
|
|
"""
|
|
results = database.query(
|
|
"""
|
|
select
|
|
[SOH].[SOHNUM_0]
|
|
,[SOH].[XX4S_DCCODE_0] -- Ship to code
|
|
,[SOH].[BPDNAM_0]
|
|
,[SOH].[BPDADDLIG_0]
|
|
,[SOH].[BPDADDLIG_1]
|
|
,[SOH].[BPDCTY_0]
|
|
,[SOH].[BPDSAT_0]
|
|
,[SOH].[BPDPOSCOD_0]
|
|
,[SOH].[BPCORD_0]
|
|
from [PROD].[SORDER] as [SOH]
|
|
where
|
|
[SOH].[SOHNUM_0] = :sohnum
|
|
and [SOH].[STOFCY_0] not in (
|
|
'PMW'
|
|
,'YOA'
|
|
)
|
|
""",
|
|
sohnum=sohnum,
|
|
)
|
|
return results.first()
|
|
|
|
|
|
def parse_source_date(source_date_str: str) -> datetime.date:
|
|
"""
|
|
Parse a Source Logistics date string into a Python date.
|
|
"""
|
|
return datetime.datetime.strptime(source_date_str, "%m/%d/%Y").date()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|