From 8aa1213f6150bdf5921655b895ae99f7e3d40ee3 Mon Sep 17 00:00:00 2001 From: Chen Asraf Date: Fri, 1 May 2026 00:04:57 +0300 Subject: [PATCH] feat: create board fix, create tags, main cmd flow feat: create tags on board feat: main flow asks for task instead of import by default fix: create board requires color --- Makefile | 13 +++++ common.py | 90 +++++++++++++++++++++++++++++ create_board.py | 43 ++++++++++++-- create_tags.py | 132 ++++++++++++++++++++++++++++++++++++++++++ import.py | 120 +++++++++++++++++++++++++++++++++++++- main.py | 150 +++++++++++++++++++++++++++++++++++++----------- 6 files changed, 507 insertions(+), 41 deletions(-) create mode 100644 create_tags.py diff --git a/Makefile b/Makefile index 9f3b3f4..7f9ddb8 100644 --- a/Makefile +++ b/Makefile @@ -1,9 +1,22 @@ all: @poetry run python main.py +install: + @command -v poetry >/dev/null 2>&1 || { echo "poetry not found. Install from https://python-poetry.org/docs/#installation"; exit 1; } + @poetry install --no-root + @if [ ! -f .env ]; then \ + echo ".env not found ā€” copying from .env.example"; \ + cp .env.example .env; \ + echo "Edit .env with your Nextcloud credentials before running."; \ + fi + @poetry run python -c "import requests, inquirer" && echo "Install OK ā€” run 'make' to start." + create_board: @poetry run python create_board.py +create_tags: + @poetry run python create_tags.py + list_boards: @poetry run python list_boards.py diff --git a/common.py b/common.py index 20772a0..5ecc30b 100644 --- a/common.py +++ b/common.py @@ -6,10 +6,100 @@ import os from typing import Callable, Iterable, Optional, TypeVar from urllib.parse import urlparse +import inquirer import requests T = TypeVar("T") +# ---------- color presets / picker ---------- + +DECK_COLOR_PRESETS: list[tuple[str, str]] = [ + ("Nextcloud Blue", "0082c9"), + ("Red", "e74c3c"), + ("Orange", "f39c12"), + ("Yellow", "f1c40f"), + ("Green", "27ae60"), + ("Teal", "1abc9c"), + ("Purple", "9b59b6"), + ("Pink", "e91e63"), + ("Gray", "95a5a6"), +] + + +def select_existing_board( + session: requests.Session, + base_url: str, + message: str = "Which board?", +) -> tuple[int, str]: + """ + Fetch boards via the Deck API and let the user pick one interactively. + Returns (board_id, board_title). + """ + # Lazy import to avoid circular module load. + from list_boards import list_boards + + boards = list_boards(session, base_url) + if not boards: + raise RuntimeError("No boards found on this instance.") + + choices: list[str] = [] + board_map: dict[str, tuple[int, str]] = {} + for board in boards: + bid = board.get("id") + title = board.get("title", "Untitled") + label = f"{title} (ID: {bid})" + choices.append(label) + board_map[label] = (bid, title) + + answer = inquirer.prompt( + [inquirer.List("board", message=message, choices=choices)] + ) + if not answer: + raise RuntimeError("Board selection cancelled") + + return board_map[answer["board"]] + + +def interactive_color_selection(message: str = "Pick a color") -> str: + """ + Show interactive menu to pick a hex color from presets, or enter a custom hex. + Returns 6-char lowercase hex string (no '#'). + """ + choices = [f"{name} (#{hex_})" for name, hex_ in DECK_COLOR_PRESETS] + custom = "Custom hex..." + choices.append(custom) + + answer = inquirer.prompt( + [inquirer.List("color", message=message, choices=choices)] + ) + if not answer: + raise RuntimeError("Color selection cancelled") + + selected = answer["color"] + if selected == custom: + custom_answer = inquirer.prompt( + [ + inquirer.Text( + "hex", + message="Enter hex color (6 chars, no '#')", + default="0082c9", + ) + ] + ) + if not custom_answer: + raise RuntimeError("Color entry cancelled") + hex_value = custom_answer["hex"].strip().lstrip("#") + if len(hex_value) != 6 or not all( + c in "0123456789abcdefABCDEF" for c in hex_value + ): + raise RuntimeError(f"Invalid hex color: {hex_value}") + return hex_value.lower() + + for name, hex_ in DECK_COLOR_PRESETS: + if selected == f"{name} (#{hex_})": + return hex_ + raise RuntimeError(f"Unexpected color selection: {selected}") + # ---------- URL & session helpers ---------- diff --git a/create_board.py b/create_board.py index 941e492..c6c0119 100644 --- a/create_board.py +++ b/create_board.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse import json import os +import re import sys from typing import Iterable @@ -28,21 +29,32 @@ def get_stack_names() -> list[str]: """ Get stack names from NEW_BOARD_COLUMNS env var (comma-separated), or fall back to DEFAULT_STACKS if not set. + + Tolerates shell-style escapes that may leak into the value (e.g., + "šŸž\\ Bugs\\,šŸ“‹\\ To\\ Do") by stripping single-backslash escapes + from each piece after splitting on commas. """ columns_env = os.getenv("NEW_BOARD_COLUMNS") if columns_env: - # Split by comma and strip whitespace from each column name - return [col.strip() for col in columns_env.split(",") if col.strip()] + names = [] + for col in columns_env.split(","): + # Remove backslash escapes: "\" -> ""; trailing "\" -> "" + unescaped = re.sub(r"\\(.?)", r"\1", col).strip() + if unescaped: + names.append(unescaped) + return names return DEFAULT_STACKS -def create_board(session: requests.Session, base_url: str, title: str) -> dict: +def create_board( + session: requests.Session, base_url: str, title: str, color: str = "0082c9" +) -> dict: """ POST /boards -> { id, title, ... } """ url = f"{base_url}/index.php/apps/deck/api/v1.0/boards" headers = {"OCS-APIRequest": "true", "Content-Type": "application/json"} - payload = {"title": title} + payload = {"title": title, "color": color} resp = session.post(url, headers=headers, data=json.dumps(payload)) if resp.status_code not in (200, 201): raise RuntimeError( @@ -94,6 +106,10 @@ def main() -> None: "--board-name", help="New board name/title (e.g., 'Team Kanban')", ) + parser.add_argument( + "--color", + help="Board color as 6-char hex (no '#'), e.g., 0082c9", + ) parser.add_argument( "--no-default-stacks", action="store_true", @@ -125,11 +141,28 @@ def main() -> None: print(f"āŒ {e}", file=sys.stderr) sys.exit(2) + # Resolve board color + try: + color_raw = resolve_arg( + args.color, + ["NEXTCLOUD_BOARD_COLOR", "BOARD_COLOR"], + prompt_text="Board color hex (6 chars, no '#') [0082c9]: ", + cast=str, + allow_empty=True, + ) + color = (color_raw or "").strip().lstrip("#") or "0082c9" + if len(color) != 6 or not all(c in "0123456789abcdefABCDEF" for c in color): + raise ValueError(f"Invalid hex color: {color}") + color = color.lower() + except Exception as e: + print(f"āŒ {e}", file=sys.stderr) + sys.exit(2) + session = build_session(username, password) # Create board try: - board = create_board(session, base_url, board_name) + board = create_board(session, base_url, board_name, color) board_id = board.get("id") if board_id is None: raise RuntimeError(f"Board created but no id returned: {board}") diff --git a/create_tags.py b/create_tags.py new file mode 100644 index 0000000..69e5749 --- /dev/null +++ b/create_tags.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +from __future__ import annotations +import argparse +import json +import sys + +import inquirer +import requests + +from common import ( + add_domain_and_auth_args, + build_session, + interactive_color_selection, + resolve_domain_and_auth, + select_existing_board, +) + + +def create_label( + session: requests.Session, + base_url: str, + board_id: int, + title: str, + color: str, +) -> dict: + """ + POST /boards/{board_id}/labels -> { id, title, color, ... } + """ + url = f"{base_url}/index.php/apps/deck/api/v1.0/boards/{board_id}/labels" + headers = {"OCS-APIRequest": "true", "Content-Type": "application/json"} + payload = {"title": title, "color": color} + resp = session.post(url, headers=headers, data=json.dumps(payload)) + if resp.status_code not in (200, 201): + raise RuntimeError( + f"Deck API error creating label '{title}': {resp.status_code} {resp.text}" + ) + return resp.json() + + +def prompt_tag_name() -> str | None: + """ + Prompt for a tag name. Empty input ends the loop. + Returns the trimmed name, or None to stop. + """ + answer = inquirer.prompt( + [ + inquirer.Text( + "name", + message="Tag name (empty to finish)", + ) + ] + ) + if answer is None: + return None + name = (answer.get("name") or "").strip() + return name or None + + +def interactive_create_tags_loop( + session: requests.Session, base_url: str, board_id: int, board_title: str +) -> list[dict]: + """ + Loop prompting for tag name + color and creating labels until the user enters + an empty name. Returns the list of created label dicts. + """ + print(f"\nCreating tags on '{board_title}' (ID: {board_id}).") + print("Press Enter on an empty name to finish.\n") + + created: list[dict] = [] + while True: + try: + name = prompt_tag_name() + except KeyboardInterrupt: + print("\nAborted.") + break + + if not name: + break + + try: + color = interactive_color_selection(f"Pick a color for '{name}'") + except Exception as e: + print(f"āŒ {e} ā€” skipping this tag.") + continue + + try: + label = create_label(session, base_url, int(board_id), name, color) + created.append(label) + print(f" āœ”ļø Created tag '{name}' (#{color}) [id: {label.get('id')}]") + except Exception as e: + print(f" āŒ {e}") + + print("\nSummary") + print("-------") + print(f"Created {len(created)} tag(s) on board {board_id}:") + for label in created: + print(f" - [{label.get('id')}] {label.get('title')} #{label.get('color')}") + return created + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Interactively create tags (labels) on a Nextcloud Deck board." + ) + add_domain_and_auth_args(parser) + args = parser.parse_args() + + try: + base_url, username, password = resolve_domain_and_auth( + cli_domain=args.domain, + cli_username=args.username, + cli_password=args.password, + ) + except Exception as e: + print(f"āŒ {e}", file=sys.stderr) + sys.exit(2) + + session = build_session(username, password) + + try: + board_id, board_title = select_existing_board( + session, base_url, "Which board do you want to add tags to?" + ) + except Exception as e: + print(f"āŒ {e}", file=sys.stderr) + sys.exit(1) + + interactive_create_tags_loop(session, base_url, int(board_id), board_title) + + +if __name__ == "__main__": + main() diff --git a/import.py b/import.py index 168867b..c02993b 100644 --- a/import.py +++ b/import.py @@ -36,6 +36,88 @@ def create_card( 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." @@ -88,6 +170,14 @@ def main() -> None: 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: @@ -95,6 +185,7 @@ def main() -> None: 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 @@ -102,10 +193,33 @@ def main() -> None: resp = create_card( session, base_url, int(board_id), int(stack_id), title, description ) - if resp.status_code in (200, 201): - print(f"āœ”ļø Created card: {title}") - else: + 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) diff --git a/main.py b/main.py index 59fa421..b45b54c 100755 --- a/main.py +++ b/main.py @@ -13,6 +13,7 @@ import requests from common import ( add_common_args, build_session, + interactive_color_selection, resolve_arg, resolve_common, resolve_domain_and_auth, @@ -20,12 +21,17 @@ from common import ( # Import functions from existing scripts from create_board import create_board, create_stacks_in_order, get_stack_names +from create_tags import interactive_create_tags_loop from list_boards import list_boards from list_stacks import list_stacks # Import from 'import.py' using importlib since 'import' is a keyword import_module = importlib.import_module("import") create_card = import_module.create_card +fetch_board_labels = import_module.fetch_board_labels +build_label_index = import_module.build_label_index +parse_tags_field = import_module.parse_tags_field +apply_tags_to_card = import_module.apply_tags_to_card def interactive_board_selection( @@ -82,9 +88,11 @@ def interactive_board_selection( if not board_name: raise RuntimeError("Board name cannot be empty") + color = interactive_color_selection("Pick a board color") + # Create the board - print(f"\nā†’ Creating board '{board_name}'...") - board = create_board(session, base_url, board_name) + print(f"\nā†’ Creating board '{board_name}' (color #{color})...") + board = create_board(session, base_url, board_name, color) board_id = board.get("id") if board_id is None: raise RuntimeError(f"Board created but no id returned: {board}") @@ -208,6 +216,14 @@ def import_csv_file( print(f"\nā†’ Importing cards from '{csv_file}'...") + # Pre-fetch labels so we can resolve tags by name. + 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) + imported = 0 skipped = 0 @@ -216,6 +232,7 @@ def import_csv_file( for row in reader: title = row.get("title", "").strip() description = row.get("description", "").strip() + tag_names = parse_tags_field(row.get("tags") or "") if not title: skipped += 1 @@ -225,14 +242,38 @@ def import_csv_file( resp = create_card( session, base_url, int(board_id), int(stack_id), title, description ) - if resp.status_code in (200, 201): - print(f" āœ”ļø Created card: {title}") - imported += 1 - else: + if resp.status_code not in (200, 201): print( f" āŒ Failed to create card '{title}': {resp.status_code} {resp.text}" ) skipped += 1 + continue + + print(f" āœ”ļø Created card: {title}") + imported += 1 + + 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 Exception as e: print(f" āŒ Failed to create card '{title}': {e}") skipped += 1 @@ -240,6 +281,74 @@ def import_csv_file( print(f"\nāœ”ļø Import complete! {imported} cards imported, {skipped} skipped.") +def run_import_csv_action( + session: requests.Session, base_url: str, board_id: int, board_name: str +) -> None: + """ + Interactive flow: pick stack, pick CSV, confirm, import. + """ + stack_id, stack_name = interactive_stack_selection(session, base_url, board_id) + print(f"ā†’ Selected column: {stack_name} (ID: {stack_id})\n") + + csv_file = interactive_csv_selection() + + confirm_answer = inquirer.prompt( + [ + inquirer.Confirm( + "confirm", + message=f"Import '{csv_file}' to '{stack_name}' in board '{board_name}'?", + default=True, + ) + ] + ) + if not confirm_answer or not confirm_answer["confirm"]: + print("Import cancelled.") + return + + import_csv_file(session, base_url, board_id, stack_id, csv_file) + + +def interactive_action_menu( + session: requests.Session, base_url: str, board_id: int, board_name: str +) -> None: + """ + Loop showing actions to perform on the selected board until the user quits. + """ + actions = { + "Import CSV file": lambda: run_import_csv_action( + session, base_url, board_id, board_name + ), + "Create tags": lambda: interactive_create_tags_loop( + session, base_url, board_id, board_name + ), + "Quit": None, + } + + while True: + answer = inquirer.prompt( + [ + inquirer.List( + "action", + message=f"What do you want to do on '{board_name}'?", + choices=list(actions.keys()), + ) + ] + ) + if not answer or answer["action"] == "Quit": + return + + handler = actions[answer["action"]] + if handler is None: + return + + try: + handler() + except KeyboardInterrupt: + print("\nAction cancelled.") + except Exception as e: + print(f"āŒ {e}", file=sys.stderr) + + def main() -> None: parser = argparse.ArgumentParser( description="Unified interactive tool for importing CSV files into Nextcloud Deck." @@ -308,7 +417,7 @@ def main() -> None: sys.exit(1) else: # Interactive mode - print("Welcome to Nextcloud Deck CSV Importer!\n") + print("Welcome to Nextcloud Deck Tools!\n") # Resolve domain and auth (can still use ENV or will prompt) try: @@ -324,35 +433,10 @@ def main() -> None: session = build_session(username, password) try: - # Step 1: Board selection/creation board_id, board_name = interactive_board_selection(session, base_url) print(f"\nā†’ Working with board: {board_name} (ID: {board_id})\n") - # Step 2: Stack selection - stack_id, stack_name = interactive_stack_selection( - session, base_url, board_id - ) - print(f"ā†’ Selected column: {stack_name} (ID: {stack_id})\n") - - # Step 3: CSV file selection - csv_file = interactive_csv_selection() - - # Step 4: Confirm and import - confirm_q = [ - inquirer.Confirm( - "confirm", - message=f"Import '{csv_file}' to '{stack_name}' in board '{board_name}'?", - default=True, - ) - ] - confirm_answer = inquirer.prompt(confirm_q) - if not confirm_answer or not confirm_answer["confirm"]: - print("Import cancelled.") - sys.exit(0) - - # Import! - import_csv_file(session, base_url, board_id, stack_id, csv_file) - + interactive_action_menu(session, base_url, board_id, board_name) except Exception as e: print(f"āŒ {e}", file=sys.stderr) sys.exit(1)