Produits Entr'ouvert » w.c.s.

from webtest import Upload

def test_backoffice_cards_import_data_from_csv(pub, studio):

    user = create_user(pub)

    CardDef.wipe()

    carddef = CardDef()

    carddef.name = 'test'

    carddef.fields = [

        fields.StringField(id='1', label='Test', type='string', varname='string'),

        fields.ItemField(id='2', label='List', type='item', varname='list',

                         items=['item1', 'item2'])

    ]

    carddef.backoffice_submission_roles = user.roles

    carddef.workflow_roles = {'_editor': user.roles[0]}

    carddef.store()

    carddef.data_class().wipe()

    app = login(get_app(pub))

    resp = app.get(carddef.get_url())

    assert 'import-csv' in resp

    resp = resp.click('Import data from a CSV file')

    resp.forms[0]['file'] = Upload('test.csv', b'', 'text/csv')

    resp = resp.forms[0].submit()

    assert 'Invalid CSV file.' in resp

    resp.forms[0]['file'] = Upload('test.csv', b'data1,data2', 'text/csv')

    resp = resp.forms[0].submit()

    assert 'File misses fields: list, string.' in resp

    resp.forms[0]['file'] = Upload('test.csv', b'string,list\ndata1,item1', 'text/csv')

    resp = resp.forms[0].submit().follow()

    assert 'Data imported successfully.' in resp

import csv

from django.utils.encoding import force_text

from ..qommon.form import Form, FileWidget

    _q_exports = ['', 'csv', 'xls', 'ods', 'json', 'export', 'map', 'geojson', 'add',

                  ('import-csv', 'import_csv')]

    import_csv_message = N_(

        'You can add data to this card by uploading a file containing '

        'var names in columns captions.')

    import_csv_success_message = N_(

        'Data imported successfully.'

    )

    def import_csv(self):

        form = Form(enctype='multipart/form-data', use_tokens=False)

        form.add(FileWidget, 'file', title=_('File'), required=False)

        form.add_submit('submit', _('Submit'))

        form.add_submit('cancel', _('Cancel'))

        if form.get_widget('cancel').parse():

            return redirect('.')

        if form.is_submitted() and not form.has_errors():

            try:

                return self.import_csv_submit(form)

            except ValueError:

                pass

        get_response().breadcrumb.append(('import_csv', _('Import CSV')))

        html_top('data_management', _('Import CSV'))

        r = TemplateIO(html=True)

        r += htmltext('<h2>%s</h2>') % _('Import CSV')

        r += htmltext('<p>%s</p>') % _(self.import_csv_message)

        r += form.render()

        return r.getvalue()

    def import_csv_submit(self, form):

        if form.get_widget('file').parse():

            data = form.get_widget('file').parse().fp.read()

            text_data = force_text(data)

        else:

            form.set_error('file', _('You have to enter a file.'))

            raise ValueError()

        carddef_fields = self.formdef.get_varname_fields()

        carddef_fields_names = [f.varname for f in carddef_fields]

        reader = csv.reader(text_data.splitlines(), delimiter=',')

        try:

            caption = next(reader)

        except StopIteration:

            form.set_error('file', _('Invalid CSV file.'))

            raise ValueError()

        if len(caption) < len(carddef_fields_names):

            form.set_error('file', _('File contains less columns that fields with varnames.'))

            raise ValueError()

        missing_fields = set(carddef_fields_names) - set(caption)

        if missing_fields:

            form.set_error('file', _('File misses fields: %s.' % ', '.join(missing_fields)))

            raise ValueError()

        data_class = self.formdef.data_class()

        for csv_line in reader:

            data_line = dict(zip(caption, csv_line))

            data_instance = data_class()

            data = {}

            for field in carddef_fields:

                value = data_line[field.varname]

                field_id = str(field.id)

                if field.convert_value_from_anything:

                    value = field.convert_value_from_anything(value)

                data[field_id] = value

                if field.store_display_value:

                    display_value = field.store_display_value(data, field_id)

                    data['%s_display' % field_id] = display_value

                if value and field.store_structured_value:

                    structured_value = field.store_structured_value(data, field_id)

                    if structured_value:

                        if isinstance(structured_value, dict) and structured_value.get('id'):

                            formdata.data[field_id] = str(structured_value.get('id'))

                        data['%s_structured' % field_id] = structured_value

            data_instance.data = data

            data_instance.just_created()

            data_instance.store()

        get_session().message = ('info', self.import_csv_success_message)

        return redirect('.')

        if get_publisher().has_site_option('studio') and ('import-csv', 'import_csv') in self._q_exports:

            r += htmltext('<li><a rel="popup" href="import-csv">%s</a></li>') % _('Import data from a CSV file')

    def get_varname_fields(self):

        return [field for field in self.fields or [] if isinstance(field, fields.WidgetField) and field.varname]