#!/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", "CA": "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 if len(sscc) != 20: pprint.pprint(f'{sscc} is not the correct length of an SSCC (20)') else: 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): pprint.pprint(file_path) 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 = '' 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) if line_info['XX4S_LUDF3_0'].strip() == '': pprint.pprint(f'Blank ASIN detected: Shipment:{record.shipment_id} PO:{record.p_o} Item:{package.sku} {package.package_sscc}') 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), UOM_MAP[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] ,case [ITM].[SAU_0] when 'CS' then 'CA' else [ITM].[SAU_0] end [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()