refactor: extract common & use env vars more

.gitignore

@@ -3,3 +3,4 @@ poetry.lock
 .env.keys
 .envrc
 *.csv
+__pycache__

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
+

import.py

@@ -1,97 +1,117 @@
 #!/usr/bin/env python3
-import csv
+from __future__ import annotations
-import requests
-import json
 import argparse
-import getpass
+import csv
-import os
+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):
+def create_card(
-    url = f"https://{domain}/index.php/apps/deck/api/v1.0/boards/{board_id}/stacks/{stack_id}/cards"
+    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,
     }
-
+    return session.post(url, headers=headers, data=json.dumps(payload))
-    response = session.post(url, headers=headers, data=json.dumps(payload))
-    return response
 
 
-def main():
+def main() -> None:
     parser = argparse.ArgumentParser(
-        description="Import tasks from a CSV file into Nextcloud Deck"
+        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",
     )
+    # 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"
+        "--csv-file",
-    )
+        help="Path to the CSV file containing tasks (expects headers: title, description)",
-    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)",
     )
 
     args = parser.parse_args()
 
-    # Prompt for credentials if missing
+    # Resolve common fields (ENV -> CLI -> prompt)
-    username = (
+    try:
-        os.environ["NEXTCLOUD_USERNAME"]
+        base_url, board_id, username, password = resolve_common(
-        or args.username
+            cli_domain=args.domain,
-        or input("Nextcloud username: ")
+            cli_board_id=args.board_id,
-    )
+            cli_username=args.username,
-    password = (
+            cli_password=args.password,
-        os.environ["NEXTCLOUD_PASSWORD"]
+        )
-        or args.password
+    except Exception as e:
-        or getpass.getpass("Nextcloud password (or app password): ")
+        print(f"❌ {e}", file=sys.stderr)
-    )
+        sys.exit(2)
 
-    session = requests.Session()
+    # Resolve script-specific fields
-    session.auth = (username, password)
+    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:
+    session = build_session(username, password)
-        reader = csv.DictReader(csvfile)
 
-        for row in reader:
+    # Read & import
-            title = row.get("title", "").strip()
+    try:
-            description = row.get("description", "").strip()
+        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:
+                resp = create_card(
-                print(f"Skipping row with missing title: {row}")
+                    session, base_url, int(board_id), int(stack_id), title, description
-                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}"
                 )
+                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__":

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
 
-
+from common import (
-def normalize_base_url(domain_or_url: str) -> str:
+    add_common_args,
-    parsed = urlparse(domain_or_url)
+    build_session,
-    if parsed.scheme in ("http", "https"):
+    resolve_common,
-        return domain_or_url.rstrip("/")
+)
-    return f"https://{domain_or_url.strip().strip('/')}"
 
 
 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",
+        "--json",
-        required=True,
+        action="store_true",
-        help="Nextcloud instance (e.g., example.com or https://example.com)",
+        help="Output raw JSON instead of a table",
-    )
-    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"
     )
 
     args = parser.parse_args()
 
-    # Prompt for missing creds
+    # Resolve common inputs (ENV -> CLI -> prompt)
-    username = (
+    try:
-        os.environ["NEXTCLOUD_USERNAME"]
+        base_url, board_id, username, password = resolve_common(
-        or args.username
+            cli_domain=args.domain,
-        or input("Nextcloud username: ")
+            cli_board_id=args.board_id,
-    )
+            cli_username=args.username,
-    password = (
+            cli_password=args.password,
-        os.environ["NEXTCLOUD_PASSWORD"]
+        )
-        or args.password
+    except Exception as e:
-        or getpass.getpass("Nextcloud password (or app password): ")
+        print(f"❌ {e}", file=sys.stderr)
-    )
+        sys.exit(2)
 
-    base_url = normalize_base_url(args.domain)
+    session = build_session(username, password)
-
-    session = requests.Session()
-    session.auth = (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)