Produits Entr'ouvert » Passerelle

# passerelle.apps.csvdatasource

# Copyright (C) 2015  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import django.apps

class AppConfig(django.apps.AppConfig):

    name = 'csvdatasource'

    def get_after_urls(self):

        from . import urls

        return urls.urlpatterns

default_app_config = 'csvdatasource.AppConfig'

from django.utils.text import slugify

from django import forms

from .models import CsvDataSource

class CsvDataSourceForm(forms.ModelForm):

    class Meta:

        model = CsvDataSource

        exclude = ('slug', 'users')

    def save(self, commit=True):

        if not self.instance.slug:

            self.instance.slug = slugify(self.instance.title)

        return super(CsvDataSourceForm, self).save(commit=commit)

# -*- coding: utf-8 -*-

from __future__ import unicode_literals

from django.db import models, migrations

class Migration(migrations.Migration):

    dependencies = [

        ('base', '0001_initial'),

    ]

    operations = [

        migrations.CreateModel(

            name='CsvDataSource',

            fields=[

                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),

                ('title', models.CharField(max_length=50)),

                ('slug', models.SlugField()),

                ('description', models.TextField()),

                ('csv_file', models.FileField(upload_to=b'csv', verbose_name='CSV File')),

                ('columns_keynames', models.CharField(help_text='ex: id,text,data1,data2', max_length=256, verbose_name='Column keynames')),

                ('skip_header', models.BooleanField(default=False, verbose_name='Skip first line')),

                ('users', models.ManyToManyField(to='base.ApiUser', blank=True)),

            ],

            options={

                'verbose_name': 'CSV File',

            },

            bases=(models.Model,),

        ),

    ]

import re

import csv

from django.db import models

from django.utils.translation import ugettext_lazy as _

from django.core.urlresolvers import reverse

from passerelle.base.models import BaseResource

class CsvDataSource(BaseResource):

    csv_file = models.FileField(_('CSV File'), upload_to='csv')

    columns_keynames = models.CharField(max_length=256,

                            verbose_name=_('Column keynames'),

                            help_text=_('ex: id,text,data1,data2'))

    skip_header = models.BooleanField(_('Skip first line'), default=False)

    category = _('Data Sources')

    class Meta:

        verbose_name = _('CSV File')

    @classmethod

    def get_verbose_name(cls):

        return cls._meta.verbose_name

    @classmethod

    def get_icon_class(cls):

        return 'grid'

    @classmethod

    def is_enabled(cls):

        return True

    @classmethod

    def get_add_url(cls):

        return reverse('csvdatasource-add')

    def get_absolute_url(self):

        return reverse('csvdatasource-detail', kwargs={'slug': self.slug})

    def get_data(self, filter_criteria=None):

        def filter_row(row, titles, filter_criteria):

            if 'text' in titles:

                col = titles.index('text')

                if filter_criteria not in unicode(row[col], 'utf-8'):

                    return False

            return True

        content = self.csv_file.read()

        if not content:

            return None

        dialect = csv.Sniffer().sniff(content[:1024])

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

        if self.skip_header:

            reader.next()

        titles = [t.strip() for t in self.columns_keynames.split(',')]

        indexes = [titles.index(t) for t in titles if t]

        caption = [titles[i] for i in indexes]

        data = []

        for row in reader:

            if filter_criteria and not filter_row(row, titles, filter_criteria):

                continue

            try:

                line = [row[i] for i in indexes]

                data.append(dict(zip(caption, line)))

            except IndexError:

                continue

        return data

{% extends "passerelle/base.html" %}

{% load i18n passerelle %}

{% block more-user-links %}

{{ block.super }}

{% if object.id %}

<a href="{% url 'csvdatasource-detail' slug=object.slug %}">{{ object.title }}</a>

{% endif %}

{% endblock %}

{% block appbar %}

<h2>CSV - {{ object.title }}</h2>

{% if perms.csvdatasource.change_csvdatasource %}

<a rel="popup" class="button" href="{% url 'csvdatasource-edit' slug=object.slug %}">{% trans 'edit' %}</a>

{% endif %}

{% if perms.csvdatasource.delete_csvdatasource %}

<a rel="popup" class="button" href="{% url 'csvdatasource-delete' slug=object.slug %}">{% trans 'delete' %}</a>

{% endif %}

{% endblock %}

{% block content %}

<p>

  {% blocktrans with file=object.csv_file %}File: {{ file }} {% endblocktrans %}

</p>

<div>

<h3>{% trans 'Endpoints' %}</h3>

<ul>

  <li>{% trans "Returning all file lines: "%}

    <a href="{% url "csvdatasource-data" slug=object.slug %}">{%  url "csvdatasource-data" slug=object.slug %}</a>

  </li>

  <li>{% trans "Returning all lines containing 'abc' in 'text' column if defined : "%}

    <a href="{% url "csvdatasource-data" slug=object.slug %}?q=abc">{%  url "csvdatasource-data" slug=object.slug %}?q=abc</a>

  </li>

</ul>

</div>

{% if perms.base.view_accessright %}

<div>

<h3>{% trans "Security" %}</h3>

<p>

  {% trans 'Access is limited to the following API users:' %}

</p>

{% access_rights_table resource=object permission='can_access' %}

{% endif %}

</div>

{% endblock %}

from django.conf.urls import patterns, include, url

from django.contrib.auth.decorators import login_required

from passerelle.urls_utils import decorated_includes, required, app_enabled

from views import *

