Spaces:
Sleeping
Sleeping
File size: 3,982 Bytes
227a7da |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 |
from config import *
import pandas as pd
import requests
def build_skill_query(skill):
skill_query = skill['preferredLabel']['de'] +" " + skill['preferredLabel']['en']+" "+ skill['description']['de']+ " "+ skill['description']['en']
if skill['borderConcept']['broaderHierarchyConcept']:
broader_hierarchy_concept_str = ", ".join(skill['borderConcept']['broaderHierarchyConcept'])
skill_query += " " + broader_hierarchy_concept_str
else:
pass
if skill['borderConcept']['broaderSkill']:
broaderSkill_str = ", ".join(skill['borderConcept']['broaderSkill'])
skill_query += " " + broaderSkill_str
else:
pass
if skill['alternativeLabel']['de']:
alternativeLabel_de_str = ", ".join(skill['alternativeLabel']['de'])
skill_query += " " + alternativeLabel_de_str
else:
pass
if skill['alternativeLabel']['en']:
alternativeLabel_en_str = ", ".join(skill['alternativeLabel']['en'])
skill_query += " " + alternativeLabel_en_str
else:
pass
return skill_query
def build_occupation_query(occupation):
occupation_name_de = occupation['preferredLabel'].get('de','')
occupation_dsp = occupation['description'].get('de','').get('literal','')
occupation_query = occupation_name_de +" " + occupation['preferredLabel'].get('en','')+" "+ occupation['description'].get('de','').get('literal','') + " "+ occupation_dsp
'''
if occupation['_links']['broaderIscoGroup']:
for group in occupation['_links']['broaderIscoGroup']:
occupation_query += " " + group['title']
else:
pass
'''
return occupation_name_de,occupation_dsp,occupation_query
# Get occupations from a CSV
def get_occupations_from_csv(file_path):
df = pd.read_csv(file_path)
return df[['preferredLabel', 'conceptUri']].set_index('preferredLabel').to_dict()['conceptUri']
def get_oauth_token():
# API endpoint URL
token_url = "https://rest.arbeitsagentur.de/oauth/gettoken_cc"
# Client credentials
client_id = "38053956-6618-4953-b670-b4ae7a2360b1"
client_secret = "c385073c-3b97-42a9-b916-08fd8a5d1795"
grant_type = "client_credentials"
# Prepare request data
payload = {
"client_id": client_id,
"client_secret": client_secret,
"grant_type": grant_type
}
# Send request and get response
response = requests.post(token_url, data=payload)
# Check if the request was successful
if response.status_code == 200:
return response.json().get("access_token")
else:
print("Token request failed:", response.text)
return None
def query_weiterbildungssuche_api(token, params):
# Set API URL
api_url = "https://rest.arbeitsagentur.de/infosysbub/wbsuche/pc/v2/bildungsangebot"
# Prepare request headers
headers = {
"Authorization": f"Bearer {token}"
}
# Send GET request
response = requests.get(api_url, headers=headers, params=params)
if response.status_code == 200:
return response.json()
else:
print("API request failed:", response.text)
return None
def get_courses_from_BA(ids):
# Get OAuth token
token = get_oauth_token()
if token:
# Set query parameters
params = {
"ids": list(ids)
}
# Use token to query the API
result = query_weiterbildungssuche_api(token, params)
return result
def get_occupation_detial(oc_uri):
uri = f'https://ec.europa.eu/esco/api/resource/occupation?selectedVersion=v1.0.9&language=en&uri={oc_uri}'
try:
occupation_response = requests.get(uri)
occupation_response.raise_for_status()
occupation_json = occupation_response.json()
return occupation_json
except requests.RequestException as e:
print(f"Error while fetching skill details: {e}")
|