helloasso-paheko-adapter/helloasso_paheko_adapter/adapter.py

import uuid
import csv
import fileinput
from datetime import datetime
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 from_helloasso_members_csv_to_paheko_normalized():
    reader = csv.reader([i for i in fileinput.input()], delimiter=";")
    column_line = []
    paheko_users = []

    for i, line in enumerate(reader):
        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, indent=4))


from pprint import pprint
from helloasso_paheko_adapter.helloasso import HelloassoClient, Organization
from helloasso_paheko_adapter.paheko import PahekoClient
import vcr

def from_helloasso_payments_api_to_paheko(env, org_slug):
    ha_client = HelloassoClient(env.HELLOASSO_API_CLIENT_ID, env.HELLOASSO_API_CLIENT_SECRET)
    pk_client = PahekoClient(env.PAHEKO_API_CLIENT_ID, env.PAHEKO_API_CLIENT_SECRET)
    ha_org = Organization(ha_client, org_slug)

    # 1. get latest adhesion periode
    period_id = "1" # or "fee_id"
    print(pk_client.get_current_membership_period())

    # 2. list payments
    with vcr.use_cassette('tmp/vcr_cassettes/list_payments.yaml'):
        payments = ha_org.list_payments()['data']
    pprint(payments)