· 9 years ago · Jun 02, 2017, 07:18 PM
1import csv
2import numpy
3import pandas
4import datetime,time
5import pyodbc
6import string
7import os
8import sqlalchemy
9import ftplib
10import sys
11import fileinput
12import fnmatch
13import pdb
14import itertools
15import urllib
16import sqlalchemy.pool as pool
17
18
19""" Define FTP's hostname, username, password, directory, and filematch type"""
20
21
22hostname = r'clients.redacted'
23username = r'lg'
24password = r'n'
25ftpdirectory =r'/Usr/'
26filematch = r'*.txt'
27
28
29
30"""Specify directories"""
31
32
33originaldirectory = r'C:\Users\al\Documents\psi_ftp\original'
34cleandirectory = r'C:\Users\al\Documents\psi_ftp\clean'
35
36
37
38
39"""Get filenames from directories to two lists, 'Original' and 'Clean'"""
40
41
42od=[]
43cd=[]
44table=None
45
46
47
48
49"""Explicitly list columns from file that we want to keep"""
50
51
52v0 = u"ID"
53v1 = u"Enrolid"
54v2 = u"ImportDate"
55v3 = u"PassFailBoolean"
56v4 = u"Grades"
57mylist = [v0, v1, v2, v3, v4]
58
59
60dsn='red-DEV-red';UID='red';PWD='Password1'
61
62
63
64
65params = urllib.quote(r'DRIVER={SQL SERVER};'
66 r'SERVER=red-red-red;DATABASE=red;'
67 r'DATABASE=red;'
68 r'Trusted_Connection=yes;')
69
70
71mypool = pool.QueuePool(getconn,max_overflow=10, pool_size=5)
72
73'''Make a direct connection to a database'''
74
75def create_table():
76 cnxn = pyodbc.connect(
77 r'DRIVER={SQL red};'
78 r'SERVER=red-red-red;DATABASE=red;'
79 r'DATABASE=red;'
80 r'Trusted_Connection=red;')
81
82
83 cursor = cnxn.cursor()
84
85
86 cursor.execute("""if not exists (select * from sysobjects where name ='PSAgradeimport' and xtype = 'U') CREATE TABLE PSAgradeimport(
87
88
89 ID nvarchar(255),
90
91
92 Enrolid nvarchar(255),
93
94
95 ImportDate nvarchar(255),
96
97
98 Unknown nvarchar(255),
99
100
101 Grades nvarchar(255)
102
103
104
105 )""")
106
107
108 cnxn.commit() #Must commit changes or else does not work
109
110
111 cnxn.close()
112
113
114 print ("Creating Table - check that table is correct, rerun script for import")
115
116
117
118
119def import_ftp_files():
120
121
122 ftp = ftplib.FTP(hostname)
123
124 ftp.login(username,password)
125
126 ftp.cwd(ftpdirectory)
127
128 ftp.set_pasv(False)
129
130 os.chdir(originaldirectory) #test """ftp.voidcmd("NOOP")"""
131
132 files=[] #list too hold files
133
134 for filename in ftp.nlst(filematch): #Return a list of file names
135
136 """Get File Name, specifically the date portion, add date to a List"""
137
138 filenamedate = filename[0:13] #Get File Name in Format: YYYYMMDD, e.g. 20130531 - NEEDS TO BE REVISITED IT's SLICING DATA (NUMPY) the 43:50 reps chars in string
139
140 files.append(filenamedate)
141
142 mypath = os.path.join(originaldirectory, filename) #Get complete file path and file name
143
144 """If the file does not exist, then downloads the file from FTP to local directory"""
145
146 if (os.path.exists(mypath) == False): #File does not exist in the local directory, then write data
147
148 fhandle = open(mypath, 'wb')
149
150 print ("Getting " + mypath)
151
152 ftp.retrbinary('RETR ' + filename, fhandle.write) #Retrieve a file in binary transfer mode - as found on https://stackoverflow.com/questions/11573817/how-to-download-a-file-via-ftp-with-python-ftplib
153
154 fhandle.close()
155
156 elif (os.path.exists(mypath) == True): #File exists in the local directory
157
158 filesize = os.stat(filename) #Get stats for filename
159
160 #print filesize.st_size #Test to see file size
161
162 if (filesize.st_size==0): # Check if file size is 0, if it is then replace with updated file on FTP
163
164 print ("File size is 0, replacing with updated FTP " + mypath)
165
166 fhandle = open(mypath, 'wb')
167
168 ftp.retrbinary('RETR ' + filename, fhandle.write) #Retrieve a file in binary transfer mode
169
170 fhandle.close()
171
172 else:
173
174 print ("File ", filename, " Already Exists, Skipping to next Download")
175
176 else:
177
178 print ("Unknown Error")
179
180
181
182
183def write_files(myfilename):
184
185 myinputfilelocation = os.path.join(originaldirectory, myfilename)
186
187 myoutputfilelocation = os.path.join(cleandirectory, myfilename)
188
189 myinputfile = open(myinputfilelocation, 'rb',
190
191 mode='rb', encoding='ascii', errors='replace')
192
193 myoutputfile = open(myoutputfilelocation, 'wb',
194
195 mode='wb', encoding='utf-8', errors='ignore')
196
197 for line in myinputfile:
198
199 try:
200
201 newline=line.encode('ascii', errors='replace')
202
203 except UnicodeDecodeError:
204
205 print ("decode error")
206
207 myoutputfile.write(newline)
208
209 myinputfile.close()
210
211 myoutputfile.close()
212
213
214
215def opencsv_writesql(myinputfile):
216
217
218 table = PSAgradeimport
219
220 with open(os.path.join(cleandirectory, myinputfile)) as f:
221
222 cf = csv.DictReader(f, fieldnames =['mylist'])
223
224 for row in cf:
225
226 print ("row['mylist']")
227
228 try:
229
230 table.insert().values(**row).execute()
231
232 except:
233
234 print ("error with this row")
235
236
237def prepare_data(maindataframe):
238
239 print("Before Cleaning Data")
240
241 df1 = pd.dataframe(maindataframe,columns=['col1','col2','col3','col4','col5'])
242
243 df1 = [['col1','col2','col5']] = df[['col1','col2','col5']].apply(pd.to_numeric)
244
245 df1= [['col3']] = df[['col3']].apply(pd.to_datetime)
246
247
248
249
250
251if __name__ == "__main__":
252
253 """Prepare DB connection for SQLAlchemy"""
254
255#engine=sqlalchemy.create_engine('mssql+pyodbc://ASC-DEV-SQL07/NASMdatamart? ? driver=OD‌​BC+Driver+11+for+SQL‌​+Server,trusted_connection=Yes', creator=getconn)
256 engine=sqlalchemy.create_engine('redacted', creator=getconn)
257 metadata = sqlalchemy.MetaData(bind=engine)
258 try:
259 PSAgradeimport = sqlalchemy.Table('PSAgradeimport', metadata, autoload=True)
260 except:
261 create_table()
262 metadata.create_all(engine)
263 conn = engine.connect()
264
265
266
267
268 """Get files from FTP"""
269
270import_ftp_files() #Update files in entire directory (i.e. get new or deleted files)
271ftp.close()
272
273
274
275
276
277
278
279
280 """Compare two lists of filenames comparing original data to cleaned data"""
281
282od=os.listdir(originaldirectory) #print
283print("Filenames in original directory include:", od)
284cd=os.listdir(cleandirectory)
285print("Filenames in clean directory include:", cd)
286od_length = len(od) #E.g. 127 files
287
288
289
290
291
292 """Check if file needs to be cleaned"""
293
294
295 for i in range(od_length):
296 print od[i] #Prints all the file names in original directory
297 myfilename = od[i] #Get current file name in list
298 if myfilename in cd:
299 print "File is already cleaned"
300 else:
301 print "File needs to be cleaned"
302 """Enter functions to run through data cleaning"""
303 write_files(myfilename)
304 """Setup dataframe to clean data"""
305 mynewoutputfile = os.path.join(cleandirectory, myfilename)
306 mydataframe = pandas.io.parsers.read_table(mynewoutputfile, sep=',', quotechar='"', header=0, index_col=0, error_bad_lines=True, warn_bad_lines=True, encoding='latin-1') # , encoding='utf-8', encoding='latin-1', warn_bad_lines=False, names=['CallReportNum','ReportVersion', 'CallDateAndTimeStart'], index_col=False, escapechar='\n', names=['CallReportNum','ReportVersion', 'CallDateAndTimeStart'], index_col=False, escapechar='\n'
307 """Get back clean dataframe consisiting of only the columns we want"""
308 dfclean = pandas.DataFrame(data=mydataframe, columns=mylist) #print dfclean[mylist[3]]
309 clean_data(myfilename) #Clean the data
310 opencsv_writesql(myfilename) #Write data to MS SQL database