Produits Entr'ouvert » w.c.s.

from webtest import Upload

from quixote import cleanup, get_publisher, get_response

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

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

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

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

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

        fields.FileField(id='5', label='File')

    ]

    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 == 'Test,List,Date,Boolean,File\r\nvalue,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'Test,List,Date\ndata1,item1,invalid',

                                   'text/csv')

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

    assert 'CSV file contains less columns than card fields.' in resp.text

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

                                   b'Test,List,Date,Boolean,File\ndata1,item1,invalid,Yes,',

                                   'text/csv')

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

    assert 'time data \'invalid\' does not match format \'%Y-%m-%d\'' in resp.text

    data = [b'Test,List,Date,Boolean,File']

    for i in range(1, 150):

        data.append(b'data%d,item%d,2020-01-%02d,%d,filename-%d' % (i, i, i%31+1, i%2,i))

    resp.forms[0]['file'] = Upload('test.csv',b'\n'.join(data),

                                   'text/csv')

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

    assert 'Importing data into ' 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 ..qommon.afterjobs import AfterJob

from wcs import fields

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

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

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

    import_supporting_fields = ['StringField', 'TextField', 'DateField',

                                'BoolField', 'ItemField', 'ItemsField',

                                'PasswordField']

    def data_sample_csv(self):

        carddef_fields = self.formdef.get_all_fields()

        output = StringIO()

        csv_output = csv.writer(output)

        csv_output.writerow([f.label 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'

            elif f.__class__.__name__ not in self.import_supporting_fields:

                value = ''

            else:

                value = 'value'

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

        context = {}

        try:

            job = AfterJob.get(get_request().form.get('job'))

            get_response().add_javascript(['jquery.js', 'afterjob.js'])

            context['job'] = job

        except KeyError:

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

            context['form'] = form

        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=context)

    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_all_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):

            raise ValueError(_('CSV file contains less columns than card fields.'))

        data_lines = []

        for csv_line in reader:

            data_line = {}

            for i, field in enumerate(self.formdef.get_all_fields()):

                value = csv_line[i]

                # skip empty values

                if not value:

                    continue

                # skip unsupported field types

                if field.__class__.__name__ not in self.import_supporting_fields:

                    continue

                field_id = str(field.id)

                if field.convert_value_from_str:

                    value = field.convert_value_from_str(value)

                data_line[field_id] = value

                if field.store_display_value:

                    display_value = field.store_display_value(data_line, field_id)

                    data_line['%s_display' % field_id] = display_value

                if value and field.store_structured_value:

                    structured_value = field.store_structured_value(data_line, 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_line['%s_structured' % field_id] = structured_value

            data_lines.append(data_line)

        class ImportAction:

            def __init__(self, data_class, lines):

                self.data_class = data_class

                self.lines = lines

            def execute(self, job):

                for item in self.lines:

                    data_instance = self.data_class()

                    data_instance.data = item

                    data_instance.just_created()

                    data_instance.store()

                    data_instance.perform_workflow()

        action = ImportAction(self.formdef.data_class(), data_lines)

        job = get_response().add_after_job(str(N_('Importing data into \'%s\'') % self.formdef.name),

                                           action.execute)

        job.store()

        return redirect('import-csv?job=%s' % job.id)

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

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

{% load i18n %}

{% block appbar %}

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

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

</div>

{% endblock %}

{% block content %}

{% if job %}

<dl class="job-status">

  <dt>{{ job.label }}</dt>

  <dd><span class="afterjob" id="{{ job.id }}">{{ job.status }}</span></dd>

</dl>

<div class="done">

  <a data-redirect-auto="true" href=".">{% trans "Back to Listing" %}</a>

</div>

{% else %}

<p>{% trans "You can add data to this card by uploading a file." %}</p>

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

{{ form.render|safe }}

{% endif %}

{% endblock %}