public_urlpatterns = patterns('',

    url(r'^(?P<slug>[\w,-]+)/$', CsvDataSourceDetailView.as_view(), name='csvdatasource-detail'),

    url(r'^(?P<slug>[\w,-]+)/data$', CsvDataView.as_view(), name='csvdatasource-data'),

)

management_urlpatterns = patterns('',

    url(r'^add$', CsvDataSourceCreateView.as_view(), name='csvdatasource-add'),

    url(r'^(?P<slug>[\w,-]+)/edit$', CsvDataSourceUpdateView.as_view(), name='csvdatasource-edit'),

    url(r'^(?P<slug>[\w,-]+)/delete$', CsvDataSourceDeleteView.as_view(), name='csvdatasource-delete'),

)

urlpatterns = required(

    app_enabled('csvdatasource'),

    patterns('',

        url(r'^csvdatasource/', include(public_urlpatterns)),

        url(r'^manage/csvdatasource/',

            decorated_includes(login_required, include(management_urlpatterns))),

    )

)

import json

from django.core.urlresolvers import reverse

from django.views.generic.edit import CreateView, UpdateView, DeleteView

from django.views.generic.detail import DetailView, SingleObjectMixin

from django.views.generic.base import View

from passerelle import utils

from .models import CsvDataSource

from .forms import CsvDataSourceForm

class CsvDataSourceDetailView(DetailView):

    model = CsvDataSource

class CsvDataSourceCreateView(CreateView):

    model = CsvDataSource

    template_name = 'passerelle/manage/service_form.html'

    form_class = CsvDataSourceForm

class CsvDataSourceUpdateView(UpdateView):

    model = CsvDataSource

    template_name = 'passerelle/manage/service_form.html'

    form_class = CsvDataSourceForm

class CsvDataSourceDeleteView(DeleteView):

    model = CsvDataSource

    template_name = 'passerelle/manage/service_confirm_delete.html'

    def get_success_url(self):

        return reverse('manage-home')

class CsvDataView(View, SingleObjectMixin):

    model = CsvDataSource

    @utils.protected_api('can_access')

    @utils.to_json('api')

    def get(self, request, *args, **kwargs):

        obj = self.get_object()

        return obj.get_data(request.GET.get('q'))

    'csvdatasource',

li.grid a { background-image: url(icons/icon-grid.png); }

li.grid a:hover { background-image: url(icons/icon-grid-hover.png); }

import pytest

from StringIO import StringIO

from django.core.files import File

data = """121;69981;DELANOUE;Eliot;H

525;6;DANIEL WILLIAMS;Shanone;F

253;67742;MARTIN;Sandra;F

511;38142;MARCHETTI;Lucie;F

235;22;MARTIN;Sandra;F

620;52156;ARNAUD;Mathis;H

902;36;BRIGAND;Coline;F

2179;48548;THEBAULT;Salima;F

3420;46;WILSON-LUZAYADIO;Anaelle;F

1421;55486;WONE;Fadouma;F

2841;51;FIDJI;Zakia;F

2431;59;BELICARD;Sacha;H

4273;60;GOUBERT;Adrien;H

4049;64;MOVSESSIAN;Dimitri;H

4605;67;ABDOU BACAR;Kyle;H

4135;22231;SAVERIAS;Marius;H

4809;75;COROLLER;Maelys;F

5427;117;KANTE;Aliou;H

7017;118;ANGELOTTI;Esther;F

116642;118;ZAHMOUM;Yaniss;H

"""

from csvdatasource.models import CsvDataSource

pytestmark = pytest.mark.django_db

def test_unfiltered_data():

    csv = CsvDataSource.objects.create(csv_file=File(StringIO(data), 'data.csv'),

                                       columns_keynames='field,,another_field,')

    result = csv.get_data()

    for item in result:

        assert 'field' in item

        assert 'another_field' in item

def test_good_filter_data():

    filter_criteria = 'ak'

    csv = CsvDataSource.objects.create(csv_file=File(StringIO(data), 'data.csv'),

                                       columns_keynames=',id,,text,')

    result = csv.get_data(filter_criteria)

    assert len(result)

    for item in result:

        assert 'id' in item

        assert 'text' in item

        assert filter_criteria in item['text']

def test_bad_filter_data():

    filter_criteria = 'bad'

    csv = CsvDataSource.objects.create(csv_file=File(StringIO(data), 'data.csv'),

                                       columns_keynames=',id,,text,')

    result = csv.get_data(filter_criteria)

    assert len(result) == 0

def test_useless_filter_data():

    csv = CsvDataSource.objects.create(csv_file=File(StringIO(data), 'data.csv'),

                                       columns_keynames='id,,nom,prenom,sexe')

    result = csv.get_data('Ali')

    assert len(result) == 20

def test_columns_keynames_with_spaces():

    csv = CsvDataSource.objects.create(csv_file=File(StringIO(data), 'data.csv'),

                                       columns_keynames='id , , nom,text , ')

    result = csv.get_data('Yaniss')

    assert len(result) == 1

def test_skipped_header_data():

    csv = CsvDataSource.objects.create(csv_file=File(StringIO(data), 'data.csv'),

                                       columns_keynames=',id,,text,',

                                       skip_header=True)

    result = csv.get_data('Eliot')

    assert len(result) == 0

def test_data():

    csv = CsvDataSource.objects.create(csv_file=File(StringIO(data), 'data.csv'),

                        columns_keynames='fam,id,, text,sexe ')

    result = csv.get_data('Sacha')

    assert result[0] == {'id': '59', 'text': 'Sacha',

                         'fam': '2431', 'sexe': 'H'}