import os import argparse import boto3 from botocore.exceptions import NoCredentialsError, ClientError import logging from datetime import datetime, timezone from urllib.parse import quote import tkinter as tk from tkinter import filedialog, messagebox, scrolledtext, ttk import threading import queue # --- S3 Logic (Adapted from previous scripts) --- # Note: Logging is now redirected to the GUI's text area. class QueueHandler(logging.Handler): """A custom logging handler that puts messages into a queue.""" def __init__(self, log_queue): super().__init__() self.log_queue = log_queue def emit(self, record): self.log_queue.put(self.format(record)) def get_s3_client(log_queue): """Initializes and returns a Boto3 S3 client.""" try: s3_client = boto3.client('s3') s3_client.list_buckets() return s3_client except NoCredentialsError: log_queue.put("ERROR: AWS credentials not found. Please configure them.") return None except ClientError as e: log_queue.put(f"ERROR: An AWS client error occurred: {e}") return None def get_bucket_region(s3_client, bucket_name, log_queue): """Retrieves the AWS region for a bucket.""" try: response = s3_client.get_bucket_location(Bucket=bucket_name) region = response.get('LocationConstraint') return region if region is not None else 'us-east-1' except ClientError as e: log_queue.put(f"ERROR: Could not get bucket region: {e}") return None def sync_folder_to_s3(local_folder, bucket_name, delete_extra_files, log_queue): """Syncs a local folder to an S3 bucket.""" s3_client = get_s3_client(log_queue) if not s3_client: return if not os.path.isdir(local_folder): log_queue.put(f"ERROR: Local directory not found: {local_folder}") return log_queue.put(f"Starting sync from '{local_folder}' to S3 bucket '{bucket_name}'...") try: paginator = s3_client.get_paginator('list_objects_v2') s3_objects = {obj['Key']: obj['LastModified'] for page in paginator.paginate(Bucket=bucket_name) if "Contents" in page for obj in page['Contents']} except ClientError as e: log_queue.put(f"ERROR: Could not list S3 objects: {e}") return local_files, upload_count, skip_count = set(), 0, 0 for root, _, files in os.walk(local_folder): for filename in files: local_path = os.path.join(root, filename) s3_key = os.path.relpath(local_path, local_folder).replace(os.path.sep, '/') local_files.add(s3_key) local_mtime = datetime.fromtimestamp(os.path.getmtime(local_path), tz=timezone.utc) if s3_key not in s3_objects or local_mtime > s3_objects[s3_key]: try: log_queue.put(f"Uploading: {s3_key}") s3_client.upload_file(local_path, bucket_name, s3_key) upload_count += 1 except ClientError as e: log_queue.put(f"ERROR: Failed to upload {s3_key}: {e}") else: skip_count += 1 delete_count = 0 if delete_extra_files: log_queue.put("Checking for files to delete from S3...") to_delete = [{'Key': key} for key in s3_objects if key not in local_files] if to_delete: for i in range(0, len(to_delete), 1000): chunk = to_delete[i:i + 1000] log_queue.put(f"Deleting {len(chunk)} files from S3...") s3_client.delete_objects(Bucket=bucket_name, Delete={'Objects': chunk}) delete_count += len(chunk) log_queue.put("="*30 + "\nSync Summary\n" + "="*30) log_queue.put(f" - Uploaded: {upload_count} file(s)") log_queue.put(f" - Skipped: {skip_count} file(s) (up-to-date)") if delete_extra_files: log_queue.put(f" - Deleted: {delete_count} file(s) from S3") log_queue.put("Sync complete.") def list_s3_buckets(log_queue=None): """Lists all available S3 buckets.""" try: s3_client = boto3.client('s3') response = s3_client.list_buckets() buckets = [bucket['Name'] for bucket in response.get('Buckets', [])] return buckets except NoCredentialsError: if log_queue: log_queue.put("ERROR: AWS credentials not found. Please configure them.") return [] except ClientError as e: if log_queue: log_queue.put(f"ERROR: Could not list buckets: {e}") return [] def list_files_and_generate_urls(bucket_name, log_queue): """Lists files in an S3 bucket and generates their URLs.""" s3_client = get_s3_client(log_queue) if not s3_client: return region = get_bucket_region(s3_client, bucket_name, log_queue) if not region: return log_queue.put(f"Bucket is in region: {region}") log_queue.put("Listing files and generating URLs...") base_url = f"https://{bucket_name}.s3.{region}.amazonaws.com/" file_count = 0 try: paginator = s3_client.get_paginator('list_objects_v2') for page in paginator.paginate(Bucket=bucket_name): if "Contents" in page: for obj in page['Contents']: encoded_key = quote(obj['Key']) file_url = f"{base_url}{encoded_key}" log_queue.put(f"File: {obj['Key']}\nURL: {file_url}\n") file_count += 1 log_queue.put("="*30) log_queue.put(f"Found {file_count} file(s) in '{bucket_name}'.") except ClientError as e: log_queue.put(f"ERROR: An error occurred: {e}") # --- GUI Application --- class S3ToolApp: def __init__(self, root): self.root = root self.root.title("S3 Sync & Lister Tool") self.root.geometry("750x600") self.log_queue = queue.Queue() # --- UI Frames --- control_frame = tk.Frame(root, padx=10, pady=10) control_frame.pack(fill='x') sync_frame = tk.LabelFrame(control_frame, text="Sync Local Folder to S3", padx=10, pady=10) sync_frame.pack(fill='x', expand=True, side='left', padx=(0, 5)) list_frame = tk.LabelFrame(control_frame, text="List S3 Bucket Files", padx=10, pady=10) list_frame.pack(fill='x', expand=True, side='right', padx=(5, 0)) output_frame = tk.Frame(root, padx=10, pady=10) output_frame.pack(fill='both', expand=True) # --- Common Widgets --- bucket_label_frame = tk.Frame(sync_frame) bucket_label_frame.pack(fill='x', pady=(0, 5)) tk.Label(bucket_label_frame, text="S3 Bucket Name:").pack(side='left') self.refresh_button = tk.Button(bucket_label_frame, text="Refresh Buckets", command=self.refresh_buckets) self.refresh_button.pack(side='right') self.bucket_combobox = ttk.Combobox(sync_frame, width=37, state='readonly') self.bucket_combobox.pack(fill='x', pady=(0, 10)) # Load buckets on startup self.refresh_buckets() # --- Sync Widgets --- folder_frame = tk.Frame(sync_frame) folder_frame.pack(fill='x', pady=(0, 10)) tk.Label(folder_frame, text="Local Folder:").pack(side='left') self.folder_path = tk.StringVar() tk.Entry(folder_frame, textvariable=self.folder_path, width=30).pack(side='left', fill='x', expand=True) tk.Button(folder_frame, text="Browse...", command=self.browse_folder).pack(side='right') self.delete_var = tk.BooleanVar() tk.Checkbutton(sync_frame, text="Delete files in S3 not present locally", variable=self.delete_var).pack(anchor='w') self.sync_button = tk.Button(sync_frame, text="Sync to S3", command=self.start_sync_thread) self.sync_button.pack(pady=(10,0)) # --- List Widgets --- tk.Label(list_frame, text="Select bucket from dropdown in left panel.").pack(pady=(15,0)) self.list_button = tk.Button(list_frame, text="List Files & Generate URLs", command=self.start_list_thread) self.list_button.pack(pady=10) # --- Output Text Area --- self.log_text = scrolledtext.ScrolledText(output_frame, state='disabled', wrap=tk.WORD, height=20) self.log_text.pack(fill='both', expand=True) self.root.after(100, self.process_queue) def browse_folder(self): folder = filedialog.askdirectory() if folder: self.folder_path.set(folder) def refresh_buckets(self): """Refresh the list of available S3 buckets.""" self.bucket_combobox.config(state='normal') self.bucket_combobox.set('Loading buckets...') self.bucket_combobox.config(state='disabled') self.refresh_button.config(state='disabled') def load_buckets(): buckets = list_s3_buckets(self.log_queue) self.root.after(0, lambda: self.update_bucket_list(buckets)) thread = threading.Thread(target=load_buckets) thread.daemon = True thread.start() def update_bucket_list(self, buckets): """Update the combobox with the list of buckets.""" self.bucket_combobox.config(state='normal') if buckets: self.bucket_combobox['values'] = buckets self.bucket_combobox.set('') self.bucket_combobox.config(state='readonly') else: self.bucket_combobox.set('No buckets found or error') self.bucket_combobox['values'] = [] self.bucket_combobox.config(state='disabled') self.refresh_button.config(state='normal') def start_sync_thread(self): local_folder = self.folder_path.get() bucket_name = self.bucket_combobox.get() if not local_folder or not bucket_name: messagebox.showerror("Error", "Please provide both a local folder and a bucket name.") return self.toggle_buttons(False) thread = threading.Thread(target=self.run_task, args=(sync_folder_to_s3, local_folder, bucket_name, self.delete_var.get(), self.log_queue)) thread.daemon = True thread.start() def start_list_thread(self): bucket_name = self.bucket_combobox.get() if not bucket_name: messagebox.showerror("Error", "Please select a bucket name from the dropdown.") return self.toggle_buttons(False) thread = threading.Thread(target=self.run_task, args=(list_files_and_generate_urls, bucket_name, self.log_queue)) thread.daemon = True thread.start() def run_task(self, task_func, *args): self.clear_log() try: task_func(*args) finally: self.log_queue.put("TASK_COMPLETE") # Signal to re-enable buttons def toggle_buttons(self, enabled): state = 'normal' if enabled else 'disabled' self.sync_button.config(state=state) self.list_button.config(state=state) def process_queue(self): try: while True: msg = self.log_queue.get_nowait() if msg == "TASK_COMPLETE": self.toggle_buttons(True) else: self.log_text.config(state='normal') self.log_text.insert(tk.END, msg + '\n') self.log_text.config(state='disabled') self.log_text.see(tk.END) except queue.Empty: pass self.root.after(100, self.process_queue) def clear_log(self): self.log_text.config(state='normal') self.log_text.delete(1.0, tk.END) self.log_text.config(state='disabled') if __name__ == "__main__": root = tk.Tk() app = S3ToolApp(root) root.mainloop()