Coverage for members/api_helper.py : 32%

import datetime 

import logging 

import csv 

from collections import OrderedDict 

from itertools import zip_longest 

import io 

from .models import EmailAddress, Erfa, Member, BalanceTransactionLog, DeliveryNumber, get_alien 

from .countryfield import get_country_dict 

def uglify_date(date): 

return "{:%d.%m.%y}".format(date) 

class ErfaCSVExporter: 

def __init__(self, erfas=None): 

self.file = None 

self._erfas = erfas 

self._writer = None 

def export(self): 

self._setup_writer() 

for erfa in self._erfas: 

self._writer.writerow(self._get_row(erfa)) 

def _get_writer(self): 

return writer(self.file, delimiter='\t') 

def _setup_writer(self): 

if not self.file: 

self.file = io.StringIO() 

self._writer = self._get_writer() 

@staticmethod 

def _get_row(erfa): 

return erfa.short_name, erfa.long_name 

class Openslides: 

warnings = [] 

logs = [] 

def do_export(self, file): 

writer = csv.writer(file, delimiter=',', quotechar="'") 

writer.writerow(['Titel', 'Vorname', 'Nachname', 'Gliederungsebene', 'Teilnehmernummer', 'Gruppen', 'Kommentar', 

'Ist aktiv', 'Ist anwesend', 'Ist ein Gremium', 'Initiales Passwort', 'E-Mail', 'Benutzername', 

'Geschlecht', 'Stimmgewicht']) 

for member in Member.objects.members_only().filter(is_active=True): 

writer.writerow(self._get_member(member)) 

@staticmethod 

def _get_member(member): 

return ( 

'', # Titel 

member.first_name, # Vorname 

member.last_name, # Nachname 

'', # Gliederungsebene 

'', # Teilnehmernummer 

'Supporters' if member.membership_type == Member.MEMBERSHIP_TYPE_SUPPORTER else 'Members', # Gruppen 

'', # Kommentar 

'1', # Ist aktiv 

'', # Ist anwesend 

'', # Ist ein Gremium 

member.initial_password, # Initiales Passwort 

'', # E-Mail 

member.username, # Benutzername 

'', # Geschlecht 

'1', # Stimmgewicht 

) 

class Vereinstisch: 

warnings = [] 

logs = [] 

def __init__(self, member_set=None): 

self.member_set = member_set 

@staticmethod 

def _str2date(string): 

if string == '': 

return None 

return datetime.datetime.strptime(string, '%d.%m.%Y').date() 

def do_export(self, file): 

assert self.member_set is not None, "We need members if we want to export them" 

writer = csv.writer(file, delimiter='\t') 

for member in self.member_set: 

writer.writerow(self._get_member(member)) 

@staticmethod 

def _get_member(member): 

date_format = '%d.%m.%Y' 

emails_queryset = member.emailaddress_set.order_by('-is_primary') 

return ( 

member.chaos_number, 

member.first_name, 

member.last_name, 

member.address_1, 

member.address_2, 

member.address_3, 

member.address_country, 

member.erfa.short_name, 

member.address_unknown, 

member.erfa.has_doppelmitgliedschaft, 

member.membership_reduced, 

member.is_active, 

member.notification_consent, 

member.membership_type, 

member.membership_start.strftime(date_format), 

member.fee_last_paid.strftime(date_format) if member.fee_last_paid else '', 

member.fee_paid_until.strftime(date_format), 

member.last_update.strftime(date_format), 

member.account_balance, 

','.join(list(emails_queryset.values_list('email_address', flat=True))), 

','.join(list(emails_queryset.values_list('gpg_key_id', flat=True))), 

member.comment 

) 

def do_import(self, file): 

# Taken from https://doku.ccc.de/index.php?title=Vereinstisch&oldid=32075#Exportformat_.28Vereinstisch_-.3E_Office.29_ab_Q4.2F2019 

fieldnames = [ 

'new_member', 

'chaos_number', 

'first_name', 

'last_name', 

'address_1', 

'address_2', 

'address_3', 

'address_country', 

'erfa', 

'address_unknown', 

'email', 

'gpg_key_id', 

'membership_type', 

'account_balance', 

'fee_paid_until', 

'membership_reduced', 

'notification_consent', 

'changed', 

'amount_paid', 

'membership_end', 

'Kommentar', 

] 

reader = csv.DictReader(file, fieldnames=fieldnames, delimiter='\t') 

for line_num, row in enumerate(reader): 

new_member = row['new_member'].lower() == 'true' 

chaos_number = row['chaos_number'] 

if not new_member: 

try: 

chaos_number = int(chaos_number) 

except ValueError: 

warn = 'Row {}: Chaos number {} is illegal. New member created'.format(line_num, chaos_number) 

new_member = True 

logging.warning(warn) 

self.warnings.append(warn) 

first_name = row['first_name'] 

last_name = row['last_name'] 

address_1 = row['address_1'] 

address_2 = row['address_2'] 

address_3 = row['address_3'] 

address_country = row['address_country'] 

if address_country not in get_country_dict(): 

warn = 'Row {}: Illegal country code: {}. Defaulted to DE for Germany.'.format(line_num, address_country) 

