942 lines
30 KiB
Python
942 lines
30 KiB
Python
#!/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
|
||
|
||
def todoist_task_url(task: Dict[str, Any]) -> str:
|
||
"""
|
||
Best-effort URL to the Todoist task.
|
||
- If Todoist provides a direct url field, use it.
|
||
- Otherwise fall back to the canonical web UI endpoint.
|
||
"""
|
||
# sometimes present in some payloads (completed/rest)
|
||
url = task.get("url") or task.get("task_url") or task.get("taskUrl")
|
||
if isinstance(url, str) and url.strip():
|
||
return url.strip()
|
||
|
||
tid = task.get("id")
|
||
if tid is None:
|
||
return "https://todoist.com/"
|
||
return f"https://todoist.com/showTask?id={tid}"
|
||
|
||
|
||
def build_tududi_description_from_todoist(task: Dict[str, Any], task_name: str) -> str:
|
||
"""
|
||
Tududi description format:
|
||
|
||
<Task name>
|
||
|
||
[Todoist task](...)
|
||
|
||
<Todoist description>
|
||
"""
|
||
parts: List[str] = []
|
||
|
||
# Task name (explicit, even if Tududi already has it)
|
||
parts.append(task_name.strip())
|
||
|
||
# Todoist link
|
||
tid = task.get("id")
|
||
if tid:
|
||
parts.append(f"\n[Todoist task](https://todoist.com/showTask?id={tid})")
|
||
|
||
# Todoist description
|
||
td_desc = task.get("description")
|
||
if isinstance(td_desc, str) and td_desc.strip():
|
||
parts.append("\n" + td_desc.strip())
|
||
|
||
return "\n\n".join(parts).strip()
|
||
|
||
# ----------------------------
|
||
# 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)
|
||
|
||
# Tududi note: Todoist link + Todoist description
|
||
tududi_description = build_tududi_description_from_todoist(task, name)
|
||
|
||
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=tududi_description,
|
||
note=tududi_description,
|
||
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)
|