#!/usr/bin/env python3 """Home-Assistant-Brücke: reicht Einträge der HA-Liste »Hobba Eingang« an Hobba weiter (über den Kurzbefehl „Hobba-Connector", wie hobba-mcp.py) — und hakt sie wieder ab, wenn HA Entwarnung gibt. Zwei Eintrags-Arten auf derselben Liste: - „" → legt die Aufgabe in Hobbas Eingang an und merkt sich Titel → Hobba-ID (~/.hobba-mcp/ha-bridge-map.json). - „erledigt: " → hakt die gemerkte Aufgabe ab (Entwarnung, z. B. Batterie wieder voll oder Anzeige hat geflappt). Ohne gemerkte ID oder bei längst abgehakter Aufgabe passiert still nichts — die Entwarnung darf nie Schaden anrichten. Jede Aufgabe bekommt einen Tag (12.09.2026). Der Grund steht in Sebastians eigener Karte: *„Ich vergesse gerne in den Eingang zu gucken. Ich schaue zumeist nur auf Heute."* „Heute" zeigt aber nur, was einen Tag hat — und was eine Maschine schreibt, hatte bisher keinen. Die Batteriewarnung lag damit im Eingang und wurde nie gesehen. Reihenfolge: 1. Das Fälligkeitsdatum des HA-Eintrags, wenn eines gesetzt ist. 2. Sonst `standardtag` aus der Konfiguration (Vorgabe „heute"). 3. `"standardtag": ""` schaltet es ab — dann bleibt alles wie vorher. Ein im Titel mitgeschriebenes Datum („Filter wechseln morgen") liest Hobba ohnehin selbst; das ausdrückliche Feld sticht es, weil HA mehr weiß. Titel-Dedup (für periodische Re-Checks in HA): Trifft ein Titel ein, zu dem die gemerkte Aufgabe in Hobba noch OFFEN ist, wird nichts Neues angelegt — der Briefkasten-Eintrag fliegt trotzdem raus. Geprüft wird am lebenden Bestand (Snapshot), nicht nur am Merkzettel: Eine von Hand abgehakte Aufgabe blockiert kein echtes neues Vorkommnis. Sicherheitsmodell — bewusst eng gehalten: - Der Mac HOLT (Polling). Kein Lauschport, keine eingehende Verbindung, HA bleibt LAN-only. - Der Token liegt in ~/.hobba-mcp/ha-bridge.conf (0600, lokal, NICHT in iCloud). Härtung: Token eines eigenen HA-Benutzers ohne Admin-Rechte. - Titel/Notizen wandern als JSON-Datei an den Kurzbefehl (subprocess mit Argumentliste) — Gerätenamen können keine Shell erreichen. - Ein Eintrag wird erst aus HA entfernt, NACHDEM Hobba ihn bestätigt hat (Entwarnungen ausgenommen, siehe oben — sie sind idempotent). Einrichten (einmalig): mkdir -p ~/.hobba-mcp cp "/Doku/Werkzeuge/hobba-ha-bridge.py" ~/.hobba-mcp/ printf '{"url": "http://homeassistant.local:8123", "token": "", "liste": "todo.hobba_eingang", "standardtag": "heute"}' > ~/.hobba-mcp/ha-bridge.conf chmod 600 ~/.hobba-mcp/ha-bridge.conf Token: HA → Profil → Sicherheit → Langlebige Zugriffs-Token. Der launchd-Job (de.rinnhome.hobba.ha-bridge) ruft das Skript alle 5 Minuten. """ import datetime import json import os import subprocess import sys import urllib.request import uuid CONF = os.path.expanduser("~/.hobba-mcp/ha-bridge.conf") MAP = os.path.expanduser("~/.hobba-mcp/ha-bridge-map.json") ERLEDIGT = "erledigt:" SHORTCUT = "Hobba-Connector" TMP = "/tmp/hobba-connector" TIMEOUT = 90 def lade_conf(): try: with open(CONF, encoding="utf-8") as handle: conf = json.load(handle) except FileNotFoundError: sys.exit(f"Keine Konfiguration: {CONF} fehlt (siehe Kopfkommentar).") if os.name == "posix" and (os.stat(CONF).st_mode & 0o077): sys.exit(f"{CONF} ist zu offen — bitte: chmod 600 {CONF}") for key in ("url", "token"): if not conf.get(key): sys.exit(f"{CONF}: Feld „{key}“ fehlt.") conf.setdefault("liste", "todo.hobba_eingang") return conf def lade_map(): try: with open(MAP, encoding="utf-8") as handle: return json.load(handle) except (FileNotFoundError, json.JSONDecodeError): return {} def sichere_map(zuordnung): with open(MAP, "w", encoding="utf-8") as handle: json.dump(zuordnung, handle, ensure_ascii=False, indent=1) def ha(conf, pfad, payload=None): """Ein Aufruf der HA-REST-API. Nur die zwei todo-Dienste — mehr darf dieses Skript gar nicht können.""" daten = json.dumps(payload).encode() if payload is not None else None anfrage = urllib.request.Request( conf["url"].rstrip("/") + pfad, data=daten, method="POST", headers={"Authorization": "Bearer " + conf["token"], "Content-Type": "application/json"}) with urllib.request.urlopen(anfrage, timeout=15) as antwort: return json.load(antwort) def hole_eintraege(conf): antwort = ha(conf, "/api/services/todo/get_items?return_response", {"entity_id": conf["liste"], "status": "needs_action"}) listen = antwort.get("service_response", {}) return listen.get(conf["liste"], {}).get("items", []) def tag_fuer(eintrag, conf): """Welcher Tag an die Aufgabe — HA zuerst, dann die Vorgabe. HA liefert `due` je nach Eintrag als „2026-10-01" oder „2026-10-01T09:00:00"; Hobba will „JJJJ-MM-TT", also die ersten zehn Zeichen. Alles, was nicht danach aussieht, wird verworfen statt geraten — ein falscher Tag ist schlimmer als keiner, weil er die Aufgabe an einem Tag zeigt, an dem sie niemand erwartet. **Wirklich geprüft, nicht nur die Form.** „2026-13-99" hat zehn Zeichen und Bindestriche an den richtigen Stellen, ist aber kein Datum; Hobba wiese es zurück, das Anlegen schlüge fehl, und weil der HA-Eintrag erst **nach** der Bestätigung entfernt wird, versuchte es jeder Lauf aufs Neue. Eine unbrauchbare Fälligkeit darf höchstens ignoriert werden. """ faellig = (eintrag.get("due") or "").strip() if len(faellig) >= 10: kopf = faellig[:10] try: datetime.date.fromisoformat(kopf) return kopf except ValueError: pass # `.get` mit Vorgabe, nicht `or`: Eine ausdrückliche leere Zeichenkette # heißt „kein Tag" und darf nicht auf „heute" zurückfallen. return (conf.get("standardtag", "heute") or "").strip() def entferne(conf, uid): ha(conf, "/api/services/todo/remove_item", {"entity_id": conf["liste"], "item": uid}) def aufgabe_offen(schnappschuss, task_id): """Steht die Aufgabe irgendwo im Bestand und ist nicht abgehakt? Der Schnappschuss zeigt Listen UND Projekte — auch eine aus dem Eingang wegsortierte Aufgabe bleibt damit als offen erkennbar.""" quellen = list((schnappschuss.get("listen") or {}).values()) quellen += [p.get("aufgaben") or [] for p in schnappschuss.get("projekte") or []] for aufgaben in quellen: for aufgabe in aufgaben: if aufgabe.get("id") == task_id: return not aufgabe.get("erledigt") return False def an_hobba(befehl): """Wie hobba-mcp.py: JSON-Datei, `shortcuts run`, Antwort auf stdout.""" os.makedirs(TMP, exist_ok=True) name = uuid.uuid4().hex + ".json" pfad = os.path.join(TMP, name) try: with open(pfad, "w", encoding="utf-8") as handle: json.dump(befehl, handle, ensure_ascii=False) lauf = subprocess.run(["shortcuts", "run", SHORTCUT, "-i", name], capture_output=True, text=True, timeout=TIMEOUT, cwd=TMP) if lauf.returncode != 0: raise RuntimeError(lauf.stderr.strip() or "shortcuts run scheiterte") antwort = json.loads(lauf.stdout.strip() or "{}") if not antwort.get("ok"): raise RuntimeError(json.dumps(antwort, ensure_ascii=False)) return antwort finally: try: os.remove(pfad) except OSError: pass def main(): conf = lade_conf() eintraege = hole_eintraege(conf) if not eintraege: return zuordnung = lade_map() schnappschuss = None for eintrag in eintraege: titel = (eintrag.get("summary") or "").strip() if not titel: entferne(conf, eintrag["uid"]) continue if titel.lower().startswith(ERLEDIGT): # Entwarnung: gemerkte Aufgabe abhaken. Fehler (schon manuell # abgehakt, gelöscht, Karte offen) sind hier folgenlos — die # Kennung fliegt trotzdem raus, sonst hakt jeder Lauf erneut. ziel = titel[len(ERLEDIGT):].strip() task_id = zuordnung.pop(ziel, None) if task_id: try: an_hobba({"befehl": "aufgabe_erledigen", "id": task_id}) print(f"✓ abgehakt: {ziel}") except (RuntimeError, subprocess.TimeoutExpired) as fehler: print(f"Entwarnung ohne Wirkung ({ziel}): {fehler}") sichere_map(zuordnung) entferne(conf, eintrag["uid"]) continue # Titel-Dedup: Ist die gemerkte Aufgabe noch offen, wäre eine # zweite nur Lärm (HA darf periodisch nachprüfen und denselben # Titel erneut schicken). Der Schnappschuss wird höchstens einmal # je Lauf geholt — und nur, wenn ein bekannter Titel eintrifft. if titel in zuordnung: if schnappschuss is None: schnappschuss = an_hobba({"befehl": "snapshot"}) if aufgabe_offen(schnappschuss, zuordnung[titel]): entferne(conf, eintrag["uid"]) print(f"= schon offen, übersprungen: {titel}") continue # Abgehakt oder verschwunden: Merkzettel-Leiche entsorgen, # das neue Vorkommnis ist echt. del zuordnung[titel] sichere_map(zuordnung) befehl = {"befehl": "aufgabe_anlegen", "text": titel} notiz = (eintrag.get("description") or "").strip() if notiz: befehl["notiz"] = notiz tag = tag_fuer(eintrag, conf) if tag: befehl["tag"] = tag antwort = an_hobba(befehl) # wirft bei Fehlern — Eintrag bleibt in HA task_id = (antwort.get("aufgabe") or {}).get("id") if task_id: zuordnung[titel] = task_id sichere_map(zuordnung) entferne(conf, eintrag["uid"]) print(f"→ Hobba: {titel}") if __name__ == "__main__": main()