init project running

a scraper for jobs in the region with connected telegram bot to send my new jobs every day

05_System/feed_filter.py

@@ -0,0 +1,247 @@
+#!/usr/bin/env python3
+"""Jobsuche: Feed-Fetcher mit Regional-/Strukturfilter.
+
+Holt die konfigurierten Job-Quellen (Atom + Personio-XML + Arbeitsagentur-
+REST-API), filtert regional (Dresden + Umland) und gibt das Ergebnis als
+JSON auf stdout aus.
+Nur Python-Stdlib. Fehler einzelner Feeds werden abgefangen (status=error),
+das Skript endet immer mit Exit-Code 0.
+"""
+import json
+import re
+import urllib.request
+import xml.etree.ElementTree as ET
+from urllib.parse import urlencode
+from datetime import datetime, timezone
+
+UA = "Mozilla/5.0 (X11; Linux x86_64) JobRadar/1.0"
+TIMEOUT = 25
+
+# Region: Dresden + Großraum bis ~60 km (inkl. Richtung Leipzig/Chemnitz/Bautzen)
+REGION = re.compile(
+    r"Dresden|Meißen|Meissen|Radebeul|Coswig|Freital|Pirna|Heidenau|Riesa|"
+    r"Großenhain|Kamenz|Bautzen|Dippoldiswalde|Freiberg|Chemnitz|Leipzig|"
+    r"Görlitz|Sächsische Schweiz|Bad Schandau|Sebnitz|Neustadt|Ottendorf|"
+    r"Wilsdruff|Tharandt|Bannewitz|Kreischa|Dohna|Glashütte|Weinböhla|Nossen|"
+    r"Lommatzsch|Lausitz|Sachsen|Saxony",
+    re.IGNORECASE,
+)
+
+FEEDS = [
+    {
+        "name": "greenjobs",
+        "kind": "atom",
+        "url": ("https://www.greenjobs.de/angebote/neueste.html"
+                "?zeitraum=28&feed=atom&ort=Dresden&umkreis=60"),
+    },
+    {
+        "name": "eejobs",
+        "kind": "atom",
+        "url": ("https://www.eejobs.de/angebote/neueste.html"
+                "?zeitraum=28&feed=atom&ort=Dresden&umkreis=60"),
+    },
+    {
+        "name": "wandelbots",
+        "kind": "personio",
+        "company": "wandelbots",
+        "url": "https://wandelbots.jobs.personio.de/xml",
+    },
+    {
+        "name": "heliatek",
+        "kind": "personio",
+        "company": "heliatek",
+        "url": "https://heliatek.jobs.personio.de/xml",
+    },
+    {
+        "name": "arbeitsagentur",
+        "kind": "arbeitsagentur",
+        "url": "https://rest.arbeitsagentur.de/jobboerse/jobsuche-service/pc/v6/jobs",
+    },
+]
+
+ATOM_NS = {"a": "http://www.w3.org/2005/Atom"}
+
+# ── Arbeitsagentur-Jobbörse (REST-API, kein Feed) ──────────────────────────
+BA_URL = "https://rest.arbeitsagentur.de/jobboerse/jobsuche-service/pc/v6/jobs"
+BA_API_KEY = "jobboerse-jobsuche"
+# Suchbegriffe = Titelsuche, abgestimmt aufs Suchprofil (Rollen-Keywords).
+BA_QUERIES = [
+    "Linux", "Systemadministrator", "IT-Administrator", "DevOps", "Python",
+    "Netzwerk", "Embedded", "Infrastruktur", "Security", "Kubernetes",
+]
+BA_FIXED = {
+    "angebotsart": "1",        # 1 = ARBEIT
+    "wo": "Dresden",
+    "umkreis": "60",
+    "pav": "false",            # keine privaten Arbeitsvermittlungen
+    "zeitarbeit": "false",
+    "arbeitszeit": "vz;tz",    # Vollzeit + Teilzeit
+    "befristung": "2",         # 2 = unbefristet
+    "veroeffentlichtseit": "7",
+}
+# Hart-Ausschluss: Rollen, die nie "klar IT" sein können (Suchprofil).
+BA_HARD_BLOCK = re.compile(
+    r"Pflege|Handwerk|Callcenter|Buchhaltung|Controlling|Lohnbuchhaltung|"
+    r"Vertrieb|Außen|Aussen|Marketing|Recruiting|Personalwesen|"
+    r"Kassierer|Verkäufer|Reinigung",
+    re.IGNORECASE,
+)
+# Nur markieren, nicht ausschließen (Agent prüft Ausnahme: klar IT + klar öko).
+BA_SOFT_BLOCK = re.compile(r"Java|PHP|ABAP|SAP|COBOL", re.IGNORECASE)
+
+
+def fetch(url: str, headers: dict | None = None) -> bytes:
+    hdrs = {"User-Agent": UA, "Accept": "*/*"}
+    if headers:
+        hdrs.update(headers)
+    req = urllib.request.Request(url, headers=hdrs)
+    with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
+        return resp.read()
+
+
+def parse_atom(data: bytes) -> list:
+    root = ET.fromstring(data)
+    items = []
+    for e in root.findall("a:entry", ATOM_NS):
+        link_el = e.find("a:link", ATOM_NS)
+        href = link_el.attrib.get("href", "") if link_el is not None else ""
+        title = (e.findtext("a:title", default="", namespaces=ATOM_NS) or "").strip()
+        updated = (e.findtext("a:updated", default="", namespaces=ATOM_NS) or "").strip()
+        items.append({"title": title, "url": href, "updated": updated})
+    return items
+
+
+def parse_personio(data: bytes, company: str) -> list:
+    root = ET.fromstring(data)
+    items = []
+    for p in root:
+        def f(tag: str) -> str:
+            el = p.find(tag)
+            return (el.text or "").strip() if el is not None else ""
+
+        office = f("office")
+        if office and not REGION.search(office):
+            continue
+        jid = f("id")
+        items.append({
+            "title": f("name"),
+            "office": office,
+            "department": f("department"),
+            "employmentType": f("employmentType"),
+            "seniority": f("seniority"),
+            "schedule": f("schedule"),
+            "createdAt": f("createdAt"),
+            "keywords": f("keywords"),
+            "url": f"https://{company}.jobs.personio.de/job/{jid}" if jid else "",
+        })
+    return items
+
+
+def parse_arbeitsagentur() -> tuple:
+    """Arbeitsagentur-Jobbörse: alle Suchbegriffe abfragen, deduplizieren.
+
+    Gibt (items, errors) zurück; einzelne Suchanfragen werden isoliert
+    abgefangen, damit ein API-Fehler nicht die ganze Quelle kippt.
+    """
+    by_ref: dict = {}   # referenznummer -> item
+    errors: list = []
+    for q in BA_QUERIES:
+        params = urlencode({"was": q, **BA_FIXED})
+        page, total, collected = 1, None, 0
+        while True:
+            url = f"{BA_URL}?{params}&page={page}&size=100"
+            try:
+                data = fetch(url, headers={"X-API-Key": BA_API_KEY})
+                payload = json.loads(data)
+            except Exception as ex:  # noqa: BLE001
+                errors.append(f"{q!r} (Seite {page}): {type(ex).__name__}: {ex}")
+                break
+            if total is None:
+                total = payload.get("maxErgebnisse") or 0
+            results = payload.get("ergebnisliste") or []
+            for j in results:
+                ref = (j.get("referenznummer") or "").strip()
+                title = (j.get("stellenangebotsTitel") or "").strip()
+                if not ref or not title:
+                    continue
+                if BA_HARD_BLOCK.search(title):
+                    continue
+                if ref not in by_ref:
+                    locs = j.get("stellenlokationen") or []
+                    orts, region = [], ""
+                    for loc in locs:
+                        adr = loc.get("adresse") or {}
+                        ort = (adr.get("ort") or "").strip()
+                        if ort and ort not in orts:
+                            orts.append(ort)
+                        if not region:
+                            region = (adr.get("region") or "").strip()
+                    ver = ((j.get("veroeffentlichungszeitraum") or {}).get("von")
+                           or (j.get("datumErsteVeroeffentlichung") or ""))
+                    eintritt = (j.get("eintrittszeitraum") or {}).get("von") or ""
+                    by_ref[ref] = {
+                        "title": title,
+                        "firma": (j.get("firma") or "").strip(),
+                        "ort": " / ".join(orts),
+                        "region": region,
+                        "entfernung": j.get("entfernung"),
+                        "vertragsdauer": j.get("vertragsdauer") or "",
+                        "vollzeit": j.get("arbeitszeitVollzeit"),
+                        "verguetung": j.get("verguetungsangabe") or "",
+                        "veroeffentlichtAm": ver,
+                        "eintrittAb": eintritt,
+                        "hauptberuf": (j.get("hauptberuf") or "").strip(),
+                        "referenznummer": ref,
+                        "url": f"https://www.arbeitsagentur.de/jobsuche/jobdetail/{ref}",
+                        "matchedQueries": [q],
+                    }
+                else:
+                    by_ref[ref]["matchedQueries"].append(q)
+                soft = BA_SOFT_BLOCK.findall(title)
+                if soft:
+                    hint = by_ref[ref].setdefault("exclusion_hint", [])
+                    for m in soft:
+                        if m not in hint:
+                            hint.append(m)
+            collected += len(results)
+            if total == 0 or collected >= total or page >= 5:
+                break
+            page += 1
+    items = sorted(
+        by_ref.values(),
+        key=lambda i: (i["entfernung"] if isinstance(i["entfernung"], (int, float)) else 99,
+                       i["title"].lower()),
+    )
+    return items, errors
+
+
+def main() -> None:
+    out = {"generated_at": datetime.now(timezone.utc).isoformat(), "feeds": {}}
+    for fd in FEEDS:
+        res = {"name": fd["name"], "kind": fd["kind"], "url": fd["url"]}
+        try:
+            if fd["kind"] == "arbeitsagentur":
+                items, errors = parse_arbeitsagentur()
+                res["status"] = "ok" if (items or not errors) else "error"
+                if errors:
+                    res["errors"] = errors
+            else:
+                data = fetch(fd["url"])
+                if fd["kind"] == "atom":
+                    items = parse_atom(data)
+                else:
+                    items = parse_personio(data, fd.get("company", ""))
+                res["status"] = "ok"
+            res["count"] = len(items)
+            res["items"] = items
+        except Exception as ex:  # noqa: BLE001 - jedes Feed-Problem einzeln abfangen
+            res["status"] = "error"
+            res["error"] = f"{type(ex).__name__}: {ex}"
+            res["count"] = 0
+            res["items"] = []
+        out["feeds"][fd["name"]] = res
+    print(json.dumps(out, ensure_ascii=False, indent=1))
+
+
+if __name__ == "__main__":
+    main()