from flask import Flask, render_template, send_from_directory, jsonify, request, send_file, session, redirect, url_for import os from io import BytesIO import base64 import json from werkzeug.security import generate_password_hash, check_password_hash from functools import wraps import secrets # Import image generator try: from image_generator import ProductImageGenerator, image_to_base64 IMAGE_GENERATOR_AVAILABLE = True print("✓ Image generation available") except ImportError as e: IMAGE_GENERATOR_AVAILABLE = False print(f"Warning: Could not import image generator: {e}") print("Image generation endpoints will be disabled.") except Exception as e: IMAGE_GENERATOR_AVAILABLE = False print(f"Error loading image generator: {e}") print("Image generation endpoints will be disabled.") # Get the directory where this script is located basedir = os.path.abspath(os.path.dirname(__file__)) # Initialize Flask with explicit paths app = Flask(__name__, template_folder=os.path.join(basedir, 'templates'), static_folder=os.path.join(basedir, 'static')) # Configure session - generate a random secret key app.secret_key = secrets.token_hex(32) app.config['SESSION_COOKIE_HTTPONLY'] = True app.config['SESSION_COOKIE_SAMESITE'] = 'Lax' # Configure static folders app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0 # Disable caching during development # Support for subdirectory deployments (e.g., /product-finder/) # Auto-detect production environment based on Python path import sys if 'virtualenv/product-finder' in sys.executable or '/home/bmdwtjuw' in sys.executable: # Running on production server app.config['APPLICATION_ROOT'] = '/product-finder' print("[OK] Production environment detected - using /product-finder") else: # Local development app.config['APPLICATION_ROOT'] = os.environ.get('APPLICATION_ROOT', '/') print(f"[OK] Development environment - using {app.config['APPLICATION_ROOT']}") # ===== SUBDIRECTORY DEPLOYMENT MIDDLEWARE ===== class PrefixMiddleware: """ Middleware to handle subdirectory deployments. This ensures Flask knows about the /product-finder prefix when deployed. """ def __init__(self, app, prefix=''): self.app = app self.prefix = prefix.rstrip('/') # Remove trailing slash if present def __call__(self, environ, start_response): # Only apply prefix if not already in SCRIPT_NAME and prefix is set if self.prefix and self.prefix != '/': # Check if the URL path starts with our prefix path = environ.get('PATH_INFO', '') script_name = environ.get('SCRIPT_NAME', '') # If prefix not already in SCRIPT_NAME, add it if not script_name.startswith(self.prefix): environ['SCRIPT_NAME'] = self.prefix + script_name # Remove prefix from PATH_INFO if it's there if path.startswith(self.prefix): environ['PATH_INFO'] = path[len(self.prefix):] return self.app(environ, start_response) # Apply middleware if APPLICATION_ROOT is set to a subdirectory application_root = app.config.get('APPLICATION_ROOT', '/') if application_root and application_root != '/': app.wsgi_app = PrefixMiddleware(app.wsgi_app, prefix=application_root) print(f"✓ PrefixMiddleware applied for subdirectory: {application_root}") # Make base URL available to all templates @app.context_processor def inject_base_url(): """Inject base URL into all templates for relative paths""" return { 'base_url': request.script_root or '', 'url_for': url_for } # ===== AUTHENTICATION HELPERS ===== def login_required(f): """Decorator to require login for routes""" @wraps(f) def decorated_function(*args, **kwargs): if 'user_id' not in session: return redirect(url_for('login_page')) # Check if user is active users = load_users() if session['user_id'] >= len(users): session.clear() return redirect(url_for('login_page', message='User not found')) user = users[session['user_id']] if not user.get('active', True): session.clear() return redirect(url_for('login_page', message='Account is inactive')) return f(*args, **kwargs) return decorated_function def get_current_user(): """Get the currently logged in user""" if 'user_id' not in session: return None users = load_users() if session['user_id'] >= len(users): return None return users[session['user_id']] def can_user(permission, location=None): """ Check if current user has a specific permission. WordPress-style permission checking. Args: permission (str): Permission name (e.g., 'manage_users', 'create_quotes') location (str, optional): Check permission for specific location. If None, uses current session location. If 'global', checks global permissions only. Returns: bool: True if user has permission, False otherwise Examples: can_user('manage_users') # Check global permission can_user('create_quotes') # Check at current location can_user('view_reports', 'LINDS') # Check at specific location """ user = get_current_user() if not user: return False # Check if user is active if not user.get('active', True): return False # Check global permissions first global_permissions = user.get('permissions', {}) if permission in global_permissions: return global_permissions[permission] is True # If location is 'global', only check global permissions if location == 'global': return False # Determine which location to check check_location = location if location else session.get('currentLocation') if not check_location: return False # Check location-specific permissions location_settings = user.get('locationSettings', {}) if check_location in location_settings: loc_permissions = location_settings[check_location].get('permissions', {}) if permission in loc_permissions: return loc_permissions[permission] is True return False def user_has_any_permission(permissions, location=None): """ Check if user has ANY of the provided permissions. Args: permissions (list): List of permission names location (str, optional): Location to check Returns: bool: True if user has at least one permission """ return any(can_user(perm, location) for perm in permissions) def user_has_all_permissions(permissions, location=None): """ Check if user has ALL of the provided permissions. Args: permissions (list): List of permission names location (str, optional): Location to check Returns: bool: True if user has all permissions """ return all(can_user(perm, location) for perm in permissions) def permission_required(permission, location=None): """ Decorator to require specific permission for a route. Similar to @login_required but checks permissions. Args: permission (str): Required permission name location (str, optional): Location to check permission for Example: @app.route('/admin/users') @login_required @permission_required('manage_users') def manage_users_page(): return render_template('user_manager.html') """ def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): if not can_user(permission, location): return render_template('access_denied.html', required_permission=permission), 403 return f(*args, **kwargs) return decorated_function return decorator def get_user_permissions(location=None): """ Get all permissions for the current user. Args: location (str, optional): Get permissions for specific location. If None, returns global + current location. Returns: dict: Dictionary of permission_name: True/False """ user = get_current_user() if not user: return {} permissions = {} # Get global permissions global_perms = user.get('permissions', {}) permissions.update(global_perms) # Get location-specific permissions if location != 'global': check_location = location if location else session.get('currentLocation') if check_location: location_settings = user.get('locationSettings', {}) if check_location in location_settings: loc_perms = location_settings[check_location].get('permissions', {}) permissions.update(loc_perms) return permissions # ===== USER MANAGEMENT FUNCTIONS (moved here for use in auth) ===== USERS_FILE = os.path.join(basedir, 'data', 'users.json') def load_users(): """Load users from JSON file""" if os.path.exists(USERS_FILE): try: with open(USERS_FILE, 'r') as f: return json.load(f) except: return [] return [] def save_users(users): """Save users to JSON file""" with open(USERS_FILE, 'w') as f: json.dump(users, f, indent=2) # ===== AUTHENTICATION ROUTES ===== @app.route('/login') def login_page(): """Serve the login page""" # If already logged in, redirect to main app if 'user_id' in session: return redirect(url_for('index')) return render_template('login.html') @app.route('/select-location') def select_location_page(): """Serve the location selection page""" if 'user_id' not in session: return redirect(url_for('login_page')) return render_template('select_location.html') @app.route('/api/login', methods=['POST']) def login(): """Authenticate user and create session""" try: data = request.json username = data.get('username') password = data.get('password') if not username or not password: return jsonify({ 'status': 'error', 'message': 'Username and password are required' }), 400 users = load_users() # Find user by username user_index = None user = None for i, u in enumerate(users): if u['username'] == username: user_index = i user = u break if not user: return jsonify({ 'status': 'error', 'message': 'Invalid username or password' }), 401 # Check if user is active if not user.get('active', True): return jsonify({ 'status': 'error', 'message': 'Account is inactive. Please contact an administrator.' }), 403 # Verify password if not check_password_hash(user['password'], password): return jsonify({ 'status': 'error', 'message': 'Invalid username or password' }), 401 # Create session session['user_id'] = user_index session['username'] = user['username'] session['defaultLocation'] = user.get('defaultLocation') # Get accessible locations accessible_locations = [] if 'locationSettings' in user: for loc_code, settings in user['locationSettings'].items(): if settings.get('accessible', False): accessible_locations.append(loc_code) session['accessibleLocations'] = accessible_locations # Check if user needs to select a location requiresLocationSelection = len(accessible_locations) > 1 # If only one location or no accessible locations, auto-select default if not requiresLocationSelection: session['currentLocation'] = user.get('defaultLocation') return jsonify({ 'status': 'success', 'message': 'Login successful', 'requiresLocationSelection': requiresLocationSelection }) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 @app.route('/api/logout', methods=['POST']) def logout(): """Log out user and clear session""" session.clear() return jsonify({ 'status': 'success', 'message': 'Logged out successfully' }) @app.route('/api/session', methods=['GET']) def get_session(): """Get current session information""" if 'user_id' not in session: return jsonify({ 'status': 'error', 'message': 'Not logged in' }), 401 return jsonify({ 'status': 'success', 'username': session.get('username'), 'defaultLocation': session.get('defaultLocation'), 'currentLocation': session.get('currentLocation'), 'accessibleLocations': session.get('accessibleLocations', []) }) @app.route('/api/select-location', methods=['POST']) def select_location(): """Select a location for the current session""" if 'user_id' not in session: return jsonify({ 'status': 'error', 'message': 'Not logged in' }), 401 try: data = request.json location = data.get('location') # Verify user has access to this location accessible = session.get('accessibleLocations', []) if location not in accessible: return jsonify({ 'status': 'error', 'message': 'You do not have access to this location' }), 403 session['currentLocation'] = location return jsonify({ 'status': 'success', 'message': 'Location selected', 'currentLocation': location }) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 # ===== PERMISSION CHECK ENDPOINT ===== @app.route('/api/check-permission', methods=['POST']) @login_required def check_permission(): """Check if current user has specific permission(s)""" try: data = request.json permission = data.get('permission') location = data.get('location') # Optional, defaults to current location if not permission: return jsonify({ 'status': 'error', 'message': 'Permission name required' }), 400 has_permission = can_user(permission, location) return jsonify({ 'status': 'success', 'hasPermission': has_permission, 'permission': permission, 'location': location if location else session.get('currentLocation', 'global') }) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 @app.route('/api/user-permissions', methods=['GET']) @login_required def get_user_permissions_endpoint(): """Get all permissions for current user""" try: location = request.args.get('location') # Optional query parameter permissions = get_user_permissions(location) return jsonify({ 'status': 'success', 'permissions': permissions, 'location': location if location else session.get('currentLocation', 'global') }) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 # ===== MAIN APPLICATION ROUTES ===== @app.route('/') def index(): """Serve the quiz page as the main page or redirect to login""" # Simple test to see if app is running if 'username' not in session: return redirect(url_for('login_page')) # If user hasn't selected a location yet, redirect to selection if not session.get('currentLocation') and len(session.get('accessibleLocations', [])) > 1: return redirect(url_for('select_location_page')) return render_template('index2.html') @app.route('/test') def test_route(): """Simple test route to verify app is running - NO AUTH REQUIRED""" import sys info = { 'status': 'OK', 'message': 'Flask app is running!', 'python': sys.version, 'application_root': app.config.get('APPLICATION_ROOT'), 'middleware': type(app.wsgi_app).__name__, 'routes': len(list(app.url_map.iter_rules())) } return f""" App Test

