""" Flask Web Interface for Image Restyling Single-file application with login and restyling form """ from flask import Flask, render_template_string, request, redirect, url_for, session, flash, jsonify, send_file from datetime import timedelta, datetime from functools import wraps from werkzeug.middleware.proxy_fix import ProxyFix from werkzeug.utils import secure_filename import os import zipfile import csv import subprocess import json import shutil from io import BytesIO app = Flask(__name__) app.secret_key = 'darveys-image-restyling-secret-key-2024' app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(days=7) # Middleware to handle URL prefix class PrefixMiddleware(object): def __init__(self, app, prefix=''): self.app = app self.prefix = prefix def __call__(self, environ, start_response): if environ['PATH_INFO'].startswith(self.prefix): environ['PATH_INFO'] = environ['PATH_INFO'][len(self.prefix):] environ['SCRIPT_NAME'] = self.prefix return self.app(environ, start_response) else: start_response('404', [('Content-Type', 'text/plain')]) return [b"Not Found"] app.wsgi_app = PrefixMiddleware(app.wsgi_app, prefix='/image-restyling') # File upload configuration UPLOAD_FOLDER = '/var/www/html/Open-AI/images_in' ALLOWED_EXTENSIONS = {'zip'} MAX_FILE_SIZE = 20 * 1024 * 1024 # 20 MB app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER app.config['MAX_CONTENT_LENGTH'] = MAX_FILE_SIZE # Hardcoded credentials USERNAME = 'darveys' PASSWORD = 'darveys@openai9876' # Helper function to check allowed file def allowed_file(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS # Login required decorator def login_required(f): @wraps(f) def decorated_function(*args, **kwargs): if 'logged_in' not in session: flash('Please login to access this page.', 'warning') return redirect(url_for('login')) return f(*args, **kwargs) return decorated_function # HTML Templates LOGIN_TEMPLATE = """ Login - Image Restyling

Image Restyling

Login to access the restyling interface

{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% endwith %}
Username is required
Password is required
""" DASHBOARD_TEMPLATE = """ Image Restyling Dashboard

Image Restyling Dashboard

Logout

Restyle Images

Configure your image restyling parameters

{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% endwith %}
0%
0/1000 characters
Prompt is required (minimum 10 characters)
Please select a quality option

Job Status Monitor

Check the processing status of your recent jobs

