From 0b115cd5753ad273c716a5fdb421e094792857c2 Mon Sep 17 00:00:00 2001 From: Davide Di Ruscio Date: Tue, 30 Dec 2025 00:39:10 +0100 Subject: [PATCH] Initial commit --- README | 14 + main.py | 894 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 908 insertions(+) create mode 100644 main.py diff --git a/README b/README index e69de29..dbb525a 100644 --- a/README +++ b/README @@ -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. + diff --git a/main.py b/main.py new file mode 100644 index 0000000..26339b6 --- /dev/null +++ b/main.py @@ -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)