✓ Flask App is Running!

Go to Login

""" @app.route('/quiz') @login_required def quiz(): """Serve the quiz page""" return render_template('index2.html') @app.route('/image-test') @login_required def image_test(): """Serve the image generation test page""" return render_template('image_test.html') @app.route('/css/') def serve_css(filename): """Serve CSS files""" return send_from_directory(os.path.join(basedir, 'css'), filename) @app.route('/js/') def serve_js(filename): """Serve JavaScript files""" return send_from_directory(os.path.join(basedir, 'js'), filename) @app.route('/images/') def serve_images(filename): """Serve image files""" return send_from_directory(os.path.join(basedir, 'images'), filename) @app.route('/data/') def serve_data(filename): """Serve data files (JSON)""" return send_from_directory(os.path.join(basedir, 'data'), filename) # API endpoint for future expansion (e.g., saving user selections) @app.route('/api/save-selection', methods=['POST']) def save_selection(): """API endpoint to save user selections""" data = request.json # Here you could save to a database or file # For now, just return success return jsonify({ 'status': 'success', 'message': 'Selection saved', 'data': data }) @app.route('/api/get-products', methods=['GET']) def get_products(): """API endpoint to get product data""" # This could fetch from a database in the future return jsonify({ 'status': 'success', 'products': [] }) # ===== DYNAMIC IMAGE GENERATION ENDPOINTS ===== @app.route('/api/product-image/') def generate_product_image(product_code): """ Generate a product image with specified configuration. URL Parameters: color: Color name (default: 'white') hinge: Hinge side - 'left' or 'right' (default: 'right') material: Material type (default: 'aluminum') format: Return format - 'image' or 'json' (default: 'image') Examples: /api/product-image/cobrai?color=black&hinge=left /api/product-image/cobrai?color=bronze&hinge=right&format=json """ if not IMAGE_GENERATOR_AVAILABLE: return jsonify({ 'error': 'Image generation not available. Please install Pillow: pip install Pillow' }), 500 try: # Get parameters color = request.args.get('color', 'white').lower() hinge = request.args.get('hinge', 'right').lower() material = request.args.get('material', 'aluminum').lower() return_format = request.args.get('format', 'image').lower() use_cache = request.args.get('cache', 'true').lower() == 'true' # Initialize generator generator = ProductImageGenerator() # Generate image image = generator.generate_product_image( product_code=product_code, color=color, hinge=hinge, material=material, use_cache=use_cache ) # Return based on format if return_format == 'json': # Return base64 encoded image in JSON img_base64 = image_to_base64(image, format='PNG') return jsonify({ 'status': 'success', 'productCode': product_code, 'color': color, 'hinge': hinge, 'material': material, 'image': img_base64, 'format': 'base64' }) else: # Return raw image file img_io = BytesIO() image.save(img_io, 'PNG', optimize=True) img_io.seek(0) return send_file(img_io, mimetype='image/png') except ValueError as e: return jsonify({'error': str(e)}), 404 except FileNotFoundError as e: return jsonify({'error': str(e)}), 404 except Exception as e: return jsonify({'error': f'Error generating image: {str(e)}'}), 500 @app.route('/api/product-config/') def get_product_image_config(product_code): """ Get the image configuration for a product. Returns JSON with available colors, hardware positions, and other metadata. Example: /api/product-config/cobrai """ if not IMAGE_GENERATOR_AVAILABLE: return jsonify({ 'error': 'Image generation not available. Please install Pillow: pip install Pillow' }), 500 try: generator = ProductImageGenerator() config = generator.get_product_config(product_code) if not config: return jsonify({ 'error': f'Configuration for product {product_code} not found' }), 404 return jsonify({ 'status': 'success', 'productCode': product_code, 'config': config }) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/clear-image-cache', methods=['POST']) def clear_image_cache(): """ Clear the image cache. Optional JSON body: { "productCode": "cobrai" } - Clear cache for specific product Example: POST /api/clear-image-cache POST /api/clear-image-cache with body: {"productCode": "cobrai"} """ if not IMAGE_GENERATOR_AVAILABLE: return jsonify({ 'error': 'Image generation not available.' }), 500 try: data = request.json or {} product_code = data.get('productCode') generator = ProductImageGenerator() generator.clear_cache(product_code) return jsonify({ 'status': 'success', 'message': f'Cache cleared for {product_code}' if product_code else 'All cache cleared' }) except Exception as e: return jsonify({'error': str(e)}), 500 # ===== END DYNAMIC IMAGE GENERATION ENDPOINTS ===== # ===== USER MANAGEMENT ENDPOINTS ===== @app.route('/users') @login_required @permission_required('manage_users') def user_manager(): """Serve the user management page""" return render_template('user_manager.html') @app.route('/api/users', methods=['GET']) @login_required @permission_required('manage_users') def get_users(): """Get all users""" users = load_users() return jsonify({ 'status': 'success', 'users': users }) @app.route('/api/users', methods=['POST']) @login_required @permission_required('manage_users') def add_user(): """Add a new user with hashed password""" try: data = request.json # Validate required fields if not data.get('username') or not data.get('password') or not data.get('defaultLocation'): return jsonify({ 'status': 'error', 'message': 'Username, password, and default location are required' }), 400 users = load_users() # Check if username already exists if any(user['username'] == data['username'] for user in users): return jsonify({ 'status': 'error', 'message': 'Username already exists' }), 400 # Hash the password using pbkdf2:sha256 (secure) hashed_password = generate_password_hash(data['password'], method='pbkdf2:sha256') # Get location settings and ensure default location is marked as accessible location_settings = data.get('locationSettings', {}) default_location = data['defaultLocation'] # Ensure default location exists in settings and is marked as accessible if default_location not in location_settings: location_settings[default_location] = {} location_settings[default_location]['accessible'] = True # Create new user new_user = { 'username': data['username'], 'password': hashed_password, 'defaultLocation': default_location, 'locationSettings': location_settings, 'active': data.get('active', True) # Default to active } users.append(new_user) save_users(users) return jsonify({ 'status': 'success', 'message': 'User added successfully', 'users': users }) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 @app.route('/api/users//active', methods=['PATCH']) @login_required @permission_required('manage_users') def toggle_user_active(user_index): """Toggle user active status""" try: data = request.json users = load_users() if user_index < 0 or user_index >= len(users): return jsonify({ 'status': 'error', 'message': 'Invalid user index' }), 400 # Update active status users[user_index]['active'] = data.get('active', True) save_users(users) return jsonify({ 'status': 'success', 'message': 'User status updated', 'users': users }) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 @app.route('/api/users/', methods=['DELETE']) @login_required @permission_required('manage_users') def delete_user(user_index): """Delete a specific user by index""" try: users = load_users() if user_index < 0 or user_index >= len(users): return jsonify({ 'status': 'error', 'message': 'Invalid user index' }), 400 users.pop(user_index) save_users(users) return jsonify({ 'status': 'success', 'message': 'User deleted successfully', 'users': users }) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 @app.route('/api/users//change-password', methods=['POST']) @login_required @permission_required('manage_users') def change_user_password(user_index): """Change a user's password with current user verification""" try: data = request.json new_password = data.get('newPassword') current_user_password = data.get('currentUserPassword') # Validate required fields if not new_password or not current_user_password: return jsonify({ 'status': 'error', 'message': 'New password and current user password are required' }), 400 # Validate password length if len(new_password) < 6: return jsonify({ 'status': 'error', 'message': 'Password must be at least 6 characters long' }), 400 # Get current logged-in user current_user = get_current_user() if not current_user: return jsonify({ 'status': 'error', 'message': 'Not authenticated' }), 401 # Verify current user's password if not check_password_hash(current_user['password'], current_user_password): return jsonify({ 'status': 'error', 'message': 'Your password is incorrect. Password change denied for security reasons.' }), 403 # Load users and validate index users = load_users() if user_index < 0 or user_index >= len(users): return jsonify({ 'status': 'error', 'message': 'Invalid user index' }), 400 # Hash the new password hashed_password = generate_password_hash(new_password, method='pbkdf2:sha256') # Update the target user's password target_username = users[user_index]['username'] users[user_index]['password'] = hashed_password save_users(users) return jsonify({ 'status': 'success', 'message': f'Password changed successfully for user "{target_username}"' }) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 @app.route('/api/users/clear', methods=['DELETE']) def clear_all_users(): """Clear all users""" try: save_users([]) return jsonify({ 'status': 'success', 'message': 'All users cleared' }) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 @app.route('/api/users/download') def download_users(): """Download users as JSON file""" try: users = load_users() if not users: return jsonify({ 'status': 'error', 'message': 'No users to download' }), 400 # Create JSON file in memory json_data = json.dumps(users, indent=2) # Send as downloadable file return send_file( BytesIO(json_data.encode()), mimetype='application/json', as_attachment=True, download_name='users.json' ) except Exception as e: return jsonify({ 'status': 'error', 'message': str(e) }), 500 # ===== END USER MANAGEMENT ENDPOINTS ===== # Error handlers @app.errorhandler(404) def page_not_found(e): return render_template('404.html'), 404 @app.errorhandler(500) def internal_error(e): return "Internal Server Error", 500 if __name__ == '__main__': # Create necessary directories os.makedirs(os.path.join(basedir, 'images'), exist_ok=True) os.makedirs(os.path.join(basedir, 'templates'), exist_ok=True) os.makedirs(os.path.join(basedir, 'css'), exist_ok=True) os.makedirs(os.path.join(basedir, 'js'), exist_ok=True) os.makedirs(os.path.join(basedir, 'data'), exist_ok=True) # Run the application # Set debug=False for production # Port 8080 is used instead of 5000 (Windows reserves port 5000) app.run(debug=True, host='0.0.0.0', port=8080)