Tag all Vikunja tasks with #POPS label; fix Vikunja URL

- On create_task(), look up the 'POPS' label (create it if missing) and
  attach it to every task. Label ID is cached for the process lifetime.
- Corrected VIKUNJA_URL typo (tasks → task)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Martin Hohenberg
2026-06-19 07:58:20 +02:00
parent a2e8c9e194
commit 5dfa700916
2 changed files with 35 additions and 2 deletions

View File

@@ -6,7 +6,7 @@ MYSQL_PASSWORD=
STL_UPLOAD_DIR=/data/stl STL_UPLOAD_DIR=/data/stl
TZ=Europe/Berlin TZ=Europe/Berlin
FILAMENTDB_URL=https://web.filamentdb.orb.local FILAMENTDB_URL=https://web.filamentdb.orb.local
VIKUNJA_URL=https://vikunja.example.com VIKUNJA_URL=https://task.example.com
VIKUNJA_API_TOKEN= VIKUNJA_API_TOKEN=
VIKUNJA_PROJECT_ID= VIKUNJA_PROJECT_ID=
VAT_RATE=19 VAT_RATE=19

View File

@@ -6,6 +6,8 @@ import httpx
logger = logging.getLogger("pops") logger = logging.getLogger("pops")
_pops_label_id: int | None = None
def _base() -> str: def _base() -> str:
return os.environ.get("VIKUNJA_URL", "").rstrip("/") return os.environ.get("VIKUNJA_URL", "").rstrip("/")
@@ -22,6 +24,33 @@ def _project_id() -> int:
return int(os.environ.get("VIKUNJA_PROJECT_ID", 0)) return int(os.environ.get("VIKUNJA_PROJECT_ID", 0))
def _get_or_create_pops_label() -> int | None:
"""Return the ID of the 'POPS' label, creating it if it doesn't exist."""
global _pops_label_id
if _pops_label_id is not None:
return _pops_label_id
try:
r = httpx.get(f"{_base()}/api/v1/labels", headers=_headers(), timeout=5)
r.raise_for_status()
for label in r.json() or []:
if label.get("title") == "POPS":
_pops_label_id = label["id"]
return _pops_label_id
# Not found — create it
r = httpx.put(
f"{_base()}/api/v1/labels",
headers=_headers(),
json={"title": "POPS"},
timeout=5,
)
r.raise_for_status()
_pops_label_id = r.json()["id"]
return _pops_label_id
except Exception:
logger.warning("Vikunja: failed to get or create POPS label")
return None
def get_tasks() -> list[dict]: def get_tasks() -> list[dict]:
try: try:
r = httpx.get( r = httpx.get(
@@ -38,11 +67,15 @@ def get_tasks() -> list[dict]:
def create_task(title: str) -> dict | None: def create_task(title: str) -> dict | None:
label_id = _get_or_create_pops_label()
payload: dict = {"title": title}
if label_id:
payload["labels"] = [{"id": label_id}]
try: try:
r = httpx.put( r = httpx.put(
f"{_base()}/api/v1/projects/{_project_id()}/tasks", f"{_base()}/api/v1/projects/{_project_id()}/tasks",
headers=_headers(), headers=_headers(),
json={"title": title}, json=payload,
timeout=5, timeout=5,
) )
r.raise_for_status() r.raise_for_status()