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', varname='string'),

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

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

        fields.DateField(id='3', label='Date', varname='date'),

        fields.BoolField(id='4', label='Boolean', varname='boolean')

    ]

    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')

    assert 'Download sample file' in resp

    sample_resp = resp.click('Download sample file')

    today = datetime.date.today()

    assert sample_resp.text == 'string,list,date,boolean\r\nstring-value,list-value,%s,Yes\r\n' % today

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

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

    assert 'Invalid file format.' in resp

    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 contains less columns that fields with varnames.' in resp

    resp.forms[0]['file'] = Upload('test.csv',

                                   b'string,list,date,boolean\ndata1,item1,invalid,Yes',

                                   'text/csv')

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

    assert 'invalid date value: invalid' in resp.text

    resp.forms[0]['file'] = Upload('test.csv',

                                   b'string,list,date,boolean\ndata1,item1,2020-01-01,Yes',

                                   'text/csv')

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

    assert 'Data imported successfully.' in resp

import csv

import datetime

from django.utils.encoding import force_text

from django.utils.six import StringIO

from ..qommon.form import Form, FileWidget

from ..qommon import template

from wcs import fields

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

                  ('import-csv', 'import_csv'), ('data-sample-csv', 'data_sample_csv')]

    def data_sample_csv(self):

        carddef_fields = self.formdef.get_varname_fields()

        output = StringIO()

        csv_output = csv.writer(output)

        csv_output.writerow([f.varname for f in carddef_fields])

        sample_line = []

        for f in carddef_fields:

            if isinstance(f, fields.DateField):

                value = datetime.date.today()

            elif isinstance(f, fields.BoolField):

                value = _('Yes')

            elif isinstance(f, fields.EmailField):

                value = 'foo@example.com'

            else:

                value = '%s-value' % f.varname

            sample_line.append(value)

        csv_output.writerow(sample_line)

        response = get_response()

        response.set_content_type('text/plain')

        response.set_header('content-disposition', 'attachment; filename=%s-sample.csv' % self.formdef.url_name)

        return output.getvalue()

    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 as e:

                form.set_error('file', str(e))

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

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

        return template.QommonTemplateResponse(

                templates=['wcs/backoffice/card-data-import-form.html'],

                context={'form': form})

    def import_csv_submit(self, form):

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

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

            if b'\0' in content:

                raise ValueError(_('Invalid file format.'))

        else:

            raise ValueError(_('You have to enter a file.'))

        for charset in ('utf-8', 'iso-8859-15'):

            try:

                content = content.decode(charset)

                break

            except UnicodeDecodeError:

                continue

        try:

            dialect = csv.Sniffer().sniff(content)

        except csv.Error:

            dialect = None

        carddef_fields = self.formdef.get_varname_fields()

        carddef_fields_names = [f.varname for f in carddef_fields]

        reader = csv.reader(content.splitlines(), dialect=dialect)

        try:

            caption = next(reader)

        except StopIteration:

            raise ValueError(_('Invalid CSV file.'))

        if len(caption) < len(carddef_fields_names):

            raise ValueError(_('File contains less columns that fields with varnames.'))

        missing_fields = set(carddef_fields_names) - set(caption)

        if missing_fields:

            raise ValueError(_('File misses fields: %s.') % ', '.join(missing_fields))

        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', N_('Data imported successfully.'))

        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]

{% extends "wcs/backoffice/base.html" %}

{% load i18n %}

{% block appbar %}

<div id="appbar" class="highlight">

<h2>{% trans "Import CSV" %}</h2>

</div>

{% endblock %}

{% block content %}

<p>{% trans "You can add data to this card by uploading a file containing var names in columns captions." %}</p>

<p><a href="data-sample-csv">{% trans 'Download sample file' %}</a> {% trans "for this card" %}</p>

{{ form.render|safe }}

{% endblock %}