Tracking Article Processing Charges (APCs) for a given institution

SAVE_CSV = False  # flag to determine whether to save the output as a CSV file 

# input
ror_id = "https://ror.org/02fa3aq29"
publication_year = 2024
publication_types = ["article", "review"]
publication_oa_statuses = ["gold", "hybrid"]

import requests

# construct the url using the provided ror id
url = f"https://api.openalex.org/institutions?filter=ror:{ror_id}"

# send a get request to the constructed url
response = requests.get(url)

# parse the response json data
json_data = response.json()

# extract the institution id from the first result
institution_id = json_data["results"][0]["id"]  # https://openalex.org/I98251732

import numpy as np
import pandas as pd

def get_works_by_institution(institution_id, publication_year, publication_types, page=1, items_per_page=50):
    # construct the api url with the given institution id, publication year, publication types, page number, and items per page
    url = f"https://api.openalex.org/works?filter=corresponding_institution_ids:{institution_id},publication_year:{publication_year},type:{'|'.join(publication_types)},oa_status:{'|'.join(publication_oa_statuses)}&page={page}&per-page={items_per_page}"

    # send a GET request to the api and parse the json response
    response = requests.get(url)
    json_data = response.json()

    # convert the json response to a dataframe
    df_json = pd.DataFrame.from_dict(json_data["results"])

    next_page = True
    if df_json.empty: # check if the dataframe is empty (i.e., no more pages available)
        next_page = False

    # if there are more pages, recursively fetch the next page
    if next_page:
        df_json_next_page = get_works_by_institution(institution_id, publication_year, publication_types, page=page+1, items_per_page=items_per_page)
        df_json = pd.concat([df_json, df_json_next_page])

    return df_json

df_works = get_works_by_institution(institution_id, publication_year, publication_types)
if SAVE_CSV:
    df_works.to_csv(f"institution_works_{publication_year}.csv", index=True)

df_works

# extract 'value_usd' from 'apc_list' if it is a dictionary (i.e. 'apc_list' exists in the work record); otherwise, set to null
df_works["apc_list_usd"] = df_works["apc_list"].apply(lambda apc_list: apc_list["value_usd"] if isinstance(apc_list, dict) else np.nan)

# extract 'id' and 'name' from 'source' within 'primary_location' if 'source' exists; otherwise, set to null
df_works["source_id"] = df_works["primary_location"].apply(lambda location: location["source"]["id"] if location["source"] else np.nan)
df_works["source_name"] = df_works["primary_location"].apply(lambda location: location["source"]["display_name"] if location["source"] else np.nan)

# extract 'host_organization' from 'source' within 'primary_location' if 'source' exists; otherwise, set to null
df_works["source_host_organization"] = df_works["primary_location"].apply(lambda location: location["source"]["host_organization"] if location["source"] else np.nan)

# extract 'issn' and 'issn_l' from 'source' within 'primary_location' if 'source' exists; otherwise, set to null
df_works["source_issn"] = df_works["primary_location"].apply(lambda location: location["source"]["issn"] if location["source"] else np.nan)
df_works["source_issn_l"] = df_works["primary_location"].apply(lambda location: location["source"]["issn_l"] if location["source"] else np.nan)

# calculate the average apc where 'apc_list_usd' is not null
apc_mean = df_works[df_works["apc_list_usd"].notnull()]["apc_list_usd"].mean()

# fill null values in 'apc_list_usd' with the calculated average
df_works["apc_list_usd"] = df_works["apc_list_usd"].fillna(apc_mean)

# fill null values in 'source_id', 'source_name', 'source_issn' and 'source_issn_l'
df_works["source_id"] = df_works["source_id"].fillna("unknown source")
df_works["source_name"] = df_works["source_name"].fillna("unknown source")
df_works["source_host_organization"] = df_works["source_host_organization"].fillna("unknown source")
df_works["source_issn"] = df_works["source_issn"].fillna("unknown source")
df_works["source_issn_l"] = df_works["source_issn_l"].fillna("unknown source")

import re

CHUNK_SIZE = 5

