Initial commit
commit
d1c161deb9
|
|
@ -0,0 +1,3 @@
|
|||
Product,ProductDescription,QtyShip,Version,Expiration,OrderID,PO,REF,OrderDate,ShipTo,ShipAdd1,ShipAdd2,ShipCity,ShipStat,ShipZip,Tracking,ShipDate,Weight,PackageID,Price,ShippingCharge
|
||||
64416,18 Ct ASIAN PEAR HARMONY TEA,1,2262024,,484705,,5.27534E+12,2/26/2024,Cody Echauri,13391 Southwest Morgan Road,NULL,Sherwood,OR,97140,1Z1231321321TEST,2/26/24 5:30 PM,10,242824,3.95,10
|
||||
08216,6/18 ASIAN PEAR HARMONY TEA,1,2262024,,484705,,5.27534E+12,2/26/2024,Cody Echauri,13391 Southwest Morgan Road,NULL,Sherwood,OR,97140,1Z1231321321TEST,2/26/24 5:30 PM,10,242824,22.5,10
|
||||
|
|
|
@ -0,0 +1,22 @@
|
|||
Goal:
|
||||
Take shipping files from Source and turn them into X3 Sales Orders via ZSHPORD
|
||||
Use same files and turn them into X3 Deliveries for the above orders via ZSDHTRK
|
||||
|
||||
Problems:
|
||||
Source currently doesn't have site information or tax information on their export
|
||||
Until we have the actual data we'll build a skeleton.
|
||||
|
||||
How does Source handle multiple lots on a line?
|
||||
|
||||
---------------------------------------------------------------------------------------
|
||||
|
||||
step 1 read in file
|
||||
2. create import record for orders
|
||||
3. drop folder X3 can reach, schedule recurring import
|
||||
4. create import record for shipment
|
||||
5. do the same thing for step 3
|
||||
6. rearrange files on ftpo so that they are processed
|
||||
|
||||
----------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
Product,ProductDescription,QtyShip,Version,Expiration,OrderID,PO,REF,OrderDate,ShipTo,ShipAdd1,ShipAdd2,ShipCity,ShipStat,ShipZip,Tracking,ShipDate,Weight,PackageID,Price,ShippingCharge
|
||||
64416,18 Ct ASIAN PEAR HARMONY TEA,1,2262024,,484705,,5.27534E+12,2/26/2024,Cody Echauri,13391 Southwest Morgan Road,NULL,Sherwood,OR,97140,1Z1231321321TEST,2/26/24 5:30 PM,10,242824,3.95,10
|
||||
08216,6/18 ASIAN PEAR HARMONY TEA,1,2262024,,484705,,5.27534E+12,2/26/2024,Cody Echauri,13391 Southwest Morgan Road,NULL,Sherwood,OR,97140,1Z1231321321TEST,2/26/24 5:30 PM,10,242824,22.5,10
|
||||
|
|
|
@ -0,0 +1,389 @@
|
|||
#!/usr/bin/env python3
|
||||
import csv
|
||||
import pprint
|
||||
import dataclasses
|
||||
import datetime
|
||||
import decimal
|
||||
import functools
|
||||
import pathlib
|
||||
import re
|
||||
import shutil
|
||||
import typing
|
||||
|
||||
import records # type: ignore
|
||||
|
||||
import yamamotoyama # type: ignore
|
||||
import yamamotoyama.x3_imports # type: ignore
|
||||
|
||||
SFTP_HOST = "s-8ade4d252cc44c50b.server.transfer.us-west-1.amazonaws.com"
|
||||
SFTP_USERNAME = "yumiddleware2023"
|
||||
SSH_DIRECTORY = edi_940.THIS_DIRECTORY / "ssh" #TODO fixme
|
||||
SSH_KNOWN_HOSTS_FILE = str(SSH_DIRECTORY / "known_hosts")
|
||||
SSH_KEY_FILENAME = str(SSH_DIRECTORY / "id_ed25519")
|
||||
|
||||
THIS_DIRECTORY = pathlib.Path(__file__).parent
|
||||
INCOMING_DIRECTORY = THIS_DIRECTORY / "incoming_orders"
|
||||
SOH_IMPORT_DIRECTORY = THIS_DIRECTORY / "to_import_SOH"
|
||||
|
||||
def main():
|
||||
retrieve_x12_edi_files()
|
||||
for file in INCOMING_DIRECTORY.iterdir():
|
||||
process_files()
|
||||
shutil.move(file, INCOMING_DIRECTORY / "archive" / file.name)
|
||||
combine_zshpords()
|
||||
#TODO determine X3 processing schedule
|
||||
|
||||
def sftp_server() -> paramiko.SFTPClient:
|
||||
with paramiko.SSHClient() as ssh_client:
|
||||
ssh_client.load_system_host_keys()
|
||||
ssh_client.load_host_keys(SSH_KNOWN_HOSTS_FILE)
|
||||
ssh_client.set_missing_host_key_policy(paramiko.client.RejectPolicy)
|
||||
ssh_client.connect(
|
||||
hostname=SFTP_HOST, username=SFTP_USERNAME, key_filename=SSH_KEY_FILENAME
|
||||
)
|
||||
with ssh_client.open_sftp() as sftp_connection:
|
||||
yield sftp_connection
|
||||
|
||||
def retrieve_x12_edi_files():
|
||||
"""
|
||||
Connect to S3 bucket & pull down files.
|
||||
"""
|
||||
with sftp_server() as sftp_connection:
|
||||
sftp_connection.chdir("/yu-edi-transfer/source-logi/dev/ecomm-inbound")#TODO set to prod
|
||||
for filename in sftp_connection.listdir():
|
||||
if edi_945.SOURCE_945_FILENAME_RE.match(filename):
|
||||
sftp_connection.get(filename, edi_945.X12_DIRECTORY / filename)
|
||||
new_filename = f"/yu-edi-transfer/source-logi/dev/ecomm-processed/{filename}"#TODO set to prod
|
||||
sftp_connection.rename(filename, new_filename)
|
||||
|
||||
def combine_zshpords():
|
||||
"""
|
||||
Collect all ZSHPORD imports into a single file for easy import.
|
||||
"""
|
||||
archive_directory = INCOMING_DIRECTORY / "archive"
|
||||
archive_directory.mkdir(exist_ok=True)
|
||||
with (INCOMING_DIRECTORY / "ZSHPORD.dat").open(
|
||||
"a", encoding="utf-8", newline="\n"
|
||||
) as combined_import_file:
|
||||
for individual_import_filename in INCOMING_DIRECTORY.glob(
|
||||
"ZSHPORD_*.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 process_files(file):
|
||||
with open(file) as source_file:
|
||||
csv_reader = csv.reader(source_file)
|
||||
sales_order = SalesOrder()
|
||||
for num, row in enumerate(csv_reader):
|
||||
if num == 0:
|
||||
continue #skip header lines
|
||||
# pprint.pprint(row)
|
||||
if num == 1: #gather header information
|
||||
order_id = row[5]
|
||||
order_date = row[8]
|
||||
customer_name = row[9]
|
||||
# shipadd1 = row[9] # address information is not stored in X3
|
||||
# shipadd2 = row[10]
|
||||
# shipcity = row[11]
|
||||
# shipstate = row[12]
|
||||
# shipzip = row[13]
|
||||
tracking = row[14]
|
||||
weight = row[16]
|
||||
ship_charge = row[20]
|
||||
taxes = "?" #TODO fixme
|
||||
ship_site = "?" #TODO fixme
|
||||
discount = "?" #TODO fixme
|
||||
sales_order.header.cusordref = order_id
|
||||
sales_order.header.orddat = datetime.datetime.strptime(order_date,'%m/%d/%Y').strftime('%Y%m%d') #TODO strftim this
|
||||
sales_order.header.stofcy = ship_site
|
||||
sales_order.header.bpdnam = customer_name
|
||||
sales_order.header.invdtaamt_5 = ship_charge
|
||||
sales_order.header.invdtaamt_7 = '0.33' #discount
|
||||
sales_order.header.invdtaamt_8 = '0.51'#taxes
|
||||
|
||||
#gather line data
|
||||
line_product = row[0]
|
||||
line_qty = row[2]
|
||||
line_lot = row[3]
|
||||
line_price = row[19]
|
||||
sales_order.append(
|
||||
SalesOrderDetail(
|
||||
itmref=line_product,
|
||||
qty=line_qty,
|
||||
gropri=line_price
|
||||
)
|
||||
)
|
||||
time_stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
with yamamotoyama.x3_imports.open_import_file(
|
||||
SOH_IMPORT_DIRECTORY / f"ZSHPORD_{time_stamp}.dat"
|
||||
) as import_file:
|
||||
sales_order.output(import_file)
|
||||
|
||||
@dataclasses.dataclass
|
||||
class SalesOrderDetail:
|
||||
"""
|
||||
Information that goes on ann order detail line, taken from ZSHPORD template.
|
||||
"""
|
||||
|
||||
itmref: str = ""
|
||||
itmrefbpc: str = ""
|
||||
itmdes: str = ""
|
||||
qty: int = 0
|
||||
gropri: decimal.Decimal = decimal.Decimal()
|
||||
discrgval_1: decimal.Decimal = decimal.Decimal()
|
||||
zamaztax: decimal.Decimal = decimal.Decimal()
|
||||
star91: str = ""
|
||||
star92: str = ""
|
||||
|
||||
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(
|
||||
[
|
||||
"D",
|
||||
self.itmref,
|
||||
self.itmrefbpc,
|
||||
self.qty,
|
||||
self.gropri,
|
||||
self.discrgval_1,
|
||||
self.zamaztax,
|
||||
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, SalesOrderDetail):
|
||||
# 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()
|
||||
# else:
|
||||
# emailtext = str(self.sohnum +' '+str(self.itmref))
|
||||
# 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)
|
||||
# raise NotImplementedError # TODO
|
||||
|
||||
# result = get()
|
||||
# self.soplin = result.SOPLIN_0
|
||||
# self.itmdes = result.ITMDES1_0
|
||||
# self.sau = result.SAU_0
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class SalesOrderHeader:
|
||||
"""
|
||||
Information that goes on an order header, taken from ZSHPORD template.
|
||||
"""
|
||||
|
||||
sohnum: str = ""
|
||||
sohtyp: str = "WEB"
|
||||
bpcord: str = "STSHOPIFY" #TODO are they handling YU?
|
||||
bpcinv: str = "STSHOPIFY" #TODO are they handling YU?
|
||||
bpcpyr: str = "STSHOPIFY" #TODO are they handling YU?
|
||||
bpaadd: str = "BL001"
|
||||
orddat: datetime.date = datetime.date(1753, 1, 1)
|
||||
cusordref: str = ""
|
||||
cur: str = "USD"
|
||||
alltyp: int = 2 #Detailed
|
||||
salfcy: str = "ECS" #TODO are they handling YU?
|
||||
stofcy: str = "" #TODO need to be set from file
|
||||
pte: str = "USPREPAY" #TODO needs checking
|
||||
vacbpr: str = "NTX" #TODO needs checking
|
||||
dlvpio: int = 1 #normal
|
||||
mdl: str = "GRN"#TODO any way to tell how they were sent?
|
||||
yshppaymth: int = 1 #prepaid freight
|
||||
bpcnam: str = "ST SHOPIFY Stashtea.com" #TODO does this need to be set
|
||||
bpdnam: str = ""
|
||||
bpdaddlig_0: str = "999 ANYSTREET"
|
||||
bpdaddlig_1: str = ""
|
||||
bpdaddlig_2: str = ""
|
||||
bpdcty: str = "POMONA"
|
||||
bpdsat: str = "CA"
|
||||
bpdposcod: str = "91768"
|
||||
bpdcry: str = "US"
|
||||
ybpdweb: str = ""
|
||||
ybpdtel: str = ""
|
||||
ybpcweb: str = ""
|
||||
yamaorder: str = ""
|
||||
ygiftwrap: int = 0
|
||||
invdtaamt_5: decimal.Decimal = decimal.Decimal()
|
||||
invdtaamt_7: decimal.Decimal = decimal.Decimal()
|
||||
invdtaamt_8: decimal.Decimal = decimal.Decimal()
|
||||
yimport: int = 0
|
||||
pjt: str = ""
|
||||
yedinotes: str = ""
|
||||
|
||||
def convert_to_strings(self) -> typing.List[str]:
|
||||
"""
|
||||
Convert to X3 import line
|
||||
"""
|
||||
return yamamotoyama.x3_imports.convert_to_strings(
|
||||
[
|
||||
"H",
|
||||
self.sohnum,
|
||||
self.sohtyp,
|
||||
self.bpcord,
|
||||
self.bpcinv,
|
||||
self.bpcpyr,
|
||||
self.bpaadd,
|
||||
self.orddat,
|
||||
self.cusordref,
|
||||
self.cur,
|
||||
self.alltyp,
|
||||
self.salfcy,
|
||||
self.stofcy,
|
||||
self.pte,
|
||||
self.vacbpr,
|
||||
self.dlvpio,
|
||||
self.mdl,
|
||||
self.yshppaymth,
|
||||
self.bpcnam,
|
||||
self.bpdnam,
|
||||
self.bpdaddlig_0,
|
||||
self.bpdaddlig_1,
|
||||
self.bpdaddlig_2,
|
||||
self.bpdcty,
|
||||
self.bpdsat,
|
||||
self.bpdposcod,
|
||||
self.bpdcry,
|
||||
self.ybpdweb,
|
||||
self.ybpdtel,
|
||||
self.ybpcweb,
|
||||
self.yamaorder,
|
||||
self.ygiftwrap,
|
||||
self.invdtaamt_5,
|
||||
self.invdtaamt_7,
|
||||
self.invdtaamt_8,
|
||||
self.yimport,
|
||||
self.pjt,
|
||||
self.yedinotes
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
class SalesOrderDetailList:
|
||||
"""
|
||||
List of shipment details
|
||||
"""
|
||||
|
||||
_details: typing.List[SalesOrderDetail]
|
||||
_item_set: typing.Set[str]
|
||||
|
||||
def __init__(self):
|
||||
self._details = []
|
||||
self._item_set = set()
|
||||
|
||||
def append(
|
||||
self,
|
||||
salesorder_detail: SalesOrderDetail,
|
||||
):
|
||||
"""
|
||||
Append
|
||||
"""
|
||||
itmref = salesorder_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)
|
||||
#salesorder_detail.fill()
|
||||
#salesorder_detail.append(shipment_subdetail)
|
||||
self._details.append(salesorder_detail)
|
||||
|
||||
def __iter__(self):
|
||||
return iter(self._details)
|
||||
|
||||
|
||||
class SalesOrder:
|
||||
"""
|
||||
sales order both header & details
|
||||
"""
|
||||
|
||||
header: SalesOrderHeader
|
||||
details: SalesOrderDetailList
|
||||
|
||||
def __init__(self):
|
||||
self.header = SalesOrderHeader()
|
||||
self.details = SalesOrderDetailList()
|
||||
|
||||
def append(
|
||||
self,
|
||||
salesorder_detail: SalesOrderDetail,
|
||||
):
|
||||
"""
|
||||
Add detail information.
|
||||
"""
|
||||
self.details.append(salesorder_detail)
|
||||
|
||||
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())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -0,0 +1,643 @@
|
|||
#!/usr/bin/env python3
|
||||
import csv
|
||||
import pprint
|
||||
import dataclasses
|
||||
import datetime
|
||||
import decimal
|
||||
import functools
|
||||
import pathlib
|
||||
import re
|
||||
import shutil
|
||||
import typing
|
||||
|
||||
import records # type: ignore
|
||||
|
||||
import yamamotoyama # type: ignore
|
||||
import yamamotoyama.x3_imports # type: ignore
|
||||
|
||||
THIS_DIRECTORY = pathlib.Path(__file__).parent
|
||||
INCOMING_DIRECTORY = THIS_DIRECTORY / "incoming_shipments"
|
||||
SDH_IMPORT_DIRECTORY = THIS_DIRECTORY / "to_import_SDH"
|
||||
|
||||
def main():
|
||||
process_files()
|
||||
#TODO shutil the files around, archive on the ftp
|
||||
#TODO determine X3 processing schedule
|
||||
|
||||
def find_so_from_po(cust_po):
|
||||
with yamamotoyama.get_connection('test') as db_connection:#TODO remove 'test'
|
||||
return db_connection.query(
|
||||
"""
|
||||
select
|
||||
SOHNUM_0
|
||||
from FY23TEST.SORDER --TODO change to PROD
|
||||
where
|
||||
SOHTYP_0 = 'WEB'
|
||||
and BPCORD_0 = 'STSHOPIFY'
|
||||
and CUSORDREF_0 = :order
|
||||
""",
|
||||
order=cust_po,
|
||||
).first()["SOHNUM_0"]
|
||||
|
||||
def process_files():
|
||||
for file in INCOMING_DIRECTORY.iterdir():
|
||||
with open(file) as source_file:
|
||||
csv_reader = csv.reader(source_file)
|
||||
warehouse_shipment = WarehouseShipment()
|
||||
for num, row in enumerate(csv_reader):
|
||||
if num == 0:
|
||||
continue #skip header lines
|
||||
if num == 1: #gather header information
|
||||
sohnum = find_so_from_po(row[5])
|
||||
order_date = row[8]
|
||||
customer_name = row[9]
|
||||
# shipadd1 = row[9] # address information is not stored in X3
|
||||
# shipadd2 = row[10]
|
||||
# shipcity = row[11]
|
||||
# shipstate = row[12]
|
||||
# shipzip = row[13]
|
||||
tracking = row[14]
|
||||
weight = row[16]
|
||||
ship_charge = row[20]
|
||||
taxes = "?" #TODO fixme
|
||||
ship_site = "?" #TODO fixme
|
||||
discount = "?" #TODO fixme
|
||||
warehouse_shipment.sohnum = sohnum
|
||||
#warehouse_shipment.header.sohnum = sohnum
|
||||
warehouse_shipment.header.shidat = datetime.datetime.strptime(order_date,'%m/%d/%Y')
|
||||
warehouse_shipment.header.ylicplate = tracking
|
||||
warehouse_shipment.header.growei = weight
|
||||
#gather line data
|
||||
#TODO how are multiple lots processed?
|
||||
line_product = row[0]
|
||||
line_qty = row[2]
|
||||
line_lot = row[3]
|
||||
line_price = row[19]
|
||||
subdetail = WarehouseShipmentSubDetail(
|
||||
qtypcu=-1 * int(line_qty),
|
||||
lot=line_lot,
|
||||
)
|
||||
warehouse_shipment.append(
|
||||
WarehouseShipmentDetail(
|
||||
sohnum=sohnum,
|
||||
itmref=line_product,
|
||||
qty=int(line_qty),
|
||||
),
|
||||
subdetail,
|
||||
)
|
||||
time_stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
with yamamotoyama.x3_imports.open_import_file(
|
||||
SDH_IMPORT_DIRECTORY / f"ZSHIP945S_{warehouse_shipment.sohnum}_{time_stamp}.dat"
|
||||
) as import_file:
|
||||
warehouse_shipment.output(import_file)
|
||||
|
||||
@dataclasses.dataclass
|
||||
class WarehouseShipmentSubDetail:
|
||||
"""
|
||||
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 WarehouseShipmentDetail:
|
||||
"""
|
||||
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[WarehouseShipmentSubDetail] = dataclasses.field(
|
||||
default_factory=list
|
||||
)
|
||||
|
||||
def append(self, subdetail: WarehouseShipmentSubDetail):
|
||||
"""
|
||||
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, WarehouseShipmentDetail):
|
||||
return self.itmref == item.itmref
|
||||
return False
|
||||
|
||||
def fill(self):
|
||||
"""
|
||||
Set soplin & itmdes from itmref & sohnum
|
||||
"""
|
||||
|
||||
def get() -> records.Record:
|
||||
with yamamotoyama.get_connection('test') as database:#TODO remove test
|
||||
how_many = (
|
||||
database.query(
|
||||
"""
|
||||
select
|
||||
count(*) as [how_many]
|
||||
from [FY23TEST].[SORDERP] as [SOP] --TODO change to PROD
|
||||
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 [FY23TEST].[SORDERP] as [SOP] --TODO change to PROD
|
||||
where
|
||||
[SOP].[SOHNUM_0] = :sohnum
|
||||
and [SOP].[ITMREF_0] = :itmref
|
||||
order by
|
||||
[SOP].[SOPLIN_0]
|
||||
""",
|
||||
sohnum=self.sohnum,
|
||||
itmref=self.itmref,
|
||||
).first()
|
||||
else:
|
||||
raise NotImplementedError # TODO
|
||||
|
||||
result = get()
|
||||
self.soplin = result.SOPLIN_0
|
||||
self.itmdes = result.ITMDES1_0
|
||||
self.sau = result.SAU_0
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class WarehouseShipmentHeader:
|
||||
"""
|
||||
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 WarehouseShipmentDetailList:
|
||||
"""
|
||||
List of shipment details
|
||||
"""
|
||||
|
||||
_details: typing.List[WarehouseShipmentDetail]
|
||||
_item_set: typing.Set[str]
|
||||
|
||||
def __init__(self):
|
||||
self._details = []
|
||||
self._item_set = set()
|
||||
|
||||
def append(
|
||||
self,
|
||||
shipment_detail: WarehouseShipmentDetail,
|
||||
shipment_subdetail: WarehouseShipmentSubDetail,
|
||||
):
|
||||
"""
|
||||
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 WarehouseShipment:
|
||||
"""
|
||||
Warehosue shipment, both header & details
|
||||
"""
|
||||
|
||||
header: WarehouseShipmentHeader
|
||||
details: WarehouseShipmentDetailList
|
||||
_sohnum: str
|
||||
|
||||
def __init__(self):
|
||||
self.header = WarehouseShipmentHeader()
|
||||
self._sohnum = ""
|
||||
self.details = WarehouseShipmentDetailList()
|
||||
|
||||
def append(
|
||||
self,
|
||||
shipment_detail: WarehouseShipmentDetail,
|
||||
shipment_subdetail: WarehouseShipmentSubDetail,
|
||||
):
|
||||
"""
|
||||
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('test') as db_connection:#TODO remove 'test'
|
||||
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 [FY23TEST].[SORDER] as [SOH]--TODO change back to PROD
|
||||
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__":
|
||||
main()
|
||||
Loading…
Reference in New Issue