179 lines
6.2 KiB
Python
179 lines
6.2 KiB
Python
|
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
openclaw_skill.py — OpenClaw Telegram-Bot-Skill: mailbox.org Kalender + Aufgaben
|
||
|
|
Registrierung in OpenClaw: /register skill mailbox-cal
|
||
|
|
|
||
|
|
Unterstützte Befehle (via Telegram-Chat an den Bot):
|
||
|
|
/termine [heute|morgen|woche] — Kalender-Ereignisse anzeigen
|
||
|
|
/aufgaben [offen|alle] — Aufgabenliste anzeigen
|
||
|
|
/neu_aufgabe <Text> — Neue Aufgabe anlegen
|
||
|
|
/sync — manuellen vdirsyncer-Sync auslösen
|
||
|
|
/status — Sync-Status abfragen
|
||
|
|
|
||
|
|
Voraussetzungen:
|
||
|
|
- khal, todoman, vdirsyncer installiert (install.sh ausgeführt)
|
||
|
|
- OpenClaw mit Python-Skill-API >= 0.4
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import subprocess
|
||
|
|
import shlex
|
||
|
|
import logging
|
||
|
|
from datetime import datetime, date, timedelta
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
# ─────────────────────────────────────────────
|
||
|
|
# Hilfsfunktionen
|
||
|
|
# ─────────────────────────────────────────────
|
||
|
|
|
||
|
|
def _run(cmd: list[str], timeout: int = 30) -> tuple[str, str, int]:
|
||
|
|
"""Führt einen Subprozess aus und gibt (stdout, stderr, returncode) zurück."""
|
||
|
|
try:
|
||
|
|
result = subprocess.run(
|
||
|
|
cmd,
|
||
|
|
capture_output=True,
|
||
|
|
text=True,
|
||
|
|
timeout=timeout,
|
||
|
|
)
|
||
|
|
return result.stdout.strip(), result.stderr.strip(), result.returncode
|
||
|
|
except subprocess.TimeoutExpired:
|
||
|
|
return "", "Timeout nach {}s".format(timeout), 1
|
||
|
|
except FileNotFoundError as e:
|
||
|
|
return "", str(e), 127
|
||
|
|
|
||
|
|
|
||
|
|
def _khal_agenda(days: int = 1, start: Optional[date] = None) -> str:
|
||
|
|
"""Gibt khal-Agenda als String zurück."""
|
||
|
|
start_str = (start or date.today()).strftime("%Y-%m-%d")
|
||
|
|
end_str = (date.today() + timedelta(days=days)).strftime("%Y-%m-%d")
|
||
|
|
stdout, stderr, rc = _run(
|
||
|
|
["khal", "list", "--format", "{start-time} {title}", start_str, end_str]
|
||
|
|
)
|
||
|
|
if rc != 0:
|
||
|
|
logger.warning("khal list Fehler: %s", stderr)
|
||
|
|
return "⚠️ khal Fehler: " + stderr
|
||
|
|
return stdout or "Keine Termine gefunden."
|
||
|
|
|
||
|
|
|
||
|
|
def _todo_list(all_tasks: bool = False) -> str:
|
||
|
|
"""Gibt todoman-Aufgabenliste als String zurück."""
|
||
|
|
cmd = ["todo", "list"]
|
||
|
|
if all_tasks:
|
||
|
|
cmd.append("--all")
|
||
|
|
stdout, stderr, rc = _run(cmd)
|
||
|
|
if rc != 0:
|
||
|
|
return "⚠️ todoman Fehler: " + stderr
|
||
|
|
return stdout or "Keine Aufgaben."
|
||
|
|
|
||
|
|
|
||
|
|
def _todo_new(title: str) -> str:
|
||
|
|
"""Legt eine neue Aufgabe an."""
|
||
|
|
if not title.strip():
|
||
|
|
return "⚠️ Kein Aufgabentitel angegeben."
|
||
|
|
stdout, stderr, rc = _run(["todo", "new", "--", title])
|
||
|
|
if rc != 0:
|
||
|
|
return "⚠️ Fehler beim Anlegen: " + stderr
|
||
|
|
return "✅ Aufgabe angelegt: " + title
|
||
|
|
|
||
|
|
|
||
|
|
def _vdirsyncer_sync() -> str:
|
||
|
|
"""Löst manuellen vdirsyncer-Sync aus."""
|
||
|
|
stdout, stderr, rc = _run(["vdirsyncer", "sync"], timeout=60)
|
||
|
|
if rc != 0:
|
||
|
|
return "⚠️ Sync-Fehler:\n" + stderr
|
||
|
|
return "✅ Sync erfolgreich.\n" + stdout
|
||
|
|
|
||
|
|
|
||
|
|
def _sync_status() -> str:
|
||
|
|
"""Prüft systemd-Timer-Status."""
|
||
|
|
stdout, _, rc = _run(
|
||
|
|
["systemctl", "--user", "status", "vdirsyncer-sync.timer", "--no-pager", "-l"]
|
||
|
|
)
|
||
|
|
if rc not in (0, 3):
|
||
|
|
return "⚠️ Timer nicht gefunden oder Fehler."
|
||
|
|
# Nur relevante Zeilen extrahieren
|
||
|
|
lines = [
|
||
|
|
line for line in stdout.splitlines()
|
||
|
|
if any(kw in line for kw in ("Active:", "Trigger:", "Last trigger:"))
|
||
|
|
]
|
||
|
|
return "\n".join(lines) or stdout[:500]
|
||
|
|
|
||
|
|
|
||
|
|
# ─────────────────────────────────────────────
|
||
|
|
# OpenClaw Skill-Handler
|
||
|
|
# ─────────────────────────────────────────────
|
||
|
|
|
||
|
|
def handle(command: str, args: str, context: dict) -> str:
|
||
|
|
"""
|
||
|
|
Haupt-Einstiegspunkt für OpenClaw.
|
||
|
|
|
||
|
|
:param command: Befehlsname ohne führenden Slash, z. B. "termine"
|
||
|
|
:param args: Argumente als String, z. B. "woche"
|
||
|
|
:param context: OpenClaw-Kontext (user_id, chat_id, etc.)
|
||
|
|
:return: Antwort-String für Telegram
|
||
|
|
"""
|
||
|
|
cmd = command.lower().strip()
|
||
|
|
arg = args.lower().strip()
|
||
|
|
|
||
|
|
if cmd == "termine":
|
||
|
|
if arg in ("morgen",):
|
||
|
|
start = date.today() + timedelta(days=1)
|
||
|
|
result = _khal_agenda(days=1, start=start)
|
||
|
|
return "📅 *Termine morgen:*\n" + result
|
||
|
|
elif arg in ("woche", "7"):
|
||
|
|
result = _khal_agenda(days=7)
|
||
|
|
return "📅 *Termine diese Woche:*\n" + result
|
||
|
|
else:
|
||
|
|
result = _khal_agenda(days=1)
|
||
|
|
return "📅 *Termine heute:*\n" + result
|
||
|
|
|
||
|
|
elif cmd == "aufgaben":
|
||
|
|
all_tasks = arg in ("alle", "all", "fertig")
|
||
|
|
result = _todo_list(all_tasks=all_tasks)
|
||
|
|
label = "alle Aufgaben" if all_tasks else "offene Aufgaben"
|
||
|
|
return "📝 *{}:*\n{}".format(label.capitalize(), result)
|
||
|
|
|
||
|
|
elif cmd == "neu_aufgabe":
|
||
|
|
return _todo_new(args.strip())
|
||
|
|
|
||
|
|
elif cmd == "sync":
|
||
|
|
return "🔄 Starte Sync...\n" + _vdirsyncer_sync()
|
||
|
|
|
||
|
|
elif cmd == "status":
|
||
|
|
return "⏱ *Sync-Status:*\n" + _sync_status()
|
||
|
|
|
||
|
|
else:
|
||
|
|
return (
|
||
|
|
"❓ Unbekannter Befehl. Verfügbare Befehle:\n"
|
||
|
|
" /termine [heute|morgen|woche]\n"
|
||
|
|
" /aufgaben [offen|alle]\n"
|
||
|
|
" /neu_aufgabe <Text>\n"
|
||
|
|
" /sync\n"
|
||
|
|
" /status"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# ─────────────────────────────────────────────
|
||
|
|
# Standalone-Test (ohne OpenClaw)
|
||
|
|
# ─────────────────────────────────────────────
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
import sys
|
||
|
|
|
||
|
|
logging.basicConfig(level=logging.DEBUG)
|
||
|
|
tests = [
|
||
|
|
("termine", "heute"),
|
||
|
|
("termine", "woche"),
|
||
|
|
("aufgaben", "offen"),
|
||
|
|
("sync", ""),
|
||
|
|
("status", ""),
|
||
|
|
]
|
||
|
|
for cmd, arg in tests:
|
||
|
|
print("\n" + "=" * 50)
|
||
|
|
print(f"CMD: {cmd!r} ARG: {arg!r}")
|
||
|
|
print(handle(cmd, arg, {}))
|