Produits Entr'ouvert » Combo

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

# chrono - agendas system

# Copyright (C) 2016-2017  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 csv

import datetime

from itertools import groupby

from django.core.management.base import BaseCommand

from django.utils.text import slugify

from chrono.agendas.models import Agenda, Event

def csv_get_dialect(filename):

    with open(filename, 'rb') as fd:

        content = fd.read()

        content = content.decode('utf-8-sig', 'replace').encode('utf-8')

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

        return dialect

def csv_get_dict_data(filename, fieldnames=[], skip_header=True):

    with open(filename, 'rb') as fd:

        kwargs = {'fieldnames': fieldnames}

        kwargs['dialect'] = csv_get_dialect(filename)

        reader = csv.DictReader(fd, **kwargs)

        if skip_header:

            reader.next()

        return list(reader)

    return False

class BaseLoader(object):

    def __init__(self, filename):

        self.filename = filename

    def create(self):

        raise NotImplemented

class ConcertoLoader(BaseLoader):

    name = 'concerto'

    fieldnames = [

        'ID_LIE', 'DAT_JOUR_TMP', 'HEU_DEBUT_TMP', 'HEU_FIN_TMP',

        'NB_PLACESPOSS_TMP', 'NB_PLACESOCC_TMP', 'NB_PLACESRESFUT_TMP',

        'NB_PLACESRES_TMP', 'NB_PLACESLIBRES_TMP', 'NB_DUREE_TMP',

        'LIB_NOM_LIE', 'NB_PLACESRESFUTPOND_TMP']

    def __init__(self, filename, **kwargs):

        super(ConcertoLoader, self).__init__(filename)

    def create(self):

        data = csv_get_dict_data(self.filename, fieldnames=self.fieldnames)

        for nursery, events in groupby(data, lambda x: x['LIB_NOM_LIE']):

            nursery_slug = slugify(nursery)

            agenda, True = Agenda.objects.get_or_create(slug=nursery_slug, label=nursery)

            # purge all

            Event.objects.filter(agenda=agenda).delete()

            for event in events:

                event_day = datetime.datetime.strptime(event['DAT_JOUR_TMP'], '%Y%m%d')

                event_time = datetime.datetime.strptime(event['HEU_DEBUT_TMP'], '%H%M')

                event_datetime = datetime.datetime.combine(event_day, event_time.time())

                # XXX: value such as '-' OR '-76,24'

                try:

                    event_places = int(event['NB_PLACESLIBRES_TMP'].split(',')[0])

                except ValueError:

                    event_places = 0

                Event.objects.create(agenda=agenda, start_datetime=event_datetime, places=event_places)

class Command(BaseCommand):

    help = "Import events from csv file"

    def add_arguments(self, parser):

        parser.add_argument('loader', type=str, help='type of file')

        parser.add_argument('filename', type=str, help='name of the file to import')

    def handle(self, filename, loader, **options):

        for Loader in BaseLoader.__subclasses__():

            if Loader.name == loader:

                obj = Loader(filename)

                obj.create()

                break

        else:

            self.stdout.write('Invalid loader: %s' % loader)

ID_LIE;DAT_JOUR_TMP;HEU_DEBUT_TMP;HEU_FIN_TMP;NB_PLACESPOSS_TMP;NB_PLACESOCC_TMP;NB_PLACESRESFUT_TMP;NB_PLACESRES_TMP;NB_PLACESLIBRES_TMP;NB_DUREE_TMP;LIB_NOM_LIE;NB_PLACESRESFUTPOND_TMP

16;20170314;1430;1500;45;46;2;49;-3;4;Crèche Collective du BARON;-27,675

16;20170314;1500;1530;45;46;2;49;-3;4;Crèche Collective du BARON;-27,675

16;20170314;1530;1600;45;46;2;49;-3;4;Crèche Collective du BARON;-27,675

16;20170313;1600;1630;45;42,75;0;44,75;2,25;4;Crèche Collective du BARON;

16;20170313;1630;1700;45;42,75;0;44,75;2,25;4;Crèche Collective du BARON;

16;20170313;1700;1730;45;20,25;0;22,25;24,75;4;Crèche Collective du BARON;

16;20170313;1730;1800;45;20,25;0;22,25;24,75;4;Crèche Collective du BARON;

17;20170313;1700;1730;72;37;0;35;35;4;Crèche Collective des BLOSSIERES;

17;20170313;1730;1800;72;37;0;35;35;4;Crèche Collective des BLOSSIERES;

17;20170314;1000;1030;72;66,25;0;80,5;5,75;4;Crèche Collective des BLOSSIERES;

17;20170314;1030;1100;72;66,25;0;80,5;5,75;4;Crèche Collective des BLOSSIERES;

17;20170314;1300;1330;72;65;0;75;7;4;Crèche Collective des BLOSSIERES;

17;20170314;1330;1400;72;65;0;75;7;4;Crèche Collective des BLOSSIERES;

from django.utils.dateparse import parse_datetime

def test_concerto_import():

    filepath = os.path.join(os.path.dirname(__file__), 'data', 'concerto.csv')

    output = get_output_of_command('import_events', 'toto', filepath)

    assert output == 'Invalid loader: toto\n'

    call_command('import_events', 'concerto', filepath)

    assert Agenda.objects.count() == 2

    bloissieres = Agenda.objects.get(slug='creche-collective-des-blossieres')

    assert Event.objects.filter(agenda=bloissieres).count() == 6

    dt = parse_datetime('2017-03-14 12:30:00+00:00')

    assert bloissieres.event_set.get(start_datetime=dt).places == 7

    baron = Agenda.objects.get(slug='creche-collective-du-baron')

    assert Event.objects.filter(agenda=baron).count() == 7

    dt = parse_datetime('2017-03-14 14:30:00+00:00')

    assert baron.event_set.get(start_datetime=dt).places == -3