From 09fc24b8f9b6556f3ab596dcbaad7b5a7ace4f76 Mon Sep 17 00:00:00 2001 From: Chen Asraf Date: Tue, 23 Sep 2025 02:32:58 +0300 Subject: [PATCH] refactor: extract common & use env vars more --- .gitignore | 1 + common.py | 207 +++++++++++++++++++++++++++++++++++++++++++++++++ import.py | 150 +++++++++++++++++++---------------- list_stacks.py | 71 +++++++---------- 4 files changed, 321 insertions(+), 108 deletions(-) create mode 100644 common.py diff --git a/.gitignore b/.gitignore index cfcfa45..473e878 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ poetry.lock .env.keys .envrc *.csv +__pycache__ diff --git a/common.py b/common.py new file mode 100644 index 0000000..20772a0 --- /dev/null +++ b/common.py @@ -0,0 +1,207 @@ +#!/usr/bin/env python3 +from __future__ import annotations +import argparse +import getpass +import os +from typing import Callable, Iterable, Optional, TypeVar +from urllib.parse import urlparse + +import requests + +T = TypeVar("T") + +# ---------- URL & session helpers ---------- + + +def normalize_base_url(domain_or_url: str) -> str: + """ + Accepts either a bare domain (example.com) or a full URL (https://example.com[/...]). + Returns a normalized base URL without trailing slash (e.g., https://example.com). + """ + val = (domain_or_url or "").strip().rstrip("/") + if not val: + return val + parsed = urlparse(val) + if parsed.scheme in ("http", "https"): + return val + return f"https://{val}" + + +def build_session(username: str, password: str) -> requests.Session: + s = requests.Session() + s.auth = (username, password) + return s + + +# ---------- argparse helpers ---------- + + +def add_common_args(parser: argparse.ArgumentParser) -> None: + """ + Adds commonly shared CLI flags. Not marked as required because we support env vars + prompts. + """ + parser.add_argument( + "--domain", + help="Nextcloud instance domain or base URL (e.g., example.com or https://example.com)", + ) + parser.add_argument( + "--board-id", + type=int, + help="Deck Board ID", + ) + parser.add_argument( + "--username", + help="Nextcloud username (if not provided, will be prompted)", + ) + parser.add_argument( + "--password", + help="Nextcloud password or app password (if not provided, will be prompted)", + ) + + +# ---------- resolution helpers ---------- + + +def _first_env_match(keys: Iterable[str]) -> Optional[str]: + for k in keys: + v = os.getenv(k) + if v is not None and str(v).strip() != "": + return v + return None + + +def resolve_arg( + args_value: Optional[str], + env_keys: Iterable[str], + *, + prompt_text: Optional[str] = None, + secret: bool = False, + cast: Optional[Callable[[str], T]] = None, + allow_empty: bool = False, +) -> T | str: + """ + Resolution order: ENV -> CLI arg -> prompt (if provided) -> error. + - env_keys: list/iterable of env var names to try in order. + - secret: use getpass for prompt if True. + - cast: optional converter (e.g., int). Cast is applied to whichever source was used. + - allow_empty: if True, an empty prompt is accepted (returns empty string). + """ + raw = _first_env_match(env_keys) + if raw is None and args_value is not None: + raw = args_value + + if raw is None: + if prompt_text is not None: + raw = ( + getpass.getpass(prompt_text) if secret else input(prompt_text) + ).strip() + if not allow_empty and raw == "": + raise ValueError(f"Missing value for: {', '.join(env_keys)}") + else: + # No prompt provided -> treat as missing. + raise ValueError( + f"Missing required value. Provide one of env({', '.join(env_keys)}) or CLI option." + ) + + return cast(raw) if cast else raw + + +def resolve_common( + *, + cli_domain: Optional[str], + cli_board_id: Optional[int], + cli_username: Optional[str], + cli_password: Optional[str], +) -> tuple[str, int, str, str]: + """ + Resolves common inputs from ENV/CLI/prompt: + Returns: (base_url, board_id, username, password) + """ + domain = resolve_arg( + None if cli_domain is None else str(cli_domain), + ["NEXTCLOUD_DOMAIN", "NEXTCLOUD_BASE_URL"], + prompt_text="Nextcloud domain (e.g., example.com or https://example.com): ", + secret=False, + cast=str, + ) + base_url = normalize_base_url(domain) + + board_id = resolve_arg( + None if cli_board_id is None else str(cli_board_id), + ["NEXTCLOUD_BOARD_ID", "BOARD_ID"], + prompt_text="Deck Board ID: ", + secret=False, + cast=lambda s: int(s.strip()), + ) + + username = resolve_arg( + cli_username, + ["NEXTCLOUD_USERNAME", "NC_USERNAME"], + prompt_text="Nextcloud username: ", + secret=False, + cast=str, + ) + password = resolve_arg( + cli_password, + ["NEXTCLOUD_PASSWORD", "NC_PASSWORD"], + prompt_text="Nextcloud password (or app password): ", + secret=True, + cast=str, + ) + + return base_url, board_id, username, password + + +def add_domain_and_auth_args(parser: argparse.ArgumentParser) -> None: + """ + Like add_common_args, but WITHOUT --board-id (useful for scripts that don't need a board yet). + """ + parser.add_argument( + "--domain", + help="Nextcloud instance domain or base URL (e.g., example.com or https://example.com)", + ) + parser.add_argument( + "--username", + help="Nextcloud username (if not provided, will be prompted)", + ) + parser.add_argument( + "--password", + help="Nextcloud password or app password (if not provided, will be prompted)", + ) + + +def resolve_domain_and_auth( + *, + cli_domain: str | None, + cli_username: str | None, + cli_password: str | None, +) -> tuple[str, str, str]: + """ + Resolve base_url, username, password via ENV -> CLI -> prompt. + Returns: (base_url, username, password) + """ + domain = resolve_arg( + None if cli_domain is None else str(cli_domain), + ["NEXTCLOUD_DOMAIN", "NEXTCLOUD_BASE_URL"], + prompt_text="Nextcloud domain (e.g., example.com or https://example.com): ", + secret=False, + cast=str, + ) + base_url = normalize_base_url(domain) + + username = resolve_arg( + cli_username, + ["NEXTCLOUD_USERNAME", "NC_USERNAME"], + prompt_text="Nextcloud username: ", + secret=False, + cast=str, + ) + password = resolve_arg( + cli_password, + ["NEXTCLOUD_PASSWORD", "NC_PASSWORD"], + prompt_text="Nextcloud password (or app password): ", + secret=True, + cast=str, + ) + return base_url, username, password + diff --git a/import.py b/import.py index 0a66aef..168867b 100644 --- a/import.py +++ b/import.py @@ -1,97 +1,117 @@ #!/usr/bin/env python3 -import csv -import requests -import json +from __future__ import annotations import argparse -import getpass -import os +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, domain, board_id, stack_id, title, description): - url = f"https://{domain}/index.php/apps/deck/api/v1.0/boards/{board_id}/stacks/{stack_id}/cards" +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 = { + payload: Dict[str, Any] = { "title": title, "type": "plain", "order": 0, "description": description, "duedate": None, } - - response = session.post(url, headers=headers, data=json.dumps(payload)) - return response + return session.post(url, headers=headers, data=json.dumps(payload)) -def main(): +def main() -> None: parser = argparse.ArgumentParser( - description="Import tasks from a CSV file into Nextcloud Deck" - ) - parser.add_argument( - "--domain", - required=True, - help="Nextcloud instance domain (e.g., example.com or https://example.com)", - ) - parser.add_argument( - "--board-id", - required=True, - type=int, - help="Board ID where the cards will be created", + 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", - required=True, type=int, - help="Stack ID where the cards will be added", + help="Deck Stack ID where cards will be created", ) parser.add_argument( - "--csv-file", required=True, help="Path to the CSV file containing tasks" - ) - parser.add_argument( - "--username", help="Nextcloud username (if not provided, will be prompted)" - ) - parser.add_argument( - "--password", - help="Nextcloud password or app password (if not provided, will be prompted)", + "--csv-file", + help="Path to the CSV file containing tasks (expects headers: title, description)", ) args = parser.parse_args() - # Prompt for credentials if missing - username = ( - os.environ["NEXTCLOUD_USERNAME"] - or args.username - or input("Nextcloud username: ") - ) - password = ( - os.environ["NEXTCLOUD_PASSWORD"] - or args.password - or getpass.getpass("Nextcloud password (or app password): ") - ) + # 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) - session = requests.Session() - session.auth = (username, password) + # 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) - with open(args.csv_file, newline="", encoding="utf-8") as csvfile: - reader = csv.DictReader(csvfile) + session = build_session(username, password) - for row in reader: - title = row.get("title", "").strip() - description = row.get("description", "").strip() + # 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() + if not title: + print(f"Skipping row with missing title: {row}") + continue - if not title: - print(f"Skipping row with missing title: {row}") - continue - - response = create_card( - session, args.domain, args.board_id, args.stack_id, title, description - ) - - if response.status_code in (200, 201): - print(f"✔️ Successfully created card: {title}") - else: - print( - f"❌ Failed to create card: {title} - Error: {response.status_code} {response.text}" + 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: + print(f"❌ Failed: {title} — {resp.status_code} {resp.text}") + 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__": diff --git a/list_stacks.py b/list_stacks.py index 58ff33e..44d85c0 100644 --- a/list_stacks.py +++ b/list_stacks.py @@ -1,18 +1,15 @@ #!/usr/bin/env python3 +from __future__ import annotations import argparse import json import sys import requests -from urllib.parse import urlparse -import getpass -import os - -def normalize_base_url(domain_or_url: str) -> str: - parsed = urlparse(domain_or_url) - if parsed.scheme in ("http", "https"): - return domain_or_url.rstrip("/") - return f"https://{domain_or_url.strip().strip('/')}" +from common import ( + add_common_args, + build_session, + resolve_common, +) def list_stacks(session: requests.Session, base_url: str, board_id: int): @@ -24,50 +21,38 @@ def list_stacks(session: requests.Session, base_url: str, board_id: int): return resp.json() -def main(): +def main() -> None: parser = argparse.ArgumentParser( description="List stacks (id + name) for a Nextcloud Deck board." ) + # Common flags (domain, board-id, username, password) + add_common_args(parser) + + # Script-specific flags parser.add_argument( - "--domain", - required=True, - help="Nextcloud instance (e.g., example.com or https://example.com)", - ) - parser.add_argument( - "--board-id", required=True, type=int, help="Deck Board ID to list stacks from" - ) - parser.add_argument( - "--username", help="Nextcloud username (if not provided, will be prompted)" - ) - parser.add_argument( - "--password", - help="Nextcloud password or app password (if not provided, will be prompted)", - ) - parser.add_argument( - "--json", action="store_true", help="Output raw JSON instead of a table" + "--json", + action="store_true", + help="Output raw JSON instead of a table", ) args = parser.parse_args() - # Prompt for missing creds - username = ( - os.environ["NEXTCLOUD_USERNAME"] - or args.username - or input("Nextcloud username: ") - ) - password = ( - os.environ["NEXTCLOUD_PASSWORD"] - or args.password - or getpass.getpass("Nextcloud password (or app password): ") - ) + # Resolve common inputs (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) - base_url = normalize_base_url(args.domain) - - session = requests.Session() - session.auth = (username, password) + session = build_session(username, password) try: - stacks = list_stacks(session, base_url, args.board_id) + stacks = list_stacks(session, base_url, int(board_id)) except Exception as e: print(f"❌ Failed to fetch stacks: {e}", file=sys.stderr) sys.exit(1) @@ -80,7 +65,7 @@ def main(): print("No stacks found.") return - print(f"Stacks for board {args.board_id} at {base_url}:") + print(f"Stacks for board {board_id} at {base_url}:") print("-" * 60) print(f"{'ID':<8} {'NAME'}") print("-" * 60)