logging.warning(warn) 

self.warnings.append(warn) 

address_country = 'DE' 

erfa = row['erfa'] 

try: 

erfa = Erfa.objects.get(short_name=erfa) 

except Erfa.DoesNotExist: 

warn = 'Row {}: Erfa {} does not exist. Member set to Alien'.format(line_num, erfa) 

logging.warning(warn) 

self.warnings.append(warn) 

erfa = get_alien() 

address_unknown = row['address_unknown'] 

try: 

address_unknown = int(address_unknown) 

except ValueError: 

warn = 'Row {}: Address unknown counter of {} is illegal. It will be ignored.'.format(line_num, address_unknown) 

logging.warning(warn) 

self.warnings.append(warn) 

address_unknown = None 

email = row['email'].split(',') 

gpg_key_id = row['gpg_key_id'].split(',') 

membership_type = row['membership_type'] 

# 'account_balance' will only be changed through applying 'amount_paid' 

# 'fee_paid_until' will only be set by the office. See Member.execute_payment_if_due(). 

membership_reduced = row['membership_reduced'].lower() == 'true' 

notification_consent = row['notification_consent'].lower() == 'true' 

changed = row['changed'] 

try: 

changed = self._str2date(changed) 

except ValueError: 

warn = 'Row {}: Illegal date: {}. Date set to today.'.format(line_num, changed) 

logging.warning(warn) 

self.warnings.append(warn) 

changed = datetime.date.today() 

amount_paid = row['amount_paid'] 

try: 

amount_paid = int(amount_paid) 

except ValueError: 

warn = 'Row {}: Amount paid {} is illegal. It will be ignored.'.format(line_num, amount_paid) 

logging.warning(warn) 

self.warnings.append(warn) 

amount_paid = 0 

membership_end = row['membership_end'] 

try: 

membership_end = self._str2date(membership_end) 

except ValueError: 

warn = 'Row {}: Illegal date: {}. No date set.'.format(line_num, membership_end) 

logging.warning(warn) 

self.warnings.append(warn) 

membership_end = None 

kommentar = row['Kommentar'].strip() 

# Now create or modify the member 

member = None 

try: 

if not new_member: 

member = Member.objects.get(chaos_number=chaos_number) 

# Deleting all emails so they can be added again later 

EmailAddress.objects.filter(person=member).delete() 

except Member.DoesNotExist: 

warn = 'Row {}: Chaos number {} not found in database. Creating new member.'.format(line_num, chaos_number) 

logging.warning(warn) 

self.warnings.append(warn) 

finally: 

if not member: 

member = Member() 

member.membership_type = membership_type 

member.first_name = first_name 

member.last_name = last_name 

member.address_1 = address_1 

member.address_2 = address_2 

member.address_3 = address_3 

member.address_country = address_country 

member.erfa = erfa 

member.address_unknown = address_unknown 

member.membership_reduced = membership_reduced 

member.notification_consent = notification_consent 

member.membership_end = membership_end 

member.comment = kommentar 

member.save() 

# Save email addresses after the member was saved 

if email: 

email_pairs = zip_longest(email, gpg_key_id) 

for email_pair in email_pairs: 

EmailAddress(person=member, email_address=email_pair[0], gpg_key_id=email_pair[1] or '').save() 

if amount_paid > 0: 

member.increase_balance_by(amount_paid, reason=BalanceTransactionLog.MANUAL_BOOKING, 

booking_day=changed, comment='Vereinstisch: ' + kommentar) 

self.logs.append('Row {}: Created/updated {}.'.format(line_num, member)) 

def cashpoint_export_impl(fh, member_set): 

fieldnames = OrderedDict([ 

('chaos_number', 'chaos_number'), 

('first_name', 'first_name'), 

('last_name', 'last_name'), 

('state', None)]) 

writer = csv.DictWriter(fh, fieldnames=fieldnames.keys(), dialect='excel-tab') 

writer.writeheader() 

payday = datetime.date.today().replace(month=12, day=1) 

for member in member_set.order_by('chaos_number'): 

row = {k: getattr(member, v) for k, v in fieldnames.items() if v is not None} 

if not member.is_active: 

row['state'] = 'ruhend' 

else: 

row['state'] = 'Verzug' if member.is_payment_late(payday) else 'bezahlt' 

writer.writerow(row) 

# Mapping of country names as used by the Post to ISO 2-letter abbreviations as used in AA 

post_countries = { 

'': 'DE', 

'Österreich': 'AT', 

'Schweiz': 'CH', 

'Deutschland': 'DE', 

'Belgien': 'BE', 

'Frankreich': 'FR', 

} 

def premiumaddress_reader(csv_string): 

""" 

Interprets the reasons for returned letters from the Premiumaddress csv files and executes sensible actions in 

response. 

:param csv_string: 

:return: 

""" 

# Based on the information from: https://www.direktmarketingcenter.de/fileadmin/Download-Center/Handbuch_Premiumadress_-_Stand_Oktober_2017.pdf 

# AdrMerk 

