source: main/waeup.kofa/trunk/src/waeup/kofa/applicants/batching.py @ 17766

Last change on this file since 17766 was 17762, checked in by Henrik Bettermann, 6 months ago

ApplicantProcessor?: import applicant history
  • Property svn:keywords set to Id
File size: 19.9 KB
Line 
1## $Id: batching.py 17762 2024-05-11 19:52:43Z henrik $
2##
3## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
4## This program is free software; you can redistribute it and/or modify
5## it under the terms of the GNU General Public License as published by
6## the Free Software Foundation; either version 2 of the License, or
7## (at your option) any later version.
8##
9## This program is distributed in the hope that it will be useful,
10## but WITHOUT ANY WARRANTY; without even the implied warranty of
11## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12## GNU General Public License for more details.
13##
14## You should have received a copy of the GNU General Public License
15## along with this program; if not, write to the Free Software
16## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17##
18"""Batch processing for applicants.
19"""
20import unicodecsv as csv # XXX: csv ops should move to dedicated module.
21import grok
22from time import time
23from ast import literal_eval
24import unicodecsv
25from zope.schema import getFields
26from zope.interface import Interface
27from zope.component import queryUtility, getUtility
28from hurry.workflow.interfaces import IWorkflowState
29from zope.catalog.interfaces import ICatalog
30from waeup.kofa.interfaces import (
31    IBatchProcessor, IObjectConverter, FatalCSVError,
32    IGNORE_MARKER, DELETION_MARKER,
33    IObjectHistory, IUserAccount, DuplicationError)
34from waeup.kofa.interfaces import MessageFactory as _
35from waeup.kofa.payments.interfaces import IPayer
36from waeup.kofa.utils.batching import BatchProcessor
37from waeup.kofa.applicants.interfaces import (
38    IApplicantsContainer, IApplicant, IApplicantUpdateByRegNo,
39    IApplicantOnlinePayment)
40from waeup.kofa.applicants.workflow import  IMPORTABLE_STATES, CREATED
41
42class ApplicantsContainerProcessor(BatchProcessor):
43    """The Applicants Container Processor imports containers for applicants.
44    It does not import their content. There is nothing special about this
45    processor.
46    """
47    grok.implements(IBatchProcessor)
48    grok.provides(IBatchProcessor)
49    grok.context(Interface)
50    util_name = 'applicantscontainerprocessor'
51    grok.name(util_name)
52
53    name = _('ApplicantsContainer Processor')
54    mode = u'create'
55    iface = IApplicantsContainer
56
57    location_fields = ['code',]
58    factory_name = 'waeup.ApplicantsContainer'
59
60    def parentsExist(self, row, site):
61        return 'applicants' in site.keys()
62
63    def entryExists(self, row, site):
64        return row['code'] in site['applicants'].keys()
65
66    def getParent(self, row, site):
67        return site['applicants']
68
69    def getEntry(self, row, site):
70        if not self.entryExists(row, site):
71            return None
72        parent = self.getParent(row, site)
73        return parent.get(row['code'])
74
75    def addEntry(self, obj, row, site):
76        parent = self.getParent(row, site)
77        parent[row['code']] = obj
78        return
79
80    def delEntry(self, row, site):
81        parent = self.getParent(row, site)
82        del parent[row['code']]
83        return
84
85class ApplicantProcessor(BatchProcessor):
86    """The Applicant Processor imports application data (applicants).
87
88    In create mode `container_code` is required. If `application_number` is
89    given, an applicant with this number is created in the designated container.
90    If `application_number` is not given, a random `application_number` is
91    assigned. `applicant_id` is being determined by the system and can't be
92    imported.
93
94    In update or remove mode `container_code` and `application_number` columns
95    must not exist. The applicant object is solely localized by searching
96    the applicants catalog for `reg_number` or `applicant_id` .
97    """
98    grok.implements(IBatchProcessor)
99    grok.provides(IBatchProcessor)
100    grok.context(Interface)
101    util_name = 'applicantprocessor'
102    grok.name(util_name)
103    name = _('Applicant Processor')
104    iface = IApplicant
105    iface_byregnumber = IApplicantUpdateByRegNo
106    factory_name = 'waeup.Applicant'
107
108    mode = None
109
110    @property
111    def available_fields(self):
112        return sorted(list(set(
113            ['application_number', 'history',
114            'container_code','state','password'] + getFields(
115                self.iface).keys())))
116
117    def cheadminckHeaders(self, headerfields, mode='create'):
118        cond1 = 'container_code' in headerfields
119        cond2 = 'application_number' in headerfields
120        cond3 = 'applicant_id' in headerfields
121        cond4 = 'reg_number' in headerfields
122        if mode == 'create':
123            if not cond1:
124                raise FatalCSVError(
125                    "Need at least container_code column!")
126            if cond3:
127                raise FatalCSVError(
128                    "applicant_id can't be imported in create mode!")
129            for field in self.required_fields:
130                if not field in headerfields:
131                    raise FatalCSVError(
132                        "Need at least columns %s for import!" %
133                        ', '.join(["'%s'" % x for x in self.required_fields]))
134        if mode in ('update', 'remove'):
135            if not cond3 and not cond4:
136                raise FatalCSVError(
137                    "Need at least column reg_number or applicant_id!")
138            if cond1 or cond2:
139                raise FatalCSVError(
140                    "container_code or application_number can't be imported " +
141                    "in update or remove mode!")
142        # Check for fields to be ignored...
143        not_ignored_fields = [x for x in headerfields
144                              if not x.startswith('--')]
145        if len(set(not_ignored_fields)) < len(not_ignored_fields):
146            raise FatalCSVError(
147                "Double headers: each column name may only appear once.")
148        return True
149
150    def getLocator(self, row):
151        if row.get('container_code', None) not in (IGNORE_MARKER, None):
152            # create mode
153            return 'container_code'
154        elif row.get('applicant_id', None) not in (IGNORE_MARKER, None):
155            # update or remove mode
156            return 'applicant_id'
157        elif row.get('reg_number', None) not in (IGNORE_MARKER, None):
158            # update or remove mode
159            return 'reg_number'
160        else:
161            return None
162
163    def getParent(self, row, site):
164        result = None
165        if self.getLocator(row) == 'container_code':
166            result = site['applicants'].get(row['container_code'], None)
167        elif self.getLocator(row) == 'reg_number':
168            reg_number = row['reg_number']
169            cat = queryUtility(ICatalog, name='applicants_catalog')
170            results = list(
171                cat.searchResults(reg_number=(reg_number, reg_number)))
172            if results:
173                result = results[0].__parent__
174        elif self.getLocator(row) == 'applicant_id':
175            applicant_id = row['applicant_id']
176            cat = queryUtility(ICatalog, name='applicants_catalog')
177            results = list(
178                cat.searchResults(applicant_id=(applicant_id, applicant_id)))
179            if results:
180                result = results[0].__parent__
181        return result
182
183    def parentsExist(self, row, site):
184        return self.getParent(row, site) is not None
185
186    def getEntry(self, row, site):
187        if self.getLocator(row) == 'container_code':
188            if row.get('application_number', None) not in (IGNORE_MARKER, None):
189                if not self.parentsExist(row, site):
190                    return None
191                parent = self.getParent(row, site)
192                return parent.get(row['application_number'])
193            return None
194        if self.getLocator(row) == 'applicant_id':
195            applicant_id = row['applicant_id']
196            cat = queryUtility(ICatalog, name='applicants_catalog')
197            results = list(
198                cat.searchResults(applicant_id=(applicant_id, applicant_id)))
199            if results:
200                return results[0]
201        if self.getLocator(row) == 'reg_number':
202            reg_number = row['reg_number']
203            cat = queryUtility(ICatalog, name='applicants_catalog')
204            results = list(
205                cat.searchResults(reg_number=(reg_number, reg_number)))
206            if results:
207                return results[0]
208        return None
209
210    def entryExists(self, row, site):
211        return self.getEntry(row, site) is not None
212
213    def addEntry(self, obj, row, site):
214        parent = self.getParent(row, site)
215        parent.addApplicant(obj)
216        #parent.__parent__.logger.info(
217        #    'Applicant imported: %s' % obj.applicant_id)
218        history = IObjectHistory(obj)
219        history.addMessage(_('Application record imported'))
220        return
221
222    def delEntry(self, row, site):
223        applicant = self.getEntry(row, site)
224        if applicant is not None:
225            parent = applicant.__parent__
226            del parent[applicant.application_number]
227            #parent.__parent__.logger.info(
228            #    'Applicant removed: %s' % applicant.applicant_id)
229        pass
230
231    def updateEntry(self, obj, row, site, filename):
232        """Update obj to the values given in row.
233        """
234        items_changed = ''
235        # Remove application_number from row if empty
236        if 'application_number' in row and row['application_number'] in (
237            None, IGNORE_MARKER):
238            row.pop('application_number')
239
240        # Update applicant_id fom application_number and container code
241        # if application_number is given
242        if 'application_number' in row:
243            obj.applicant_id = u'%s_%s' % (
244                row['container_code'], row['application_number'])
245            items_changed += ('%s=%s, ' % ('applicant_id', obj.applicant_id))
246            row.pop('application_number')
247
248        # Update
249        if 'password' in row:
250            passwd = row.get('password', IGNORE_MARKER)
251            if passwd not in ('', IGNORE_MARKER):
252                if passwd.startswith('{SSHA}'):
253                    # already encrypted password
254                    obj.password = passwd
255                elif passwd == DELETION_MARKER:
256                    obj.password = None
257                else:
258                    # not yet encrypted password
259                    IUserAccount(obj).setPassword(passwd)
260                items_changed += ('%s=%s, ' % ('password', passwd))
261            row.pop('password')
262
263        # Replace entire history
264        if 'history' in row:
265            new_history = row.get('history', IGNORE_MARKER)
266            if new_history not in (IGNORE_MARKER, ''):
267                history = IObjectHistory(obj)
268                history._annotations[
269                    history.history_key] = literal_eval(new_history)
270                items_changed += ('%s=%s, ' % ('history', new_history))
271            row.pop('history')
272
273        # Update registration state
274        if 'state' in row:
275            state = row.get('state', IGNORE_MARKER)
276            if state not in (IGNORE_MARKER, ''):
277                IWorkflowState(obj).setState(state)
278                msg = _("State '${a}' set", mapping = {'a':state})
279                history = IObjectHistory(obj)
280                history.addMessage(msg)
281                items_changed += ('%s=%s, ' % ('state', state))
282            row.pop('state')
283
284        # apply other values...
285        items_changed += super(ApplicantProcessor, self).updateEntry(
286            obj, row, site, filename)
287
288        # Log actions...
289        parent = self.getParent(row, site)
290        if self.getLocator(row) == 'container_code':
291            parent.__parent__.logger.info(
292                '%s - %s - imported: %s' % (self.name, filename, items_changed))
293        else:
294            parent.__parent__.logger.info(
295                '%s - %s - updated: %s' % (self.name, filename, items_changed))
296        return items_changed
297
298    def getMapping(self, path, headerfields, mode):
299        """Get a mapping from CSV file headerfields to actually used fieldnames.
300        """
301        result = dict()
302        reader = csv.reader(open(path, 'rb'))
303        raw_header = reader.next()
304        for num, field in enumerate(headerfields):
305            if field not in ['applicant_id', 'reg_number'] and mode == 'remove':
306                continue
307            if field == u'--IGNORE--':
308                # Skip ignored columns in failed and finished data files.
309                continue
310            result[raw_header[num]] = field
311        return result
312
313    def checkConversion(self, row, mode='create'):
314        """Validates all values in row.
315        """
316        iface = self.iface
317        if self.getLocator(row) == 'reg_number' or mode == 'remove':
318            iface = self.iface_byregnumber
319        converter = IObjectConverter(iface)
320        errs, inv_errs, conv_dict =  converter.fromStringDict(
321            row, self.factory_name, mode=mode)
322        cert = conv_dict.get('course1', None)
323        if cert is not None and (mode in ('create', 'update')):
324            # course1 application category must match container's.
325            site = grok.getSite()
326            parent = self.getParent(row, site)
327            if parent is None:
328                errs.append(('container', 'not found'))
329            elif cert.application_category != parent.application_category:
330                errs.append(('course1', 'wrong application category'))
331        if 'state' in row and \
332            not row['state'] in IMPORTABLE_STATES:
333            if row['state'] not in (IGNORE_MARKER, ''):
334                errs.append(('state','not allowed'))
335            else:
336                # state is an attribute of Applicant and must not
337                # be changed if empty
338                conv_dict['state'] = IGNORE_MARKER
339        application_number = row.get('application_number', None)
340        if application_number in (IGNORE_MARKER, ''):
341                conv_dict['application_number'] = IGNORE_MARKER
342        return errs, inv_errs, conv_dict
343
344    def checkUpdateRequirements(self, obj, row, site):
345        """Checks requirements the object must fulfill when being updated.
346
347        This method is not used in case of deleting or adding objects.
348
349        Returns error messages as strings in case of requirement
350        problems.
351        """
352        # Block applicant with backdoor ...
353        if obj.state == CREATED and obj.notice != 'reset':
354            return 'Applicant is blocked.'
355        return None
356
357class ApplicantOnlinePaymentProcessor(BatchProcessor):
358    """The Applicant Online Payment Processor imports applicant payment tickets.
359    The tickets are located in the applicant container.
360
361    The `checkConversion` method checks the format of the payment identifier.
362    In create mode it does also ensures that same p_id does not exist
363    elsewhere. It must be portal-wide unique.
364
365    When adding a payment ticket, the `addEntry` method checks if a
366    payment has already been made. If so, a `DuplicationError` is raised.
367    """
368    grok.implements(IBatchProcessor)
369    grok.provides(IBatchProcessor)
370    grok.context(Interface)
371    util_name = 'applicantpaymentprocessor'
372    grok.name(util_name)
373
374    name = _('ApplicantOnlinePayment Processor')
375    iface = IApplicantOnlinePayment
376    factory_name = 'waeup.ApplicantOnlinePayment'
377
378    location_fields = ['applicant_id',]
379
380    @property
381    def available_fields(self):
382        af = sorted(list(set(
383            self.location_fields + getFields(self.iface).keys())) +
384            ['p_id',])
385        af.remove('display_item')
386        return af
387
388    def checkHeaders(self, headerfields, mode='ignore'):
389        super(ApplicantOnlinePaymentProcessor, self).checkHeaders(headerfields)
390        if mode in ('update', 'remove') and not 'p_id' in headerfields:
391            raise FatalCSVError(
392                "Need p_id for import in update and remove modes!")
393        return True
394
395    def getMapping(self, path, headerfields, mode):
396        """Get a mapping from CSV file headerfields to actually used fieldnames.
397
398        """
399        result = dict()
400        reader = unicodecsv.reader(open(path, 'rb'))
401        raw_header = reader.next()
402        for num, field in enumerate(headerfields):
403            if field not in  ['applicant_id', 'p_id'] and mode == 'remove':
404                continue
405            if field == u'--IGNORE--':
406                continue
407            result[raw_header[num]] = field
408        return result
409
410    def parentsExist(self, row, site):
411        return self.getParent(row, site) is not None
412
413    def getParent(self, row, site):
414        applicant_id = row['applicant_id']
415        cat = queryUtility(ICatalog, name='applicants_catalog')
416        results = list(
417            cat.searchResults(applicant_id=(applicant_id, applicant_id)))
418        if results:
419            return results[0]
420        return None
421
422    def getEntry(self, row, site):
423        applicant = self.getParent(row, site)
424        if applicant is None:
425            return None
426        p_id = row.get('p_id', None)
427        if p_id in (None, IGNORE_MARKER):
428            return None
429        # We can use the hash symbol at the end of p_id in import files
430        # to avoid annoying automatic number transformation
431        # by Excel or Calc
432        p_id = p_id.strip('#')
433        entry = applicant.get(p_id)
434        return entry
435
436    def entryExists(self, row, site):
437        return self.getEntry(row, site) is not None
438
439    def updateEntry(self, obj, row, site, filename):
440        """Update obj to the values given in row.
441        """
442        items_changed = super(ApplicantOnlinePaymentProcessor, self).updateEntry(
443            obj, row, site, filename)
444        applicant = self.getParent(row, site)
445        applicant.__parent__.__parent__.logger.info(
446            '%s - %s - %s - updated: %s'
447            % (self.name, filename, applicant.applicant_id, items_changed))
448        return
449
450    def samePaymentMade(self, applicant, category):
451        for key in applicant.keys():
452            ticket = applicant[key]
453            if ticket.p_state == 'paid' and\
454                ticket.p_category == category:
455                return True
456        return False
457
458    def addEntry(self, obj, row, site):
459        applicant = self.getParent(row, site)
460        p_id = row['p_id'].strip('#')
461        if not (obj.p_item and obj.p_item.startswith(
462            'Balance')) and self.samePaymentMade(applicant, obj.p_category):
463            applicant.__parent__.__parent__.logger.info(
464                '%s - %s - previous update cancelled'
465                % (self.name, applicant.applicant_id))
466            raise DuplicationError('Payment has already been made.')
467        applicant[p_id] = obj
468        return
469
470    def delEntry(self, row, site):
471        payment = self.getEntry(row, site)
472        applicant = self.getParent(row, site)
473        if payment is not None:
474            applicant.__parent__.__parent__.logger.info('%s - Payment ticket removed: %s'
475                % (applicant.applicant_id, payment.p_id))
476            del applicant[payment.p_id]
477        return
478
479    def checkConversion(self, row, mode='ignore'):
480        """Validates all values in row.
481        """
482        errs, inv_errs, conv_dict = super(
483            ApplicantOnlinePaymentProcessor, self).checkConversion(row, mode=mode)
484        # We have to check p_id.
485        p_id = row.get('p_id', None)
486        if mode == 'create' and p_id in (None, IGNORE_MARKER):
487            timestamp = ("%d" % int(time()*10000))[1:]
488            p_id = "p%s" % timestamp
489            conv_dict['p_id'] = p_id
490            return errs, inv_errs, conv_dict
491        elif p_id in (None, IGNORE_MARKER):
492            errs.append(('p_id','missing'))
493            return errs, inv_errs, conv_dict
494        else:
495            p_id = p_id.strip('#')
496            if not len(p_id) == 14:
497                errs.append(('p_id','invalid length'))
498                return errs, inv_errs, conv_dict
499        if mode == 'create':
500            cat = getUtility(ICatalog, name='payments_catalog')
501            results = list(cat.searchResults(p_id=(p_id, p_id)))
502            if len(results) > 0:
503                sids = [IPayer(payment).id for payment in results]
504                sids_string = ''
505                for id in sids:
506                    sids_string += '%s ' % id
507                errs.append(('p_id','p_id exists in %s' % sids_string))
508                return errs, inv_errs, conv_dict
509        return errs, inv_errs, conv_dict
Note: See TracBrowser for help on using the repository browser.