From 0e5e88f1a6ece0c283883fb20933c4bdb1bc7d5c Mon Sep 17 00:00:00 2001 From: Bram Jans Date: Thu, 3 Oct 2024 09:55:34 +0200 Subject: [PATCH] WC2-566 Add v1 of OU setup script from a CSV This little script was created for WFP and used to set up the South Sudan and Nigeria health pyramids. Adding to the repo mainly for traceability, can be easily adapted for other countries. --- scripts/create_health_pyramid_from_csv.py | 159 ++++++++++++++++++++++ 1 file changed, 159 insertions(+) create mode 100644 scripts/create_health_pyramid_from_csv.py diff --git a/scripts/create_health_pyramid_from_csv.py b/scripts/create_health_pyramid_from_csv.py new file mode 100644 index 0000000000..840d1f35f4 --- /dev/null +++ b/scripts/create_health_pyramid_from_csv.py @@ -0,0 +1,159 @@ +import csv +import requests + +""" +Disclaimer: This script is a work in progress. There are many ways to improve and +make this script more generic. For the time being, you can set it up as follows: + +1. Put all data in a CSV with column names: + - "partner" this will become the OU groups + - For each OrgUnitType: short_name +2. Make sure SERVER_BASE_URL and AUTH_TOKEN are correctly configured to your + desired server. +""" + +SERVER_BASE_URL = "https://server.org" + +orgunits_url = SERVER_BASE_URL + "/api/orgunits/" +groups_url = SERVER_BASE_URL + "/api/groups/" +create_org_unit_url = SERVER_BASE_URL + "/api/orgunits/create_org_unit/" +org_unit_types_url = SERVER_BASE_URL + "/api/v2/orgunittypes/" + +AUTH_TOKEN = "XXX" +headers = {"Authorization": "Bearer %s" % AUTH_TOKEN} + +SOURCE_ID = 1 # ID of the default source +COUNTRY_ID = 1 # ID of the main node of the org unit tree +CSV_NAME = "example.csv" + + +def csv_to_dict(filename): + with open(filename, newline="") as csvfile: + reader = csv.DictReader(csvfile) + return [{k: v.strip() for k, v in row.items()} for row in reader] + + +def get_group_id_by_name(groups, name): + for group in groups: + if group["name"] == name: + return group["id"] + return None + + +def find_or_create_org_unit(name, org_unit_type_id, parent_id, group_id=None): + search_url = ( + orgunits_url + + f'?limit=20&order=id&page=1&searches=[{{"validation_status":"VALID","search":"{name}","source":{SOURCE_ID},"orgUnitTypeId":"{org_unit_type_id}","orgUnitParentId":"{parent_id}"}}]' + ) + response = requests.get(search_url, headers=headers) + + result_count = response.json()["count"] + if result_count == 0: + print("CREATING") + payload = { + "id": None, + "name": name, + "org_unit_type_id": org_unit_type_id, + "groups": [group_id] if group_id else [], + "sub_source": None, + "aliases": [], + "validation_status": "VALID", + "parent_id": parent_id, + } + resp = requests.post(create_org_unit_url, headers=headers, json=payload) + return resp.json()["id"] + else: + if result_count > 1: + print(f"WARNING: Found multiple results for {name}") + print("FOUND") + return response.json()["orgunits"][0]["id"] + + +# Call the function with your csv file path +rows = csv_to_dict(CSV_NAME) + + +# Groups +groups = {} +print("Fetching group ids, create groups if needed") + +groups_in_csv = set([row["partner"] for row in rows]) +existing_groups = requests.get(groups_url, headers=headers).json()["groups"] +existing_group_names = set([g["name"] for g in existing_groups]) + +for group_name in groups_in_csv: + if group_name in existing_group_names: + groups[group_name] = get_group_id_by_name(existing_groups, group_name) + else: + print(f"\tcreating group: {group_name}") + payload = {"name": group_name, "source_ref": ""} + response = requests.post(groups_url, headers=headers, json=payload) + groups[group_name] = response.json()["id"] + +print("Groups", groups) + +# OU Types +print("Fetching OU types") +ou_types = {} +response = requests.get(org_unit_types_url, headers=headers).json()["orgUnitTypes"] +for out in response: + ou_types[out["short_name"]] = out["id"] +print("OU types", ou_types) + + +pyramid_dict = {} +# Turn the file into a pyramid. E.g. for South Sudan: +# state +# |- county +# |- payam +# |- CHC-CHP +for row in rows: + state = row["State"] + lga = row["LGA"] + ward = row["Ward"] + phc_site = row["PHC/Site"] + group_name = row["partner"] + + if state in pyramid_dict: + if lga in pyramid_dict[state]: + if ward in pyramid_dict[state][lga]: + pyramid_dict[state][lga][ward][phc_site] = group_name + else: + pyramid_dict[state][lga][ward] = {phc_site: group_name} + else: + pyramid_dict[state][lga] = {ward: {phc_site: group_name}} + else: + pyramid_dict[state] = {lga: {ward: {phc_site: group_name}}} + + +# TODO: Generalize this based on the retrieved OU types +for state in pyramid_dict.keys(): + print(state, end=" ") + state_id = find_or_create_org_unit(name=state, org_unit_type_id=ou_types["State"], parent_id=COUNTRY_ID) + + for lga in pyramid_dict[state].keys(): + print("\t" + lga, end=" ") + county_id = find_or_create_org_unit( + name=lga, + org_unit_type_id=ou_types["LGA"], + parent_id=state_id, + ) + + for ward in pyramid_dict[state][lga].keys(): + print("\t\t" + ward, end=" ") + payam_id = find_or_create_org_unit( + name=ward, + org_unit_type_id=ou_types["Ward"], + parent_id=county_id, + ) + + for phc_site in pyramid_dict[state][lga][ward].keys(): + print("\t\t\t" + phc_site, end=": ") + group_name = pyramid_dict[state][lga][ward][phc_site] + print(group_name) + find_or_create_org_unit( + name=phc_site, + org_unit_type_id=ou_types["PHC/Site"], + parent_id=payam_id, + group_id=groups[group_name], + )