""" @app.route('/') def index(): if 'logged_in' in session: return redirect(url_for('dashboard')) return redirect(url_for('login')) @app.route('/login', methods=['GET', 'POST']) def login(): if 'logged_in' in session: return redirect(url_for('dashboard')) if request.method == 'POST': username = request.form.get('username', '').strip() password = request.form.get('password', '') # Server-side validation if not username or not password: flash('Username and password are required.', 'danger') return render_template_string(LOGIN_TEMPLATE) # Check credentials if username == USERNAME and password == PASSWORD: session.permanent = True session['logged_in'] = True session['username'] = username flash('Login successful! Welcome back.', 'success') return redirect(url_for('dashboard')) else: flash('Invalid username or password. Please try again.', 'danger') return render_template_string(LOGIN_TEMPLATE) return render_template_string(LOGIN_TEMPLATE) @app.route('/dashboard', methods=['GET', 'POST']) @login_required def dashboard(): if request.method == 'POST': prompt = request.form.get('prompt', '').strip() quality = request.form.get('quality', '').strip() # Server-side validation errors = [] # Check if images were uploaded if 'upload_timestamp' not in session: errors.append('Please upload images first before submitting.') if not prompt or len(prompt) < 10: errors.append('Prompt must be at least 10 characters long.') if len(prompt) > 1000: errors.append('Prompt must not exceed 1000 characters.') if quality not in ['high', 'medium']: errors.append('Please select a valid quality option.') if errors: for error in errors: flash(error, 'danger') return render_template_string(DASHBOARD_TEMPLATE) # Get timestamp from session timestamp = session['upload_timestamp'] # Setup paths csv_path = os.path.join(app.config['UPLOAD_FOLDER'], timestamp, f'catalog-{timestamp}.csv') jobs_dir = '/var/www/html/Open-AI/jobs' os.makedirs(jobs_dir, exist_ok=True) # Save job data job_data = { 'timestamp': timestamp, 'prompt': prompt, 'quality': quality, 'csv_path': csv_path, 'status': 'pending', 'submitted_at': datetime.now().strftime("%Y-%m-%d %H:%M:%S") } job_file = os.path.join(jobs_dir, f'job_{timestamp}.json') with open(job_file, 'w') as f: json.dump(job_data, f, indent=2) # Trigger background processing try: script_path = '/var/www/html/Open-AI/restyle_batch_prod.py' subprocess.Popen([ 'python3', script_path, '--prompt', prompt, '--quality', quality, '--csv', csv_path, '--timestamp', timestamp ], cwd='/var/www/html/Open-AI') flash('Your request has been submitted and processing will begin shortly.', 'success') # Clear upload timestamp from session session.pop('upload_timestamp', None) except Exception as e: flash(f'Error starting background job: {str(e)}', 'danger') return redirect(url_for('dashboard')) return render_template_string(DASHBOARD_TEMPLATE) @app.route('/upload-images', methods=['POST']) @login_required def upload_images(): """Handle ZIP file upload and processing""" try: # Check if file is present if 'zipfile' not in request.files: return jsonify({'success': False, 'message': 'No file uploaded'}), 400 file = request.files['zipfile'] # Check if file is selected if file.filename == '': return jsonify({'success': False, 'message': 'No file selected'}), 400 # Validate file type if not allowed_file(file.filename): return jsonify({'success': False, 'message': 'Only ZIP files are allowed'}), 400 # Generate timestamp for folder name timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') timestamp_folder = os.path.join(app.config['UPLOAD_FOLDER'], timestamp) # Create timestamp folder os.makedirs(timestamp_folder, exist_ok=True) # Save uploaded ZIP file temporarily filename = secure_filename(file.filename) temp_zip_path = os.path.join(timestamp_folder, filename) file.save(temp_zip_path) # Extract images from ZIP image_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp'} extracted_files = [] with zipfile.ZipFile(temp_zip_path, 'r') as zip_ref: for file_info in zip_ref.filelist: # Skip directories if file_info.is_dir(): continue # Get file extension file_ext = os.path.splitext(file_info.filename)[1].lower() # Only extract image files if file_ext in image_extensions: # Get just the filename (flatten structure) base_filename = os.path.basename(file_info.filename) # Extract directly to timestamp folder with zip_ref.open(file_info) as source: target_path = os.path.join(timestamp_folder, base_filename) with open(target_path, 'wb') as target: target.write(source.read()) extracted_files.append(base_filename) # Remove temporary ZIP file os.remove(temp_zip_path) # Generate CSV file csv_filename = f'catalog-{timestamp}.csv' csv_path = os.path.join(timestamp_folder, csv_filename) with open(csv_path, 'w', newline='') as csvfile: writer = csv.writer(csvfile) writer.writerow(['filename']) # Header for filename in sorted(extracted_files): writer.writerow([filename]) # Store timestamp in session for form submission session['upload_timestamp'] = timestamp return jsonify({ 'success': True, 'message': 'Images uploaded successfully', 'timestamp': timestamp, 'file_count': len(extracted_files) }), 200 except zipfile.BadZipFile: return jsonify({'success': False, 'message': 'Invalid ZIP file'}), 400 except Exception as e: return jsonify({'success': False, 'message': f'Error processing file: {str(e)}'}), 500 @app.route('/check-status/', methods=['GET']) @login_required def check_status(timestamp): """Check job processing status""" try: jobs_dir = '/var/www/html/Open-AI/jobs' job_file = os.path.join(jobs_dir, f'job_{timestamp}.json') if not os.path.exists(job_file): return jsonify({ 'success': False, 'message': 'Job not found' }), 404 # Read job status with open(job_file, 'r') as f: job_data = json.load(f) # Calculate progress percentage processed = job_data.get('processed', 0) total = job_data.get('total', 0) progress = int((processed / total * 100)) if total > 0 else 0 return jsonify({ 'success': True, 'status': job_data.get('status', 'unknown'), 'processed': processed, 'total': total, 'progress': progress, 'message': job_data.get('message', ''), 'updated_at': job_data.get('updated_at', '') }), 200 except Exception as e: return jsonify({ 'success': False, 'message': f'Error checking status: {str(e)}' }), 500 @app.route('/download/', methods=['GET']) @login_required def download_images(timestamp): """Download processed images as ZIP""" try: output_dir = os.path.join('/var/www/html/Open-AI/images_out', timestamp) if not os.path.exists(output_dir): flash('No processed images found for this job.', 'danger') return redirect(url_for('dashboard')) # Check if directory has files files = [f for f in os.listdir(output_dir) if os.path.isfile(os.path.join(output_dir, f))] if not files: flash('No processed images found for this job.', 'danger') return redirect(url_for('dashboard')) # Create ZIP file in memory memory_file = BytesIO() with zipfile.ZipFile(memory_file, 'w', zipfile.ZIP_DEFLATED) as zipf: for filename in files: file_path = os.path.join(output_dir, filename) zipf.write(file_path, filename) memory_file.seek(0) return send_file( memory_file, mimetype='application/zip', as_attachment=True, download_name=f'restyled-images-{timestamp}.zip' ) except Exception as e: flash(f'Error creating download: {str(e)}', 'danger') return redirect(url_for('dashboard')) @app.route('/get-recent-jobs', methods=['GET']) @login_required def get_recent_jobs(): """Get list of recent jobs for status checking""" try: jobs_dir = '/var/www/html/Open-AI/jobs' if not os.path.exists(jobs_dir): return jsonify({'success': True, 'jobs': []}), 200 job_files = [f for f in os.listdir(jobs_dir) if f.startswith('job_') and f.endswith('.json')] jobs = [] for job_file in sorted(job_files, reverse=True)[:10]: # Last 10 jobs try: with open(os.path.join(jobs_dir, job_file), 'r') as f: job_data = json.load(f) timestamp = job_file.replace('job_', '').replace('.json', '') jobs.append({ 'timestamp': timestamp, 'status': job_data.get('status', 'unknown'), 'submitted_at': job_data.get('submitted_at', '') }) except: continue return jsonify({'success': True, 'jobs': jobs}), 200 except Exception as e: return jsonify({'success': False, 'message': str(e)}), 500 @app.route('/logout') def logout(): session.clear() flash('You have been logged out successfully.', 'success') return redirect(url_for('login')) if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=5000)