address_unknown = ['10', '11', '13', '14', '18', '19', '21', '22', '25', '30', '31', '50', '53', '55'] 

exit_through_death = ['12', ] 

new_address = ['20', '51'] 

# SdgS 

arrived = ['20', '30'] 

not_arrived = ['10', '40'] 

failed_lines = [] 

processed_lines = [] 

file_handle = io.StringIO(csv_string) 

csv_reader = csv.DictReader(file_handle, delimiter=';') 

for field in ['SdgS', 'AdrMerk', 'E_Na1', 'E_Na2', 'Kd_Info', 'NSA_Na1', 'NSA_Na2', 'NSA_Na3', 'NSA_Na4', 

'NSA_Str', 'NSA_HNr', 'NSA_PLZ', 'NSA_Ort', 'NSA_Land', 'NSA_Postf', 'NSA_PLZPostf', 

'NSA_OrtPostfach', 'NSA_LandPostfach', 'UebgID']: 

341 ↛ 342line 341 didn't jump to line 342, because the condition on line 341 was never true if field not in csv_reader.fieldnames: 

return [], ['Missing field \'{}\' in uploaded file.'.format(field)] 

for line in csv_reader: 

delivery_number = line['Kd_Info'].upper() 

# The alphabet used by the delivery numbers is: ABCDEFGHJKLMNPQRSTUVWXYZ3789 

# All other letters and numbers are mapped to another character of this alphabet 

delivery_number_substitution = { 

'I': 'J', 

'O': 'D', 

'0': 'D', 

'1': 'J', 

'2': 'Z', 

'4': 'A', 

'5': 'S', 

'6': 'G', 

} 

cleaned_delivery_number = (delivery_number_substitution.get(char, char) for char in delivery_number) 

delivery_number = ''.join(cleaned_delivery_number) 

try: 

del_num_obj = DeliveryNumber.objects.get(number__exact=delivery_number) 

except DeliveryNumber.DoesNotExist: 

failed_lines.append( 

'Delivery \'{}\' has no valid delivery number: \'{}\'.'.format(line['UebgID'], delivery_number)) 

continue 

372 ↛ 373line 372 didn't jump to line 373, because the condition on line 372 was never true if del_num_obj.returned: 

failed_lines.append( 

'Delivery number \'{}\' has previously been marked as returned.'.format(delivery_number)) 

continue 

recipient = del_num_obj.recipient 

sdgs = line['SdgS'] 

adrmerk = line['AdrMerk'] 

if (adrmerk in address_unknown and sdgs in not_arrived) or (sdgs == '20' and adrmerk in ['21', '22']): 

# Adresse ist nicht mehr aktuell 

recipient.address_unknown += 1 

recipient.save() 

processed_lines.append( 

'{}: Member {} increased address unknown counter.'.format(delivery_number, recipient.chaos_number)) 

del_num_obj.returned = True 

del_num_obj.save() 

elif adrmerk in new_address and sdgs in ['10', '20', '40']: 

# Neue Adresse 

394 ↛ 399line 394 didn't jump to line 399, because the condition on line 394 was never false if line['NSA_Postf']: 

street = line['NSA_Postf'] 

town = (line['NSA_PLZPostf'].zfill(5) + ' ' + line['NSA_OrtPostfach']).strip() 

country = line['NSA_LandPostfach'] 

else: 

street = (line['NSA_Str'] + ' ' + line['NSA_HNr']).strip() 

town = (line['NSA_PLZ'].zfill(5) + ' ' + line['NSA_Ort']).strip() 

country = line['NSA_Land'] 

403 ↛ 408line 403 didn't jump to line 408, because the condition on line 403 was never false if country not in post_countries.keys(): 

failed_lines.append( 

'{}: Country \'{}\' not in mapping table for country IDs.'.format(delivery_number, country)) 

return processed_lines, failed_lines 

recipient.first_name = line['NSA_Na1'] 

recipient.last_name = line['NSA_Na2'] 

if line['NSA_Na3']: 

recipient.address_1 = (line['NSA_Na3'] + ' ' + line['NSA_Na4']).strip() 

recipient.address_2 = street 

recipient.address_3 = town 

else: 

recipient.address_1 = street 

recipient.address_2 = town 

recipient.address_3 = '' 

recipient.address_country = post_countries[country] 

recipient.save() 

processed_lines.append('{}: Member {} updated address.'.format(delivery_number, recipient.chaos_number)) 

del_num_obj.returned = True 

del_num_obj.save() 

elif adrmerk in exit_through_death and sdgs in not_arrived: 

# Austritt durch Tod 

processed_lines.append('{}: Member {} exited by death.'.format(delivery_number, recipient.chaos_number)) 

del_num_obj.returned = True 

del_num_obj.save() 

436 ↛ 437line 436 didn't jump to line 437, because the condition on line 436 was never true if Member.objects.filter(chaos_number=recipient.chaos_number).exists(): 

member = Member.objects.get(chaos_number=recipient.chaos_number) 

member.emailtomember_set.all().delete() 

recipient.delete() 

else: 

failed_lines.append('{}: Could not determine reason for not delivering'.format(delivery_number)) 

return processed_lines, failed_lines