#!/usr/bin/env python3 """ Fetch files from SeaweedFS S3-compatible storage. This script downloads files from SeaweedFS to the local filesystem. Usage: # Download a single file python scripts/fetch_from_seaweedfs.py documents/my-file.xlsx # Download to a specific local path python scripts/fetch_from_seaweedfs.py documents/my-file.xlsx -o ./downloads/ # Download from a specific bucket python scripts/fetch_from_seaweedfs.py my-file.xlsx --bucket my-bucket # List files in a bucket python scripts/fetch_from_seaweedfs.py --list --bucket documents # List files with a prefix python scripts/fetch_from_seaweedfs.py --list --bucket documents --prefix images/ # Dry run (show what would be downloaded) python scripts/fetch_from_seaweedfs.py documents/my-file.xlsx --dry-run """ import argparse import os import sys from pathlib import Path import boto3 from botocore.config import Config from dotenv import load_dotenv # Load environment variables load_dotenv() def get_s3_client(): """Create S3 client for SeaweedFS.""" # Use SEAWEEDFS_ENDPOINT if set, otherwise build from host/port endpoint = os.getenv('SEAWEEDFS_ENDPOINT') if not endpoint: host = os.getenv('SEAWEEDFS_HOST', 'localhost') port = os.getenv('SEAWEEDFS_PORT', '8333') # Check SSL settings use_ssl = os.getenv('SEAWEEDFS_USE_SSL', '').lower() == 'true' or \ os.getenv('S3_USE_HTTPS', '').lower() == 'true' protocol = 'https' if use_ssl else 'http' endpoint = f"{protocol}://{host}:{port}" access_key = os.getenv('SEAWEEDFS_ACCESS_KEY_ID', 'dummy') secret_key = os.getenv('SEAWEEDFS_SECRET_ACCESS_KEY', 'dummy') print(f"Connecting to SeaweedFS at: {endpoint}") return boto3.client( 's3', endpoint_url=endpoint, aws_access_key_id=access_key, aws_secret_access_key=secret_key, config=Config(signature_version='s3v4'), region_name='us-east-1', ) def list_objects(s3_client, bucket: str, prefix: str = '') -> list[dict]: """List objects in a bucket with optional prefix.""" try: paginator = s3_client.get_paginator('list_objects_v2') pages = paginator.paginate(Bucket=bucket, Prefix=prefix) objects = [] for page in pages: if 'Contents' in page: objects.extend(page['Contents']) return objects except Exception as e: print(f"Error listing objects: {e}") return [] def download_file( s3_client, bucket: str, key: str, output_path: Path, dry_run: bool = False, ) -> bool: """Download a single file from SeaweedFS.""" # Determine local filename if output_path.is_dir(): local_file = output_path / Path(key).name else: local_file = output_path # Ensure parent directory exists local_file.parent.mkdir(parents=True, exist_ok=True) if dry_run: print(f"[DRY RUN] Would download:") print(f" Source: s3://{bucket}/{key}") print(f" Destination: {local_file}") return True try: # Get object metadata first head = s3_client.head_object(Bucket=bucket, Key=key) size = head.get('ContentLength', 0) content_type = head.get('ContentType', 'unknown') print(f"Downloading: s3://{bucket}/{key}") print(f" Size: {size:,} bytes") print(f" Content-Type: {content_type}") # Download file s3_client.download_file(bucket, key, str(local_file)) print(f" Saved to: {local_file}") return True except s3_client.exceptions.NoSuchKey: print(f"Error: Object not found: s3://{bucket}/{key}") return False except Exception as e: print(f"Error downloading {key}: {e}") return False def download_with_prefix( s3_client, bucket: str, prefix: str, output_dir: Path, dry_run: bool = False, ) -> tuple[int, int]: """Download all files matching a prefix.""" objects = list_objects(s3_client, bucket, prefix) if not objects: print(f"No objects found with prefix: {prefix}") return 0, 0 print(f"Found {len(objects)} object(s) with prefix '{prefix}'") success_count = 0 for obj in objects: key = obj['Key'] # Preserve directory structure relative to prefix relative_path = key[len(prefix):].lstrip('/') local_path = output_dir / relative_path if relative_path else output_dir / Path(key).name if download_file(s3_client, bucket, key, local_path, dry_run): success_count += 1 return success_count, len(objects) def get_file_as_bytes(s3_client, bucket: str, key: str) -> bytes | None: """Fetch a file and return its contents as bytes.""" try: response = s3_client.get_object(Bucket=bucket, Key=key) return response['Body'].read() except s3_client.exceptions.NoSuchKey: print(f"Error: Object not found: s3://{bucket}/{key}") return None except Exception as e: print(f"Error fetching {key}: {e}") return None def format_size(size_bytes: int) -> str: """Format file size in human-readable form.""" for unit in ['B', 'KB', 'MB', 'GB']: if size_bytes < 1024: return f"{size_bytes:.1f} {unit}" size_bytes /= 1024 return f"{size_bytes:.1f} TB" def main(): parser = argparse.ArgumentParser( description='Fetch files from SeaweedFS S3-compatible storage', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument( 'key', nargs='?', help='Object key (path) to download', ) parser.add_argument( '--bucket', '-b', default='documents', help='S3 bucket name (default: documents)', ) parser.add_argument( '--output', '-o', type=Path, default=Path('.'), help='Output path (file or directory, default: current directory)', ) parser.add_argument( '--list', '-l', action='store_true', help='List objects in the bucket', ) parser.add_argument( '--prefix', '-p', default='', help='Prefix filter for listing or batch download', ) parser.add_argument( '--recursive', '-r', action='store_true', help='Download all files matching the prefix', ) parser.add_argument( '--dry-run', '-n', action='store_true', help='Show what would be downloaded without actually downloading', ) parser.add_argument( '--stdout', action='store_true', help='Output file contents to stdout (useful for piping)', ) args = parser.parse_args() # Create S3 client try: s3_client = get_s3_client() except Exception as e: print(f"Failed to create S3 client: {e}") sys.exit(1) # List mode if args.list: print(f"\nListing objects in bucket: {args.bucket}") if args.prefix: print(f"Prefix filter: {args.prefix}") print("-" * 60) objects = list_objects(s3_client, args.bucket, args.prefix) if not objects: print("No objects found.") sys.exit(0) total_size = 0 for obj in objects: size = obj.get('Size', 0) total_size += size modified = obj.get('LastModified', '') modified_str = modified.strftime('%Y-%m-%d %H:%M:%S') if modified else 'N/A' print(f" {format_size(size):>10} {modified_str} {obj['Key']}") print("-" * 60) print(f"Total: {len(objects)} object(s), {format_size(total_size)}") sys.exit(0) # Download mode requires key or recursive with prefix if not args.key and not (args.recursive and args.prefix): parser.print_help() print("\nError: Provide an object key to download, or use --list to list objects") sys.exit(1) # Output to stdout if args.stdout: if not args.key: print("Error: --stdout requires a specific object key") sys.exit(1) content = get_file_as_bytes(s3_client, args.bucket, args.key) if content is None: sys.exit(1) sys.stdout.buffer.write(content) sys.exit(0) # Recursive download with prefix if args.recursive and args.prefix: success, total = download_with_prefix( s3_client, args.bucket, args.prefix, args.output, args.dry_run, ) print(f"\nDownloaded: {success}/{total} files") sys.exit(0 if success == total else 1) # Single file download if args.key: success = download_file( s3_client, args.bucket, args.key, args.output, args.dry_run, ) sys.exit(0 if success else 1) if __name__ == '__main__': main()