def get_source_host_organization_display_name(publisher_ids):
    def get_source_host_organization_publisher_display_name(publisher_ids):
        def get_source_host_organization_publisher_display_name_by_chunk(publisher_ids_chunk):
            # construct the api url using the chunk of publisher ids
            url = f"https://api.openalex.org/publishers?filter=ids.openalex:{'|'.join(publisher_ids_chunk)}"

            # send a GET request to the api and parse the json response
            response = requests.get(url)
            json_data = response.json()

            # convert the json response to a dataframe and return the relevant columns
            df_json = pd.DataFrame.from_dict(json_data["results"])
            return df_json[["id", "display_name"]]
        
        # check if the length of 'publisher_ids' is less than 1
        if len(publisher_ids) < 1:
            # if true, return an empty dataframe
            return pd.DataFrame()

        # split the publisher ids into chunks and apply the function to each chunk
        chunks = np.array_split(publisher_ids, np.ceil(len(publisher_ids) / CHUNK_SIZE))
        df_chunks = pd.DataFrame({"chunk": chunks})
        return pd.concat(df_chunks["chunk"].apply(get_source_host_organization_publisher_display_name_by_chunk).tolist())

    def get_source_host_organization_institution_display_name(institution_ids):
        def get_source_host_organization_institution_display_name_by_chunk(institution_ids_chunk):
            # construct the api url using the chunk of institution ids
            url = f"https://api.openalex.org/institutions?filter=id:{'|'.join(institution_ids_chunk)}"

            # send a GET request to the api and parse the json response
            response = requests.get(url)
            json_data = response.json()

            # convert the json response to a dataframe and return the relevant columns
            df_json = pd.DataFrame.from_dict(json_data["results"])
            return df_json[["id", "display_name"]]

        # check if the length of 'institution_ids' is less than 1
        if len(institution_ids) < 1:
            # if true, return an empty dataframe
            return pd.DataFrame()

        # split the institution ids into chunks and apply the function to each chunk
        chunks = np.array_split(institution_ids, np.ceil(len(institution_ids) / CHUNK_SIZE))
        df_chunks = pd.DataFrame({"chunk": chunks})
        return pd.concat(df_chunks["chunk"].apply(get_source_host_organization_institution_display_name_by_chunk).tolist())

     # filter the publisher ids to get only publisher urls
    publishers = list(filter(lambda s: re.search(r"https:\/\/openalex\.org\/P", s), publisher_ids))
    # filter the institution ids to get only institution urls
    institutions = list(filter(lambda s: re.search(r"https:\/\/openalex\.org\/I", s), publisher_ids))

    # create a dataframe with a default entry for unknown source
    df_lookup = pd.DataFrame.from_dict({"id": ["unknown source"], "display_name": ["unknown source"]})
    # concatenate the dataframes with publisher and institution display names
    df_lookup = pd.concat([df_lookup, get_source_host_organization_publisher_display_name(publishers), get_source_host_organization_institution_display_name(institutions)], ignore_index=True)
    return df_lookup

# get the display names for unique source_host_organization ids in df_works
df_lookup = get_source_host_organization_display_name(df_works["source_host_organization"].unique())
if SAVE_CSV:
    df_lookup.to_csv(f"source_host_organization_lookup.csv", index=True)

df_lookup

# update the 'source_host_organization' with the corresponding display names
df_works["source_host_organization"] = df_works["source_host_organization"].apply(lambda publisher: df_lookup[df_lookup["id"] == publisher]["display_name"].squeeze())  

# group the dataframe by 'source_id' and 'source_issn_l'
# and aggregate 'source_issn' by taking the maximum value (in this case the common issn list of strings)
# and aggregate 'source_host_organization' by taking the maximum value (in this case the common string name of the source's host organization)
# and aggregate 'source_name' by taking the maximum value (in this case the common string name of the source)
# and 'id' by counting
# and 'apc_list_usd' by summing
df_apc = df_works.groupby(["source_id", "source_issn_l"]).agg({"source_issn": "max", "source_name": "max", "source_host_organization": "max", "id": "count", "apc_list_usd": "sum"})
# rename the 'id' column to 'num_publications' and 'apc_list_usd' column to 'apc_usd'
df_apc.rename(columns={"id": "num_publications", "apc_list_usd": "apc_usd"}, inplace=True)
if SAVE_CSV:
    df_apc.to_csv(f"apc_usd_by_source.csv", index=True)

df_apc

total_apc = df_apc["apc_usd"].sum()
print(f"Estimated total (non-discounted) APC cost in {publication_year}: USD {round(total_apc, 2)}.")

# input
df_discount = pd.read_csv("discount-list.csv")

import typing

def get_discount(issn: typing.List[str] | str, apc: float) -> float:
    # check if issn is a string, if so, convert it to a list
    if isinstance(issn, str):
        issn = [issn]

    # filter the discount dataframe to get rows where 'issn' is in the provided issn list
    discount_rows = df_discount[df_discount["issn"].isin(issn)]
    
    # if no discount rows are found, return the original apc
    if discount_rows.empty:
        return apc

    # get the first row from the filtered discount rows
    discount_row = discount_rows.iloc[0]
    
    # if the discount is a flat rate, subtract the discount from the apc
    if discount_row["is_flatrate"]:
        return apc - discount_row["discount"]
    else:
        # if the discount is a percentage, apply the discount to the apc
        return apc * (1 - discount_row["discount"])

# apply the get_discount function to each row of the dataframe to calculate the discounted apc and store it in a new column 'discounted_apc_usd'
df_apc["discounted_apc_usd"] = df_apc.apply(lambda x: get_discount(issn=x["source_issn"], apc=x["apc_usd"]), axis=1)
if SAVE_CSV:
    df_apc.to_csv(f"apc_usd_with_discounts_by_source.csv", index=True)

df_apc

total_apc_discount = df_apc["discounted_apc_usd"].sum()
print(f"Estimated APC cost (including discounts) for {publication_year}: USD {round(total_apc_discount, 2)}.")

print(f"Estimated APC saving in {publication_year}: USD {round(total_apc - total_apc_discount, 2)}.")