🎨 Palette: Interactive input validation & cleanup

.Jules/palette.md

@@ -9,3 +9,7 @@
 ## 2024-05-24 - Fail Fast & Friendly
 **Learning:** In CLI tools involving APIs, cascade failures (hundreds of "Failed to X") caused by basic auth issues (401/403) are overwhelming and confusing. A dedicated "Pre-flight Check" that validates credentials *before* attempting the main workload allows for specific, actionable error messages (e.g. "Check your token at [URL]") instead of generic HTTP errors.
 **Action:** Implement a `check_api_access()` step at the start of any CLI workflow to validate permissions and provide human-readable guidance on failure.
+
+## 2024-05-25 - Interactive Input Validation
+**Learning:** In CLI interactive modes, validating input immediately (looping until valid) prevents user frustration from waiting for a script to start only to crash later. It turns a "Run -> Error -> Restart" cycle into a "Prompt -> Typo -> Correction -> Success" flow.
+**Action:** Always implement `get_validated_input` loops for critical configuration prompts in interactive CLI sessions.

main.py

@@ -23,9 +23,10 @@
 import concurrent.futures
 import ipaddress
 import socket
+import getpass
 from functools import lru_cache
 from urllib.parse import urlparse
-from typing import Dict, List, Optional, Any, Set, Sequence
+from typing import Dict, List, Optional, Any, Set, Sequence, Callable
 
 import httpx
 from dotenv import load_dotenv
@@ -140,6 +141,25 @@
     return v
 
 
+def get_validated_input(
+    prompt: str,
+    validator: Callable[[str], bool],
+    error_msg: str,
+    is_password: bool = False
+) -> str:
+    """Prompts for input until the validator returns True."""
+    while True:
+        if is_password:
+            value = getpass.getpass(prompt).strip()
+        else:
+            value = input(prompt).strip()
+
+        if validator(value):
+            return value
+
+        print(f"{Colors.FAIL}❌ {error_msg}{Colors.ENDC}")
+
+
 TOKEN = _clean_env_kv(os.getenv("TOKEN"), "TOKEN")
 
 # Default folder sources
@@ -189,7 +209,6 @@
     )
 
 _gh = httpx.Client(timeout=30)
-MAX_RESPONSE_SIZE = 10 * 1024 * 1024  # 10 MB limit for external resources
 
 # --------------------------------------------------------------------------- #
 # 3. Helpers
@@ -228,7 +247,7 @@
             try:
                 # Resolve hostname to IPs (IPv4 and IPv6)
                 # We filter for AF_INET/AF_INET6 to ensure we get IP addresses
                 addr_info = socket.getaddrinfo(hostname, None, proto=socket.IPPROTO_TCP)
                 for res in addr_info:
                     # res is (family, type, proto, canonname, sockaddr)
                     # sockaddr is (address, port) for AF_INET/AF_INET6
@@ -247,12 +266,19 @@
 
     return True
 
-def validate_profile_id(profile_id: str) -> bool:
+def is_valid_profile_id_format(profile_id: str) -> bool:
     if not re.match(r"^[a-zA-Z0-9_-]+$", profile_id):
-        log.error("Invalid profile ID format (contains unsafe characters)")
         return False
     if len(profile_id) > 64:
-        log.error("Invalid profile ID length (max 64 chars)")
+        return False
+    return True
+
+def validate_profile_id(profile_id: str) -> bool:
+    if not is_valid_profile_id_format(profile_id):
+        if not re.match(r"^[a-zA-Z0-9_-]+$", profile_id):
+            log.error("Invalid profile ID format (contains unsafe characters)")
+        elif len(profile_id) > 64:
+            log.error("Invalid profile ID length (max 64 chars)")
         return False
     return True
 
