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
This commit is contained in:
2026-05-01 00:04:57 +03:00
parent 65090b375b
commit 8aa1213f61
6 changed files with 507 additions and 41 deletions

View File

@@ -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

View File

@@ -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 ----------

View File

@@ -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: "\<c>" -> "<c>"; 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}")

132
create_tags.py Normal file
View File

@@ -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()

120
import.py
View File

@@ -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)

150
main.py
View File

@@ -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)