import requests, json
def get_origin_version(doi, count=1):
req = requests.get( f"https://api.datacite.org/dois/{doi}" )
res = req.json()
try:
origins = res["data"]["attributes"]["relatedIdentifiers"]
except:
return doi, count
else:
if origins == []: return doi, count
for i in origins:
if i["relationType"] != "IsVersionOf":
return doi, count
else:
return get_origin_version(i["relatedIdentifier"], count+1)
def get_md_from_datacite( doi ) :
"""
retrieve data research metadata from datacite
"""
req = requests.get( f"https://api.datacite.org/dois/{doi}" )
try :
res = req.json()
except :
return "error"
if "errors" in res :
return "error"
return res
def parse_value_following_instruction(key, instruction, datacite_content) :
"""
permet de récuéprer les données reçues depuis datacite avec leur propre sturcturation selon les instructions précisées manuellement
key : la clé à traiter
instruction : les instructions à appliquer pour récupérer la valeur de la clé
datacite_content : le contenu de datacite à la clé précisé
la recherche des instructions est effectuée notamment depuis leur type (string ou dict) : à revoir pour plus de cohérence
"""
buffer = {} ## pour récupérer les données avec un dictionnaire
## quand les données à récupérer sont des objets simples, on récupère simplement la valeur
if instruction == "string" or instruction == "int" :
return {key: datacite_content}
## qaund les instructions sont fomratées en dict
if isinstance(instruction, dict) :
## si past_values_w_this_key est dans les instructions
if "past_values_w_this_key" in instruction :
## pour débbugger
##print("attribute is", key)
## quand les données sont directement sous forme de dict (eg. datatype)
if instruction["past_values_w_this_key"] in datacite_content :
temp_key_to_get = instruction["past_values_w_this_key"]
buffer.update(
{ instruction["past_values_w_this_key"] : datacite_content[temp_key_to_get] }
)
## quand les données sont des listes et qu'il faut itérer dessus
else :
all_vals = []
temp_key_to_get = instruction["past_values_w_this_key"]
# itérer sur les éléments, vérifier si la clé est bien présente, si oui ajouter la valeur
[all_vals.append( item[ temp_key_to_get ]) for item in datacite_content if item.get(temp_key_to_get)]
buffer.update(
{temp_key_to_get : ",".join(all_vals)}
)
## quand il faut sortir toutes les données brutes
if "flatten_all_in_this_key" in instruction :
buffer.update(
{instruction["flatten_all_in_this_key"] : str(datacite_content) }
)
## if past_first_occ in instruction
if "past_first_occ" in instruction :
temp_key_to_get = instruction["past_first_occ"]
buffer.update(
{temp_key_to_get: datacite_content[0][temp_key_to_get]}
)
## if go_to_sub_key in the instruction
if "go_to_sub_key" in instruction :
temp_parent_key_to_get = instruction["go_to_sub_key"]
## identifier les clés enfantes à retrouver dans dataCite
temp_child_key_to_get = instruction["get_key"]
buffer.update(
{key : datacite_content[temp_parent_key_to_get][temp_child_key_to_get] }
)
else :
buffer.update(
{key : "to_do"}
)
return buffer
def parse_datacite_md(raw_datacite_mds):
"""
from json file load instruction
from DOI get datacite cotent
iterate on all datacite attributes
if data from datacite is needed get it with parse_value_following_instruction()
iterate on all datacite relationship
if data from datacite is needed get it with parse_value_following_instruction()
"""
doi_md = {
"doi" : raw_datacite_mds["data"]["id"]
}
## ____0____ from json file load instructions
with open("datacite-parser-instructions.json") as parser_file :
datacite_parser = json.load(parser_file)
## liste de tous les attributs à récupérer
attributes_to_get = datacite_parser["path-and-fields"]["attributes"].keys()
relations_to_get = datacite_parser["path-and-fields"]["relationships"].keys()
## ____1___ iterate on datacite attributes
for attribute_key in raw_datacite_mds["data"]["attributes"] :
attribute_value = raw_datacite_mds["data"]["attributes"][attribute_key]
## ne pas prendre les valeurs si elles sont nulles (sauf pour les nb)
if not isinstance(attribute_value, int) and not attribute_value :
# pour suivi print(f"{attribute_key} is empty")
continue
## si l'attribut fait parti de ceux à récupérer
if attribute_key in attributes_to_get :
## redistribuer le nom de la clé et sa valeur
value_to_add = parse_value_following_instruction(
attribute_key,
datacite_parser["path-and-fields"]["attributes"][attribute_key],
attribute_value)
doi_md.update(value_to_add)
## ____2___ iterate on datacite relations
### nb : on pourrait alléger en regroupant attribut et relationships
for relation_key in raw_datacite_mds["data"]["relationships"] :
relation_value = raw_datacite_mds["data"]["relationships"][relation_key]
## ne pas prendre les valeurs si elles sont nulles (sauf pour les nb)
if not isinstance(relation_key, int) and not relation_value :
continue
## si la relation est préciser dans les instructions
if relation_key in relations_to_get :
## to debug print("relation is", relation_key)
relation_to_add = parse_value_following_instruction(
relation_key,
datacite_parser["path-and-fields"]["relationships"][relation_key],
relation_value
)
doi_md.update(relation_to_add)
return doi_md
def from_files_load_dois(files_name) :
"""
load txt files which contains DOIs
"""
all_dois = []
for file_shortname in files_name :
file_fullname = file_shortname + "-dois.txt"
folder_path = "../0-collect-data/"
temp_loaded_dois = []
with open(folder_path + file_fullname) as f:
## attention à la fin des lignes il y a un retour à la ligne, d'où le (line)-1
## lower DOI string also
[temp_loaded_dois.append( line[ :len(line)-1].lower()) for line in f.readlines()]
print(f"\t{file_shortname:<10s}\t\t{len(temp_loaded_dois)}")
[all_dois.append(elem) for elem in temp_loaded_dois]
return all_dois