150 lines
4.4 KiB
Python
150 lines
4.4 KiB
Python
# Wir schreiben einen einfachen Task-Parser in Python, der Markdown-Dateien nach Task-Listeneinträgen durchsucht.
|
||
# Dafür nutzen wir den 'markdown' Parser und reguläre Ausdrücke.
|
||
|
||
import os
|
||
import re
|
||
from pathlib import Path
|
||
|
||
from dataclasses import dataclass
|
||
from sys import stdout
|
||
from typing import List, Optional
|
||
from datetime import datetime
|
||
|
||
import caldav
|
||
from icalendar import Todo
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv()
|
||
|
||
|
||
@dataclass
|
||
class Task:
|
||
file: str
|
||
line_number: int
|
||
raw_text: str
|
||
text: str
|
||
completed: bool
|
||
due: Optional[datetime] = None
|
||
start: Optional[datetime] = None
|
||
scheduled: Optional[datetime] = None
|
||
recurrence: Optional[str] = None
|
||
priority: Optional[str] = None
|
||
|
||
|
||
def parse_task_line(line: str, file: str, line_number: int) -> Optional[Task]:
|
||
task_pattern = re.compile(r"- \[([ +/\-?!<])\] (.+)")
|
||
match = task_pattern.match(line)
|
||
if not match:
|
||
return None
|
||
|
||
completed = match.group(1) == "x" or match.group(1) == "-"
|
||
text = match.group(2)
|
||
# Suche nach Metadaten im Text
|
||
due = extract_date(text, r"📅 (\d{4}-\d{2}-\d{2})")
|
||
if due is not None:
|
||
due = due.replace(hour=23, minute=59, second=59)
|
||
start = extract_date(text, r"🛫 (\d{4}-\d{2}-\d{2})")
|
||
scheduled = extract_date(text, r"⏳ (\d{4}-\d{2}-\d{2})")
|
||
if scheduled is not None:
|
||
scheduled = scheduled.replace(hour=23, minute=59, second=59)
|
||
recurrence = extract_text(text, r"🔁 ([^\s]+)")
|
||
priority = extract_text(text, r"(‼|❗|➖|🔽)")
|
||
clean = extract_text(text, r"([^📅🛫⏳🔁‼❗➖🔽🆔➕⛔]*)")
|
||
|
||
if due is not None and scheduled is not None:
|
||
if due < scheduled:
|
||
due = scheduled
|
||
return Task(
|
||
file=file,
|
||
line_number=line_number,
|
||
raw_text=line.strip(),
|
||
text=clean or text.strip(),
|
||
completed=completed,
|
||
due=due,
|
||
start=start,
|
||
scheduled=scheduled,
|
||
recurrence=recurrence,
|
||
priority=priority,
|
||
)
|
||
|
||
|
||
def extract_date(text: str, pattern: str) -> Optional[datetime]:
|
||
match = re.search(pattern, text)
|
||
if match:
|
||
try:
|
||
return datetime.strptime(match.group(1), "%Y-%m-%d")
|
||
except ValueError:
|
||
return None
|
||
return None
|
||
|
||
|
||
def extract_text(text: str, pattern: str) -> Optional[str]:
|
||
match = re.search(pattern, text)
|
||
if match:
|
||
return match.group(1)
|
||
return None
|
||
|
||
|
||
def parse_tasks_from_vault(vault_path: str) -> List[Task]:
|
||
tasks = []
|
||
for md_file in Path(vault_path).rglob("*.md"):
|
||
with open(md_file, "r", encoding="utf-8") as f:
|
||
for i, line in enumerate(f):
|
||
task = parse_task_line(line, str(md_file), i + 1)
|
||
if task:
|
||
tasks.append(task)
|
||
return tasks
|
||
|
||
|
||
# Für die Demonstration verwenden wir ein Testverzeichnis
|
||
tasks = parse_tasks_from_vault(os.getenv("OBSIDIAN_VAULT_PATH", "."))
|
||
|
||
# Verbindung zur Nextcloud herstellen
|
||
client = caldav.DAVClient(
|
||
url=os.getenv("NEXTCLOUD_URL"),
|
||
username=os.getenv("NEXTCLOUD_USER"),
|
||
password=os.getenv("NEXTCLOUD_PASSWORD"),
|
||
)
|
||
|
||
# Hole das erste verfügbare Kalender-Objekt (für Aufgaben, nicht Termine!)
|
||
principal = client.principal()
|
||
task_calendars = [c for c in principal.calendars() if "nicole" == str(c.name).lower()]
|
||
|
||
print(f"Found task calendars: {task_calendars}")
|
||
|
||
calendar = task_calendars[0]
|
||
|
||
## delete all created todos
|
||
for todo in calendar.todos():
|
||
ical = todo.vobject_instance
|
||
todoitem = ical.contents["vtodo"][0].contents
|
||
|
||
if "x-obsidian-task-id" in todoitem:
|
||
print(f"deleting {todoitem['summary']}")
|
||
todo.delete()
|
||
|
||
# create new todos
|
||
|
||
for t in tasks:
|
||
if not t.completed and t.scheduled is not None:
|
||
print(f"adding on {t.scheduled.date().isoformat()}: {t.text}")
|
||
stdout.flush()
|
||
todo = Todo()
|
||
todo.add("summary", t.text)
|
||
todo.add("created", datetime.now())
|
||
todo.add("description", "Generated by ObsidianSyncScript")
|
||
todo.add("X-OBSIDIAN-TASK-ID", f"{t.file}:{t.line_number}")
|
||
if t.scheduled:
|
||
todo.add("due", t.scheduled)
|
||
if t.start:
|
||
todo.add("dtstart", t.start)
|
||
if t.due:
|
||
todo.add("comment", f"due {t.due}")
|
||
# if t.priority:
|
||
# todo.add('priority', convert_priority(t.priority))
|
||
|
||
try:
|
||
calendar.add_todo(todo.to_ical().decode("utf-8"))
|
||
except Exception as e:
|
||
print(e)
|