From a47ea8c8ed9e84c5bdd8b040956e848aba98ae0f Mon Sep 17 00:00:00 2001 From: Daniel Schubert Date: Thu, 8 Oct 2026 11:33:52 +0200 Subject: [PATCH] first working version --- .gitignore | 4 ++ jev-instructions.json | 8 +++ main.py | 123 ++++++++++++++++++++++++++++++++++++++ modules/html_helper.py | 43 +++++++++++++ modules/stellenangebot.py | 62 +++++++++++++++++++ 5 files changed, 240 insertions(+) create mode 100644 .gitignore create mode 100644 jev-instructions.json create mode 100644 main.py create mode 100644 modules/html_helper.py create mode 100644 modules/stellenangebot.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..45b5552 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.env +venv/ +.idea +stellenangebote.html diff --git a/jev-instructions.json b/jev-instructions.json new file mode 100644 index 0000000..bf98d4f --- /dev/null +++ b/jev-instructions.json @@ -0,0 +1,8 @@ +{ + "type": "noul", + "instructions": "Bewerte ob ein erfahrener Webentwickler gesucht wird. Nach folgenden Skills sollten nicht gefragt werden: C#, C, Java, SAP, Typo3", + "criteria": { + "true": "Die Qualifikation ist ausreichend", + "false": "Es werden andere Programmier-Sprachen gefordert." + } +} \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..2892209 --- /dev/null +++ b/main.py @@ -0,0 +1,123 @@ +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 + + +search_strings = ["Developer", "Webentwickler", "Softwareentwickler"] +combined_response: JobSearchResponse = {"ergebnisliste": []} + +def main(): + load_dotenv() + jobs_data = [] + for search in search_strings: + jobs_data = get_jobs(search) + combined_response["ergebnisliste"].extend(jobs_data["ergebnisliste"]) + + job_refs = get_job_refs(combined_response) + jev_instructions = read_json() + print(jobs_data) + html_parts = [] + for job_ref in job_refs: + data = classify(job_ref, jev_instructions) + if data: + html_parts.append(html_helper.render_html(data)) + + html_helper.write_html(html_parts) + + +def read_json() -> dict: + with open("jev-instructions.json", 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": "Pforzheim", + "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_validation": 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_validation"]["noul"] + if score > 0.6: + return job_data, score + else: + return None + else: + print(f"Fehler: {response.status_code}") + return None + + + + +if __name__ == "__main__": + main() diff --git a/modules/html_helper.py b/modules/html_helper.py new file mode 100644 index 0000000..e5ed71a --- /dev/null +++ b/modules/html_helper.py @@ -0,0 +1,43 @@ +import json + +import markdown +from tinyhtml import h, raw, render +from stellenangebot import Stellenangebot + +def render_html(data: tuple[Stellenangebot, float]) -> str: + job_data, score = data + description = job_data.get("stellenangebotsBeschreibung", "") + description_html = markdown.markdown(description, extensions=["nl2br"]) + + job_fields = [] + for field_name, field_value in job_data.items(): + if field_name in ("stellenangebotsTitel", "stellenangebotsBeschreibung"): + continue + + if isinstance(field_value, (dict, list)): + field_value = json.dumps(field_value, ensure_ascii=False) + + job_fields.append(h("p")(f"{field_name}: {field_value}")) + + job_section = h("section")( + h("h1")(f"{job_data.get('stellenangebotsTitel', '')} - {score:.2f}"), + h("div")(raw(description_html)), + h("hr"), + *job_fields, + ) + + return render(job_section) + + +def write_html(html_parts: list[str]) -> None: + stylesheet_link = '' + html_document = render( + h("html")( + h("head")(raw(stylesheet_link)), + h("body")(*[raw(html_part) for html_part in html_parts]), + ) + ) + + with open("stellenangebote.html", mode="w", encoding="utf-8") as html_file: + html_file.write(html_document) + html_file.write("\n") diff --git a/modules/stellenangebot.py b/modules/stellenangebot.py new file mode 100644 index 0000000..48e4377 --- /dev/null +++ b/modules/stellenangebot.py @@ -0,0 +1,62 @@ +from typing import TypedDict + + +class Eintrittszeitraum(TypedDict, total=False): + von: str # Date string in format 'YYYY-MM-DD' + + +class Veroeffentlichungszeitraum(TypedDict): + von: str # Date string in format 'YYYY-MM-DD' + + +class Adresse(TypedDict): + plz: str + ort: str + region: str + land: str + + +class Stellenlokation(TypedDict): + adresse: Adresse + breite: float + laenge: float + + +class Stellenangebot(TypedDict, total=False): + stellenangebotsart: str + stellenangebotsTitel: str + quereinstiegGeeignet: bool + stellenangebotsBeschreibung: str + arbeitszeitSchichtNachtWochenende: bool + arbeitszeitTeilzeitAbend: bool + arbeitszeitTeilzeitNachmittag: bool + arbeitszeitTeilzeitVormittag: bool + arbeitszeitTeilzeitFlexibel: bool + arbeitszeitVollzeit: bool + eintrittszeitraum: Eintrittszeitraum + verguetungsangabe: str + vertragsdauer: str + istGeringfuegigeBeschaeftigung: bool + istBehinderungGefordert: bool + stellenlokationen: list[Stellenlokation] + homeofficemoeglich: bool + homeofficetyp: str + veroeffentlichungszeitraum: Veroeffentlichungszeitraum + datumErsteVeroeffentlichung: str # Date string in format 'YYYY-MM-DD' + aenderungsdatum: str # Datetime string in ISO format + istBetreut: bool + hauptberuf: str + alternativBeruf1: str + firma: str + arbeitgeberKundennummerHash: str + allianzpartnerUrl: str + allianzpartnerName: str + referenznummer: str + + +class JobSearchResult(TypedDict): + referenznummer: str + + +class JobSearchResponse(TypedDict): + ergebnisliste: list[JobSearchResult]