-
Notifications
You must be signed in to change notification settings - Fork 1
π¨ Palette: Enhance CLI Summary Table and Progress Bar #326
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. Weβll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
1a03703
f107c95
6db6069
35a7f11
ec0c410
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -313,6 +313,156 @@ | |
| print("") | ||
|
|
||
|
|
||
| def print_summary_table(results: List[Dict[str, Any]], dry_run: bool) -> None: | ||
| """Prints a nicely formatted summary table.""" | ||
| # Determine the width for the Profile ID column (min 25) | ||
| max_profile_len = max((len(r["profile"]) for r in results), default=25) | ||
| profile_col_width = max(25, max_profile_len) | ||
|
|
||
| # Calculate widths | ||
| col_widths = { | ||
| "profile": profile_col_width, | ||
| "folders": 10, | ||
| "rules": 10, | ||
| "duration": 10, | ||
| "status": 15, | ||
| } | ||
|
|
||
| if USE_COLORS: | ||
| # Unicode Box Drawing | ||
| chars = { | ||
| "tl": "β", "tm": "β¬", "tr": "β", | ||
| "bl": "β", "bm": "β΄", "br": "β", | ||
| "ml": "β", "mm": "βΌ", "mr": "β€", | ||
| "v": "β", "h": "β", | ||
| } | ||
| else: | ||
| # ASCII Fallback | ||
| chars = { | ||
| "tl": "+", "tm": "+", "tr": "+", | ||
| "bl": "+", "bm": "+", "br": "+", | ||
| "ml": "+", "mm": "+", "mr": "+", | ||
| "v": "|", "h": "-", | ||
| } | ||
|
|
||
| def _print_separator(left, mid, right): | ||
| segments = [chars["h"] * (width + 2) for width in col_widths.values()] | ||
| print(f"{chars[left]}{chars[mid].join(segments)}{chars[right]}") | ||
|
|
||
| def _print_row(profile, folders, rules, duration, status, is_header=False): | ||
| v = chars["v"] | ||
|
|
||
| # 1. Pad raw strings first (so padding is calculated on visible chars) | ||
| p_val = f"{profile:<{col_widths['profile']}}" | ||
| f_val = f"{folders:>{col_widths['folders']}}" | ||
| r_val = f"{rules:>{col_widths['rules']}}" | ||
| d_val = f"{duration:>{col_widths['duration']}}" | ||
| s_val = f"{status:<{col_widths['status']}}" | ||
|
|
||
| # 2. Wrap in color codes if needed | ||
| if is_header and USE_COLORS: | ||
| p_val = f"{Colors.BOLD}{p_val}{Colors.ENDC}" | ||
| f_val = f"{Colors.BOLD}{f_val}{Colors.ENDC}" | ||
| r_val = f"{Colors.BOLD}{r_val}{Colors.ENDC}" | ||
| d_val = f"{Colors.BOLD}{d_val}{Colors.ENDC}" | ||
| s_val = f"{Colors.BOLD}{s_val}{Colors.ENDC}" | ||
|
|
||
| print( | ||
| f"{v} {p_val} {v} {f_val} {v} {r_val} {v} {d_val} {v} {s_val} {v}" | ||
| ) | ||
|
|
||
| title_text = "DRY RUN SUMMARY" if dry_run else "SYNC SUMMARY" | ||
| title_color = Colors.CYAN if dry_run else Colors.HEADER | ||
|
|
||
| total_width = ( | ||
| 1 + (col_widths["profile"] + 2) + 1 + | ||
| (col_widths["folders"] + 2) + 1 + | ||
| (col_widths["rules"] + 2) + 1 + | ||
| (col_widths["duration"] + 2) + 1 + | ||
| (col_widths["status"] + 2) + 1 | ||
| ) | ||
|
|
||
| print("\n" + (f"{title_color}{title_text:^{total_width}}{Colors.ENDC}" if USE_COLORS else f"{title_text:^{total_width}}")) | ||
|
|
||
| _print_separator("tl", "tm", "tr") | ||
| # Header row - pad manually then print | ||
| _print_row("Profile ID", "Folders", "Rules", "Duration", "Status", is_header=True) | ||
| _print_separator("ml", "mm", "mr") | ||
|
|
||
| total_folders = 0 | ||
| total_rules = 0 | ||
| total_duration = 0.0 | ||
| success_count = 0 | ||
|
|
||
| for res in results: | ||
| # Profile | ||
| p_val = f"{res['profile']:<{col_widths['profile']}}" | ||
|
|
||
| # Folders | ||
| f_val = f"{res['folders']:>{col_widths['folders']}}" | ||
|
|
||
| # Rules | ||
| r_val = f"{res['rules']:>{col_widths['rules']},}" | ||
|
|
||
| # Duration | ||
| d_val = f"{res['duration']:>{col_widths['duration']-1}.1f}s" | ||
|
|
||
| # Status | ||
| status_label = res["status_label"] | ||
| s_val_raw = f"{status_label:<{col_widths['status']}}" | ||
| if USE_COLORS: | ||
| status_color = Colors.GREEN if res["success"] else Colors.FAIL | ||
| s_val = f"{status_color}{s_val_raw}{Colors.ENDC}" | ||
| else: | ||
| s_val = s_val_raw | ||
|
|
||
| # Delegate the actual row printing to the shared helper to avoid | ||
| # duplicating table border/spacing logic here. | ||
| _print_row(p_val, f_val, r_val, d_val, s_val) | ||
|
|
||
| total_folders += res["folders"] | ||
| total_rules += res["rules"] | ||
| total_duration += res["duration"] | ||
| if res["success"]: | ||
| success_count += 1 | ||
|
|
||
| _print_separator("ml", "mm", "mr") | ||
|
|
||
| # Total Row | ||
| total = len(results) | ||
| all_success = success_count == total | ||
|
|
||
| if dry_run: | ||
| total_status_text = "β Ready" if all_success else "β Errors" | ||
| else: | ||
| total_status_text = "β All Good" if all_success else "β Errors" | ||
|
|
||
| p_val = f"{'TOTAL':<{col_widths['profile']}}" | ||
| if USE_COLORS: | ||
| p_val = f"{Colors.BOLD}{p_val}{Colors.ENDC}" | ||
|
|
||
| f_val = f"{total_folders:>{col_widths['folders']}}" | ||
| r_val = f"{total_rules:>{col_widths['rules']},}" | ||
| d_val = f"{total_duration:>{col_widths['duration']-1}.1f}s" | ||
|
|
||
| s_val_raw = f"{total_status_text:<{col_widths['status']}}" | ||
| if USE_COLORS: | ||
| status_color = Colors.GREEN if all_success else Colors.FAIL | ||
| s_val = f"{status_color}{s_val_raw}{Colors.ENDC}" | ||
| else: | ||
| s_val = s_val_raw | ||
|
Comment on lines
+440
to
+453
|
||
|
|
||
| print( | ||
| f"{chars['v']} {p_val} " | ||
| f"{chars['v']} {f_val} " | ||
| f"{chars['v']} {r_val} " | ||
| f"{chars['v']} {d_val} " | ||
| f"{chars['v']} {s_val} {chars['v']}" | ||
| ) | ||
|
|
||
| _print_separator("bl", "bm", "br") | ||
|
|
||
|
|
||
| def _get_progress_bar_width() -> int: | ||
| """Calculate dynamic progress bar width based on terminal size. | ||
|
|
||
|
|
@@ -1630,98 +1780,98 @@ | |
| return False | ||
|
|
||
|
|
||
| def create_folder( | ||
| client: httpx.Client, profile_id: str, name: str, do: int, status: int | ||
| ) -> Optional[str]: | ||
| """ | ||
| Create a new folder and return its ID. | ||
| Attempts to read ID from response first, then falls back to polling. | ||
| """ | ||
| try: | ||
| # 1. Send the Create Request | ||
| response = _api_post( | ||
| client, | ||
| f"{API_BASE}/{profile_id}/groups", | ||
| data={"name": name, "do": do, "status": status}, | ||
| ) | ||
|
|
||
| # OPTIMIZATION: Try to grab ID directly from response to avoid the wait loop | ||
| try: | ||
| resp_data = response.json() | ||
| body = resp_data.get("body", {}) | ||
|
|
||
| # Check if it returned a single group object | ||
| if isinstance(body, dict) and "group" in body and "PK" in body["group"]: | ||
| pk = str(body["group"]["PK"]) | ||
| if not validate_folder_id(pk, log_errors=False): | ||
| log.error(f"API returned invalid folder ID: {sanitize_for_log(pk)}") | ||
| return None | ||
| log.info( | ||
| "Created folder %s (ID %s) [Direct]", | ||
| sanitize_for_log(name), | ||
| sanitize_for_log(pk), | ||
| ) | ||
| return pk | ||
|
|
||
| # Check if it returned a list containing our group | ||
| if isinstance(body, dict) and "groups" in body: | ||
| for grp in body["groups"]: | ||
| if grp.get("group") == name: | ||
| pk = str(grp["PK"]) | ||
| if not validate_folder_id(pk, log_errors=False): | ||
| log.error(f"API returned invalid folder ID: {sanitize_for_log(pk)}") | ||
| continue | ||
| log.info( | ||
| "Created folder %s (ID %s) [Direct]", | ||
| sanitize_for_log(name), | ||
| sanitize_for_log(pk), | ||
| ) | ||
| return pk | ||
| except Exception as e: | ||
| log.debug( | ||
| f"Could not extract ID from POST response: " f"{sanitize_for_log(e)}" | ||
| ) | ||
|
|
||
| # 2. Fallback: Poll for the new folder (The Robust Retry Logic) | ||
| for attempt in range(MAX_RETRIES + 1): | ||
| try: | ||
| data = _api_get(client, f"{API_BASE}/{profile_id}/groups").json() | ||
| groups = data.get("body", {}).get("groups", []) | ||
|
|
||
| for grp in groups: | ||
| if grp["group"].strip() == name.strip(): | ||
| pk = str(grp["PK"]) | ||
| if not validate_folder_id(pk, log_errors=False): | ||
| log.error(f"API returned invalid folder ID: {sanitize_for_log(pk)}") | ||
| return None | ||
| log.info( | ||
| "Created folder %s (ID %s) [Polled]", | ||
| sanitize_for_log(name), | ||
| sanitize_for_log(pk), | ||
| ) | ||
| return pk | ||
| except Exception as e: | ||
| log.warning( | ||
| f"Error fetching groups on attempt {attempt}: {sanitize_for_log(e)}" | ||
| ) | ||
|
|
||
| if attempt < MAX_RETRIES: | ||
| wait_time = FOLDER_CREATION_DELAY * (attempt + 1) | ||
| log.info( | ||
| f"Folder '{sanitize_for_log(name)}' not found yet. Retrying in {wait_time}s..." | ||
| ) | ||
| time.sleep(wait_time) | ||
|
|
||
| log.error( | ||
| f"Folder {sanitize_for_log(name)} was not found after creation and retries." | ||
| ) | ||
| return None | ||
|
|
||
| except (httpx.HTTPError, KeyError) as e: | ||
| log.error( | ||
| f"Failed to create folder {sanitize_for_log(name)}: {sanitize_for_log(e)}" | ||
| ) | ||
| return None | ||
|
|
||
|
|
||
| def push_rules( | ||
|
|
@@ -2191,7 +2341,7 @@ | |
| return | ||
|
|
||
| # Unicode Table | ||
| def line(l, m, r): return f"{Colors.BOLD}{l}{m.join('β' * (x+2) for x in w)}{r}{Colors.ENDC}" | ||
| def row(c): return f"{Colors.BOLD}β{Colors.ENDC} {c[0]:<{w[0]}} {Colors.BOLD}β{Colors.ENDC} {c[1]:>{w[1]}} {Colors.BOLD}β{Colors.ENDC} {c[2]:>{w[2]}} {Colors.BOLD}β{Colors.ENDC} {c[3]:>{w[3]}} {Colors.BOLD}β{Colors.ENDC} {c[4]:<{w[4]}} {Colors.BOLD}β{Colors.ENDC}" | ||
|
|
||
| print(f"\n{line('β', 'β', 'β')}") | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The calculation for
total_widthis repetitive and can be simplified by deriving it from thecol_widthsdictionary. This makes the code more maintainable, as it will automatically adjust if columns are added or removed.