#!/usr/bin/env python3 from __future__ import annotations import argparse import csv import json import sys import requests from typing import Dict, Any from common import ( add_common_args, build_session, normalize_base_url, resolve_arg, resolve_common, ) def create_card( session: requests.Session, base_url: str, board_id: int, stack_id: int, title: str, description: str, ) -> requests.Response: url = f"{base_url}/index.php/apps/deck/api/v1.0/boards/{board_id}/stacks/{stack_id}/cards" headers = {"OCS-APIRequest": "true", "Content-Type": "application/json"} payload: Dict[str, Any] = { "title": title, "type": "plain", "order": 0, "description": description, "duedate": None, } return session.post(url, headers=headers, data=json.dumps(payload)) def fetch_board_labels( session: requests.Session, base_url: str, board_id: int ) -> list[dict]: """ GET /boards/{board_id} -> returns labels[] from the board details. """ url = f"{base_url}/index.php/apps/deck/api/v1.0/boards/{board_id}" headers = {"OCS-APIRequest": "true", "Accept": "application/json"} resp = session.get(url, headers=headers) if resp.status_code != 200: raise RuntimeError( f"Deck API error fetching board {board_id}: {resp.status_code} {resp.text}" ) return resp.json().get("labels") or [] def build_label_index(labels: list[dict]) -> Dict[str, int]: """ Build a case-insensitive name -> labelId map. """ return { (lbl.get("title") or "").strip().lower(): int(lbl["id"]) for lbl in labels if lbl.get("title") and lbl.get("id") is not None } def parse_tags_field(value: str) -> list[str]: """ Parse a CSV 'tags' cell: comma-separated tag names; whitespace trimmed; empties dropped. """ if not value: return [] return [t.strip() for t in value.split(",") if t.strip()] def assign_label_to_card( session: requests.Session, base_url: str, board_id: int, stack_id: int, card_id: int, label_id: int, ) -> requests.Response: url = ( f"{base_url}/index.php/apps/deck/api/v1.0/boards/{board_id}" f"/stacks/{stack_id}/cards/{card_id}/assignLabel" ) headers = {"OCS-APIRequest": "true", "Content-Type": "application/json"} payload = {"labelId": int(label_id)} return session.put(url, headers=headers, data=json.dumps(payload)) def apply_tags_to_card( session: requests.Session, base_url: str, board_id: int, stack_id: int, card_id: int, tag_names: list[str], label_index: Dict[str, int], ) -> tuple[list[str], list[str]]: """ Assign labels by name to a card. Returns (applied, missing). """ applied: list[str] = [] missing: list[str] = [] for name in tag_names: label_id = label_index.get(name.lower()) if label_id is None: missing.append(name) continue resp = assign_label_to_card( session, base_url, board_id, stack_id, card_id, label_id ) if resp.status_code in (200, 201): applied.append(name) else: missing.append(f"{name} (HTTP {resp.status_code})") return applied, missing def main() -> None: parser = argparse.ArgumentParser( description="Import tasks from a CSV file into Nextcloud Deck." ) # Common flags (domain, board-id, username, password) add_common_args(parser) # Script-specific flags parser.add_argument( "--stack-id", type=int, help="Deck Stack ID where cards will be created", ) parser.add_argument( "--csv-file", help="Path to the CSV file containing tasks (expects headers: title, description)", ) args = parser.parse_args() # Resolve common fields (ENV -> CLI -> prompt) try: base_url, board_id, username, password = resolve_common( cli_domain=args.domain, cli_board_id=args.board_id, cli_username=args.username, cli_password=args.password, ) except Exception as e: print(f"❌ {e}", file=sys.stderr) sys.exit(2) # Resolve script-specific fields try: stack_id = resolve_arg( None if args.stack_id is None else str(args.stack_id), ["NEXTCLOUD_STACK_ID", "STACK_ID"], prompt_text="Deck Stack ID: ", cast=lambda s: int(s.strip()), ) csv_file = resolve_arg( args.csv_file, ["IMPORT_CSV_FILE", "CSV_FILE"], prompt_text="CSV file path: ", cast=str, ) except Exception as e: print(f"❌ {e}", file=sys.stderr) sys.exit(2) session = build_session(username, password) # Pre-fetch labels so we can resolve tags by name (cheap, one call). try: labels = fetch_board_labels(session, base_url, int(board_id)) except Exception as e: print(f"⚠️ Could not fetch board labels (tags will be skipped): {e}") labels = [] label_index = build_label_index(labels) # Read & import try: with open(csv_file, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: title = (row.get("title") or "").strip() description = (row.get("description") or "").strip() tag_names = parse_tags_field(row.get("tags") or "") if not title: print(f"Skipping row with missing title: {row}") continue resp = create_card( session, base_url, int(board_id), int(stack_id), title, description ) if resp.status_code not in (200, 201): print(f"❌ Failed: {title} — {resp.status_code} {resp.text}") continue print(f"✔️ Created card: {title}") if not tag_names: continue card = resp.json() card_id = card.get("id") if card_id is None: print(f" ⚠️ Card '{title}' has no id; skipping tags.") continue applied, missing = apply_tags_to_card( session, base_url, int(board_id), int(stack_id), int(card_id), tag_names, label_index, ) if applied: print(f" 🏷 Applied: {', '.join(applied)}") if missing: print(f" ⚠️ Unknown tags: {', '.join(missing)}") except FileNotFoundError: print(f"❌ CSV file not found: {csv_file}", file=sys.stderr) sys.exit(1) except Exception as e: print(f"❌ Error during import: {e}", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()