source: main/waeup.ikoba/trunk/src/waeup/ikoba/customers/batching.py @ 11985

Last change on this file since 11985 was 11985, checked in by Henrik Bettermann, 10 years ago

pep8
File size: 14.1 KB
Line 
1## $Id: batching.py 11891 2014-10-28 20:02:45Z henrik $
2##
3## Copyright (C) 2014 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 components for customer objects.
19
20Batch processors eat CSV files to add, update or remove large numbers
21of certain kinds of objects at once.
22
23Here we define the processors for customers specific objects like
24customers and subobjects.
25"""
26import grok
27import unicodecsv as csv  # XXX: csv ops should move to dedicated module.
28from time import time
29from datetime import datetime
30from zope.i18n import translate
31from zope.interface import Interface
32from zope.schema import getFields
33from zope.component import queryUtility, getUtility, createObject
34from zope.event import notify
35from zope.catalog.interfaces import ICatalog
36from hurry.workflow.interfaces import IWorkflowState, IWorkflowInfo
37from waeup.ikoba.interfaces import (
38    IBatchProcessor, FatalCSVError, IObjectConverter, IUserAccount,
39    IObjectHistory, IGNORE_MARKER)
40from waeup.ikoba.interfaces import IIkobaUtils
41from waeup.ikoba.interfaces import MessageFactory as _
42from waeup.ikoba.customers.interfaces import (
43    ICustomer, ICustomerUpdateByRegNo)
44from waeup.ikoba.customers.workflow import  (
45    IMPORTABLE_STATES, IMPORTABLE_TRANSITIONS)
46from waeup.ikoba.utils.batching import BatchProcessor
47
48
49class CustomerProcessor(BatchProcessor):
50    """A batch processor for ICustomer objects.
51    """
52    grok.implements(IBatchProcessor)
53    grok.provides(IBatchProcessor)
54    grok.context(Interface)
55    util_name = 'customerprocessor'
56    grok.name(util_name)
57
58    name = _('Customer Processor')
59    iface = ICustomer
60    iface_byregnumber = ICustomerUpdateByRegNo
61
62    location_fields = []
63    factory_name = 'waeup.Customer'
64
65    @property
66    def available_fields(self):
67        fields = getFields(self.iface)
68        return sorted(list(set(
69            ['customer_id', 'reg_number',
70            'password', 'state', 'transition'] + fields.keys())))
71
72    def checkHeaders(self, headerfields, mode='create'):
73        if 'state' in headerfields and 'transition' in headerfields:
74            raise FatalCSVError(
75                "State and transition can't be  imported at the same time!")
76        if not 'reg_number' in headerfields and not 'customer_id' \
77            in headerfields:
78            raise FatalCSVError(
79                "Need at least columns customer_id or reg_number " +
80                "for import!")
81        if mode == 'create':
82            for field in self.required_fields:
83                if not field in headerfields:
84                    raise FatalCSVError(
85                        "Need at least columns %s for import!" %
86                        ', '.join(["'%s'" % x for x in self.required_fields]))
87        # Check for fields to be ignored...
88        not_ignored_fields = [x for x in headerfields
89                              if not x.startswith('--')]
90        if len(set(not_ignored_fields)) < len(not_ignored_fields):
91            raise FatalCSVError(
92                "Double headers: each column name may only appear once.")
93        return True
94
95    def parentsExist(self, row, site):
96        return 'customers' in site.keys()
97
98    def getLocator(self, row):
99        if row.get('customer_id', None) not in (None, IGNORE_MARKER):
100            return 'customer_id'
101        elif row.get('reg_number', None) not in (None, IGNORE_MARKER):
102            return 'reg_number'
103        else:
104            return None
105
106    # The entry never exists in create mode.
107    def entryExists(self, row, site):
108        return self.getEntry(row, site) is not None
109
110    def getParent(self, row, site):
111        return site['customers']
112
113    def getEntry(self, row, site):
114        if not 'customers' in site.keys():
115            return None
116        if self.getLocator(row) == 'customer_id':
117            if row['customer_id'] in site['customers']:
118                customer = site['customers'][row['customer_id']]
119                return customer
120        elif self.getLocator(row) == 'reg_number':
121            reg_number = row['reg_number']
122            cat = queryUtility(ICatalog, name='customers_catalog')
123            results = list(
124                cat.searchResults(reg_number=(reg_number, reg_number)))
125            if results:
126                return results[0]
127        return None
128
129    def addEntry(self, obj, row, site):
130        parent = self.getParent(row, site)
131        parent.addCustomer(obj)
132        # Reset _curr_cust_id if customer_id has been imported
133        if self.getLocator(row) == 'customer_id':
134            parent._curr_cust_id -= 1
135        # We have to log this if state is provided. If not,
136        # logging is done by the event handler handle_customer_added
137        if 'state' in row:
138            parent.logger.info('%s - Customer record created' % obj.customer_id)
139        history = IObjectHistory(obj)
140        history.addMessage(_('Customer record created'))
141        return
142
143    def delEntry(self, row, site):
144        customer = self.getEntry(row, site)
145        if customer is not None:
146            parent = self.getParent(row, site)
147            parent.logger.info('%s - Customer removed' % customer.customer_id)
148            del parent[customer.customer_id]
149        pass
150
151    def checkUpdateRequirements(self, obj, row, site):
152        """Checks requirements the object must fulfill when being updated.
153
154        This method is not used in case of deleting or adding objects.
155
156        Returns error messages as strings in case of requirement
157        problems.
158        """
159        transition = row.get('transition', IGNORE_MARKER)
160        if transition not in (IGNORE_MARKER, ''):
161            allowed_transitions = IWorkflowInfo(obj).getManualTransitionIds()
162            if transition not in allowed_transitions:
163                return 'Transition not allowed.'
164        return None
165
166    def updateEntry(self, obj, row, site, filename):
167        """Update obj to the values given in row.
168        """
169        items_changed = ''
170
171        # Remove customer_id from row if empty
172        if 'customer_id' in row and row['customer_id'] in (None, IGNORE_MARKER):
173            row.pop('customer_id')
174
175        # Update password
176        # XXX: Take DELETION_MARKER into consideration
177        if 'password' in row:
178            passwd = row.get('password', IGNORE_MARKER)
179            if passwd not in ('', IGNORE_MARKER):
180                if passwd.startswith('{SSHA}'):
181                    # already encrypted password
182                    obj.password = passwd
183                else:
184                    # not yet encrypted password
185                    IUserAccount(obj).setPassword(passwd)
186                items_changed += ('%s=%s, ' % ('password', passwd))
187            row.pop('password')
188
189        # Update registration state
190        if 'state' in row:
191            state = row.get('state', IGNORE_MARKER)
192            if state not in (IGNORE_MARKER, ''):
193                value = row['state']
194                IWorkflowState(obj).setState(value)
195                msg = _("State '${a}' set", mapping={'a': value})
196                history = IObjectHistory(obj)
197                history.addMessage(msg)
198                items_changed += ('%s=%s, ' % ('state', state))
199            row.pop('state')
200
201        if 'transition' in row:
202            transition = row.get('transition', IGNORE_MARKER)
203            if transition not in (IGNORE_MARKER, ''):
204                value = row['transition']
205                IWorkflowInfo(obj).fireTransition(value)
206                items_changed += ('%s=%s, ' % ('transition', transition))
207            row.pop('transition')
208
209        # apply other values...
210        items_changed += super(CustomerProcessor, self).updateEntry(
211            obj, row, site, filename)
212
213        # Log actions...
214        parent = self.getParent(row, site)
215        if hasattr(obj, 'customer_id'):
216            # Update mode: the customer exists and we can get the customer_id.
217            # Create mode: the record contains the customer_id
218            parent.logger.info(
219                '%s - %s - %s - updated: %s'
220                % (self.name, filename, obj.customer_id, items_changed))
221        else:
222            # Create mode: the customer does not yet exist
223            # XXX: It seems that this never happens because customer_id
224            # is always set.
225            parent.logger.info(
226                '%s - %s - %s - imported: %s'
227                % (self.name, filename, obj.customer_id, items_changed))
228        return items_changed
229
230    def getMapping(self, path, headerfields, mode):
231        """Get a mapping from CSV file headerfields to actually used fieldnames.
232        """
233        result = dict()
234        reader = csv.reader(open(path, 'rb'))
235        raw_header = reader.next()
236        for num, field in enumerate(headerfields):
237            if field not in ['customer_id', 'reg_number'] and mode == 'remove':
238                continue
239            if field == u'--IGNORE--':
240                # Skip ignored columns in failed and finished data files.
241                continue
242            result[raw_header[num]] = field
243        return result
244
245    def checkConversion(self, row, mode='create'):
246        """Validates all values in row.
247        """
248        iface = self.iface
249        if mode in ['update', 'remove']:
250            if self.getLocator(row) == 'reg_number':
251                iface = self.iface_byregnumber
252        converter = IObjectConverter(iface)
253        errs, inv_errs, conv_dict = converter.fromStringDict(
254            row, self.factory_name, mode=mode)
255        if 'transition' in row:
256            if row['transition'] not in IMPORTABLE_TRANSITIONS:
257                if row['transition'] not in (IGNORE_MARKER, ''):
258                    errs.append(('transition', 'not allowed'))
259        if 'state' in row:
260            if row['state'] not in IMPORTABLE_STATES:
261                if row['state'] not in (IGNORE_MARKER, ''):
262                    errs.append(('state', 'not allowed'))
263                else:
264                    # State is an attribute of Customer and must not
265                    # be changed if empty.
266                    conv_dict['state'] = IGNORE_MARKER
267        try:
268            # Correct cust_id counter. As the IConverter for customers
269            # creates customer objects that are not used afterwards, we
270            # have to fix the site-wide customer_id counter.
271            site = grok.getSite()
272            customers = site['customers']
273            customers._curr_cust_id -= 1
274        except (KeyError, TypeError, AttributeError):
275                pass
276        return errs, inv_errs, conv_dict
277
278
279class CustomerProcessorBase(BatchProcessor):
280    """A base for customer subitem processor.
281
282    Helps reducing redundancy.
283    """
284    grok.baseclass()
285
286    # additional available fields
287    # beside 'customer_id', 'reg_number' and 'matric_number'
288    additional_fields = []
289
290    #: header fields additionally required
291    additional_headers = []
292
293    @property
294    def available_fields(self):
295        fields = ['customer_id', 'reg_number'] + self.additional_fields
296        return sorted(list(set(fields + getFields(
297                self.iface).keys())))
298
299    def checkHeaders(self, headerfields, mode='ignore'):
300        if not 'reg_number' in headerfields and not 'customer_id' \
301            in headerfields:
302            raise FatalCSVError(
303                "Need at least columns customer_id " +
304                "or reg_number for import!")
305        for name in self.additional_headers:
306            if not name in headerfields:
307                raise FatalCSVError(
308                    "Need %s for import!" % name)
309
310        # Check for fields to be ignored...
311        not_ignored_fields = [x for x in headerfields
312                              if not x.startswith('--')]
313        if len(set(not_ignored_fields)) < len(not_ignored_fields):
314            raise FatalCSVError(
315                "Double headers: each column name may only appear once.")
316        return True
317
318    def _getCustomer(self, row, site):
319        NON_VALUES = ['', IGNORE_MARKER]
320        if not 'customers' in site.keys():
321            return None
322        if row.get('customer_id', '') not in NON_VALUES:
323            if row['customer_id'] in site['customers']:
324                customer = site['customers'][row['customer_id']]
325                return customer
326        elif row.get('reg_number', '') not in NON_VALUES:
327            reg_number = row['reg_number']
328            cat = queryUtility(ICatalog, name='customers_catalog')
329            results = list(
330                cat.searchResults(reg_number=(reg_number, reg_number)))
331            if results:
332                return results[0]
333        return None
334
335    def parentsExist(self, row, site):
336        return self.getParent(row, site) is not None
337
338    def entryExists(self, row, site):
339        return self.getEntry(row, site) is not None
340
341    def checkConversion(self, row, mode='ignore'):
342        """Validates all values in row.
343        """
344        converter = IObjectConverter(self.iface)
345        errs, inv_errs, conv_dict = converter.fromStringDict(
346            row, self.factory_name, mode=mode)
347        return errs, inv_errs, conv_dict
348
349    def getMapping(self, path, headerfields, mode):
350        """Get a mapping from CSV file headerfields to actually used fieldnames.
351        """
352        result = dict()
353        reader = csv.reader(open(path, 'rb'))
354        raw_header = reader.next()
355        for num, field in enumerate(headerfields):
356            if field not in ['customer_id', 'reg_number', 'p_id', 'code', 'level'
357                             ] and mode == 'remove':
358                continue
359            if field == u'--IGNORE--':
360                # Skip ignored columns in failed and finished data files.
361                continue
362            result[raw_header[num]] = field
363        return result
Note: See TracBrowser for help on using the repository browser.