passerelle/passerelle/contrib/nancypoll/models.py

90 lines
2.7 KiB
Python

import csv
from django.db import models
from django.utils.encoding import force_str
from django.utils.translation import gettext_lazy as _
from passerelle.base.models import BaseResource
from passerelle.utils.api import endpoint
from passerelle.utils.jsonresponse import APIError
from passerelle.utils.models import resource_file_upload_to
COLUMN_NAMES = (
'street_start_number, street_end_number,,,street_side,,,,code,id,text,address,,,street_name,,canton,,,'
)
def to_unicode(value):
return force_str(value, 'utf-8')
class NancyPoll(BaseResource):
csv_file = models.FileField(_('CSV File'), upload_to=resource_file_upload_to)
category = _('Data Sources')
class Meta:
verbose_name = _('NancyPoll')
@classmethod
def get_verbose_name(cls):
return cls._meta.verbose_name
@endpoint(perm='OPEN')
def data(self, request, *args, **kwargs):
street_no = request.GET.get('street_no')
street_name = request.GET.get('street_name')
if not street_no or not street_name:
raise APIError('All parameters are required')
try:
street_no = int(street_no)
except ValueError:
raise APIError('Invalid street no value')
titles = [t.strip() for t in COLUMN_NAMES.split(',')]
content = self.csv_file.read()
if not content:
raise APIError('No content found')
content = force_str(content)
reader = csv.reader(content.splitlines())
next(reader)
idx_name = titles.index('street_name')
idx_side = titles.index('street_side')
idx_min = titles.index('street_start_number')
idx_max = titles.index('street_end_number')
for row in reader:
if not self.street_name_like(street_name, row[idx_name]):
continue
if not self.street_no_between(street_no, row[idx_min], row[idx_max]):
continue
if row[idx_side] == 'I' and int(street_no) % 2 == 0:
continue
if row[idx_side] == 'P' and int(street_no) % 2 == 1:
continue
return {
'data': {
'id': row[titles.index('id')],
'text': row[titles.index('text')],
'code': row[titles.index('code')],
'address': row[titles.index('address')],
'canton': row[titles.index('canton')],
}
}
raise APIError('Polling Station Not Found')
def street_name_like(self, street_name, elt):
return street_name.lower() in to_unicode(elt).lower()
def street_no_between(self, street_no, no_min, no_max):
return int(no_min) < int(street_no) <= int(no_max)