#!/usr/bin/env python3 """Sicher in S.W.A.R.M-Watchdog inbox.json pending[] appendet. Nutzung: python append_inbox.py --inbox --item '' python append_inbox.py --inbox --item '' --item '' python append_inbox.py --inbox --items-file Exit-Codes: 0 = OK, Anzahl neuer Items in stdout 1 = Datei-/JSON-Fehler 2 = Schema-Fehler (pending fehlt) """ from __future__ import annotations import argparse import datetime as dt import json import os import sys from typing import Any REQUIRED_PER_TYPE = { "frist": ["projekt", "aufgabe"], "rundgang": ["projekt", "befund"], "memory": ["key", "value"], "status": ["rowid", "status"], } VALID_STATUS = {"offen", "inarbeit", "erledigt", "blockiert"} VALID_BUCKETS = {"krit", "acht", "lauf"} VALID_QUELLE_TYP = {"mail", "whatsapp", "foto", "termin", "telefon", "link", "claude-mobil"} VALID_FOTO_EXT = {".jpg", ".jpeg", ".png", ".heic", ".webp"} def now_iso() -> str: return dt.datetime.now().replace(microsecond=0).isoformat() def _validate_foto_path(path: str) -> list[str]: errs: list[str] = [] p = str(path) ext = os.path.splitext(p)[1].lower() if ext and ext not in VALID_FOTO_EXT: errs.append(f"foto-Endung '{ext}' nicht erlaubt, erwartet {sorted(VALID_FOTO_EXT)}") if not p.startswith("fotos/"): errs.append(f"foto-Pfad '{p}' muss relativ mit 'fotos/' beginnen") return errs def validate_item(item: dict[str, Any]) -> list[str]: errors: list[str] = [] typ = item.get("typ", "") if typ not in REQUIRED_PER_TYPE: errors.append(f"Unbekannter typ '{typ}', erlaubt: {sorted(REQUIRED_PER_TYPE)}") return errors for key in REQUIRED_PER_TYPE[typ]: if not item.get(key): errors.append(f"Feld '{key}' fehlt fuer typ '{typ}'") if typ == "status" and item.get("status") not in VALID_STATUS: errors.append(f"status muss einer von {sorted(VALID_STATUS)} sein") if typ == "frist": bucket = item.get("bucket", "acht") if bucket not in VALID_BUCKETS: errors.append(f"bucket muss einer von {sorted(VALID_BUCKETS)} sein") if typ == "rundgang": rg_status = item.get("status") if rg_status and rg_status not in VALID_STATUS: errors.append(f"rundgang.status '{rg_status}' ungueltig, erwartet {sorted(VALID_STATUS)}") quelle = item.get("quelle") if quelle and quelle.get("typ") and quelle["typ"] not in VALID_QUELLE_TYP: errors.append(f"quelle.typ muss einer von {sorted(VALID_QUELLE_TYP)} sein") foto = item.get("foto") if foto: errors.extend(_validate_foto_path(foto)) fotos = item.get("fotos") if fotos is not None: if not isinstance(fotos, list): errors.append("'fotos' muss eine Liste von Pfaden sein (['fotos/.jpg', ...])") else: for i, fp in enumerate(fotos): if not isinstance(fp, str): errors.append(f"fotos[{i}] ist kein String") continue for e in _validate_foto_path(fp): errors.append(f"fotos[{i}]: {e}") return errors def normalize_item(item: dict[str, Any]) -> dict[str, Any]: """Setzt Defaults, ohne User-Werte zu ueberschreiben.""" out = dict(item) if not out.get("ts"): out["ts"] = now_iso() if not out.get("id"): ts_clean = out["ts"].replace(":", "").replace("-", "").replace("T", "-") typ = out.get("typ", "item") out["id"] = f"inbox-{typ}-{ts_clean}" if out.get("typ") == "frist": out.setdefault("bucket", "acht") out.setdefault("verantw", "Gueven Kaplangil") out.setdefault("quelleStatus", "Meldung Claude-Session") out.setdefault("quelle", {"typ": "claude-mobil", "datum": out["ts"][:10], "zeit": out["ts"][11:16], "ref": "Claude-Session"}) if out.get("typ") == "rundgang": out.setdefault("datum", out["ts"][:10]) out.setdefault("status", "offen") # Wenn 'foto' singular gesetzt ist aber keine 'fotos'-Liste existiert: in Liste ueberfuehren if out.get("foto") and not out.get("fotos"): out["fotos"] = [out["foto"]] return out def append_items(inbox_path: str, items: list[dict[str, Any]], replace_ids: list[str] | None = None) -> dict[str, Any]: if not os.path.exists(inbox_path): print(f"FEHLER: inbox.json nicht gefunden: {inbox_path}", file=sys.stderr) sys.exit(1) try: with open(inbox_path, "r", encoding="utf-8") as f: data = json.load(f) except json.JSONDecodeError as e: print(f"FEHLER: inbox.json kein valides JSON: {e}", file=sys.stderr) sys.exit(1) if "pending" not in data or not isinstance(data["pending"], list): print("FEHLER: 'pending'-Array fehlt oder ist kein Array", file=sys.stderr) sys.exit(2) replace_ids = set(replace_ids or []) removed: list[str] = [] if replace_ids: kept = [] for it in data["pending"]: iid = it.get("id") if isinstance(it, dict) else None if iid in replace_ids: removed.append(iid) else: kept.append(it) data["pending"] = kept existing_ids = {it.get("id") for it in data["pending"] if isinstance(it, dict)} appended = [] skipped_dupes = [] errors_all = [] for raw in items: item = normalize_item(raw) errs = validate_item(item) if errs: errors_all.append({"item_id": item.get("id"), "errors": errs}) continue if item["id"] in existing_ids: skipped_dupes.append(item["id"]) continue data["pending"].append(item) existing_ids.add(item["id"]) appended.append(item["id"]) if appended or removed: # Atomarer Write ueber temp-Datei tmp_path = inbox_path + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) f.write("\n") os.replace(tmp_path, inbox_path) return { "appended": appended, "removed": removed, "skipped_duplicates": skipped_dupes, "errors": errors_all, "pending_count_now": len(data["pending"]), } def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--inbox", required=True, help="Pfad zu inbox.json") ap.add_argument("--item", action="append", default=[], help="JSON-String eines Items (mehrfach)") ap.add_argument("--items-file", help="Pfad zu JSON-Array mit mehreren Items") ap.add_argument("--replace", action="append", default=[], help="ID eines bestehenden Items, das ersetzt werden soll (mehrfach)") args = ap.parse_args() items: list[dict[str, Any]] = [] for raw in args.item: try: items.append(json.loads(raw)) except json.JSONDecodeError as e: print(f"FEHLER: --item ist kein valides JSON: {e}\nWert: {raw[:200]}", file=sys.stderr) return 1 if args.items_file: with open(args.items_file, "r", encoding="utf-8") as f: loaded = json.load(f) if not isinstance(loaded, list): print("FEHLER: --items-file muss ein JSON-Array sein", file=sys.stderr) return 1 items.extend(loaded) if not items and not args.replace: print("FEHLER: mindestens ein --item / --items-file oder --replace erforderlich", file=sys.stderr) return 1 result = append_items(args.inbox, items, args.replace) print(json.dumps(result, indent=2, ensure_ascii=False)) if result["errors"]: return 2 if result["appended"] or result["removed"]: return 0 return 2 if __name__ == "__main__": sys.exit(main())