Chat Beta

6/30/2014

Selective File copying using Python

Last few days I had a chance to learn a new scripting language- Python. 

I installed python 3.4.1 on windows, as well as a free IDE called "Eric" from http://eric-ide.python-projects.org/

After learning some basics of Python for couple of hours, I started working on an assignment which is to do a selective file copy from a source and save it to a desired destination.

Problem is something like following:

We have certain files in a remote machine, that we need to download daily. These files are build files created every night by the build server. Files are saved in the disk based on the development branch that is specified in build definition.A main directory is created with a build number, then inside the directory, several other sub directories are created for different products

We are tasked with getting the latest build from the server each following morning. A DOS based batch script has been used for this purpose, and  I thought of re-writing same with this new scripting language Python

The process that I constructed to achieve this as follows:

1. Sort main level directories based on the date they are created. This way we can find out the latest added directory- which also carries the latest build number
2. Select the latest directory
3. Select what sub directories to be copied (products)
4. Copy the required files and folders to the destination
5. Exception handling
6. Logging
7. Notify member by e-mail

Several different Python commands, modules are required for this tasks to accomplish

First I had to import following libraries
import os,smtplib,subprocess


With library "os", I could use listdir() method  which returns a list containing the names of the entries in the directory given by path. So this command created me a list having main directory names("50041","50042","50043" etc) as list items

dir_namelist.sort(key=lambda x: os.stat(os.path.join(source, x)).st_ctime)

dir_namelist=os.listdir(source)

source is the path to the main directories. Ex: source=r'\\tfs-file-01\\Builds\\BranchXXX'

Next. the list is in arbitrary order. So I had to sort the list based on the date/time. Following can be used for the purpose
Or we can sort the list by using sorted()command

dir_namelist=sorted(dir_namelist, reverse=True)
This will sort the list in descending order.

"os.stat". "os.path.join() are methods from "os" library

To get the latest, I used following command
index=len(dir_namelist) gets the length of the list
mydir=dir_namelist[index-1]

or simply  mydir=dir_namelist[0]


"mydir" is actually the main directory name that we are targeting(ex:50041)

To copy the files, I used copytree() method in shutil library.
shutil.copytree(CS, dest_cs, symlinks=False, ignore=None, ignore_dangling_symlinks=False)

or robocopy in windows:

subprocess.call("robocopy /S %s %s" %source, destination)

CopyTree() recursively copy an entire directory tree rooted at source.

To log certain activity, I write to a txt file using file.write() method
logfile=open(basedir+"\\"+"log.txt", "a") . file is created with mode=a which is to "append" content to the file


To send email, smtplib library is used
sender = 'builds@test.com'
receiver = 'user@test.com'
message="""From: Auto Builder <builds@test.com>
To: Build downloader <user@test.com>
Subject:your subject 
This is an automatically generated e-mail. Do not reply

"""
E-mail can be sent using following methods
smtpObj = smtplib.SMTP('mal.test.com')
           smtpObj.sendmail(sender, receiver, message)

