Initial commit
This commit is contained in:
@@ -0,0 +1,14 @@
|
||||
This is an application to import tasks to Tududi from Todoist.
|
||||
|
||||
To use it, you need to set up a .env file with your Todoist API token and Tududi API key. Here is an example of what the .env file should look like:
|
||||
```
|
||||
TODOIST_TOKEN=your_todoist_api_token
|
||||
TUDUDI_BASE_URL=https://your_tududi_instance_url
|
||||
TUDUDI_API_KEY=your_tududi_api_key
|
||||
DRY_RUN=false
|
||||
CLEAN_FIRST=True
|
||||
USE_AREAS=False
|
||||
```
|
||||
|
||||
Make sure to replace `your_todoist_api_token`, `https://your_tududi_instance_url`, and `your_tududi_api_key` with your actual credentials.
|
||||
|
||||
|
||||
@@ -0,0 +1,894 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Todoist -> Tududi full migration (projects, labels->tags, tasks, subtasks, comments->description, completed tasks).
|
||||
|
||||
Key fixes in this version:
|
||||
- Uses Tududi `uid` (not `id`) wherever applicable (projects/tasks/tags/areas).
|
||||
- Uses `projectUid` when creating tasks (so tasks land in projects).
|
||||
- Slugs Todoist labels to ASCII-safe Tududi tag names (avoids emoji/unicode issues => "undefined").
|
||||
- CLEAN_FIRST deletes tasks/projects/tags using robust list parsing + uid/id fallback + multiple delete endpoints.
|
||||
- USE_AREAS is optional; default false. If enabled, uses POST /api/v1/areas (NOT /api/v1/area).
|
||||
|
||||
Env:
|
||||
TODOIST_TOKEN=...
|
||||
TUDUDI_BASE_URL=https://tududi.example.org
|
||||
TUDUDI_API_KEY=...
|
||||
DRY_RUN=false
|
||||
CLEAN_FIRST=false
|
||||
USE_AREAS=false
|
||||
REQUEST_SLEEP=0.25
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import json
|
||||
import random
|
||||
import re
|
||||
import unicodedata
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, Iterable, List, Optional, Tuple
|
||||
|
||||
import requests
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
# ----------------------------
|
||||
# Helpers
|
||||
# ----------------------------
|
||||
def slug_tag(name: str) -> str:
|
||||
"""
|
||||
Convert label/tag to Tududi-safe ASCII slug:
|
||||
- removes emoji/non-ascii
|
||||
- removes accents
|
||||
- keeps a-z0-9_- and collapses spaces to _
|
||||
"""
|
||||
s = unicodedata.normalize("NFKD", name)
|
||||
s = "".join(ch for ch in s if not unicodedata.combining(ch))
|
||||
s = s.encode("ascii", "ignore").decode("ascii")
|
||||
s = s.strip().lower()
|
||||
s = re.sub(r"\s+", "_", s)
|
||||
s = re.sub(r"[^a-z0-9_-]", "", s)
|
||||
return s or "tag"
|
||||
|
||||
def normalize_list(resp: Any) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Tududi may respond as:
|
||||
- [ {...}, {...} ]
|
||||
- { "projects":[...] } / { "tags":[...] } / { "tasks":[...] } / { "areas":[...] }
|
||||
- { "data":[...] } / { "items":[...] } / { "results":[...] }
|
||||
Always return a list of dicts.
|
||||
"""
|
||||
if resp is None:
|
||||
return []
|
||||
if isinstance(resp, list):
|
||||
return [x for x in resp if isinstance(x, dict)]
|
||||
if isinstance(resp, dict):
|
||||
for key in ("projects", "areas", "tags", "tasks", "data", "results", "items"):
|
||||
v = resp.get(key)
|
||||
if isinstance(v, list):
|
||||
return [x for x in v if isinstance(x, dict)]
|
||||
if "id" in resp or "uid" in resp or "name" in resp:
|
||||
return [resp]
|
||||
return []
|
||||
|
||||
|
||||
def die(msg: str, code: int = 1) -> None:
|
||||
print(f"[FATAL] {msg}", file=sys.stderr)
|
||||
raise SystemExit(code)
|
||||
|
||||
|
||||
def getenv_bool(name: str, default: bool = False) -> bool:
|
||||
v = os.getenv(name)
|
||||
if v is None:
|
||||
return default
|
||||
return v.strip().lower() in ("1", "true", "yes", "y", "on")
|
||||
|
||||
|
||||
def getenv_float(name: str, default: float) -> float:
|
||||
v = os.getenv(name)
|
||||
if v is None:
|
||||
return default
|
||||
try:
|
||||
return float(v.strip())
|
||||
except ValueError:
|
||||
return default
|
||||
|
||||
|
||||
def iso_now() -> str:
|
||||
return datetime.utcnow().isoformat(timespec="seconds") + "Z"
|
||||
|
||||
|
||||
def chunks(lst: List[Any], n: int) -> Iterable[List[Any]]:
|
||||
for i in range(0, len(lst), n):
|
||||
yield lst[i : i + n]
|
||||
|
||||
|
||||
def _sanitize_tags(tags: Optional[List[str]]) -> Optional[List[str]]:
|
||||
if not tags:
|
||||
return None
|
||||
clean: List[str] = []
|
||||
for t in tags:
|
||||
if t is None:
|
||||
continue
|
||||
s = str(t).strip()
|
||||
if not s or s.lower() in ("undefined", "null", "none"):
|
||||
continue
|
||||
clean.append(s)
|
||||
# dedup preserving order
|
||||
seen = set()
|
||||
out = []
|
||||
for s in clean:
|
||||
if s not in seen:
|
||||
seen.add(s)
|
||||
out.append(s)
|
||||
return out or None
|
||||
|
||||
|
||||
def _valid_name(x: Any) -> Optional[str]:
|
||||
if x is None:
|
||||
return None
|
||||
s = str(x).strip()
|
||||
if not s or s.lower() in ("undefined", "null", "none"):
|
||||
return None
|
||||
return s
|
||||
|
||||
|
||||
def pick_uid(obj: Dict[str, Any]) -> Optional[str]:
|
||||
"""
|
||||
Tududi often uses uid; fallback to id.
|
||||
"""
|
||||
v = obj.get("uid") or obj.get("id")
|
||||
if v is None:
|
||||
return None
|
||||
s = str(v).strip()
|
||||
return s or None
|
||||
|
||||
|
||||
# ----------------------------
|
||||
# Todoist client (API v1)
|
||||
# ----------------------------
|
||||
@dataclass
|
||||
class TodoistData:
|
||||
projects: List[Dict[str, Any]]
|
||||
labels: List[Dict[str, Any]]
|
||||
sections: List[Dict[str, Any]]
|
||||
items: List[Dict[str, Any]] # active tasks in sync payload ("items")
|
||||
notes: List[Dict[str, Any]] # comments in sync payload ("notes")
|
||||
completed_tasks: List[Dict[str, Any]] # completed tasks payload (best-effort)
|
||||
|
||||
|
||||
class TodoistClient:
|
||||
BASE = "https://api.todoist.com"
|
||||
|
||||
def __init__(self, token: str, timeout: int = 60) -> None:
|
||||
self.s = requests.Session()
|
||||
self.s.headers.update({"Authorization": f"Bearer {token}"})
|
||||
self.timeout = timeout
|
||||
|
||||
def sync_full(self) -> Dict[str, Any]:
|
||||
url = f"{self.BASE}/api/v1/sync"
|
||||
data = {"sync_token": "*", "resource_types": json.dumps(["all"])}
|
||||
r = self.s.post(url, data=data, timeout=self.timeout)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
def get_completed_tasks_by_completion_date(self, since: Optional[str] = None) -> List[Dict[str, Any]]:
|
||||
candidates = [
|
||||
(f"{self.BASE}/api/v1/tasks/completed/by_completion_date", {"since": since} if since else {}),
|
||||
(f"{self.BASE}/api/v1/tasks/completed", {"since": since} if since else {}),
|
||||
]
|
||||
for url, params in candidates:
|
||||
try:
|
||||
out = self._get_paginated(url, params=params)
|
||||
if out:
|
||||
return out
|
||||
except requests.HTTPError:
|
||||
continue
|
||||
return []
|
||||
|
||||
def _get_paginated(self, url: str, params: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
|
||||
params = dict(params or {})
|
||||
all_items: List[Dict[str, Any]] = []
|
||||
cursor: Optional[str] = None
|
||||
while True:
|
||||
p = dict(params)
|
||||
if cursor:
|
||||
p["cursor"] = cursor
|
||||
r = self.s.get(url, params=p, timeout=self.timeout)
|
||||
r.raise_for_status()
|
||||
j = r.json()
|
||||
|
||||
batch = j.get("results") or j.get("items") or j.get("data") or []
|
||||
if isinstance(batch, list):
|
||||
all_items.extend(batch)
|
||||
|
||||
cursor = j.get("next_cursor") or j.get("nextCursor")
|
||||
if not cursor:
|
||||
break
|
||||
return all_items
|
||||
|
||||
def export_everything(self) -> TodoistData:
|
||||
sync = self.sync_full()
|
||||
projects = sync.get("projects", [])
|
||||
labels = sync.get("labels", [])
|
||||
sections = sync.get("sections", [])
|
||||
items = sync.get("items", [])
|
||||
notes = sync.get("notes", [])
|
||||
completed = self.get_completed_tasks_by_completion_date()
|
||||
return TodoistData(
|
||||
projects=projects,
|
||||
labels=labels,
|
||||
sections=sections,
|
||||
items=items,
|
||||
notes=notes,
|
||||
completed_tasks=completed,
|
||||
)
|
||||
|
||||
|
||||
# ----------------------------
|
||||
# Tududi client
|
||||
# ----------------------------
|
||||
class TududiClient:
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str,
|
||||
api_key: Optional[str] = None,
|
||||
email: Optional[str] = None,
|
||||
password: Optional[str] = None,
|
||||
timeout: int = 60,
|
||||
dry_run: bool = False,
|
||||
request_sleep: float = 0.25,
|
||||
) -> None:
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.timeout = timeout
|
||||
self.dry_run = dry_run
|
||||
self.request_sleep = request_sleep
|
||||
self.s = requests.Session()
|
||||
|
||||
if api_key:
|
||||
self.s.headers.update({"Authorization": f"Bearer {api_key}"})
|
||||
elif email and password:
|
||||
self._login_cookie(email, password)
|
||||
else:
|
||||
die("Need TUDUDI_API_KEY (recommended) OR TUDUDI_EMAIL + TUDUDI_PASSWORD.")
|
||||
|
||||
self.s.headers.setdefault("Content-Type", "application/json")
|
||||
self.s.headers.setdefault("Accept", "application/json")
|
||||
|
||||
def _login_cookie(self, email: str, password: str) -> None:
|
||||
payload = {"email": email, "password": password}
|
||||
for path in ("/api/login", "/api/v1/login"):
|
||||
url = self._url(path)
|
||||
r = self.s.post(url, data=json.dumps(payload), timeout=self.timeout)
|
||||
if r.status_code < 400:
|
||||
return
|
||||
die("Tududi cookie login failed. Use an API key.")
|
||||
|
||||
def _url(self, path: str) -> str:
|
||||
if not path.startswith("/"):
|
||||
path = "/" + path
|
||||
return self.base_url + path
|
||||
|
||||
def _request_with_retry(
|
||||
self,
|
||||
method: str,
|
||||
path: str,
|
||||
payload: Optional[dict] = None,
|
||||
params: Optional[dict] = None,
|
||||
):
|
||||
url = self._url(path)
|
||||
max_attempts = 8
|
||||
base_sleep = 1.0
|
||||
|
||||
for attempt in range(1, max_attempts + 1):
|
||||
try:
|
||||
if method == "GET":
|
||||
r = self.s.get(url, params=params, timeout=self.timeout)
|
||||
elif method == "DELETE":
|
||||
if self.dry_run:
|
||||
print(f"[DRY_RUN] DELETE {path}")
|
||||
return {"_dry_run": True}
|
||||
r = self.s.delete(url, timeout=self.timeout)
|
||||
else:
|
||||
if self.dry_run:
|
||||
print(f"[DRY_RUN] POST {path} payload={payload}")
|
||||
return {"_dry_run": True}
|
||||
if payload is not None and isinstance(payload, dict):
|
||||
print(f"[DEBUG] POST {path} payload.Tags={payload.get('Tags')}")
|
||||
r = self.s.post(url, data=json.dumps(payload or {}), timeout=self.timeout)
|
||||
|
||||
# 429 -> retry
|
||||
if r.status_code == 429:
|
||||
retry_after = r.headers.get("Retry-After")
|
||||
if retry_after and retry_after.isdigit():
|
||||
sleep_s = float(retry_after)
|
||||
else:
|
||||
sleep_s = base_sleep * (2 ** (attempt - 1)) + random.uniform(0, 0.5)
|
||||
print(f"[WARN] 429 on {path}. Sleeping {sleep_s:.2f}s (attempt {attempt}/{max_attempts})")
|
||||
time.sleep(sleep_s)
|
||||
continue
|
||||
|
||||
# 4xx (except 429) -> fail fast
|
||||
if 400 <= r.status_code < 500:
|
||||
print(f"[FATAL] {r.status_code} on {path}: {r.text[:2000]}")
|
||||
if payload is not None:
|
||||
print("[FATAL] sent_payload=", json.dumps(payload or {}, ensure_ascii=False)[:2000])
|
||||
r.raise_for_status()
|
||||
|
||||
# 5xx -> retry
|
||||
if 500 <= r.status_code < 600:
|
||||
sleep_s = base_sleep * (2 ** (attempt - 1)) + random.uniform(0, 0.5)
|
||||
print(f"[WARN] {r.status_code} on {path}. Sleeping {sleep_s:.2f}s (attempt {attempt}/{max_attempts})")
|
||||
time.sleep(sleep_s)
|
||||
continue
|
||||
|
||||
r.raise_for_status()
|
||||
if r.text:
|
||||
try:
|
||||
return r.json()
|
||||
except Exception:
|
||||
return {"_raw": r.text}
|
||||
return {}
|
||||
|
||||
except (requests.ConnectionError, requests.Timeout) as e:
|
||||
sleep_s = base_sleep * (2 ** (attempt - 1)) + random.uniform(0, 0.5)
|
||||
print(f"[WARN] Network error on {path}: {e}. Sleeping {sleep_s:.2f}s (attempt {attempt}/{max_attempts})")
|
||||
time.sleep(sleep_s)
|
||||
|
||||
die(f"Request failed after retries: {method} {path}")
|
||||
|
||||
def _get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Any:
|
||||
return self._request_with_retry("GET", path, params=params)
|
||||
|
||||
def _post(self, path: str, payload: Dict[str, Any]) -> Any:
|
||||
return self._request_with_retry("POST", path, payload=payload)
|
||||
|
||||
def _delete(self, path: str) -> Any:
|
||||
return self._request_with_retry("DELETE", path)
|
||||
|
||||
def list_all(self, path: str) -> List[Dict[str, Any]]:
|
||||
return normalize_list(self._get(path))
|
||||
|
||||
# ---- entity lists ----
|
||||
def list_projects(self) -> List[Dict[str, Any]]:
|
||||
return self._get("/api/v1/projects")
|
||||
|
||||
def list_tags(self) -> List[Dict[str, Any]]:
|
||||
return self._get("/api/v1/tags")
|
||||
|
||||
def list_tasks(self) -> List[Dict[str, Any]]:
|
||||
return self._get("/api/v1/tasks")
|
||||
|
||||
def list_areas(self) -> List[Dict[str, Any]]:
|
||||
return self._get("/api/v1/areas")
|
||||
|
||||
# ---- create wrappers ----
|
||||
def create_area(self, name: str) -> Dict[str, Any]:
|
||||
# Your server: POST /api/v1/areas works; /api/v1/area does NOT.
|
||||
for path in ("/api/v1/areas", "/api/areas"):
|
||||
try:
|
||||
return self._post(path, {"name": name})
|
||||
except requests.HTTPError:
|
||||
continue
|
||||
die("Cannot create area: no supported endpoint (tried /api/v1/areas, /api/areas).")
|
||||
|
||||
def create_project(self, name: str) -> Dict[str, Any]:
|
||||
payload = {"name": name}
|
||||
for path in ("/api/v1/project", "/api/v1/projects"):
|
||||
try:
|
||||
return self._post(path, payload)
|
||||
except requests.HTTPError:
|
||||
continue
|
||||
die("Cannot create project: endpoint not found.")
|
||||
|
||||
def create_tag(self, name: str) -> Dict[str, Any]:
|
||||
for path in ("/api/v1/tag", "/api/v1/tags"):
|
||||
try:
|
||||
return self._post(path, {"name": name})
|
||||
except requests.HTTPError:
|
||||
continue
|
||||
die("Cannot create tag: endpoint not found.")
|
||||
|
||||
def create_task(
|
||||
self,
|
||||
name: str,
|
||||
project_id: Optional[int] = None,
|
||||
due_date: Optional[str] = None,
|
||||
priority: Optional[str] = None,
|
||||
tags: Optional[List[Dict[str, str]]] = None,
|
||||
description: Optional[str] = None,
|
||||
note: Optional[str] = None,
|
||||
parent_task_id: Optional[int] = None,
|
||||
is_completed: Optional[bool] = None,
|
||||
completed_at: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
payload: Dict[str, Any] = {"name": name}
|
||||
|
||||
if project_id is not None:
|
||||
payload["project_id"] = int(project_id)
|
||||
|
||||
if due_date:
|
||||
payload["due_date"] = due_date # oppure dueDate, dipende dal backend; se non va, lo togliamo
|
||||
if priority:
|
||||
payload["priority"] = priority
|
||||
|
||||
tags = _sanitize_tags(tags)
|
||||
if tags:
|
||||
clean_tags: List[Dict[str, str]] = []
|
||||
for t in tags:
|
||||
if not isinstance(t, dict):
|
||||
continue
|
||||
name = str(t.get("name") or "").strip()
|
||||
uid = str(t.get("uid") or "").strip()
|
||||
if not name or name.lower() in ("undefined", "null", "none"):
|
||||
continue
|
||||
if not uid or uid.lower() in ("undefined", "null", "none"):
|
||||
continue
|
||||
clean_tags.append({"name": name, "uid": uid})
|
||||
|
||||
if clean_tags:
|
||||
payload["Tags"] = clean_tags
|
||||
|
||||
# Tududi spesso usa "note" (tu l’hai mostrato), non "description"
|
||||
if note:
|
||||
payload["note"] = note
|
||||
if description:
|
||||
payload["description"] = description
|
||||
|
||||
if parent_task_id is not None:
|
||||
payload["parent_task_id"] = int(parent_task_id)
|
||||
|
||||
if is_completed is not None:
|
||||
payload["completed"] = is_completed
|
||||
if completed_at:
|
||||
payload["completed_at"] = completed_at
|
||||
|
||||
return self._post("/api/v1/task", payload)
|
||||
|
||||
|
||||
# ---- delete helpers ----
|
||||
def delete_best_effort(self, templates: List[str], uid: str) -> bool:
|
||||
"""
|
||||
Try multiple delete endpoints; return True if any succeeds, False if all 404.
|
||||
"""
|
||||
for tpl in templates:
|
||||
path = tpl.format(uid=uid)
|
||||
try:
|
||||
self._delete(path)
|
||||
return True
|
||||
except requests.HTTPError as e:
|
||||
resp = getattr(e, "response", None)
|
||||
if resp is not None and resp.status_code == 404:
|
||||
continue
|
||||
raise
|
||||
return False
|
||||
|
||||
|
||||
# ----------------------------
|
||||
# Cleanup via API (CLEAN_FIRST)
|
||||
# ----------------------------
|
||||
def clean_tududi(tud: TududiClient, use_areas: bool = False) -> None:
|
||||
print("[CLEAN] Starting full cleanup via API")
|
||||
|
||||
def delete_many(list_path: str, delete_templates: List[str], label: str):
|
||||
items = tud.list_all(list_path)
|
||||
print(f"[CLEAN] Deleting {len(items)} {label}")
|
||||
for it in items:
|
||||
if not isinstance(it, dict):
|
||||
continue
|
||||
uid = pick_uid(it)
|
||||
if not uid:
|
||||
continue
|
||||
ok = tud.delete_best_effort(delete_templates, uid)
|
||||
if not ok:
|
||||
print(f"[CLEAN][WARN] Could not delete {label[:-1]} uid={uid}")
|
||||
time.sleep(max(0.10, tud.request_sleep))
|
||||
|
||||
# Tasks
|
||||
delete_many(
|
||||
"/api/v1/tasks",
|
||||
[
|
||||
"/api/v1/task/{uid}",
|
||||
"/api/v1/tasks/{uid}",
|
||||
"/api/task/{uid}",
|
||||
"/api/tasks/{uid}",
|
||||
],
|
||||
"tasks",
|
||||
)
|
||||
|
||||
# Projects
|
||||
delete_many(
|
||||
"/api/v1/projects",
|
||||
[
|
||||
"/api/v1/project/{uid}",
|
||||
"/api/v1/projects/{uid}",
|
||||
"/api/project/{uid}",
|
||||
"/api/projects/{uid}",
|
||||
],
|
||||
"projects",
|
||||
)
|
||||
|
||||
# Tags
|
||||
delete_many(
|
||||
"/api/v1/tags",
|
||||
[
|
||||
"/api/v1/tag/{uid}",
|
||||
"/api/v1/tags/{uid}",
|
||||
"/api/tag/{uid}",
|
||||
"/api/tags/{uid}",
|
||||
],
|
||||
"tags",
|
||||
)
|
||||
|
||||
if use_areas:
|
||||
delete_many(
|
||||
"/api/v1/areas",
|
||||
[
|
||||
"/api/v1/area/{uid}", # some installs
|
||||
"/api/v1/areas/{uid}", # other installs
|
||||
"/api/area/{uid}",
|
||||
"/api/areas/{uid}",
|
||||
],
|
||||
"areas",
|
||||
)
|
||||
else:
|
||||
print("[CLEAN] Areas cleanup skipped (USE_AREAS=false).")
|
||||
|
||||
print("[CLEAN] Done.")
|
||||
|
||||
|
||||
# ----------------------------
|
||||
# Migration logic
|
||||
# ----------------------------
|
||||
def map_priority(todoist_priority: Optional[int]) -> Optional[str]:
|
||||
if not todoist_priority:
|
||||
return None
|
||||
return {1: "low", 2: "medium", 3: "high", 4: "urgent"}.get(int(todoist_priority), None)
|
||||
|
||||
|
||||
def normalize_due(todoist_item: Dict[str, Any]) -> Optional[str]:
|
||||
due = todoist_item.get("due")
|
||||
if not due:
|
||||
return None
|
||||
return due.get("datetime") or due.get("date")
|
||||
|
||||
|
||||
def build_comments_index(notes: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]:
|
||||
by_item: Dict[str, List[Dict[str, Any]]] = {}
|
||||
for n in notes:
|
||||
item_id = n.get("item_id") or n.get("task_id")
|
||||
if not item_id:
|
||||
continue
|
||||
by_item.setdefault(str(item_id), []).append(n)
|
||||
|
||||
for k in by_item:
|
||||
by_item[k].sort(key=lambda x: x.get("posted_at") or x.get("postedAt") or "")
|
||||
return by_item
|
||||
|
||||
|
||||
def comments_to_text(comments: List[Dict[str, Any]]) -> str:
|
||||
if not comments:
|
||||
return ""
|
||||
lines = ["\n\n---\n**Imported comments (Todoist)**\n"]
|
||||
for c in comments:
|
||||
ts = c.get("posted_at") or c.get("postedAt") or ""
|
||||
content = c.get("content") or c.get("text") or ""
|
||||
lines.append(f"- [{ts}] {content}")
|
||||
return "\n".join(lines).strip()
|
||||
|
||||
|
||||
def topo_sort_items(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||||
by_id = {str(i["id"]): i for i in items if "id" in i}
|
||||
children: Dict[str, List[str]] = {}
|
||||
roots: List[str] = []
|
||||
|
||||
for i in items:
|
||||
iid = str(i["id"])
|
||||
pid = i.get("parent_id") or i.get("parentId")
|
||||
if pid:
|
||||
children.setdefault(str(pid), []).append(iid)
|
||||
else:
|
||||
roots.append(iid)
|
||||
|
||||
out: List[Dict[str, Any]] = []
|
||||
visited: set[str] = set()
|
||||
|
||||
def dfs(node_id: str):
|
||||
if node_id in visited:
|
||||
return
|
||||
visited.add(node_id)
|
||||
out.append(by_id[node_id])
|
||||
for ch in children.get(node_id, []):
|
||||
if ch in by_id:
|
||||
dfs(ch)
|
||||
|
||||
for r in roots:
|
||||
if r in by_id:
|
||||
dfs(r)
|
||||
|
||||
for iid, obj in by_id.items():
|
||||
if iid not in visited:
|
||||
out.append(obj)
|
||||
|
||||
return out
|
||||
|
||||
|
||||
def main() -> None:
|
||||
todoist_token = os.getenv("TODOIST_TOKEN")
|
||||
if not todoist_token:
|
||||
die("Missing TODOIST_TOKEN in .env")
|
||||
|
||||
tududi_base = os.getenv("TUDUDI_BASE_URL", "").strip()
|
||||
if not tududi_base:
|
||||
die("Missing TUDUDI_BASE_URL in .env (e.g., https://tududi.example.org)")
|
||||
|
||||
tududi_key = os.getenv("TUDUDI_API_KEY")
|
||||
tududi_email = os.getenv("TUDUDI_EMAIL")
|
||||
tududi_password = os.getenv("TUDUDI_PASSWORD")
|
||||
|
||||
dry_run = getenv_bool("DRY_RUN", False)
|
||||
clean_first = getenv_bool("CLEAN_FIRST", False)
|
||||
use_areas = getenv_bool("USE_AREAS", False)
|
||||
request_sleep = getenv_float("REQUEST_SLEEP", 0.25)
|
||||
|
||||
print(f"[INFO] Starting migration @ {iso_now()} (dry_run={dry_run})")
|
||||
print("[INFO] Exporting from Todoist (sync full + completed best-effort)...")
|
||||
|
||||
tdc = TodoistClient(todoist_token)
|
||||
data = tdc.export_everything()
|
||||
|
||||
print(
|
||||
f"[INFO] Todoist: projects={len(data.projects)} labels={len(data.labels)} "
|
||||
f"sections={len(data.sections)} active_items={len(data.items)} "
|
||||
f"notes={len(data.notes)} completed_tasks={len(data.completed_tasks)}"
|
||||
)
|
||||
|
||||
tud = TududiClient(
|
||||
base_url=tududi_base,
|
||||
api_key=tududi_key,
|
||||
email=tududi_email,
|
||||
password=tududi_password,
|
||||
dry_run=dry_run,
|
||||
request_sleep=request_sleep,
|
||||
)
|
||||
|
||||
if clean_first:
|
||||
clean_tududi(tud, use_areas=use_areas)
|
||||
|
||||
# Load existing entities
|
||||
print("[INFO] Reading existing entities from Tududi...")
|
||||
existing_projects = normalize_list(tud.list_projects())
|
||||
existing_tags = normalize_list(tud.list_tags())
|
||||
existing_areas = normalize_list(tud.list_areas()) if use_areas else []
|
||||
|
||||
proj_by_name: Dict[str, Dict[str, Any]] = {}
|
||||
for p in existing_projects:
|
||||
n = _valid_name(p.get("name") if isinstance(p, dict) else None)
|
||||
if n:
|
||||
proj_by_name[n] = p
|
||||
|
||||
tag_by_name: Dict[str, Dict[str, Any]] = {}
|
||||
for t in existing_tags:
|
||||
n = _valid_name(t.get("name") if isinstance(t, dict) else None)
|
||||
if n:
|
||||
tag_by_name[n] = t
|
||||
|
||||
area_by_name: Dict[str, Dict[str, Any]] = {}
|
||||
if use_areas:
|
||||
for a in existing_areas:
|
||||
n = _valid_name(a.get("name") if isinstance(a, dict) else None)
|
||||
if n:
|
||||
area_by_name[n] = a
|
||||
|
||||
# Todoist project hierarchy info
|
||||
children_count: Dict[str, int] = {}
|
||||
for p in data.projects:
|
||||
pid = p.get("parent_id")
|
||||
if pid:
|
||||
children_count[str(pid)] = children_count.get(str(pid), 0) + 1
|
||||
|
||||
# Areas (optional)
|
||||
area_map: Dict[str, str] = {}
|
||||
if use_areas:
|
||||
print("[INFO] Creating Areas (USE_AREAS=true)...")
|
||||
for p in data.projects:
|
||||
tid = str(p.get("id"))
|
||||
name = p.get("name")
|
||||
if not name:
|
||||
continue
|
||||
if children_count.get(tid, 0) <= 0:
|
||||
continue
|
||||
if name in area_by_name:
|
||||
area_map[tid] = pick_uid(area_by_name[name]) or ""
|
||||
continue
|
||||
created = tud.create_area(name)
|
||||
cid = pick_uid(created) or pick_uid(created.get("area", {}) if isinstance(created, dict) else {}) or ""
|
||||
if cid:
|
||||
area_map[tid] = cid
|
||||
area_by_name[name] = {"uid": cid, "name": name}
|
||||
print(f"[INFO] Area: {name} -> {cid}")
|
||||
else:
|
||||
print("[INFO] Skipping Areas (USE_AREAS=false).")
|
||||
|
||||
# Projects: create ALL Todoist projects as Tududi projects (including "container" projects)
|
||||
project_map: Dict[str, str] = {}
|
||||
print("[INFO] Creating Projects...")
|
||||
for p in data.projects:
|
||||
tid = str(p["id"])
|
||||
name = p.get("name")
|
||||
if not name:
|
||||
continue
|
||||
|
||||
if name in proj_by_name:
|
||||
# prendi ID numerico se esiste, fallback su uid se proprio manca (ma dovrebbe esserci)
|
||||
existing = proj_by_name[name]
|
||||
pid = existing.get("id")
|
||||
if pid is None:
|
||||
print(f"[WARN] Existing project without numeric id? name={name} obj={existing}")
|
||||
continue
|
||||
project_map[tid] = int(pid)
|
||||
continue
|
||||
|
||||
created = tud.create_project(name)
|
||||
pid = created.get("id")
|
||||
if pid is None:
|
||||
die(f"Project created but missing id: {created}")
|
||||
project_map[tid] = int(pid)
|
||||
proj_by_name[name] = created
|
||||
print(f"[INFO] Project: {name} -> id={pid} uid={created.get('uid')}")
|
||||
time.sleep(request_sleep)
|
||||
|
||||
# Tags: create from Todoist labels using slug
|
||||
print("[INFO] Creating Tags from Todoist labels...")
|
||||
todo_label_id_to_name = {str(l["id"]): l.get("name") for l in data.labels if "id" in l and l.get("name")}
|
||||
for lbl in data.labels:
|
||||
orig = lbl.get("name")
|
||||
if not orig:
|
||||
continue
|
||||
slug = slug_tag(orig)
|
||||
if slug in tag_by_name:
|
||||
continue
|
||||
created = tud.create_tag(slug)
|
||||
cid = pick_uid(created) or pick_uid(created.get("tag", {}) if isinstance(created, dict) else {}) or ""
|
||||
if cid:
|
||||
tag_by_name[slug] = {"uid": cid, "name": slug}
|
||||
print(f"[INFO] Tag: {orig} -> {slug} ({cid})")
|
||||
time.sleep(request_sleep)
|
||||
|
||||
# Comments index
|
||||
comments_idx = build_comments_index(data.notes)
|
||||
|
||||
# Merge active + completed tasks
|
||||
all_tasks_src: List[Tuple[Dict[str, Any], bool]] = []
|
||||
for i in data.items:
|
||||
all_tasks_src.append((i, False))
|
||||
for c in data.completed_tasks:
|
||||
all_tasks_src.append((c, True))
|
||||
|
||||
items_like = [x for (x, _done) in all_tasks_src if "id" in x]
|
||||
items_sorted = topo_sort_items(items_like)
|
||||
is_done_map = {str(x["id"]): done for (x, done) in all_tasks_src if "id" in x}
|
||||
|
||||
created_task_map: Dict[str, int] = {}
|
||||
|
||||
|
||||
def get_project_for_task(task: Dict[str, Any]) -> Optional[int]:
|
||||
pid = task.get("project_id") or task.get("projectId")
|
||||
if pid is None:
|
||||
return None
|
||||
return project_map.get(str(pid))
|
||||
|
||||
def get_project_uid_for_task(task: Dict[str, Any]) -> Optional[str]:
|
||||
pid = task.get("project_id") if isinstance(task, dict) else None
|
||||
if pid is None:
|
||||
pid = task.get("projectId") if isinstance(task, dict) else None
|
||||
if pid is None:
|
||||
return None
|
||||
return project_map.get(str(pid))
|
||||
|
||||
def get_tags_for_task(task: Dict[str, Any]) -> List[Dict[str, str]]:
|
||||
lbls = task.get("labels") or task.get("label_ids") or task.get("labelIds") or []
|
||||
names: List[str] = []
|
||||
|
||||
for l in lbls:
|
||||
if isinstance(l, int) or (isinstance(l, str) and l.isdigit()):
|
||||
nm = todo_label_id_to_name.get(str(l))
|
||||
if nm:
|
||||
names.append(nm)
|
||||
elif isinstance(l, str):
|
||||
names.append(l)
|
||||
|
||||
# Build Tududi Tags objects: [{"name": "...", "uid": "..."}]
|
||||
tags_obj: List[Dict[str, str]] = []
|
||||
seen = set()
|
||||
|
||||
for nm in names:
|
||||
nm2 = str(nm).strip()
|
||||
if not nm2 or nm2.lower() in ("undefined", "null", "none"):
|
||||
continue
|
||||
|
||||
slug = slug_tag(nm2)
|
||||
|
||||
tinfo = tag_by_name.get(slug)
|
||||
if not tinfo:
|
||||
continue
|
||||
|
||||
uid = tinfo.get("uid") or tinfo.get("id")
|
||||
if not uid:
|
||||
continue
|
||||
|
||||
key = (slug, str(uid))
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
|
||||
tags_obj.append({"name": slug, "uid": str(uid)})
|
||||
|
||||
return tags_obj
|
||||
|
||||
|
||||
print(f"[INFO] Creating tasks (active+completed): {len(items_sorted)}")
|
||||
for task in items_sorted:
|
||||
tid = str(task["id"])
|
||||
name = task.get("content") or task.get("name") or task.get("title")
|
||||
if not name:
|
||||
continue
|
||||
|
||||
parent_id = task.get("parent_id") or task.get("parentId")
|
||||
tud_parent_id = created_task_map.get(str(parent_id)) if parent_id else None
|
||||
|
||||
due = normalize_due(task)
|
||||
prio = map_priority(task.get("priority"))
|
||||
project_uid = get_project_uid_for_task(task)
|
||||
tags = get_tags_for_task(task)
|
||||
|
||||
desc = task.get("description") or ""
|
||||
comments = comments_idx.get(tid, [])
|
||||
desc_extra = comments_to_text(comments)
|
||||
if desc_extra:
|
||||
desc = (desc + "\n" + desc_extra).strip()
|
||||
|
||||
is_completed = is_done_map.get(tid, False)
|
||||
completed_at = task.get("completed_at") or task.get("completedAt")
|
||||
|
||||
created = tud.create_task(
|
||||
name=name,
|
||||
project_id=project_uid,
|
||||
due_date=due,
|
||||
priority=prio,
|
||||
tags=tags if tags else None,
|
||||
description=desc if desc else None,
|
||||
parent_task_id=tud_parent_id,
|
||||
is_completed=is_completed if is_completed else None,
|
||||
completed_at=completed_at,
|
||||
)
|
||||
|
||||
new_id = None
|
||||
if isinstance(created, dict):
|
||||
new_id = created.get("id")
|
||||
if new_id is None and isinstance(created.get("task"), dict):
|
||||
new_id = created["task"].get("id")
|
||||
|
||||
if new_id is not None:
|
||||
created_task_map[tid] = int(new_id)
|
||||
|
||||
|
||||
if len(created_task_map) % 50 == 0:
|
||||
print(f"[INFO] Created tasks: {len(created_task_map)}")
|
||||
|
||||
time.sleep(request_sleep)
|
||||
|
||||
print(f"[DONE] Migration finished. Created/seen tasks: {len(created_task_map)} @ {iso_now()}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except KeyboardInterrupt:
|
||||
die("Interrupted by user.", 130)
|
||||
Reference in New Issue
Block a user