119 lines
3.4 KiB
Python
119 lines
3.4 KiB
Python
import base64
|
|
import json
|
|
import os
|
|
import sys
|
|
import requests
|
|
from modules import html_helper
|
|
from dotenv import load_dotenv
|
|
from modules.stellenangebot import Stellenangebot, JobSearchResponse
|
|
|
|
load_dotenv()
|
|
|
|
defapi_token = os.getenv("DEF_API_TOKEN")
|
|
arbeitsagentur_base_url = "https://rest.arbeitsagentur.de/"
|
|
defapi_url = "https://api.defapi.org/api/v1/decisions"
|
|
timeout = 60
|
|
jev_instructions_file = "jev-instructions.json"
|
|
search_strings = ["Developer", "Webentwickler", "Softwareentwickler"]
|
|
wo = "Pforzheim"
|
|
combined_response: JobSearchResponse = {"ergebnisliste": []}
|
|
|
|
|
|
def read_json(file_name) -> dict:
|
|
with open(file_name, mode="r", encoding="utf-8") as read_file:
|
|
jev_instructions = json.load(read_file)
|
|
read_file.close()
|
|
return jev_instructions
|
|
|
|
|
|
def get_job_refs(jobs_data: JobSearchResponse) -> list:
|
|
job_ids = []
|
|
for job in jobs_data["ergebnisliste"]:
|
|
job_ids.append(job["referenznummer"])
|
|
return job_ids
|
|
|
|
|
|
def get_jobs(search_string: str) -> JobSearchResponse:
|
|
jobs_path = "jobboerse/jobsuche-service/pc/v6/jobs"
|
|
|
|
headers = {
|
|
"X-API-Key": "jobboerse-jobsuche"
|
|
}
|
|
|
|
params = {
|
|
"angebotsart": "1",
|
|
"was": search_string,
|
|
"wo": wo,
|
|
"umkreis": "50",
|
|
"arbeitszeit": "ho;vz;tz",
|
|
"page": "1",
|
|
"size": "50"
|
|
}
|
|
|
|
url = arbeitsagentur_base_url + jobs_path
|
|
response = requests.get(url, headers=headers, params=params, timeout=timeout)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
print(f"Fehler: {response.status_code}")
|
|
sys.exit(response.text)
|
|
|
|
|
|
def fetch_job_data(job_ref: str) -> Stellenangebot:
|
|
encoded_ref = base64.b64encode(job_ref.encode("utf-8")).decode("utf-8")
|
|
job_path = f"{arbeitsagentur_base_url}/jobboerse/jobsuche-service/pc/v4/jobdetails/{encoded_ref}"
|
|
headers = {
|
|
"X-API-Key": "jobboerse-jobsuche"
|
|
}
|
|
response = requests.get(job_path, headers=headers)
|
|
return response.json()
|
|
|
|
|
|
def classify(job_ref: str, jev_instructions: dict) -> tuple[Stellenangebot, float] | None:
|
|
job_data = fetch_job_data(job_ref)
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {defapi_token}"
|
|
}
|
|
|
|
data = {
|
|
"model": "typesafe/jev-1.13",
|
|
"state": job_data["stellenangebotsBeschreibung"],
|
|
"questions": {
|
|
"job_classification": jev_instructions
|
|
}}
|
|
|
|
response = requests.post(defapi_url, headers=headers, json=data, timeout=timeout)
|
|
|
|
if response.status_code == 200:
|
|
response_data = response.json()
|
|
answers = response_data["answers"]
|
|
score = answers["job_classification"]["noul"]
|
|
if score > 0.6:
|
|
return job_data, score
|
|
else:
|
|
return None
|
|
else:
|
|
print(f"Fehler: {response.status_code}")
|
|
return None
|
|
|
|
|
|
def main():
|
|
for search in search_strings:
|
|
jobs_data = get_jobs(search)
|
|
combined_response["ergebnisliste"].extend(jobs_data["ergebnisliste"])
|
|
|
|
jev_instructions = read_json(jev_instructions_file)
|
|
html_parts = []
|
|
for job_ref in get_job_refs(combined_response):
|
|
data = classify(job_ref, jev_instructions)
|
|
if data:
|
|
html_parts.append(html_helper.render_html(data))
|
|
|
|
html_helper.write_html(html_parts)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|