#! /usr/bin/env python ############################################################# # # # Author: Bertrand Neron # # Organization:'Biological Software and Databases' Group, # # Institut Pasteur, Paris. # # Distributed under GPLv2 Licence. Please refer to the # # COPYING.LIB document. # # # ############################################################# import os , sys MOBYLEHOME = None if os.environ.has_key('MOBYLEHOME'): MOBYLEHOME = os.environ['MOBYLEHOME'] if not MOBYLEHOME: sys.exit( 'MOBYLEHOME must be defined in your environment' ) if ( os.path.join( MOBYLEHOME , 'Src' ) ) not in sys.path: sys.path.append( os.path.join( MOBYLEHOME , 'Src' ) ) import shutil import re from random import randint from time import time from Mobyle.ConfigManager import Config from Mobyle.MobyleError import MobyleError from Mobyle.Transaction import Transaction from lxml import etree class UpdateTransaction( Transaction ): def getAllJobsID(self): return [ node.get( 'id') for node in self._root.findall( 'jobList/job' ) ] def setJobID( self , oldID , newID ): """ set the ID of the job corresponding oldID to newID @param oldID: the ID of a job in this transaction @type oldID: string @param oldID: the new ID of a job in this transaction @type oldID: string """ jobNode = self._getJobNode( oldID ) jobNode.set( 'id' , newID ) self._setModified( True ) def removeJob(self , jobID ): """ remove the entry corresponding to this jobID @param jobID: the identifier of the job in the session ( the url without index.xml ) @type jobID: string @return: the job corresponding to jobID as a dict @rtype: dictionary @raise ValueError: if the jobID does not match any entry in this session """ jobNode = self._getJobNode(jobID) jobListNode = jobNode.getparent() jobListNode.remove( jobNode ) ## remove the ref of this job from the data dataUsedNodes = jobNode.findall( 'dataUsed' ) for dataUsedNode in dataUsedNodes: dataID = dataUsedNode.get( 'ref' ) try: dataNode = self._getDataNode( dataID ) except ValueError: continue else: try: usedByNode = dataNode.xpath( 'usedBy[@ref="%s"]' %jobID )[0] except IndexError: continue else: dataNode.remove( usedByNode ) dataProducedNodes = jobNode.findall( 'dataProduced' ) for dataProducedNode in dataProducedNodes: dataID = dataProducedNode.get( 'ref' ) try: dataNode = self._getDataNode( dataID ) except ValueError: continue else: try: producedByNode = dataNode.xpath( 'producedBy[@ref="%s"]' %jobID )[0] except IndexError: continue else: dataNode.remove( producedByNode ) self._setModified( True ) def setTicket( self , ticket_id, exp_date ): """ set the currently allocated ticket and its expiration date @param ticket_id: the ticket @type ticket_id: string @param exp_date: the expiration date @type exp_date: date """ tck = self._root.find( 'ticket' ) if tck is not None: self._setModified( False ) else: ticketNode = etree.Element( "ticket" ) ticket_idNode = etree.Element( "id" ) ticket_idNode.text = str( ticket_id ) exp_dateNode = etree.Element( "exp_date" ) exp_dateNode.text = str( exp_date ) ticketNode.append( ticket_idNode ) ticketNode.append( exp_dateNode ) self._root.append( ticketNode ) self._setModified( True ) def updateData( self , old_jobs_repository , new_jobs_repository ): modified = False usedBy_nodes = self._root.findall( 'dataList/data/usedBy' ) for usedBy_node in usedBy_nodes: old_job_url = usedBy_node.get( 'ref' ) match = re.search( '(.*)/(.*/.*)$' , old_job_url ) if match: jobs_repository = match.group(1) if jobs_repository == old_jobs_repository: job = match.group(2) new_job_url = "%s/%s" %( new_jobs_repository , job ) usedBy_node.set( 'ref' , new_job_url ) logger.debug( "usedBy node has been updated from %s to %s"%( old_job_url , new_job_url )) else: dataNode = usedBy_node.getparent() dataNode.remove( usedBy_node ) logger.debug( "usedBy node referencing job %s was removed" % old_job_url ) modified =True else: raise Exception( "usedBy node have a ref %s witch not match standard job url" %old_job_url) producedBy_nodes = self._root.findall( 'dataList/data/producedBy' ) for producedBy_node in producedBy_nodes: old_job_url = producedBy_node.get( 'ref' ) match = re.search( '(.*)/(.*/.*)$' , old_job_url ) if match: jobs_repository = match.group(1) if jobs_repository == old_jobs_repository: job = match.group(2) new_job_url = "%s/%s" %( new_jobs_repository , job ) producedBy_node.set( 'ref' , new_job_url ) logger.debug( "producedBy as been updated from %s to %s"%( old_job_url , new_job_url )) else: dataNode = producedBy_node.getparent() dataNode.remove( producedBy_node ) logger.debug( "producedBy node referencing job %s was removed" % old_job_url ) modified =True else: raise Exception( "producedBy node have a ref %s witch not match standard job url" %old_job_url) self._setModified( modified ) class SessionUpdater(object): def __init__(self , config , logger , old_jobs_repository_url ): self.cfg = config self.log = logger self.old_jobs_repository_url = old_jobs_repository_url.rstrip( '/' ) self.new_jobs_repository_url = self.cfg.results_url() self.skipped_sessions = [] def _on_error(self , err ): self.log.warning( "%s : %s" ( err.filename , err )) def update( self , repository , force = False ,dry_run = False): sessions_paths = os.walk( repository , onerror = self._on_error ) for path in sessions_paths: session_path = path[0] if '.session.xml' in path[2]: session_key = os.path.split( path[0] )[-1] self.fix( session_path , session_key , force = force , dry_run = dry_run ) # def rollback(self ): # session_dir_mtime = os.path.getmtime( session_path ) # session_dir_atime = os.path.getatime( session_path ) # index_mtime = os.path.getmtime( os.path.join( session_path , '.session.xml') ) # index_atime = os.path.getatime( os.path.join( session_path , '.session.xml') ) # index_path = os.path.join( session_path , '.session.xml') # # bckp_path = os.path.join( session_path , '.session.xml.bckp') # def fix( self , session_path , session_key , force = False ,dry_run = False): self.log.info( "====================" ) if session_path.find( '/authentified/') != -1 : session_type = 'authentified' else: session_type = 'anonymous' self.log.info( "updating %s session %s" %( session_type , session_key ) ) session_dir_mtime = os.path.getmtime( session_path ) session_dir_atime = os.path.getatime( session_path ) index_mtime = os.path.getmtime( os.path.join( session_path , '.session.xml') ) index_atime = os.path.getatime( os.path.join( session_path , '.session.xml') ) index_path = os.path.join( session_path , '.session.xml') try: os.utime( session_path , ( session_dir_atime , session_dir_mtime ) ) except Exception , err : self.log.warning( "cannot modify the mtime of %s : %s : fix this problem before rerun updater" %(session_path , err ) ) return bckp_path = os.path.join( session_path , '.session.xml.bckp') if not os.path.exists( bckp_path ): self.log.debug( "make %s back up" % session_path ) if not dry_run: try: shutil.copy( index_path , bckp_path ) except Exception ,err: self.log.warning( "cannot make %s back up " % index_path) else: os.utime( bckp_path , ( index_atime , index_mtime ) ) os.utime( session_path , ( session_dir_atime , session_dir_mtime ) ) else: pass try: ut = UpdateTransaction( index_path , UpdateTransaction.WRITE ) except Exception , err: self.log.warning( "cannot load session %s : %s : skipped" %( session_path , err ) ) return if session_type == 'authentified' : ticket_id = str(randint(0,1000000)) t = time() + 3600 ut.setTicket( ticket_id, t ) self.log.debug('set ticket ticket_id = %s , time = %d' %(ticket_id , t )) all_jobsID = ut.getAllJobsID() for id in all_jobsID: match = re.search( '(.*)/(.*/.*)$' , id ) if match: jobs_repository = match.group(1) if jobs_repository == self.old_jobs_repository_url: job = match.group(2) new_job_url = "%s/%s" %( self.new_jobs_repository_url , job ) self.log.debug( "update jobid %s by %s" %(id , new_job_url ) ) ut.setJobID( id , new_job_url ) else: self.log.debug( "remove job %s" %id ) if jobs_repository.find( 'http://mobyle.pasteur.fr/Mobyle/Results' ) != -1: self.log.warning( ' -- ATTENTION vielle session: %s -- ' % session_path ) ut.removeJob( id ) else: raise Exception( "job have an id %s witch not match standard job url" % id ) ut.updateData( self.old_jobs_repository_url , self.cfg.results_url() ) ut.commit() os.utime( index_path , ( index_atime , index_mtime ) ) os.utime( session_path , ( session_dir_atime , session_dir_mtime ) ) if __name__ == "__main__": from optparse import OptionParser parser = OptionParser( ) parser.add_option( "-s" , "--sessions", action="store", type = 'string', dest="sessions", help="comma separated list of sessions to update." ) parser.add_option( "--old-jobs-repos", action="store", type = 'string', dest="old_jobs_repository", help="the previous mobyle jobs repository url." ) parser.add_option( "-l" , "--log", action="store", type = 'string', dest = "log_file", help= "Path to the Logfile where put the logs.") parser.add_option( "-n" , "--dry-run", action="store_true", dest = "dry_run", default = False , help= "don't actually do anything.") parser.add_option( "-f" , "--force", action="store_true", dest = "force", default = False , help= "force to update session even if it was already updated.") parser.add_option("-v", "--verbose", action="count", dest="verbosity", default= 0, help="increase the verbosity level. There is 4 levels: Error messages (default), Warning (-v), Info (-vv) and Debug.(-vvv)") options, args = parser.parse_args() # if not options.old_host : # parser.error("option --old-host is mandatory") if not options.old_jobs_repository: parser.error("option --old-jobs-repos is mandatory") import logging logger = logging.getLogger( 'sessionUpdater' ) if options.log_file is None: handler = logging.StreamHandler( sys.stderr ) else: try: handler = logging.FileHandler( options.log_file , 'a' ) except(IOError , OSError) , err: print >> sys.stderr , "cannot log messages in %s: %s"%( options.log_file , err ) sys.exit(1) if options.verbosity < 2: formatter = logging.Formatter( '%(filename)-10s : %(levelname)-8s : %(asctime)s : %(message)s' , '%a, %d %b %Y %H:%M:%S' ) else: formatter = logging.Formatter( '%(filename)-10s : %(levelname)-8s : L %(lineno)d : %(asctime)s : %(message)s' , '%a, %d %b %Y %H:%M:%S' ) handler.setFormatter( formatter ) logger.addHandler( handler) if options.verbosity == 0: logger.setLevel( logging.WARNING ) elif options.verbosity == 1: logger.setLevel( logging.INFO ) else: logger.setLevel( logging.DEBUG ) config = Config() session_updater = SessionUpdater( config , logger , options.old_jobs_repository) session_repository = config.user_sessions_path() if options.sessions is None: session_updater.update( session_repository , force = options.force , dry_run = options.dry_run ) else: sessions = options.sessions.split( ',' ) for session in sessions: session_path = os.path.join( session_repository , session ) session_updater.update( session_path , force = options.force , dry_run = options.dry_run )