helloasso-paheko-adapter/csv_helloasso_to_paheko_csv.py

import uuid
import csv
import fileinput
from datetime import datetime
from pprint import pprint
from dataclasses import dataclass
import json

INPUT_FIELDS_MAPPING = {
    "last_name": "Nom",
    "first_name_1": "Prénom",
    "date": "\ufeffDate",
    "email_1": "Email acheteur",
    "first_name_2": "Champ complémentaire 1 Prénom conjoint",
    "address": "Champ complémentaire 2 ADRESSE",
    "postal_code": "Champ complémentaire 3 CODE POSTAL",
    "city": "Champ complémentaire 4 VILLE",
    "birth_date": "Champ complémentaire 9 DATE DE NAISSANCE",
    "email_2": "Champ complémentaire 6 EMAIL",
    "job": "Champ complémentaire 7 PROFESSION",
    "skills": "Champ complémentaire 8 CENTRE D'INTÉRÊTS / COMPÉTENCES",
    "payment_method": "Moyen de paiement",
    "status": "Status",
    "tarif": "Tarif",
}

def get_matching_keys(src_dict, value):
    return [k for k,v in src_dict.items() if v == value]

def load_csv_row_to_dict(field_mapping, first_row, row):
    final_dict = {}
    for i, cell in enumerate(row):
        keys = get_matching_keys(field_mapping, first_row[i])
        if len(keys) == 0: continue
        final_dict[keys[0]] = cell
    return final_dict

def get_id():
    return str(uuid.uuid4()).split("-")[0]

@dataclass
class PahekoMember:
    name: str
    email: str
    phone: str
    status: str

def import_data():
    reader = csv.reader([i for i in fileinput.input()], delimiter=";")
    column_line = []
    paheko_users = []

    for i, line in enumerate(reader):
        print(i)
        if i == 0:
            column_line = line
            continue

        ha_membership: dict = load_csv_row_to_dict(INPUT_FIELDS_MAPPING, column_line, line)

        def get_email(ha_ms):
            if ha_ms["email_1"] == None: return ha_ms["email_2"]
            if ha_ms["email_2"] == None: return ha_ms["email_1"]
            if ha_ms['email_2'] != ha_ms['email_2']:
                return ha_ms["email_2"]
            return ha_ms["email_1"]

        def format_date_time_french(raw_date):
            return datetime.strptime(raw_date, "%d/%m/%Y %H:%M:%S")

        def format_date_french(raw_date):
            return datetime.strptime(raw_date, "%d/%m/%Y")

        def format_string(subj):
            return subj.strip()

        def format_name(subj):
            return subj[0:1].upper() + subj[1:].lower()

        def format_mode(subj):
            return subj

        if format_string(ha_membership['status']) != "Validé":
            continue

        # then normalize dict
        paheko_user: dict = {
            'id': get_id(),
            'first_name': format_name(format_string(ha_membership['first_name_1'])),
            'last_name': format_name(format_string(ha_membership['last_name'])),
            'mode_adhesion': format_mode(ha_membership['tarif']),
            'email': format_string(get_email(ha_membership)),
            'date': format_date_time_french(ha_membership['date']),
            'birth_date': format_date_french(ha_membership['birth_date']) if ha_membership['birth_date'] and ha_membership['birth_date'].strip() != '' else None,
            'linked_users': []
        }
        keys_to_copy = ['job', 'skills', 'address', 'postal_code', 'city']
        for key in keys_to_copy:
            if ha_membership[key].strip() == '':
                paheko_user[key] = None
                continue
            paheko_user[key] = format_string(ha_membership[key])

        linked_user = None
        if ha_membership["first_name_2"].strip() != '':
            # we count as two membership
            linked_user = {
                'id': get_id(),
                'first_name': format_name(format_string(ha_membership['first_name_2'])),
                'linked_users': [paheko_user['id']]
            }
            copy_from_parent_user = ['last_name', 'address', 'postal_code', 'city', 'date']
            for k in copy_from_parent_user:
                linked_user[k] = paheko_user[k]
            paheko_user["linked_users"].append(linked_user['id'])
        paheko_users.append(paheko_user)
        if linked_user:
            paheko_users.append(linked_user)

    # pprint(paheko_users, sort_dicts=False)
    print(json.dumps(paheko_users, sort_keys=True, default=str))

    """
    Une fois qu'on est bon au niveau de la liste d'adhesion qu'on veut ajouter
    on regarde si on a pas déjà les adhesions dans helloasso
    on télécharge tout les adherents et leurs participations à des activités et cotisations
    on regarde les adherents sur leur addresse emails
    on regarde si l'id de payment n'a pas déjà été traité

    on regarde quelle est la période d'adhesion en cours
    première étape:
        on regarde si le member existe déjà à partir de son email
        si non, on ajoute le member
        ensuite on regarde si le membre a déjà adhéré pour cette période
        on regarde également si il n'y a pas un id existant helloasso (sanity check)
        si oui, on discard
        sinon, on ajoute la cotisation pour cette année
    """
        

import_data()