first working version

This commit is contained in:
2026-10-08 11:33:52 +02:00
commit a47ea8c8ed
5 changed files with 240 additions and 0 deletions
+4
View File
@@ -0,0 +1,4 @@
.env
venv/
.idea
stellenangebote.html
+8
View File
@@ -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."
}
}
+123
View File
@@ -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()
+43
View File
@@ -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 = '<link rel="stylesheet" href="https://cdn.simplecss.org/simple.min.css">'
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")
+62
View File
@@ -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]