Initial GLPI daily report workflow

tools/inspect_n8n_execution.py

@@ -0,0 +1,81 @@
+import json
+import sqlite3
+import sys
+
+
+def decode_flatted(table):
+    memo = {}
+
+    def decode(value):
+        if isinstance(value, str) and value.isdigit():
+            return decode_index(int(value))
+        if isinstance(value, list):
+            return [decode(item) for item in value]
+        if isinstance(value, dict):
+            return {key: decode(item) for key, item in value.items()}
+        return value
+
+    def decode_index(index):
+        if index in memo:
+            return memo[index]
+
+        raw = table[index]
+
+        if isinstance(raw, dict):
+            obj = {}
+            memo[index] = obj
+            obj.update({key: decode(value) for key, value in raw.items()})
+            return obj
+
+        if isinstance(raw, list):
+            obj = []
+            memo[index] = obj
+            obj.extend(decode(value) for value in raw)
+            return obj
+
+        memo[index] = raw
+        return raw
+
+    return decode_index(0)
+
+
+def main():
+    if len(sys.argv) not in (3, 4):
+        raise SystemExit("usage: inspect_n8n_execution.py <database.sqlite> <execution_id> [node_name]")
+
+    db_path = sys.argv[1]
+    execution_id = sys.argv[2]
+
+    con = sqlite3.connect(db_path)
+    row = con.execute(
+        "select data from execution_data where executionId = ?",
+        (execution_id,),
+    ).fetchone()
+
+    if not row:
+        raise SystemExit(f"execution {execution_id} not found")
+
+    decoded = decode_flatted(json.loads(row[0]))
+    result_data = decoded.get("resultData", {})
+
+    if len(sys.argv) == 4:
+        node_name = sys.argv[3]
+        run_data = result_data.get("runData", {})
+        node_runs = run_data.get(node_name)
+
+        if node_runs is None:
+            print("available nodes:")
+            for name in run_data:
+                print("-", name)
+            raise SystemExit(f"node {node_name!r} not found")
+
+        print(json.dumps(node_runs, ensure_ascii=False, indent=2)[:12000])
+        return
+
+    print("lastNodeExecuted:", result_data.get("lastNodeExecuted"))
+    print("error:")
+    print(json.dumps(result_data.get("error"), ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()