364 lines
14 KiB
Python
364 lines
14 KiB
Python
from csv import reader
|
|
import shutil
|
|
from os import path
|
|
from datetime import datetime
|
|
|
|
from django.db import models, transaction
|
|
from django.conf import settings
|
|
from django.utils.timezone import make_aware, get_current_timezone
|
|
|
|
SEXES = (
|
|
('G', 'Homme'),
|
|
('F', 'Femme'),
|
|
)
|
|
|
|
BOOLEANS = {'O': True,
|
|
'N': False
|
|
}
|
|
|
|
INVOICES_DIR = 'factures'
|
|
|
|
INPUT_DATE_FORMAT = '%Y-%m-%d'
|
|
|
|
class PersonneManager(models.Manager):
|
|
|
|
attribute_mappings = {
|
|
'ID_PER': 'id',
|
|
'LIB_NOM_PERSONNE': 'nom',
|
|
'LIB_PRENOM_PERSONNE': 'prenom',
|
|
'TYP_SEXE': 'sexe',
|
|
'NUM_TELDOM_PER': 'telephone_domicile',
|
|
'NUM_TELPORT_PER': 'telephone_portable',
|
|
'NUM_TELPROF_PER': 'telephone_portable_professionnel',
|
|
'NUM_ALLOC_PER': 'numero_allocation',
|
|
'LIB_NOM_ALL': 'regime',
|
|
'LIB_NOM_CAI': 'cai',
|
|
'LIB_PAYS': 'pays',
|
|
'COD_POSTAL': 'code_postal',
|
|
'LIB_COMMUNE': 'commune',
|
|
'NUM_VOIE_ADR': 'voie',
|
|
'NUM_BAT_ADR': 'batiment',
|
|
'LIB_NOM_BTQ': 'btq',
|
|
'NUM_APPT_ADR': 'appartement',
|
|
'LIB_NOM_RUE': 'rue',
|
|
'LIB_COMP_ADR': 'complement_adresse',
|
|
}
|
|
|
|
def synchronize(self, source_file):
|
|
|
|
with open(source_file) as source:
|
|
csv_reader = reader(source, delimiter = ';')
|
|
caption = csv_reader.next()
|
|
attrs = map(lambda e: self.attribute_mappings[e], caption)
|
|
objs = ()
|
|
|
|
for row in csv_reader:
|
|
params = dict(zip(attrs, row))
|
|
|
|
# transform all inputs to unicode
|
|
for key, value in params.iteritems():
|
|
params[key] = value.decode('utf-8')
|
|
|
|
objs += (Personne(**params), )
|
|
|
|
self.all().delete()
|
|
self.bulk_create(objs)
|
|
|
|
class FamilleManager(models.Manager):
|
|
|
|
attribute_mappings = {
|
|
'ID_FAM': 'id',
|
|
'LOGIN': 'login',
|
|
'PASSWORD': 'password',
|
|
'COD_SECRET_FAM': 'code_secret',
|
|
'QF_CCAS': 'qf_ccas',
|
|
'QF_VO': 'qf_vo',
|
|
'LIB_PAYS_ADR': 'pays',
|
|
'COD_POSTAL_ADR': 'code_postal',
|
|
'LIB_COMMUNE_ADR': 'commune',
|
|
'NUM_VOIE_ADR': 'voie',
|
|
'NUM_BAT_ADR': 'batiment',
|
|
'LIB_NOM_BTQ': 'btq',
|
|
'NUM_APPT_ADR': 'appartement',
|
|
'LIB_NOM_RUE': 'rue',
|
|
'LIB_COMP_ADR': 'complement_adresse',
|
|
'NB_ENFANTS_CHARGE': 'nombre_enfants',
|
|
'ID_PER1': 'pers1',
|
|
'ID_PER2': 'pers2'
|
|
}
|
|
|
|
|
|
def synchronize(self, source_file):
|
|
|
|
with open(source_file) as source:
|
|
csv_reader = reader(source, delimiter = ';')
|
|
caption = csv_reader.next()
|
|
attrs = map(lambda e: self.attribute_mappings[e], caption)
|
|
fields = [field.name for field in self.model._meta.fields]
|
|
families = ()
|
|
families_links = ()
|
|
existing_families = dict((fam.id, fam) for fam in self.all())
|
|
links = ()
|
|
|
|
for row in csv_reader:
|
|
params = dict(zip(attrs, row))
|
|
obj_id = int(params['id'])
|
|
pers1 = params.pop('pers1')
|
|
pers2 = params.pop('pers2')
|
|
families_links += ((obj_id, pers1, pers2),)
|
|
changed = False
|
|
|
|
# transform all inputs to unicode
|
|
for key, value in params.iteritems():
|
|
params[key] = unicode(value, 'utf-8')
|
|
|
|
if obj_id not in existing_families:
|
|
families += (Famille(**params),)
|
|
else:
|
|
obj = existing_families.pop(obj_id)
|
|
for field in fields:
|
|
if getattr(obj, field) != params[field]:
|
|
changed = True
|
|
setattr(obj, field, params[field])
|
|
if changed:
|
|
obj.save()
|
|
|
|
if existing_families:
|
|
self.filter(id__in=existing_families).delete()
|
|
|
|
if families:
|
|
self.bulk_create(families)
|
|
|
|
for link in families_links:
|
|
family, pers1, pers2 = link
|
|
for person in (pers1, pers2):
|
|
try:
|
|
member = Personne.objects.get(pk=person)
|
|
famille = self.get(pk=family)
|
|
links += (LiaisonParentFamille(famille = famille,
|
|
personne = member),)
|
|
except ValueError:
|
|
pass
|
|
|
|
LiaisonParentFamille.objects.all().delete()
|
|
LiaisonParentFamille.objects.bulk_create(links)
|
|
|
|
class FactureManager(models.Manager):
|
|
|
|
attribute_mappings = {
|
|
'ID_FAC': 'id',
|
|
'ID_FAM': 'famille',
|
|
'MNT_FACTURE_FAC': 'montant',
|
|
'TYP_MONNAIE_FAC': 'devise',
|
|
'MNT_REGLE_FAC': 'montant_regle',
|
|
'MNT_SOLDE_FAC': 'solde',
|
|
'DAT_GENERATION_FAC': 'date_generation',
|
|
'DAT_LIMITEPAIE_FAC': 'date_limite_paie',
|
|
'LIB_PRENOM_PER_PAYEUR': 'prenom_payeur',
|
|
'LIB_NOM_PER_PAYEUR': 'nom_payeur',
|
|
'ON_PRELEVAUTO_INS': 'prelevement_automatique',
|
|
'DAT_DEBUT_PGE': 'debut_pge',
|
|
'DAT_FIN_PGE': 'fin_pge',
|
|
'ID_PYR': 'id_payeur_par_defaut',
|
|
'DAT_REGLEMENT': 'date_reglement',
|
|
'DAT_PERCEPTION_FAC': 'date_passage_perception'
|
|
}
|
|
|
|
def synchronize(self, source_file):
|
|
|
|
with open(source_file) as source:
|
|
source_pdf_directory = path.join(path.dirname(source_file), INVOICES_DIR)
|
|
csv_reader = reader(source, delimiter = ';')
|
|
caption = csv_reader.next()
|
|
attrs = map(lambda e: self.attribute_mappings[e], caption)
|
|
objs = ()
|
|
|
|
for row in csv_reader:
|
|
params = dict(zip(attrs, row))
|
|
# transform all inputs to unicode
|
|
for key, value in params.iteritems():
|
|
params[key] = unicode(value, 'utf-8')
|
|
|
|
params['prelevement_automatique'] = params['prelevement_automatique'] == 'O' or False
|
|
for date_type in ('date_generation', 'date_limite_paie',
|
|
'debut_pge', 'fin_pge', 'date_reglement',
|
|
'date_passage_perception'):
|
|
try:
|
|
params[date_type] = datetime.strptime(params[date_type], INPUT_DATE_FORMAT)
|
|
params[date_type] = make_aware(params[date_type], get_current_timezone())
|
|
except ValueError:
|
|
params.pop(date_type)
|
|
params['active'] = path.exists(settings.INVOICES_LOCATION_PATTERN.format(invoice_id = params['id']))
|
|
family = Famille.objects.get(pk = params.pop('famille'))
|
|
params['famille'] = family
|
|
objs += (Facture(**params), )
|
|
|
|
self.all().delete()
|
|
self.bulk_create(objs)
|
|
|
|
class KidManager(models.Manager):
|
|
|
|
attribute_mappings = {
|
|
'ID_FAM': 'famille',
|
|
'ID_PER': 'id_enfant',
|
|
'LIB_NOM_PER': 'nom',
|
|
'LIB_PRENOM_PER': 'prenom',
|
|
'TYP_SEXE_PER': 'sexe',
|
|
'DAT_NAISSANCE': 'date_naissance'
|
|
}
|
|
|
|
def synchronize(self, source_file):
|
|
|
|
with open(source_file) as source:
|
|
csv_reader = reader(source, delimiter = ';')
|
|
caption = csv_reader.next()
|
|
attrs = map(lambda e: self.attribute_mappings[e], caption)
|
|
objs = ()
|
|
|
|
|
|
for row in csv_reader:
|
|
params = dict(zip(attrs, row))
|
|
# transform all inputs to unicode
|
|
for key, value in params.iteritems():
|
|
params[key] = value.decode('utf-8')
|
|
|
|
family_id = params.pop('famille')
|
|
|
|
try:
|
|
params['date_naissance'] = datetime.strptime(params['date_naissance'],
|
|
INPUT_DATE_FORMAT)
|
|
except ValueError:
|
|
params.pop('date_naissance')
|
|
|
|
try:
|
|
|
|
family = Famille.objects.get(pk = family_id)
|
|
except Famille.DoesNotExist:
|
|
print "La famille %s de l'enfant kid %s n'existe pas(%s)" % (family_id, params['id_enfant'], row)
|
|
continue
|
|
params['famille'] = family
|
|
|
|
objs += (Enfant(**params), )
|
|
|
|
self.all().delete()
|
|
self.bulk_create(objs)
|
|
|
|
class Personne(models.Model):
|
|
id = models.AutoField(primary_key = True)
|
|
nom = models.CharField(max_length = 256, null = True, blank = True)
|
|
prenom = models.CharField(max_length = 256, null = True, blank = True)
|
|
sexe = models.CharField(choices = SEXES, max_length = 1, null = True)
|
|
telephone_domicile = models.CharField(max_length = 32, null = True, blank = True)
|
|
telephone_portable = models.CharField(max_length = 32, null = True, blank = True)
|
|
telephone_portable_professionnel = models.CharField(max_length = 32, null = True, blank = True)
|
|
numero_allocation = models.CharField(max_length = 32, null = True, blank = True)
|
|
regime = models.CharField(max_length = 32, null = True, blank = True)
|
|
cai = models.CharField(max_length = 32, null = True, blank = True)
|
|
pays = models.CharField(max_length = 32, null = True, blank = True)
|
|
code_postal = models.CharField(max_length = 16, null = True, blank = True)
|
|
commune = models.CharField(max_length = 32, null = True, blank = True)
|
|
voie = models.CharField(max_length = 32, null = True, blank = True)
|
|
batiment = models.CharField(max_length = 32, null = True, blank = True)
|
|
btq = models.CharField(max_length = 32, null = True, blank = True)
|
|
appartement = models.CharField(max_length = 32, null = True, blank = True)
|
|
rue = models.CharField(max_length = 64, null = True, blank = True)
|
|
complement_adresse = models.CharField(max_length = 64, null = True, blank = True)
|
|
email = models.EmailField(null = True, blank = True)
|
|
name_id = models.CharField(max_length = 256, null = True, blank = True, db_index = True)
|
|
objects = PersonneManager()
|
|
|
|
def __unicode__(self):
|
|
return u'%s %s' % (self.prenom, self.nom)
|
|
|
|
def is_federated(self):
|
|
return self.name_id is not None and len(self.name_id) > 0
|
|
is_federated.boolean = True
|
|
|
|
class LiaisonParentFamille(models.Model):
|
|
famille = models.ForeignKey('Famille')
|
|
personne = models.ForeignKey(Personne)
|
|
|
|
class LiaisonNameidFamille(models.Model):
|
|
"""
|
|
model used to grant permissions to a nameid to see all informations about
|
|
the family(children, invoices, etc)
|
|
"""
|
|
name_id = models.CharField(max_length = 256, db_index = True)
|
|
famille = models.ForeignKey('Famille', null=True)
|
|
link_date = models.DateTimeField(auto_now_add = True)
|
|
|
|
class Meta:
|
|
|
|
unique_together = ('name_id', 'famille')
|
|
|
|
class Famille(models.Model):
|
|
id = models.AutoField(primary_key = True)
|
|
login = models.CharField(max_length = 32, null = True, blank = True)
|
|
password = models.CharField(max_length = 128, null = True, blank = True)
|
|
code_secret = models.CharField(max_length = 128, null = True, blank = True)
|
|
qf_ccas = models.CharField(max_length = 128, null = True, blank = True)
|
|
qf_vo = models.CharField(max_length = 128, null = True, blank = True)
|
|
pays = models.CharField(max_length = 128, null = True, blank = True)
|
|
code_postal = models.CharField(max_length = 16, null = True, blank = True)
|
|
commune = models.CharField(max_length = 32, null = True, blank = True)
|
|
voie = models.CharField(max_length = 32, null = True, blank = True)
|
|
batiment = models.CharField(max_length = 32, null = True, blank = True)
|
|
btq = models.CharField(max_length = 32, null = True, blank = True)
|
|
appartement = models.CharField(max_length = 32, null = True, blank = True)
|
|
rue = models.CharField(max_length = 64, null = True, blank = True)
|
|
complement_adresse = models.CharField(max_length = 64, null = True, blank = True)
|
|
nombre_enfants = models.IntegerField(default = 0)
|
|
personnes = models.ManyToManyField(Personne, through = LiaisonParentFamille)
|
|
objects = FamilleManager()
|
|
|
|
def __unicode__(self):
|
|
personnes = u' et '.join([u'%s (%s)' % (p, p.id) for p in self.personnes.all()])
|
|
return u'Famille %s de %s' % (self.id, personnes)
|
|
|
|
|
|
class Facture(models.Model):
|
|
id = models.AutoField(primary_key = True)
|
|
famille = models.ForeignKey(Famille)
|
|
montant = models.DecimalField(max_digits = 6, decimal_places = 2)
|
|
devise = models.CharField(max_length = 3)
|
|
montant_regle = models.DecimalField(max_digits = 6, decimal_places = 2)
|
|
solde = models.DecimalField(max_digits = 6, decimal_places = 2)
|
|
date_generation = models.DateField()
|
|
date_limite_paie = models.DateField()
|
|
prenom_payeur = models.CharField(max_length = 128)
|
|
nom_payeur = models.CharField(max_length = 128)
|
|
prelevement_automatique = models.BooleanField()
|
|
debut_pge = models.DateField()
|
|
fin_pge = models.DateField()
|
|
paye = models.BooleanField(default = False)
|
|
objects = FactureManager()
|
|
active = models.BooleanField(default = False)
|
|
statut_tipi = models.CharField(max_length=16, blank=True, null=True)
|
|
date_reponse_tipi = models.DateTimeField(null=True, blank=True)
|
|
date_reglement = models.DateTimeField(null=True, blank=True)
|
|
id_payeur_par_defaut = models.CharField(max_length=128, null=True, blank=True)
|
|
date_envoi_dernier_mail = models.DateTimeField(null=True, blank=True)
|
|
date_passage_perception = models.DateTimeField(null=True, blank=True)
|
|
|
|
def __unicode__(self):
|
|
return 'Facture %s (Famille %s)' % (self.id, self.famille.id)
|
|
|
|
class Enfant(models.Model):
|
|
id_enfant = models.CharField(max_length=64)
|
|
famille = models.ForeignKey(Famille)
|
|
prenom = models.CharField(max_length = 128, null = True, blank = True)
|
|
nom = models.CharField(max_length = 128, null = True, blank = True)
|
|
sexe = models.CharField(choices = SEXES, max_length = 1, null = True)
|
|
date_naissance = models.DateField(null = True, blank = True)
|
|
objects = KidManager()
|
|
|
|
def __unicode__(self):
|
|
return u'%s %s' % (self.prenom, self.nom)
|
|
|
|
class InvoiceNotificationEmail(models.Model):
|
|
invoice_number = models.CharField(max_length=64)
|
|
date_sent = models.DateTimeField(auto_now_add=True)
|
|
|
|
class Meta:
|
|
get_latest_by = 'date_sent'
|