add --notification

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,15 +347,49 @@ 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)
 
     global_end_epoch = time.time()
     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)