#!/usr/bin/env python3 """ Batch Image Processing Script for Ghostprint ============================================= This script processes all PNG and JPEG images in a folder using the Ghostprint Invisible Armor protection. Usage: python batch_process_images.py [output_folder] Examples: # Process all images in 'my_images' folder, save to 'my_images_protected' python batch_process_images.py ./my_images # Process all images in 'my_images' folder, save to custom output folder python batch_process_images.py ./my_images ./protected_output # Process with custom settings python batch_process_images.py ./my_images --strength 4.0 --creator "My Name" """ import os import sys import base64 import json import argparse from pathlib import Path import requests # Default server URL (change if your server runs on a different port) DEFAULT_SERVER_URL = "http://localhost:7860" def encode_image_file(file_path): """Read and base64 encode an image file.""" with open(file_path, 'rb') as f: return base64.b64encode(f.read()).decode('utf-8') def find_images(folder_path): """Find all PNG and JPEG images in a folder.""" image_extensions = {'.png', '.jpg', '.jpeg', '.PNG', '.JPG', '.JPEG'} folder = Path(folder_path) images = [] for ext in image_extensions: images.extend(folder.glob(f'*{ext}')) return sorted(images) def batch_process_folder(input_folder, output_folder=None, creator="Anonymous", ai_disclaimer="", strength=3.0, focus=3.5, frequency_strategy="auto", server_url=DEFAULT_SERVER_URL, batch_size=10): """ Process all images in a folder using batch API. Args: input_folder: Path to folder containing images output_folder: Path to save protected images (defaults to input_folder + "_protected") creator: Artist/creator name ai_disclaimer: Optional AI disclaimer text strength: Armor strength (1.0-10.0) focus: Focus parameter (1.0-10.0) frequency_strategy: 'auto', 'low', 'high', 'hybrid', or 'scramble' server_url: URL of the Ghostprint server batch_size: Number of images to process per API call """ # Validate input folder input_path = Path(input_folder) if not input_path.exists() or not input_path.is_dir(): print(f"Error: Input folder '{input_folder}' does not exist or is not a directory.") return False # Set up output folder if output_folder is None: output_folder = str(input_path) + "_protected" output_path = Path(output_folder) output_path.mkdir(parents=True, exist_ok=True) # Find all images images = find_images(input_folder) if not images: print(f"No PNG or JPEG images found in '{input_folder}'") return False print(f"\n{'='*60}") print(f"Ghostprint Batch Image Protection") print(f"{'='*60}") print(f"Input folder: {input_folder}") print(f"Output folder: {output_folder}") print(f"Found {len(images)} images") print(f"Server URL: {server_url}") print(f"Strength: {strength}") print(f"Focus: {focus}") print(f"Strategy: {frequency_strategy}") print(f"{'='*60}\n") # Process images in batches total_successful = 0 total_failed = 0 for i in range(0, len(images), batch_size): batch = images[i:i+batch_size] batch_num = (i // batch_size) + 1 total_batches = (len(images) + batch_size - 1) // batch_size print(f"\nProcessing batch {batch_num}/{total_batches} ({len(batch)} images)...") # Prepare batch request images_data = [] for img_path in batch: try: file_content_b64 = encode_image_file(img_path) images_data.append({ "file_content": file_content_b64, "filename": img_path.name }) except Exception as e: print(f" ✗ Failed to read {img_path.name}: {e}") total_failed += 1 if not images_data: continue # Make API request try: payload = { "images": images_data, "creator": creator, "ai_disclaimer": ai_disclaimer, "strength": strength, "focus": focus, "frequency_strategy": frequency_strategy } response = requests.post( f"{server_url}/batch-protect-images", json=payload, timeout=300 # 5 minute timeout for batch processing ) if response.status_code != 200: print(f" ✗ Batch request failed: HTTP {response.status_code}") print(f" {response.text}") total_failed += len(images_data) continue result = response.json() if not result.get('success'): print(f" ✗ Batch processing failed: {result.get('error')}") total_failed += len(images_data) continue # Save results for item in result['results']: if item['success']: try: output_file = output_path / item['filename'] output_data = base64.b64decode(item['output_file_content']) with open(output_file, 'wb') as f: f.write(output_data) print(f" ✓ {item['original_filename']} → {item['filename']}") # Show format conversion info if item.get('format_converted'): print(f" (Converted to PNG for optimal protection)") total_successful += 1 except Exception as e: print(f" ✗ Failed to save {item['filename']}: {e}") total_failed += 1 else: print(f" ✗ {item['filename']}: {item.get('error')}") total_failed += 1 except requests.exceptions.Timeout: print(f" ✗ Batch request timed out (try reducing batch size)") total_failed += len(images_data) except Exception as e: print(f" ✗ Batch request failed: {e}") total_failed += len(images_data) # Summary print(f"\n{'='*60}") print(f"Processing Complete") print(f"{'='*60}") print(f"Total images: {len(images)}") print(f"Successful: {total_successful}") print(f"Failed: {total_failed}") print(f"Output folder: {output_folder}") print(f"{'='*60}\n") return total_failed == 0 def main(): parser = argparse.ArgumentParser( description="Batch process images with Ghostprint Invisible Armor", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Basic usage python batch_process_images.py ./my_images # Custom output folder python batch_process_images.py ./my_images ./protected_output # Custom settings python batch_process_images.py ./my_images --strength 4.0 --creator "Artist Name" # Use different server python batch_process_images.py ./my_images --server http://localhost:8080 """ ) parser.add_argument('input_folder', help='Folder containing images to protect') parser.add_argument('output_folder', nargs='?', help='Folder to save protected images (optional)') parser.add_argument('--creator', default='Anonymous', help='Artist/creator name') parser.add_argument('--disclaimer', default='', help='AI disclaimer text') parser.add_argument('--strength', type=float, default=3.0, help='Armor strength (1.0-10.0, default: 3.0)') parser.add_argument('--focus', type=float, default=3.5, help='Focus parameter (1.0-10.0, default: 3.5)') parser.add_argument('--strategy', default='auto', choices=['auto', 'low', 'high', 'hybrid', 'scramble'], help='Frequency strategy (default: auto)') parser.add_argument('--server', default=DEFAULT_SERVER_URL, help=f'Server URL (default: {DEFAULT_SERVER_URL})') parser.add_argument('--batch-size', type=int, default=10, help='Number of images per batch (default: 10)') args = parser.parse_args() # Check if server is running try: response = requests.get(args.server, timeout=5) print(f"✓ Connected to Ghostprint server at {args.server}") except Exception as e: print(f"✗ Cannot connect to Ghostprint server at {args.server}") print(f" Make sure the server is running: python server.py") return 1 # Process images success = batch_process_folder( input_folder=args.input_folder, output_folder=args.output_folder, creator=args.creator, ai_disclaimer=args.disclaimer, strength=args.strength, focus=args.focus, frequency_strategy=args.strategy, server_url=args.server, batch_size=args.batch_size ) return 0 if success else 1 if __name__ == '__main__': sys.exit(main())