wes/Python_Examples
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Found more code examples from STM. Used claude to generate a summary (CLAUDE.md)

Browse Source
This commit is contained in:
wes
2026-02-11 06:05:20 -05:00
parent cf0edba776
commit c4b970cbde
33 changed files with 6077 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
stm_etl/FTP/mf_850.py
+399
View File
@@ -0,0 +1,399 @@
'''
Created on Dec 2, 2014
There was code in the original RPG/CL programs to e-mail Moe and Malcolm
if the pattern "OPALESCENT" comes through on an order. I think this is probably
best handled within Jomar and not here.
@author: Wes
'''
from alert import send_alert_html
import datetime, logging, sys, os, StringIO
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
RECIPIENTS = ['admin@example.com']
SUBJECT = 'New Maharam 850 Processed.'
MESSAGE = None
TEST_MODE = False
logger = None
LOG_LEVEL = logging.DEBUG
engine = create_engine("mssql+pyodbc://jsisbprod", echo=False)
Base = declarative_base(engine)
metadata = Base.metadata
CR_DATA_LENGTHS = [1, 8, 5]
HR_DATA_LENGTHS = [1, 12, 8, 30, 30, 30, 30, 2, 15, 30, 30, 30, 1]
DR_DATA_LENGTHS = [1, 3, 15, 4, 6, 3, 9, 9, 8, 50, 50, 50, 50]
RR_DATA_LENGTHS = [1, 80]
SR_DATA_LENGTHS = [1, 5]
S_TYPE_CODE = 0 # len 1
S_LINES = 1 # len 5
########################################################################
class OrderHeader(Base):
__tablename__ = 'CSPEDI50H' #EDI 850 Header Table
__table_args__ = {'autoload':True}
class OrderDetail(Base):
__tablename__ = 'CSPEDI50D' #EDI 850 Detail Table
__table_args__ = {'autoload':True}
class ItemXref(Base):
__tablename__ = 'TXUYAT01' #Customer Item -> Sunbury Item reference table
__table_args__ = {'autoload':True}
#----------------------------------------------------------------------
def loadSession():
global engine
Session = sessionmaker(bind=engine)
session = Session()
return session
def initializeLogger(name='log'):
global logger
global MESSAGE
#logger = logging.getLogger(__name__)
logger = logging.getLogger(name)
if len(logger.handlers) > 0:
return logger
logger.setLevel(LOG_LEVEL)
handler = logging.FileHandler('D:\\FTP\\logs\\' + name + '.log')
handler.setLevel(LOG_LEVEL)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
#Also send logs > DEBUG to stdout
ch = logging.StreamHandler(sys.stdout)
ch.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
ch.setFormatter(formatter)
logger.addHandler(ch)
#Capture log messages in a string for e-mail alert
MESSAGE = StringIO.StringIO()
sh = logging.StreamHandler(MESSAGE)
sh.setLevel(logging.DEBUG)
sh.setFormatter(formatter)
logger.addHandler(sh)
'''
Splits the supplied string into len(lengths) elements.
The length of element i will be equal to lengths[i]
'''
def split_string_by_position(a_string, lengths):
result = []
position = 0
for length in lengths:
result.append(a_string[position:position+length])
position = position+length
return result
def trim_elements(a_list):
result = []
for ele in a_list:
result.append(ele.strip())
return result
def getSunburyCust(hccb, addr, city):
addressList = ['45 RASONS CT', '74 HORSEBLOCK ROAD', '175 MUFFIN LANE']
cityList = ['HAUPPAUGE', 'YAPHANK', 'CUYAHOGA FALLS']
if (len(hccb) > 0):
return 659000
else:
if(addr in addressList[0]) and (city in cityList[0]):
return 343900
elif(addr in addressList[1]) and (city in cityList[1]):
return 3439
elif(addr in addressList[2]) and (city in cityList[2]):
return 343905
else:
#logger.warning('Encountered new bill to address in transmission. Using will-advise ship-to (343999)')
return 343999
def lookUpSunburyItem(session, v_item, v_col, v_num):
if len(str(v_num)) > 4:
ven = str(v_num)[:-2]
else:
ven = v_num
#This query retrieves a list of every item we have for this Pattern + Customer combo
q = session.query(ItemXref).order_by(ItemXref.UACCOL).filter(ItemXref.UACPAT.like(v_item + "%"), ItemXref.UACCNR == ven)
#q = session.query(ItemXref).order_by(ItemXref.UACCOL).filter(ItemXref.UACPAT.like(v_item), ItemXref.UACCNR == ven)
for row in q:
col = int(row.UACCOL.split('/')[0])
#Iterate over results till we find matching color #
if(int(v_col) == col):
logger.info('Maharam Pattern Number: ' + str(v_item) + ' Color: ' + str(v_col) + ' = Sunbury Item: ' + row.UACANR)
return row.UACANR
#If we made it to here the pattern/color combo does not exist in the xref table...
logger.warn('Failed to find a Sunbury item number for: ' + str(v_item) + ' - ' + str(v_col))
return ''
#raise RuntimeError('Failed to find a Sunbury item number for: ' + str(v_item) + ' - ' + str(v_col))
def processMaharamPo(inFile):
global MESSAGE
os.chdir('D:\FTP')
initializeLogger('mf_850_interface')
logger.info('****************************************')
#Read downloaded 850 file into list
logger.info('Loading ' + inFile)
f = open(inFile, 'r')
lines = f.readlines()
f.close()
# Last character on each line is '\n'; trim it off
logger.debug('Doctoring up transmitted lines...')
for idx, val in enumerate(lines):
lines[idx] = val[:-1]
# Validate transmission length
logger.debug('Validating transmission length...')
sr = split_string_by_position(lines[len(lines) - 1], SR_DATA_LENGTHS)
if(sr[0] == 'S'):
s_lines = int(sr[1])
if not(s_lines == (len(lines) - 2)):
logger.error('Transmission Corrupt: Number of records actually transmitted does not match what they say they transmitted...')
raise ValueError('Transmission Corrupt: Number of records actually transmitted does not match what they say they transmitted...')
else:
logger.info('They transmitted ' + str((len(lines) - 2)) + ' lines and we received ' + str(s_lines) + ' lines. Good.')
else:
logger.error('Transmission Corrupt: Last record in transmission is not a summary record...')
raise ValueError('Transmission Corrupt: Last record in transmission is not a summary record...')
# Parse Control Record
logger.info('Parsing Control Record:')
cr = split_string_by_position(lines[0], CR_DATA_LENGTHS)
if not(cr[0] == 'T'):
logger.error('Transmission Corrupt: Missing Control Record...')
raise ValueError('Transmission Corrupt: Missing Control Record...')
xmit_date = datetime.datetime.strptime(cr[1], "%m%d%Y")
mah_xmit_seq = int(cr[2])
logger.info('\t Transmission Date: ' + xmit_date.strftime('%m/%d/%Y'))
logger.info('\t Maharam Transmission Sequence #: ' + str(mah_xmit_seq))
# Parse Header Record
logger.info('')
logger.info('Parsing Header Record:')
hr = split_string_by_position(lines[1], HR_DATA_LENGTHS)
if not(hr[0] == 'H'):
logger.error('Transmission Corrupt: Missing Header Record...')
raise ValueError('Transmission Corrupt: Missing Header Record...')
hr = trim_elements(hr) # Trim white space off each data element; we don't need it.
#mah_po_num = int(hr[1])
mah_po_num = hr[1].strip()
po_date = datetime.datetime.strptime(hr[2], "%m%d%Y")
shp_to_name = hr[3]
shp_to_addr1 = hr[4]
shp_to_addr2 = hr[5]
shp_to_city = hr[6]
shp_to_state = hr[7]
shp_to_zip = hr[8]
shp_via = hr[9]
pymt_terms = hr[10]
requester_name = hr[11]
customer_ord_code = hr[12]
logger.info('\t Maharam PO #: ' + str(mah_po_num))
logger.info('\t PO Date: ' + po_date.strftime('%m/%d/%Y'))
logger.info('\t Ship To: ' + shp_to_name)
logger.info('\t Ship to Address 1: ' + shp_to_addr1)
logger.info('\t Ship to Address 2: ' + shp_to_addr2)
logger.info('\t Ship to City: ' + shp_to_city)
logger.info('\t Ship to State: ' + shp_to_state)
logger.info('\t Ship to Zip: ' + str(shp_to_zip))
logger.info('\t Ship Via: ' + shp_via)
logger.info('\t Payment Terms: ' + pymt_terms)
logger.info('\t Requested By: ' + requester_name)
logger.info('\t Custom Order Code: ' + customer_ord_code)
mah_po_line = []
stm_pattern_name = []
stm_color = []
mah_style = []
mah_color = []
yards_ord = []
price = []
req_ship_date = []
line_comment_1 = []
line_comment_2 = []
line_comment_3 = []
line_comment_4 = []
po_comments = []
# Process rest of transmission, may contain any number of D or R record types.
for x in range(2, len(lines) - 1):
rtype = lines[x][:1]
if(rtype == 'D'):
dr = split_string_by_position(lines[x], DR_DATA_LENGTHS)
mah_po_line.append(int(dr[1]))
stm_pattern_name.append(dr[2])
stm_color.append(int(dr[3]))
mah_style.append(int(dr[4]))
mah_color.append(dr[5])
yards_ord.append(float(dr[6]))
price.append(float(dr[7]))
#req_ship_date.append(datetime.datetime.strptime(dr[8], "%m%d%Y"))
line_comment_1.append(dr[9])
line_comment_2.append(dr[10])
line_comment_3.append(dr[11])
line_comment_4.append(dr[12])
elif(rtype == 'R'):
rr = split_string_by_position(lines[x], RR_DATA_LENGTHS)
po_comments.append(rr[1].strip())
else:
logger.error('Transmission Corrupt: Encountered something other than D or R record type in body of transmission...')
raise ValueError('Transmission Corrupt: Encountered something other than D or R record type in body of transmission...')
logger.info('')
logger.info('Detail Records:')
for i in range (0, len(mah_po_line)):
#Recreated from original RPG code (MF850I) modification from 03/03/09. Special price for certain colors of "HIGH FIDELITY"
#They finally ordered some of this and transmitted a price of $18.65; which is higher than the $17.68 and also correct (according to Moe)...
#Going to ignore this going forward.
#if(stm_pattern_name[i] == 'HIGH FIDELITY '):
#if(stm_color[i] == 302) or (stm_color[i] == 6103):
#price[i] = 17.68
logger.info('\t Maharam PO Line #: ' + str(mah_po_line[i]))
logger.info('\t STM Pattern Name: ' + stm_pattern_name[i])
logger.info('\t STM Color #: ' + str(stm_color[i]))
logger.info('\t Maharam Style: ' + str(mah_style[i]))
logger.info('\t Maharam Color: ' + str(mah_color[i]))
logger.info('\t Yards Ordered: ' + str(yards_ord[i]))
logger.info('\t Price/Yard: ' + str(price[i]))
#logger.info('\t Requested Ship Date: ' + req_ship_date[i].strftime('%m/%d/%Y'))
logger.info('\t Detail Line Comment 1: ' + line_comment_1[i])
logger.info('\t Detail Line Comment 2: ' + line_comment_2[i])
logger.info('\t Detail Line Comment 3: ' + line_comment_3[i])
logger.info('\t Detail Line Comment 4: ' + line_comment_4[i])
logger.info('')
logger.info('PO Comments:')
if(len(po_comments) == 0):
logger.info('\t ***No Data***')
else:
for comment in po_comments:
logger.info(comment)
logger.debug('Looking up customer account number...')
sunburyCust = getSunburyCust(customer_ord_code, shp_to_addr1, shp_to_city)
if sunburyCust == 343999:
logger.warning('Encountered new bill to address in transmission. Using will-advise ship-to (343999)')
logger.info('Sunbury Customer Number: ' + str(sunburyCust))
'''
Query server to figure out Jomar internal sequential item number
'''
logger.debug('Querying customer XREF table to get Sunbury item numbers...')
internal_item = []
session = loadSession()
for p, c in zip(mah_style, mah_color):
internal_item.append(lookUpSunburyItem(session, str(p), c, sunburyCust))
#Create Object to be persisted to database
header = OrderHeader()
#Set up constant values in order header record
header.UBNR = '001' #Business Unit
header.UPLT = '00' #Plant #
header.ULOAD = 'N' #Flag to allow order to be processed in Jomar EDI workbench
header.UBLA = 'SO' #Jomar order type "SO" = Sales Order
header.UDAT = int(po_date.strftime('%Y%m%d')) #Transaction Date -> Order Date
header.FOB = ' ' #This customer doesn't transmit this information
header.UIHR = mah_po_num
header.UIHS = int(po_date.strftime('%Y%m%d')) #PO date
header.UKDN = sunburyCust #Ship to customer #
header.UKDR = str(sunburyCust)[:4] #Bill to customer #
header.XTX5 = pymt_terms #Payment terms
header.XTX6 = 'N/A' #Jomar "notes" field sticking order message here
header.URLRV = '0' #An ID column of sorts, in case the same PO is transmitted twice
header.FLX01 = '00'
header.UNA2 = shp_to_name
header.UNA3 = shp_to_addr1
header.UNA4 = shp_to_addr2
header.UORT = shp_to_city
header.USTA = shp_to_state
header.ULND = 'US'
header.UPLZ = shp_to_zip
header.UVT1 = '380'
#Create Detail objects
line = 1
orderLines = []
for item, qty, price, sPat, sCol, mPat, mCol in zip(internal_item, yards_ord, price, stm_pattern_name, stm_color, mah_style, mah_color):
oline = OrderDetail()
oline.VBNR = '001'
oline.VPLT = '00'
oline.VQUALT = '01'
oline.VUOM = 'YDS'
oline.VBLA = 'SO'
oline.VLFN = line
oline.VMGS = qty
oline.VPR2 = price
oline.VANR = item
oline.VKDR = str(sunburyCust)[:4]
oline.VKDL = sunburyCust
oline.VIHR = mah_po_num
oline.VRLRV = '0'
oline.VFLX01 = sPat
oline.VFLX02 = sCol
oline.VFLX03 = mPat
oline.VFLX04 = mCol
oline.VVT1 = '380'
line = line + 1
orderLines.append(oline)
#Check for duplicate PO(s) already in database; increment URLRV/VRLRV accordingly if so
logger.debug('Checking if this PO already exists in Jomar 850 tables...')
dups = session.query(OrderHeader).filter(OrderHeader.UBNR == '001', OrderHeader.UPLT == '00', OrderHeader.UKDR == header.UKDR, OrderHeader.UIHR == header.UIHR).count()
if(dups > 0):
logger.warning('PO ' + str(header.UIHR) + ' already in 850 tables. Adding this anyway with sequence number ' + str(dups) + ' it may be deleted in Jomar if not needed.')
header.URLRV = str(dups)
header.XTX6 = '**POSSIBLE DUPLICATE TRANSMISSION** ' + header.XTX6
for line in orderLines:
line.VRLRV = str(dups)
#submit
session.add(header)
for line in orderLines:
session.add(line)
if not TEST_MODE:
logger.info('Writing records to database...')
session.commit()
else:
logger.warning('**RUNNING IN TEST MODE. ROLLING BACK TRANSACTION**')
session.close()
try:
if not TEST_MODE:
logger.info('Sending e-mail alert...')
send_alert_html(RECIPIENTS, SUBJECT, attachments=[inFile], message=MESSAGE.getvalue())
except:
logger.error('Unable to send e-mail. Is the internet out?')
if not TEST_MODE:
logger.info('Moving ' + inFile + ' to D:\\FTP\\OLD\\MF\\')
os.rename(inFile, 'OLD\\MF\\' + inFile.split('\\')[-1])
MESSAGE.truncate(0)
if __name__ == '__main__':
if(len(sys.argv) < 2):
pass
else:
processMaharamPo(sys.argv[1])
sys.exit()