@@ -376,9 +402,9 @@
             log.critical(f"{Colors.FAIL}❌ Authentication Failed: The API Token is invalid.{Colors.ENDC}")
             log.critical(f"{Colors.FAIL}   Please check your token at: https://controld.com/account/manage-account{Colors.ENDC}")
         elif code == 403:
             log.critical(f"{Colors.FAIL}🚫 Access Denied: Token lacks permission for Profile {profile_id}.{Colors.ENDC}")
         elif code == 404:
             log.critical(f"{Colors.FAIL}🔍 Profile Not Found: The ID '{profile_id}' does not exist.{Colors.ENDC}")
             log.critical(f"{Colors.FAIL}   Please verify the Profile ID from your Control D Dashboard URL.{Colors.ENDC}")
         else:
             log.error(f"API Access Check Failed ({code}): {e}")
@@ -854,7 +880,7 @@
         return success_count == len(folder_data_list)
 
     except Exception as e:
         log.error(f"Unexpected error during sync for profile {profile_id}: {sanitize_for_log(e)}")
         return False
 
 # --------------------------------------------------------------------------- #
@@ -881,17 +907,31 @@
         if not profile_ids:
             print(f"{Colors.CYAN}ℹ Profile ID is missing.{Colors.ENDC}")
             print(f"{Colors.CYAN}  You can find this in the URL of your profile in the Control D Dashboard.{Colors.ENDC}")
-            p_input = input(f"{Colors.BOLD}Enter Control D Profile ID:{Colors.ENDC} ").strip()
-            if p_input:
-                profile_ids = [p.strip() for p in p_input.split(",") if p.strip()]
+
+            def validate_profile_input(value: str) -> bool:
+                if not value:
+                    return False
+                ids = [p.strip() for p in value.split(",") if p.strip()]
+                return bool(ids) and all(is_valid_profile_id_format(pid) for pid in ids)
+
+            p_input = get_validated_input(
+                f"{Colors.BOLD}Enter Control D Profile ID:{Colors.ENDC} ",
+                validate_profile_input,
+                "Invalid ID(s). Must be alphanumeric (including - and _), max 64 chars. Comma-separate for multiple."
+            )
+            profile_ids = [p.strip() for p in p_input.split(",") if p.strip()]
 
         if not TOKEN:
             print(f"{Colors.CYAN}ℹ API Token is missing.{Colors.ENDC}")
             print(f"{Colors.CYAN}  You can generate one at: https://controld.com/account/manage-account{Colors.ENDC}")
-            import getpass
-            t_input = getpass.getpass(f"{Colors.BOLD}Enter Control D API Token:{Colors.ENDC} ").strip()
-            if t_input:
-                TOKEN = t_input
+
+            t_input = get_validated_input(
+                f"{Colors.BOLD}Enter Control D API Token:{Colors.ENDC} ",
+                lambda x: bool(x),
+                "Token cannot be empty.",
+                is_password=True
+            )
+            TOKEN = t_input
 
     if not profile_ids and not args.dry_run:
         log.error("PROFILE missing and --dry-run not set. Provide --profiles or set PROFILE env.")
@@ -925,7 +965,7 @@
                 })
                 continue
 
             log.info("Starting sync for profile %s", profile_id)
             status = sync_profile(
                 profile_id,
                 folder_urls,
@@ -1015,11 +1055,11 @@
         status_color = Colors.GREEN if res['success'] else Colors.FAIL
 
         print(
             f"{res['profile']:<{profile_col_width}} | "
             f"{res['folders']:>10} | "
             f"{res['rules']:>10,} | "
             f"{res['duration']:>9.1f}s | "
             f"{status_color}{res['status_label']:<15}{Colors.ENDC}"
         )
         total_folders += res['folders']
         total_rules += res['rules']
@@ -1045,12 +1085,12 @@
     total_status_color = Colors.GREEN if all_success else Colors.FAIL
 
     print(
         f"{Colors.BOLD}"
         f"{'TOTAL':<{profile_col_width}} | "
         f"{total_folders:>10} | "
         f"{total_rules:>10,} | "
         f"{total_duration:>9.1f}s | "
         f"{total_status_color}{total_status_text:<15}{Colors.ENDC}"
     )
     print("=" * table_width + "\n")