The complete script looks like following 

 import os , smtplib, subprocess  
   
 #define variables for source and destination flder paths  
   
 #source='E:\\Python\\excercises\\test'  
 #destination='E:\\Python\\excercises\\dest'  
   
 #Create a Pythn list to hold all the folder names avalable under the source  
 dir_namelist=os.listdir(source)  
   
 #Sort the list descending order  
 dir_namelist=sorted(dir_namelist, reverse=True)  
   
 #Getting the first list item which actualy represents the latest folder name available  
 mydir=dir_namelist[0]  
     
   
 #Constructing Path to the latest build available in the build server.  
 final_source=os.path.join(source, mydir)  
   
 installerfolder=final_source+"\\_Installers"  
 CS=installerfolder+"\\cs"  
 SMweb=installerfolder+"\\SM.web"  
 SMwin=installerfolder+"\\SM.win"  
   
 basedir=os.path.join(destination, mydir)  
   
 class utilities:   
   i=1   
   j=1   
   #Define variables to hold email parameters  
   SMTPServer='mail.xxx.com'  
   sender = 'builds@downloadSL.com'  
   receiver = 'xxx@xx.lk'  
   def logging(self, logstr, b_dir, myd):   
     basedir=b_dir     
     mydir=myd  
     if os.path.exists(basedir):  
       logfile=open(basedir+"\\"+"log.txt", "a+")        
       if self.i==1:  
         logfile.write("Latest available successful build is " + mydir + "\n\n")   
         logfile.write("Files are copied to: " +basedir +"\n\n")  
         logfile.write("Following files copied:\n")  
         logfile.write("------------------------------\n")  
         self.i+=1  
       logfile.write(logstr)  
     else:  
       print("basedir is not available")  
         
     
   def sendmail(self, message):   
     try:  
       smtpObj = smtplib.SMTP(self.SMTPServer)  
       smtpObj.sendmail(self.sender, self.receiver, message)    
     except Exception as e:  
        print("e-mail sending error occured: " + e)  
         
   def readlog(self, b_dir):  
     basedir=b_dir  
     if os.path.exists(basedir):  
       log=open(basedir+"\\"+"log.txt", "r")  
       if os.stat(basedir+"\\"+"log.txt").st_size!=0:  
         logstr=log.read()  
         return str(logstr)  
         log.close()  
     else:  
      return "empty log file"   
      log.close()   
     
   def recursive(self):  
              
         myd=dir_namelist[self.j]  
         final_source=os.path.join(source, myd)  
         instfolder=final_source+"\\_Installers"   
         b_dir=os.path.join(destination, myd)  
         pCS=instfolder+"\\cs"  
         pSMweb=instfolder+"\\SM.web"  
         pSMwin=instfolder+"\\SM.win"   
         self.j+=1         
         main(myd, instfolder, b_dir, pCS, pSMweb, pSMwin)  
           
 #initiate an object from "helper" class  
 help=utilities()  
   
 #e-mail messages  
 def m_success(b_dir, myd):  
   mydir=myd  
   msgsuccess = "From: Auto Builder " + "\n" + \  
       "To: Build downloader " + "\n" + \  
       "Subject: " +mydir + " downloaded successfully." + "\n\n" + \  
        ""+help.readlog(b_dir)  
     
     
   return (msgsuccess)  
   
 def m_notavail(myd):  
   mydir=myd  
   msgnotavail="From: Auto Builder " + "\n" + \  
       "To: Build downloader &lt;xxx@x.lk&gt;" + "\n" + \  
       "Subject:" +mydir + " _Installer folder is not available. Retrying previous..." + "\n\n" + \  
       "This is a system generated e-mail. Do not reply"  
   return (msgnotavail)  
   
 def m_exists(myd):   
   mydir=myd  
   msgexists="From: Auto Builder " + "\n" + \  
         "To: Build downloader &lt;xxx@x.lk&gt;" + "\n" + \  
         "Subject:" +mydir + "  build already exists." + "\n\n" + \  
         "This is a system generated e-mail. Do not reply"  
   return (msgexists)      
     
 def m_initiated(myd):    
   mydir=myd   
   msginitiated="From: Auto Builder " + "\n" + \  
       "To: Build downloader &lt;xxx@x.lk&gt;" + "\n" + \  
       "Subject:" +mydir + " build download started..." + "\n\n" + \  
       "This is a system generated e-mail. Do not reply"  
   return (msginitiated)   
   
 def m_retry(myd):  
     
   msgretry="From: Auto Builder " + "\n" + \  
         "To: Build downloader &lt;xxx@x.lk&gt;" + "\n" + \  
         "Subject:It seems that latest build folders are not available. retrying one before...\n\n" + \  
         "This is a system generated e-mail. Do not reply"  
   return (msgretry)   
   
 def main(myd, instfolder, b_dir, pCS, pSMweb, pSMwin):  
   try:  
     if myd is not None and instfolder is not None and b_dir is not None :  
       mydir=myd  
       installerfolder=instfolder  
       basedir=b_dir  
     if pCS is not None and pSMweb is not None and pSMwin is not None:  
       CS=pCS  
       SMweb=pSMweb  
       SMwin=pSMwin  
         
     if not os.path.exists(installerfolder):  
       print("installer folder is not available for build "+ mydir)   
            
       help.sendmail (m_notavail(myd))       
       print ("Successfully sent email")  
       help.recursive()  
         
     elif os.path.exists(basedir) :    
       print( mydir + " build already exists")    
         
       help.sendmail(m_exists(myd))       
       print ("Successfully sent email")  
         
         
     else:   
       print("Latest available successful build is " + mydir)    
       flag=0  
       #create the base directory  
       os.mkdir(basedir)  
       #create robocopy log file  
       open(basedir+"\\"+"robolog.txt", "w+")   
       #download CS  
       dest_cs=basedir + "\\cs"   
       if os.path.exists(installerfolder+"\\cs"):  
         help.sendmail(m_initiated(myd))  
         print("CS build copy initiated...")        
         subprocess.call("robocopy /S "+ CS+ " " + dest_cs + " /LOG+:" + basedir + "\\robolog.txt")        
         help.logging("- CS\n", b_dir, myd)      
         print("CS build copy completed")  
       else:  
         #os.mkdir(basedir)  
         help.logging("- CS files NOT found \n", b_dir, myd)  
         print("CS files NOT found")  
         flag+=1  
          
       #download Web  
       dest_web=basedir + "\\SM.web"   
       if os.path.exists(installerfolder+"\\SM.web"):  
         print("SM Web build copy initiated...")       
           
         subprocess.call("robocopy /S "+ SMweb+ " " + dest_web + " /LOG+:" + basedir + "\\robolog.txt")  
         help.logging("- SM Web\n", b_dir, myd)  
         print("SM Web build copy competed")  
       else:  
         help.logging("- SM.Web files NOT found\n", b_dir, myd)  
         print("SM.Web files NOT found")  
         flag+=1  
       #download win  
       dest_win=basedir + "\\SM.win"    
       if os.path.exists(installerfolder+"\\SM.win"):      
         print("SM Win build copy initiated...")     
           
         subprocess.call("robocopy /S "+ SMwin+ " " + dest_win + " /LOG+:" + basedir + "\\robolog.txt")  
         help.logging("- SM Win\n", b_dir, myd)  
         print("SM Win build copy completed")  
       else:  
         help.logging("- SM.Win files NOT found \n\n", b_dir, myd)  
         print("SM.Win files NOT found")  
         flag+=1  
       if flag==3:  
         print('It seems that latest build folders are not available. retrying one before...\n\n')  
         help.logging("It seems that latest build folders are not available. retrying one before...\n", b_dir, myd)  
         help.sendmail(m_retry(myd))  
         help.recursive()  
         flag=0  
       else:  
         #sending mail      
         print ("Successfully sent email")      
         help.sendmail(m_success(b_dir,myd ))   
         
       
   except Exception as e:  
       #exception handling  
       print ("Error occured:" , e)        
       msgerr = "From: Auto Builder " + "\n" + \  
             "To: Build downloader &lt;xxx@x.lk&gt;" + "\n" + \  
             "Subject: " +str(e) + " : An error has occured." + "\n\n" + \  
              "This is a system generated e-mail. Do not reply"  
       help.sendmail(msgerr)  
       help.logging(str(e), b_dir, myd)  
       help.recursive()  
       #input("Press to exit")  
         
 if __name__=='__main__':  
   
   main(mydir, installerfolder, basedir, CS, SMweb, SMwin)  

No comments:

Post a Comment