source: mauRepo/MolesManager/trunk/src/libs/migration/processor/dataEntity.py @ 8325

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/mauRepo/MolesManager/trunk/src/libs/migration/processor/dataEntity.py@8325
Revision 8325, 10.7 KB checked in by mnagni, 7 years ago (diff)

Great improvement on how SA alchemy session is managed

Line 
1'''
2Created on 15 Nov 2011
3
4@author: mnagni
5'''
6from MolesManager.moles3epb import Moles3EPB
7from ea_model.ceda_metadatamodel.ceda_observationcollection.ceda_observationcollection import \
8    CEDA_ObservationCollection
9from libs.migration.MigrationEPB import MigrationEPB
10from libs.migration.exception.exceptions import MigrationObjectException, NoAssociatedAuthor,\
11    migrationObjectDescription, NoAssociatedDeployments
12from libs.migration.processor.commons import findDeploymentsInDE,\
13    createMD_Identifier, extractContent,\
14    hasAtomDocumentSameHash, createCI_Citation, createCI_Date, findPublishedDate,\
15    isoDateTimeStringToTimeDate, findUpdatedDate, createDate,\
16    calculateHash, extractUpdateFrequency, findDOIInMigrationDocument,\
17    getAtomDocumentHash
18from libs.migration.processor.deployment import DeploymentProcessor
19from ea_model.moles3_4.utilities.mo_publicationstatevalue import MO_PublicationStateValue
20from logging import StreamHandler
21import logging
22from ea_model.iso_19115_2006_metadata_corrigendum.citation_and_responsible_party_information.ci_datetypecode import CI_DateTypeCode
23from MolesManager.ceda_guid import CedaGUID
24from ea_model.iso_19115_2006_metadata_corrigendum.maintenance_information.md_maintenancefrequencycode import MD_MaintenanceFrequencyCode
25from libs.migration.processor.EPBRepo import EPBRepo
26
27class DataEntityProcessor(object):
28    log = logging.getLogger('DataEntityProcessor')
29    log.addHandler(StreamHandler())
30    log.setLevel(logging.INFO) 
31    '''
32        Processes a DataEntityMigration item. Note that each DataEntity is associated to a "dataent_xxxx" file in Moles2
33    '''       
34    def __init__(self, dataEntityMigration):
35        '''
36            Initializes the class
37            @param _dataEntityMigration: the DataEntityMigration instance
38        '''
39        if dataEntityMigration is None:
40            raise MigrationObjectException("DataEntityProcessor cannot process an None item")
41        self._dataEntityMigration = dataEntityMigration       
42   
43    def _processCitation(self, ceda_observationCollection):
44        # TDB - Check that if is an update or not!
45        contentDict = extractContent(self._dataEntityMigration)
46        if not contentDict.has_key('citation'):
47            DataEntityProcessor.log.info("The migration object "+ migrationObjectDescription(self._dataEntityMigration) \
48                                         + " has not associated cedacat:citation")
49        else:
50            ci_dates = []
51            doc_date = findPublishedDate(self._dataEntityMigration)           
52            if doc_date:
53                i_date = createDate(isoDateTimeStringToTimeDate(doc_date))           
54                ci_dates.append(createCI_Date(CI_DateTypeCode.cl_publication, date = i_date))
55
56            doc_date = findUpdatedDate(self._dataEntityMigration)           
57            if doc_date:
58                i_date = createDate(isoDateTimeStringToTimeDate(doc_date))               
59                ci_dates.append(createCI_Date(CI_DateTypeCode.cl_revision, date = i_date))
60               
61            i_citation = createCI_Citation(title = 'ceda_moles2_citation', date=ci_dates)
62            newIdentifier = createMD_Identifier(code = contentDict['citation'], authority=i_citation)
63            ceda_observationCollection.identifier.append(newIdentifier)
64   
65    def _getObservationCollection(self):
66        return EPBRepo.moles3EPB.search(CEDA_ObservationCollection, self._dataEntityMigration.ceda_observation_coll_id)
67   
68    def _processObservationCollection(self):                                   
69        #Moles3 object exists...
70        if self._dataEntityMigration.ceda_observation_coll_id: 
71            # ...and the data entity document has not changed
72            if hasAtomDocumentSameHash(self._dataEntityMigration):
73                return self._getObservationCollection()
74            else:
75                self.updateObservationCollection()
76
77        #... does not exist so create it
78        return self.createNewObservationCollection()
79
80    def updateObservationCollection(self):
81        """
82            Updated the existing CEDA_ObservationCollection instance binded to the self._dataEntityMigration object.
83            @return: the persisted and updated CEDA_ObservationCollection element
84        """       
85        ceda_observationCollection = self._getObservationCollection()
86        # TBD
87        #self._processCitation(ceda_observationCollection)
88        return ceda_observationCollection
89
90    def _update(self): 
91        # ...and the data entity document has not changed
92        if hasAtomDocumentSameHash(self._dataEntityMigration):
93                return self._getObservationCollection()
94        else:
95                return self.updateObservationCollection() 
96
97    def _create(self): 
98        """
99            Creates a new CEDA_ObservationCollection instance in the Moles3DB using the self._dataEntityMigration object.
100            If successful adds the new instance ID to the related DataEntityMigration object
101            @return: the persisted CEDA_ObservationCollection element
102        """
103        ceda_observationCollection = CEDA_ObservationCollection()
104        self._processCitation(ceda_observationCollection)
105        #self._processResult(ceda_observationCollection)
106        ceda_observationCollection.publicationState = MO_PublicationStateValue.cl_working
107                 
108        EPBRepo.moles3EPB.persistInstance(ceda_observationCollection)
109       
110        EPBRepo.migrationEPB.updateMigrationObject(self._dataEntityMigration, {'ceda_observation_coll_id': ceda_observationCollection.id})
111
112       
113        #Adds the CedaGUID
114        ceda_guid = CedaGUID()
115        ceda_guid.id = calculateHash(self._dataEntityMigration.data_ent_id)
116        ceda_guid.ceda_observationcollection = ceda_observationCollection.id
117        EPBRepo.moles3EPB.persistInstance(ceda_guid)
118        DataEntityProcessor.log.info("GUID for this ObservationCollection: %s" % (ceda_guid.id))
119       
120        return ceda_observationCollection
121
122    def _processResultAccumulation(self, ceda_observation):       
123        ceda_observation = EPBRepo.moles3EPB.loadAttributes(ceda_observation, "resultAccumulation")       
124        if ceda_observation.resultAccumulation is None:
125            updateFrequency = extractUpdateFrequency(self._dataEntityMigration)
126            if updateFrequency:
127                resultAccumulation = MD_MaintenanceFrequencyCode.from_string(updateFrequency)
128                EPBRepo.moles3EPB.updateCedaObject(ceda_observation, {'resultAccumulation': resultAccumulation})
129
130    def _processDOI(self, deploymentMigration, ceda_observation, deProcessor, single_deployment):       
131        doi = findDOIInMigrationDocument(deploymentMigration)                                                             
132        if single_deployment:
133            if doi is None:
134                doi = findDOIInMigrationDocument(self._dataEntityMigration)
135                    #collection_identifier = Moles3EPB.extractCollectionIdentifierByTitle(MD_CODE_MOLES2_CITATION, self.migrationSessions.molesSession)
136                    #if collection_identifier.count()==1:
137                    #    ceda_observation.identifier.append(collection_identifier.first())
138               
139        deProcessor.assignDOI(ceda_observation, doi)
140
141    def _processDeploymentMigration(self, deploymentMigration, single_deployment):                                     
142        deProcessor = DeploymentProcessor(self._dataEntityMigration, deploymentMigration)
143        try:
144            DataEntityProcessor.log.info("Processing deployment: %s" % (migrationObjectDescription(deploymentMigration)))
145            ceda_observation = deProcessor.process()
146            try:           
147                self._processResultAccumulation(ceda_observation)
148            except Exception as ex:
149                pass   
150            try:                           
151                self._processDOI(deploymentMigration, ceda_observation, deProcessor, single_deployment)
152            except Exception as ex:
153                pass               
154
155                if not hasAtomDocumentSameHash(deploymentMigration):
156                    doc_hash = getAtomDocumentHash(deploymentMigration.docStatus, deploymentMigration.docType, deploymentMigration.docOwner, deploymentMigration.docName)
157                    EPBRepo.migrationEPB.updateMigrationObject(deploymentMigration, {'doc_hash': doc_hash})               
158        except NoAssociatedAuthor as ex:
159            raise ex                 
160        except Exception as ex:
161            #self.migrationSessions.molesSession.rollback()
162            #self.migrationSessions.migrationSession.rollback()               
163            raise MigrationObjectException(ex)                     
164       
165        return ceda_observation
166   
167    def process(self):
168        cedaObservationCollection = None
169        exs = []
170        DataEntityProcessor.log.info("Processing dataEntity: %s" % (migrationObjectDescription(self._dataEntityMigration)))
171        try :
172            if self._dataEntityMigration.ceda_observation_coll_id:
173                cedaObservationCollection = self._update()
174            else:
175                cedaObservationCollection = self._create()
176        except Exception as ex:
177            exs.append(ex)
178            return exs       
179
180       
181        #retrieves the associated deployment links from the data_entity
182        deploymentsLinks = findDeploymentsInDE(self._dataEntityMigration)
183       
184        #retrieves the DataEntityMigration sorted by creation date
185        deploymentMigrations = EPBRepo.migrationEPB.getAllDeploymentsMigrationByDataEntitySortedByDate( \
186                                                self._dataEntityMigration, deploymentsLinks)
187        howManydm = 0
188        if deploymentMigrations:
189            howManydm = len(deploymentMigrations)
190        if howManydm == 0:
191            exs.append(NoAssociatedDeployments(self._dataEntityMigration))
192        for deploymentMigration in deploymentMigrations:
193            try:
194                ceda_observation = self._processDeploymentMigration(deploymentMigration, howManydm == 1)
195               
196                #Check if a doi has been already assigned     
197                cedaObservationCollection = EPBRepo.moles3EPB.loadAttributes(cedaObservationCollection, 'member')                           
198                member = cedaObservationCollection.member
199                if member and not (ceda_observation in member):
200                    EPBRepo.moles3EPB.updateCedaObject(cedaObservationCollection, {'member': ceda_observation})
201                                     
202            except Exception as ex:
203                exs.append(ex)
204            except RuntimeError as er:
205                print er               
206        #self.migrationSessions.molesSession.commit()                       
207               
208        return exs
Note: See TracBrowser for help on using the repository browser.