creator.py

# Import the necessary libraries
import json
import requests
import pandas as pd

# Load the configuration settings from config.json
with open('config.json', 'r', encoding='utf-8') as f:
    config = json.load(f)

# Define the necessary variables
OCL_API_URL = config['OCL_API_URL']
SOURCE_ID = config['SOURCE_ID']
COLLECTION_ID = config['COLLECTION_ID']
OCL_TOKEN = config['OCL_TOKEN']
ORG_ID = config['ORG_ID']

ORG_ID = "MSF"
SOURCE_ID = "MSF"
COLLECTION_ID = "MentalHealth"

# Headers for the API request
HEADERS = {
    "Authorization": f"Token {OCL_TOKEN}",
    "Content-Type": "application/json"
}

# Create the necessary OCL concepts for the provided CSV file
csv_file = "cleaned_concepts.csv"

# Create a function to check if a concept external ID already exists in OCL collection TEST - If not, add it as a reference 
def check_and_add_reference(concept_details):
    concept_id = concept_details["id"]
    url = f"https://api.openconceptlab.org/orgs/{ORG_ID}/collections/{COLLECTION_ID}/concepts/{concept_id}"
    response = requests.get(url, headers=HEADERS, timeout=30)
    if response.status_code == 200:
        print(f"Concept '{concept['name']}' already exists in OCL collection - Skipping")
    else:
        # Add the newly created concept to the OCL collection as a reference
        reference_query = {
            "data": {
                "expressions": [
                    concept_details["url"]
                ]
            },
            "cascade": {
                "method": "sourcetoconcepts",
                "cascade_levels": "*",
                "map_types": "Q-AND-A,CONCEPT-SET",
                "return_map_types": "*"
            }
        }
        collection_api_url = f"https://api.openconceptlab.org/orgs/{ORG_ID}/collections/{COLLECTION_ID}/references/?transformReferences=extensional"
        response = requests.put(collection_api_url, json=reference_query, headers=HEADERS)
        if response.status_code == 200:
            print(f"Concept '{concept['name']}' added to OCL collection as a reference")
        else:
            print(f"Failed to add concept '{concept['name']}' to OCL collection - Skipping reference")

def create_concepts(csv_file):
    concepts = []

    # Load the CSV file
    df = pd.read_csv(csv_file)

    # Iterate over each row in the CSV file
    for _, row in df.iterrows():
        concept = {
            "name":  row["Name"],
            "external_id": row["External ID"] if not pd.isnull(row["External ID"]) else None,
        }
        concepts.append(concept)

    return concepts, df

concepts, df = create_concepts(csv_file)


# Write a query to create each concept in OCL using Open Concept Lab API
for concept in concepts:
    source_query = {
    "concept_class": "Misc",
    "datatype": "N/A",
    "names": [
            {
                "name": concept["name"],
                "locale": "en",
                "locale_preferred": "true",
                "name_type": "Fully-Specified"
            }
        ]
    }
    # Make a POST request to the Open Concept Lab API to create the concept
    # If the concept creation is successful, update the column "External ID" with the OCL concept External ID in the CSV file
    # if column "External ID" is empty, create the OCL concept, otherwise skip the concept creation
    # Save the updated CSV file with the OCL concept External IDs if concept creation is successful
    if concept["external_id"] is None:
        source_api_url = f"https://api.openconceptlab.org/orgs/{ORG_ID}/sources/{SOURCE_ID}/concepts/"
        response = requests.post(source_api_url, json=source_query, headers=HEADERS)
        if response.status_code == 201:
            concept_details = response.json()
            concept["external_id"] = concept_details["external_id"]
            df.loc[df["Name"] == concept["name"], "External ID"] = concept["external_id"] 
            # Update the CSV file with external IDs if concept creation is successful
            df.to_csv(csv_file, index=False)
            print(f"Concept '{concept['name']}' created successfully with External ID: {concept['external_id']}")
            check_and_add_reference(concept_details)
            
        else:
            print(f"Failed to create concept: {concept['name']}")
            print(f"Status code: {response.status_code}")
    else:
        print(f"Concept {concept['name']} already exists with external ID: {concept['external_id']}")

print("Concepts created and added successfully!")