nextcloud-deck-tools/import.py

#!/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()