From a1856a7d2a9c8f7c9385bdc5401611302f41de04 Mon Sep 17 00:00:00 2001 From: badra001 Date: Fri, 17 Jul 2026 16:29:18 -0400 Subject: [PATCH] add --notification --- .../purge_check_s3_buckets.py.new | 78 ++++++++++++++++--- 1 file changed, 68 insertions(+), 10 deletions(-) diff --git a/local-app/python-tools/cross-organization/purge_check_s3_buckets.py.new b/local-app/python-tools/cross-organization/purge_check_s3_buckets.py.new index 5c41bbab..eadf2a8b 100755 --- a/local-app/python-tools/cross-organization/purge_check_s3_buckets.py.new +++ b/local-app/python-tools/cross-organization/purge_check_s3_buckets.py.new @@ -7,7 +7,10 @@ import json import queue import time import argparse +import getpass +import smtplib from datetime import datetime, timezone +from email.message import EmailMessage from collections import defaultdict from concurrent.futures import ThreadPoolExecutor @@ -18,10 +21,29 @@ from rich.table import Table from rich.progress import Progress, BarColumn, TextColumn, TimeElapsedColumn, SpinnerColumn # --- VERSIONING --- -__version__ = "2.6.0" +__version__ = "2.7.0" console = Console() +def dispatch_progress_email(recipient_email, subject_str, body_str): + """Dispatches a plain-text status tracking email via localhost SMTP.""" + try: + sender_user = getpass.getuser() + sender_host = "localhost" + sender_address = f"{sender_user}@{sender_host}" + + msg = EmailMessage() + msg.set_content(body_str) + msg["Subject"] = subject_str + msg["From"] = sender_address + msg["To"] = recipient_email + + with smtplib.SMTP("localhost", 25) as server: + server.send_message(msg) + except Exception as email_err: + # Fail gracefully to keep terminal execution threads uninterrupted + console.print(f"[bold yellow]Warning: Notification pipeline failed to dispatch email: {email_err}[/bold yellow]") + def build_assumed_role_session(target_account_id, role_name, region_name, profile=None): """Assumes the cross-account role in the target member account and returns an authorized session.""" base_session = boto3.Session(profile_name=profile) if profile else boto3.Session() @@ -156,6 +178,7 @@ def main(): parser.add_argument("--profile", help="AWS configuration context configuration handle target") parser.add_argument("--output", default=".", help="Folder destination directory path to save logs (Default: current directory)") parser.add_argument("--execute", action="store_true", help="Arm structural execution destructive deletions") + parser.add_argument("--notification", help="Optional target email address to receive incremental milestone updates") args = parser.parse_args() dry_run = not args.execute @@ -218,8 +241,11 @@ def main(): console.print(f"Target Bucket Filter Regex : [cyan]{args.filter_bucket}[/cyan]") console.print(f"Discovered Filter Targets : [green]{len(targets)} Buckets[/green]") console.print(f"Safe Execution Dry-Run Mode: [white]{dry_run}[/white]") + if args.notification: + console.print(f"Notification Pipeline Target: [white]{args.notification}[/white]") console.print(f"[bold blue]----------------------------------------------------------------------[/bold blue]") + total_targets_count = len(targets) if not targets: console.print("[bold yellow]Warning: Zero buckets matched filtering configurations criteria rules. Safely exiting.[/bold yellow]") sys.exit(0) @@ -237,7 +263,7 @@ def main(): TextColumn("[bold green]Bucket {task.completed} of {task.total}[/bold green]"), TimeElapsedColumn() ) - macro_task_id = overall_progress.add_task("Purge Fleet", total=len(targets)) + macro_task_id = overall_progress.add_task("Purge Fleet", total=total_targets_count) lane_progress = Progress( TextColumn("[bold blue]Lane #{task.fields[lane_idx]}[/bold blue]"), @@ -248,7 +274,7 @@ def main(): TimeElapsedColumn() ) - render_lanes_count = min(args.workers, len(targets)) + render_lanes_count = min(args.workers, total_targets_count) lane_task_ids = {} for i in range(render_lanes_count): tid = lane_progress.add_task("Awaiting batch assignment", total=100, lane_idx=i, bucket_name="IDLE", current_test="START", total_tests="0") @@ -258,18 +284,19 @@ def main(): progress_table.add_row("[bold blue]----------------------------------------------------------------------[/bold blue]") progress_table.add_row(lane_progress) - # Convert the static targets dataset list into a fluid, stateful iterator pool targets_iterator = iter(targets) active_lane_allocations = {i: None for i in range(args.workers)} futures_map = {} + + # Milestone notification variables setup + completed_counter = 0 + next_email_milestone_pct = 10 with Live(progress_table, refresh_per_second=10, console=console): with ThreadPoolExecutor(max_workers=args.workers) as executor: - # Keep looping as long as we have jobs left in the pool OR threads actively executing work while len(futures_map) > 0 or any(v is not None for v in active_lane_allocations.values()): - # --- FIXED: Dynamically allocate upcoming buckets to open lanes immediately --- for lane_idx, active_fut in active_lane_allocations.items(): if active_fut is None: try: @@ -281,9 +308,8 @@ def main(): futures_map[fut] = next_item active_lane_allocations[lane_idx] = fut except StopIteration: - break # Inventory array completely exhausted + break - # Consume real-time progress events while not progress_queue.empty(): try: signal = progress_queue.get_nowait() @@ -301,12 +327,10 @@ def main(): except queue.Empty: break - # --- FIXED: Process completed threads and release their lease slots immediately --- completed_futures = [f for f in futures_map if f.done()] for fut in completed_futures: meta = futures_map.pop(fut) - # Identify and wipe the lease mapping to open the lane back up assigned_lane = next(k for k, v in active_lane_allocations.items() if v == fut) active_lane_allocations[assigned_lane] = None @@ -323,8 +347,28 @@ def main(): "error_count": 1, "status": f"CRITICAL_EXCEPTION: {str(fatal_fut_err)}" }) + completed_counter += 1 overall_progress.update(macro_task_id, advance=1) lane_progress.update(lane_task_ids[assigned_lane], bucket_name="IDLE", completed=0, current_test="IDLE", total_tests="0") + + # --- REFRESH EMAIL NOTIFICATION PIPELINE SNAPSHOT CHECKS --- + if args.notification and completed_counter < total_targets_count: + current_macro_pct = int((completed_counter / total_targets_count) * 100) + if current_macro_pct >= next_email_milestone_pct: + current_elapsed = time.time() - global_start_epoch + + sub_msg = f"purge progress {current_macro_pct}% bucket {completed_counter} of {total_targets_count}" + body_msg = ( + f"AWS ORGANIZATIONAL S3 METRICS STORAGE PURGE LIFECYCLE PROGRESS REPORT\n" + f"======================================================================\n" + f"Current Status : RUNNING (Incremental Milestone Reached)\n" + f"Progress Percentage : {current_macro_pct}%\n" + f"Current Bucket Index : {completed_counter} of {total_targets_count}\n" + f"Elapsed Duration : {current_elapsed:.2f} seconds\n" + f"Active Filtering Rule : {args.filter_bucket}\n" + ) + dispatch_progress_email(args.notification, sub_msg, body_msg) + next_email_milestone_pct = ((current_macro_pct // 10) + 1) * 10 time.sleep(0.05) @@ -332,6 +376,20 @@ def main(): global_end_timestamp = datetime.now(timezone.utc).strftime(time_format)[:-3] + "Z" global_total_elapsed = round(global_end_epoch - global_start_epoch, 3) + # --- DISPATCH COMPLETION EMAIL SNAPSHOT REPORT --- + if args.notification: + sub_msg = f"purge progress 100% bucket {total_targets_count} of {total_targets_count}" + body_msg = ( + f"AWS ORGANIZATIONAL S3 METRICS STORAGE PURGE LIFECYCLE PROGRESS REPORT\n" + f"======================================================================\n" + f"Current Status : CONCLUSION (All Operations Executed Safely)\n" + f"Progress Percentage : 100%\n" + f"Total Buckets Managed : {total_targets_count} of {total_targets_count}\n" + f"Total Elapsed Duration: {global_total_elapsed:.2f} seconds\n" + f"Dry Run Active Status : {dry_run}\n" + ) + dispatch_progress_email(args.notification, sub_msg, body_msg) + if args.output != "." and not os.path.exists(args.output): os.makedirs(args.output, exist_ok=True)