""" Business Routes Blueprint Contains routes for managing restaurants, markets, adventures, and beaches """ from flask import Blueprint, request, render_template, redirect, url_for, flash, jsonify, session from flask_login import login_required from sqlalchemy import create_engine, text from sqlalchemy.orm import sessionmaker import os import sys # Import config from parent directory sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from config import ADMIN_CONFIG # Create Blueprint business_bp = Blueprint('business', __name__) @business_bp.route('/markets') def markets(): """Market management page""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) # Get search and filter parameters search_term = request.args.get('search', '').strip() # Check if this is a search request (search parameter exists in URL) is_search_request = 'search' in request.args # Only load data if this is a search request if not is_search_request: # First page load, return empty results return render_template('admin/markets.html', markets=[], page=1, total_count=0, total_pages=1, has_prev=False, has_next=False, prev_num=None, next_num=None, search_term='') # This is a search request, load data from database try: # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Get the page number from request args, default to 1 page = request.args.get('page', 1, type=int) per_page = 10 offset = (page - 1) * per_page # Build the WHERE clause for filtering where_conditions = [] params = {} if search_term: where_conditions.append("(m.market_name ILIKE :search OR m.phone ILIKE :search OR m.email ILIKE :search)") params['search'] = f'%{search_term}%' where_clause = "WHERE " + " AND ".join(where_conditions) if where_conditions else "" # Get total count for pagination count_query = f"SELECT COUNT(*) FROM markets m {where_clause}" count_result = db.execute(text(count_query), params).fetchone() total_count = count_result[0] if count_result else 0 # Get markets with company and country info markets_query = f""" SELECT m.market_id, m.market_name, m.address, m.city, c.company_name, co.country_name, m.phone, m.email, m.created_at, m.updated_at FROM markets m LEFT JOIN companies c ON m.company_id = c.company_id LEFT JOIN countries co ON m.country_id = co.country_id {where_clause} ORDER BY m.market_name LIMIT :limit OFFSET :offset """ params['limit'] = per_page params['offset'] = offset markets = db.execute(text(markets_query), params).fetchall() # Calculate pagination variables total_pages = (total_count + per_page - 1) // per_page has_prev = page > 1 has_next = page < total_pages prev_num = page - 1 if has_prev else None next_num = page + 1 if has_next else None db.close() return render_template('admin/markets.html', markets=markets, page=page, total_count=total_count, total_pages=total_pages, has_prev=has_prev, has_next=has_next, prev_num=prev_num, next_num=next_num, search_term=search_term) except Exception as e: flash(f'Error loading markets: {str(e)}', 'error') return render_template('admin/markets.html', markets=[], page=1, total_count=0, total_pages=1, has_prev=False, has_next=False, prev_num=None, next_num=None, search_term=search_term) @business_bp.route('/adventures') def adventures(): """Adventure management page""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) # Get search and filter parameters search_term = request.args.get('search', '').strip() # Check if this is a search request (search parameter exists in URL) is_search_request = 'search' in request.args # Only load data if this is a search request if not is_search_request: # First page load, return empty results return render_template('admin/adventures.html', adventures=[], page=1, total_count=0, total_pages=1, has_prev=False, has_next=False, prev_num=None, next_num=None, search_term='') # This is a search request, load data from database try: # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Get the page number from request args, default to 1 page = request.args.get('page', 1, type=int) per_page = 10 offset = (page - 1) * per_page # Build the WHERE clause for filtering where_conditions = [] params = {} if search_term: where_conditions.append("(a.adventure_name ILIKE :search OR a.phone ILIKE :search OR a.email ILIKE :search)") params['search'] = f'%{search_term}%' where_clause = "WHERE " + " AND ".join(where_conditions) if where_conditions else "" # Get total count for pagination count_query = f"SELECT COUNT(*) FROM adventures a {where_clause}" count_result = db.execute(text(count_query), params).fetchone() total_count = count_result[0] if count_result else 0 # Get adventures with company and country info adventures_query = f""" SELECT a.adventure_id, a.adventure_name, a.address, a.city, c.company_name, co.country_name, a.phone, a.email, a.duration_minutes, a.max_participants, a.price, a.created_at, a.updated_at FROM adventures a LEFT JOIN companies c ON a.company_id = c.company_id LEFT JOIN countries co ON a.country_id = co.country_id {where_clause} ORDER BY a.adventure_name LIMIT :limit OFFSET :offset """ params['limit'] = per_page params['offset'] = offset adventures = db.execute(text(adventures_query), params).fetchall() # Calculate pagination variables total_pages = (total_count + per_page - 1) // per_page has_prev = page > 1 has_next = page < total_pages prev_num = page - 1 if has_prev else None next_num = page + 1 if has_next else None db.close() return render_template('admin/adventures.html', adventures=adventures, page=page, total_count=total_count, total_pages=total_pages, has_prev=has_prev, has_next=has_next, prev_num=prev_num, next_num=next_num, search_term=search_term) except Exception as e: flash(f'Error loading adventures: {str(e)}', 'error') return render_template('admin/adventures.html', adventures=[], page=1, total_count=0, total_pages=1, has_prev=False, has_next=False, prev_num=None, next_num=None, search_term=search_term) @business_bp.route('/beaches') def beaches(): """Beach management page""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) # Get search and filter parameters search_term = request.args.get('search', '').strip() # Check if this is a search request (search parameter exists in URL) is_search_request = 'search' in request.args # Only load data if this is a search request if not is_search_request: # First page load, return empty results return render_template('admin/beaches.html', beaches=[], page=1, total_count=0, total_pages=1, has_prev=False, has_next=False, prev_num=None, next_num=None, search_term='') # This is a search request, load data from database try: # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Get the page number from request args, default to 1 page = request.args.get('page', 1, type=int) per_page = 10 offset = (page - 1) * per_page # Build the WHERE clause for filtering where_conditions = [] params = {} if search_term: where_conditions.append("(bp.beach_name ILIKE :search OR bp.city ILIKE :search)") params['search'] = f'%{search_term}%' where_clause = "WHERE " + " AND ".join(where_conditions) if where_conditions else "" # Get total count for pagination count_query = f"SELECT COUNT(*) FROM beach_places bp {where_clause}" count_result = db.execute(text(count_query), params).fetchone() total_count = count_result[0] if count_result else 0 # Get beaches with company and country info beaches_query = f""" SELECT bp.beach_place_id, bp.beach_name, bp.address, bp.city, c.company_name, co.country_name, bp.enable_beach, bp.created_at, bp.updated_at FROM beach_places bp LEFT JOIN companies c ON bp.company_id = c.company_id LEFT JOIN countries co ON bp.country_id = co.country_id {where_clause} ORDER BY bp.beach_name LIMIT :limit OFFSET :offset """ params['limit'] = per_page params['offset'] = offset beaches = db.execute(text(beaches_query), params).fetchall() # Calculate pagination variables total_pages = (total_count + per_page - 1) // per_page has_prev = page > 1 has_next = page < total_pages prev_num = page - 1 if has_prev else None next_num = page + 1 if has_next else None db.close() return render_template('admin/beaches.html', beaches=beaches, page=page, total_count=total_count, total_pages=total_pages, has_prev=has_prev, has_next=has_next, prev_num=prev_num, next_num=next_num, search_term=search_term) except Exception as e: flash(f'Error loading beaches: {str(e)}', 'error') return render_template('admin/beaches.html', beaches=[], page=1, total_count=0, total_pages=1, has_prev=False, has_next=False, prev_num=None, next_num=None, search_term=search_term) @business_bp.route('/companies') def companies(): """Company management page""" # Check if user is logged in and is admin # Use the regular session check for admin users if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) # Get search and filter parameters search_term = request.args.get('search', '').strip() company_status = request.args.get('status', '').strip() selected_country_id = request.args.get('country_id', '').strip() # Check if this is a search/filter request is_search_request = 'search' in request.args or 'status' in request.args or 'country_id' in request.args # Only load data if this is a search/filter request if not is_search_request: # First page load, return empty results return render_template('admin/companies.html', companies=[], page=1, total_count=0, total_pages=1, has_prev=False, has_next=False, prev_num=None, next_num=None, search_term='', company_status='', selected_country_id='', countries=[]) # This is a search/filter request, load data from database try: # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Base query for companies with country information and business counts base_query = """ SELECT c.company_id, c.company_name, c.website, c.email AS contact_email, c.phone AS contact_phone, c.company_logo_path AS logo_path, c.create_date AS created_at, c.update_date AS updated_at, c.vat_number, c.city, c.province, c.country_id, c.company_status, co.country_name, COALESCE(beach_counts.count, 0) AS beaches_count, COALESCE(restaurant_counts.count, 0) AS restaurants_count, COALESCE(market_counts.count, 0) AS markets_count, COALESCE(adventure_counts.count, 0) AS adventures_count FROM companies c LEFT JOIN countries co ON c.country_id = co.country_id LEFT JOIN ( SELECT company_id, COUNT(*) AS count FROM beach_places GROUP BY company_id ) beach_counts ON c.company_id = beach_counts.company_id LEFT JOIN ( SELECT company_id, COUNT(*) AS count FROM restaurants GROUP BY company_id ) restaurant_counts ON c.company_id = restaurant_counts.company_id LEFT JOIN ( SELECT company_id, COUNT(*) AS count FROM markets GROUP BY company_id ) market_counts ON c.company_id = market_counts.company_id LEFT JOIN ( SELECT company_id, COUNT(*) AS count FROM adventures GROUP BY company_id ) adventure_counts ON c.company_id = adventure_counts.company_id """ # Get the page number from request args, default to 1 page = request.args.get('page', 1, type=int) per_page = 10 offset = (page - 1) * per_page # Build the WHERE clause for filtering where_conditions = [] params = {} if search_term: # Updated search to include actual column names with table alias where_conditions.append("(c.company_name ILIKE :search OR c.email ILIKE :search OR c.phone ILIKE :search OR c.vat_number ILIKE :search)") params['search'] = f'%{search_term}%' if company_status: where_conditions.append("c.company_status = :status") params['status'] = company_status if selected_country_id: where_conditions.append("c.country_id = :country_id") params['country_id'] = selected_country_id where_clause = "WHERE " + " AND ".join(where_conditions) if where_conditions else "" # Get total count for pagination count_query = f"SELECT COUNT(*) FROM companies c {where_clause}" count_result = db.execute(text(count_query), params).fetchone() total_count = count_result[0] if count_result else 0 # Get companies with pagination paginated_query = f""" {base_query} {where_clause} ORDER BY c.company_name LIMIT :limit OFFSET :offset """ params['limit'] = per_page params['offset'] = offset companies = db.execute(text(paginated_query), params).fetchall() # Calculate pagination variables total_pages = (total_count + per_page - 1) // per_page has_prev = page > 1 has_next = page < total_pages prev_num = page - 1 if has_prev else None next_num = page + 1 if has_next else None # Get countries for the filter dropdown countries = [] try: countries_result = db.execute(text("SELECT country_id, country_name FROM countries ORDER BY country_name")).fetchall() countries = [{'country_id': row[0], 'country_name': row[1]} for row in countries_result] except Exception as e: print(f"Error loading countries: {e}") db.close() print(f"Companies route - Found {len(companies)} companies, total count: {total_count}") return render_template('admin/companies.html', companies=companies, page=page, total_count=total_count, total_pages=total_pages, has_prev=has_prev, has_next=has_next, prev_num=prev_num, next_num=next_num, search_term=search_term, company_status=company_status, selected_country_id=selected_country_id, countries=countries) except Exception as e: flash(f'Error loading companies: {str(e)}', 'error') print(f"Error in companies route: {str(e)}") import traceback traceback.print_exc() return render_template('admin/companies.html', companies=[], page=1, total_count=0, total_pages=1, has_prev=False, has_next=False, prev_num=None, next_num=None, search_term=search_term, company_status=company_status, selected_country_id=selected_country_id, countries=[]) @business_bp.route('/restaurant//edit') def edit_restaurant(restaurant_id): """Edit restaurant page""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) try: # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Get restaurant details restaurant_query = """ SELECT r.*, c.company_name, co.country_name FROM restaurants r LEFT JOIN companies c ON r.company_id = c.company_id LEFT JOIN countries co ON r.country_id = co.country_id WHERE r.restaurant_id = :restaurant_id """ restaurant_result = db.execute(text(restaurant_query), {'restaurant_id': restaurant_id}).fetchone() if not restaurant_result: flash('Restaurant not found.', 'error') return redirect(url_for('business.restaurants')) # Get companies for dropdown companies_result = db.execute(text("SELECT company_id, company_name FROM companies ORDER BY company_name")).fetchall() companies = [{'company_id': row[0], 'company_name': row[1]} for row in companies_result] # Get countries for dropdown countries_result = db.execute(text("SELECT country_id, country_name FROM countries ORDER BY country_name")).fetchall() countries = [{'country_id': row[0], 'country_name': row[1]} for row in countries_result] db.close() return render_template('admin/edit_restaurant.html', restaurant=restaurant_result, companies=companies, countries=countries) except Exception as e: flash(f'Error loading restaurant: {str(e)}', 'error') return redirect(url_for('business.restaurants')) @business_bp.route('/market//edit') def edit_market(market_id): """Edit market page""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) try: # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Get market details market_query = """ SELECT m.*, c.company_name, co.country_name FROM markets m LEFT JOIN companies c ON m.company_id = c.company_id LEFT JOIN countries co ON m.country_id = co.country_id WHERE m.market_id = :market_id """ market_result = db.execute(text(market_query), {'market_id': market_id}).fetchone() if not market_result: flash('Market not found.', 'error') return redirect(url_for('business.markets')) # Get companies for dropdown companies_result = db.execute(text("SELECT company_id, company_name FROM companies ORDER BY company_name")).fetchall() companies = [{'company_id': row[0], 'company_name': row[1]} for row in companies_result] # Get countries for dropdown countries_result = db.execute(text("SELECT country_id, country_name FROM countries ORDER BY country_name")).fetchall() countries = [{'country_id': row[0], 'country_name': row[1]} for row in countries_result] db.close() return render_template('admin/edit_market.html', market=market_result, companies=companies, countries=countries) except Exception as e: flash(f'Error loading market: {str(e)}', 'error') return redirect(url_for('business.markets')) @business_bp.route('/adventure//edit') def edit_adventure(adventure_id): """Edit adventure page""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) try: # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Get adventure details adventure_query = """ SELECT a.*, c.company_name, co.country_name FROM adventures a LEFT JOIN companies c ON a.company_id = c.company_id LEFT JOIN countries co ON a.country_id = co.country_id WHERE a.adventure_id = :adventure_id """ adventure_result = db.execute(text(adventure_query), {'adventure_id': adventure_id}).fetchone() if not adventure_result: flash('Adventure not found.', 'error') return redirect(url_for('business.adventures')) # Get companies for dropdown companies_result = db.execute(text("SELECT company_id, company_name FROM companies ORDER BY company_name")).fetchall() companies = [{'company_id': row[0], 'company_name': row[1]} for row in companies_result] # Get countries for dropdown countries_result = db.execute(text("SELECT country_id, country_name FROM countries ORDER BY country_name")).fetchall() countries = [{'country_id': row[0], 'country_name': row[1]} for row in countries_result] db.close() return render_template('admin/edit_adventure.html', adventure=adventure_result, companies=companies, countries=countries) except Exception as e: flash(f'Error loading adventure: {str(e)}', 'error') return redirect(url_for('business.adventures')) @business_bp.route('/beach//edit') def edit_beach(beach_id): """Edit beach page""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) try: # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Get beach details beach_query = """ SELECT bp.*, c.company_name, co.country_name FROM beach_places bp LEFT JOIN companies c ON bp.company_id = c.company_id LEFT JOIN countries co ON bp.country_id = co.country_id WHERE bp.beach_place_id = :beach_id """ beach_result = db.execute(text(beach_query), {'beach_id': beach_id}).fetchone() if not beach_result: flash('Beach not found.', 'error') return redirect(url_for('business.beaches')) # Get companies for dropdown companies_result = db.execute(text("SELECT company_id, company_name FROM companies ORDER BY company_name")).fetchall() companies = [{'company_id': row[0], 'company_name': row[1]} for row in companies_result] # Get countries for dropdown countries_result = db.execute(text("SELECT country_id, country_name FROM countries ORDER BY country_name")).fetchall() countries = [{'country_id': row[0], 'country_name': row[1]} for row in countries_result] db.close() return render_template('admin/edit_beach.html', beach=beach_result, companies=companies, countries=countries) except Exception as e: flash(f'Error loading beach: {str(e)}', 'error') return redirect(url_for('business.beaches')) @business_bp.route('/company//edit') def edit_company(company_id): """Edit company page""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) try: # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Get company details company_query = """ SELECT c.*, co.country_name FROM companies c LEFT JOIN countries co ON c.country_id = co.country_id WHERE c.company_id = :company_id """ company_result = db.execute(text(company_query), {'company_id': company_id}).fetchone() if not company_result: flash('Company not found.', 'error') return redirect(url_for('business.companies')) # Get countries for dropdown countries_result = db.execute(text("SELECT country_id, country_name FROM countries ORDER BY country_name")).fetchall() countries = [{'country_id': row[0], 'country_name': row[1]} for row in countries_result] db.close() return render_template('admin/edit_company.html', company=company_result, countries=countries) except Exception as e: flash(f'Error loading company: {str(e)}', 'error') return redirect(url_for('business.companies')) @business_bp.route('/restaurant//update', methods=['POST']) def update_restaurant(restaurant_id): """Update restaurant""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) try: # Get form data restaurant_name = request.form.get('restaurant_name', '').strip() company_id = request.form.get('company_id', '').strip() phone = request.form.get('phone', '').strip() email = request.form.get('email', '').strip() address = request.form.get('address', '').strip() city = request.form.get('city', '').strip() country_id = request.form.get('country_id', '').strip() latitude = request.form.get('latitude', '').strip() longitude = request.form.get('longitude', '').strip() opening_time = request.form.get('opening_time', '').strip() closing_time = request.form.get('closing_time', '').strip() is_active = request.form.get('is_active', '').strip() description = request.form.get('description', '').strip() cuisine_type = request.form.get('cuisine_type', '').strip() beach_place_id = request.form.get('beach_place_id', '').strip() # Validate required fields if not restaurant_name or not company_id: flash('Restaurant name and company are required.', 'error') return redirect(url_for('business.edit_restaurant', restaurant_id=restaurant_id)) # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Update restaurant update_query = """ UPDATE restaurants SET restaurant_name = :restaurant_name, company_id = :company_id, phone = :phone, email = :email, address = :address, city = :city, country_id = :country_id, latitude = :latitude, longitude = :longitude, opening_time = :opening_time, closing_time = :closing_time, is_active = :is_active, description = :description, cuisine_type = :cuisine_type, beach_place_id = :beach_place_id, updated_at = NOW() WHERE restaurant_id = :restaurant_id """ def safe_int(value): if value: try: return int(value) except ValueError: return None return None def safe_float(value): if value: try: return float(value) except ValueError: return None return None def safe_bool(value): return value.lower() == 'on' if value else False def safe_time(value): if value: try: from datetime import datetime return datetime.strptime(value, '%H:%M').time() except ValueError: return None return None def safe_uuid(value): if value: try: import uuid return uuid.UUID(value) except ValueError: return None return None params = { 'restaurant_name': restaurant_name, 'company_id': safe_uuid(company_id), 'phone': phone if phone else None, 'email': email if email else None, 'address': address if address else None, 'city': city if city else None, 'country_id': safe_int(country_id), 'latitude': safe_float(latitude), 'longitude': safe_float(longitude), 'opening_time': safe_time(opening_time), 'closing_time': safe_time(closing_time), 'is_active': safe_bool(is_active), 'description': description if description else None, 'cuisine_type': cuisine_type if cuisine_type else None, 'beach_place_id': safe_uuid(beach_place_id), 'restaurant_id': restaurant_id } db.execute(text(update_query), params) db.commit() db.close() flash('Restaurant updated successfully.', 'success') return redirect(url_for('business.restaurants')) except Exception as e: flash(f'Error updating restaurant: {str(e)}', 'error') return redirect(url_for('business.edit_restaurant', restaurant_id=restaurant_id)) @business_bp.route('/market//update', methods=['POST']) def update_market(market_id): """Update market""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) try: # Get form data market_name = request.form.get('market_name', '').strip() company_id = request.form.get('company_id', '').strip() phone = request.form.get('phone', '').strip() email = request.form.get('email', '').strip() address = request.form.get('address', '').strip() city = request.form.get('city', '').strip() country_id = request.form.get('country_id', '').strip() latitude = request.form.get('latitude', '').strip() longitude = request.form.get('longitude', '').strip() opening_time = request.form.get('opening_time', '').strip() closing_time = request.form.get('closing_time', '').strip() delivery_available = request.form.get('delivery_available', '').strip() delivery_fee = request.form.get('delivery_fee', '').strip() min_order_amount = request.form.get('min_order_amount', '').strip() is_active = request.form.get('is_active', '').strip() description = request.form.get('description', '').strip() website = request.form.get('website', '').strip() beach_place_id = request.form.get('beach_place_id', '').strip() # Validate required fields if not market_name or not company_id: flash('Market name and company are required.', 'error') return redirect(url_for('business.edit_market', market_id=market_id)) # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Update market update_query = """ UPDATE markets SET market_name = :market_name, company_id = :company_id, phone = :phone, email = :email, address = :address, city = :city, country_id = :country_id, latitude = :latitude, longitude = :longitude, opening_time = :opening_time, closing_time = :closing_time, delivery_available = :delivery_available, delivery_fee = :delivery_fee, min_order_amount = :min_order_amount, is_active = :is_active, description = :description, website = :website, beach_place_id = :beach_place_id, updated_at = NOW() WHERE market_id = :market_id """ def safe_int(value): if value: try: return int(value) except ValueError: return None return None def safe_float(value): if value: try: return float(value) except ValueError: return None return None def safe_bool(value): return value.lower() == 'on' if value else False def safe_time(value): if value: try: from datetime import datetime return datetime.strptime(value, '%H:%M').time() except ValueError: return None return None def safe_uuid(value): if value: try: import uuid return uuid.UUID(value) except ValueError: return None return None params = { 'market_name': market_name, 'company_id': safe_uuid(company_id), 'phone': phone if phone else None, 'email': email if email else None, 'address': address if address else None, 'city': city if city else None, 'country_id': safe_int(country_id), 'latitude': safe_float(latitude), 'longitude': safe_float(longitude), 'opening_time': safe_time(opening_time), 'closing_time': safe_time(closing_time), 'delivery_available': safe_bool(delivery_available), 'delivery_fee': safe_float(delivery_fee), 'min_order_amount': safe_float(min_order_amount), 'is_active': safe_bool(is_active), 'description': description if description else None, 'website': website if website else None, 'beach_place_id': safe_uuid(beach_place_id), 'market_id': market_id } db.execute(text(update_query), params) db.commit() db.close() flash('Market updated successfully.', 'success') return redirect(url_for('business.markets')) except Exception as e: flash(f'Error updating market: {str(e)}', 'error') return redirect(url_for('business.edit_market', market_id=market_id)) @business_bp.route('/adventure//update', methods=['POST']) def update_adventure(adventure_id): """Update adventure""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) try: # Get form data adventure_name = request.form.get('adventure_name', '').strip() company_id = request.form.get('company_id', '').strip() phone = request.form.get('phone', '').strip() email = request.form.get('email', '').strip() address = request.form.get('address', '').strip() city = request.form.get('city', '').strip() country_id = request.form.get('country_id', '').strip() duration_minutes = request.form.get('duration_minutes', '').strip() max_participants = request.form.get('max_participants', '').strip() price = request.form.get('price', '').strip() # Validate required fields if not adventure_name or not company_id: flash('Adventure name and company are required.', 'error') return redirect(url_for('business.edit_adventure', adventure_id=adventure_id)) # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Update adventure update_query = """ UPDATE adventures SET adventure_name = :adventure_name, company_id = :company_id, phone = :phone, email = :email, address = :address, city = :city, country_id = :country_id, duration_minutes = :duration_minutes, max_participants = :max_participants, price = :price, updated_at = NOW() WHERE adventure_id = :adventure_id """ params = { 'adventure_name': adventure_name, 'company_id': company_id, 'phone': phone if phone else None, 'email': email if email else None, 'address': address if address else None, 'city': city if city else None, 'country_id': country_id if country_id else None, 'duration_minutes': int(duration_minutes) if duration_minutes else None, 'max_participants': int(max_participants) if max_participants else None, 'price': float(price) if price else None, 'adventure_id': adventure_id } db.execute(text(update_query), params) db.commit() db.close() flash('Adventure updated successfully.', 'success') return redirect(url_for('business.adventures')) except Exception as e: flash(f'Error updating adventure: {str(e)}', 'error') return redirect(url_for('business.edit_adventure', adventure_id=adventure_id)) @business_bp.route('/beach//update', methods=['POST']) def update_beach(beach_id): """Update beach""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) try: # Get form data beach_name = request.form.get('beach_name', '').strip() company_id = request.form.get('company_id', '').strip() beach_information = request.form.get('beach_information', '').strip() address = request.form.get('address', '').strip() city = request.form.get('city', '').strip() country_id = request.form.get('country_id', '').strip() contact_phone = request.form.get('contact_phone', '').strip() latitude = request.form.get('latitude', '').strip() longitude = request.form.get('longitude', '').strip() max_seats = request.form.get('max_seats', '').strip() top_rows = request.form.get('top_rows', '').strip() top_seats = request.form.get('top_seats', '').strip() starttime = request.form.get('starttime', '').strip() endtime = request.form.get('endtime', '').strip() opendays = request.form.get('opendays', '').strip() enable_beach = request.form.get('enable_beach', '').strip() is_double_seat = request.form.get('is_double_seat', '').strip() disable_today = request.form.get('disable_today', '').strip() has_umbrella = request.form.get('has_umbrella', '').strip() has_showers = request.form.get('has_showers', '').strip() has_lockers = request.form.get('has_lockers', '').strip() has_market = request.form.get('has_market', '').strip() has_restaurant = request.form.get('has_restaurant', '').strip() has_swimming_pool = request.form.get('has_swimming_pool', '').strip() has_parking = request.form.get('has_parking', '').strip() has_toilets = request.form.get('has_toilets', '').strip() has_free_wifi = request.form.get('has_free_wifi', '').strip() has_music = request.form.get('has_music', '').strip() has_safety_box = request.form.get('has_safety_box', '').strip() has_access_wheelchair = request.form.get('has_access_wheelchair', '').strip() has_lockers_for_baby = request.form.get('has_lockers_for_baby', '').strip() has_beach_volley = request.form.get('has_beach_volley', '').strip() has_tennis = request.form.get('has_tennis', '').strip() has_spa = request.form.get('has_spa', '').strip() has_medical = request.form.get('has_medical', '').strip() has_lifeguard = request.form.get('has_lifeguard', '').strip() pet_allow = request.form.get('pet_allow', '').strip() mask = request.form.get('mask', '').strip() # Validate required fields if not beach_name or not company_id: flash('Beach name and company are required.', 'error') return redirect(url_for('business.edit_beach', beach_id=beach_id)) # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Update beach update_query = """ UPDATE beach_places SET beach_name = :beach_name, company_id = :company_id, beach_information = :beach_information, address = :address, city = :city, country_id = :country_id, contact_phone = :contact_phone, latitude = :latitude, longitude = :longitude, max_seats = :max_seats, top_rows = :top_rows, top_seats = :top_seats, starttime = :starttime, endtime = :endtime, opendays = :opendays, enable_beach = :enable_beach, is_double_seat = :is_double_seat, disable_today = :disable_today, has_umbrella = :has_umbrella, has_showers = :has_showers, has_lockers = :has_lockers, has_market = :has_market, has_restaurant = :has_restaurant, has_swimming_pool = :has_swimming_pool, has_parking = :has_parking, has_toilets = :has_toilets, has_free_wifi = :has_free_wifi, has_music = :has_music, has_safety_box = :has_safety_box, has_access_wheelchair = :has_access_wheelchair, has_lockers_for_baby = :has_lockers_for_baby, has_beach_volley = :has_beach_volley, has_tennis = :has_tennis, has_spa = :has_spa, has_medical = :has_medical, has_lifeguard = :has_lifeguard, pet_allow = :pet_allow, mask = :mask, updated_at = NOW() WHERE beach_place_id = :beach_id """ def safe_int(value): if value: try: return int(value) except ValueError: return None return None def safe_float(value): if value: try: return float(value) except ValueError: return None return None def safe_bool(value): return value.lower() == 'on' if value else False def safe_time(value): if value: try: from datetime import datetime return datetime.strptime(value, '%H:%M').time() except ValueError: return None return None def safe_uuid(value): if value: try: import uuid return uuid.UUID(value) except ValueError: return None return None params = { 'beach_name': beach_name, 'company_id': safe_uuid(company_id), 'beach_information': beach_information if beach_information else None, 'address': address if address else None, 'city': city if city else None, 'country_id': safe_int(country_id), 'contact_phone': contact_phone if contact_phone else None, 'latitude': safe_float(latitude), 'longitude': safe_float(longitude), 'max_seats': safe_int(max_seats), 'top_rows': safe_int(top_rows), 'top_seats': safe_int(top_seats), 'starttime': safe_time(starttime), 'endtime': safe_time(endtime), 'opendays': opendays if opendays else '1234567', 'enable_beach': safe_bool(enable_beach), 'is_double_seat': safe_bool(is_double_seat), 'disable_today': safe_bool(disable_today), 'has_umbrella': safe_bool(has_umbrella), 'has_showers': safe_bool(has_showers), 'has_lockers': safe_bool(has_lockers), 'has_market': safe_bool(has_market), 'has_restaurant': safe_bool(has_restaurant), 'has_swimming_pool': safe_bool(has_swimming_pool), 'has_parking': safe_bool(has_parking), 'has_toilets': safe_bool(has_toilets), 'has_free_wifi': safe_bool(has_free_wifi), 'has_music': safe_bool(has_music), 'has_safety_box': safe_bool(has_safety_box), 'has_access_wheelchair': safe_bool(has_access_wheelchair), 'has_lockers_for_baby': safe_bool(has_lockers_for_baby), 'has_beach_volley': safe_bool(has_beach_volley), 'has_tennis': safe_bool(has_tennis), 'has_spa': safe_bool(has_spa), 'has_medical': safe_bool(has_medical), 'has_lifeguard': safe_bool(has_lifeguard), 'pet_allow': safe_bool(pet_allow), 'mask': mask if mask else None, 'beach_id': beach_id } db.execute(text(update_query), params) db.commit() db.close() flash('Beach updated successfully.', 'success') return redirect(url_for('business.beaches')) except Exception as e: flash(f'Error updating beach: {str(e)}', 'error') return redirect(url_for('business.edit_beach', beach_id=beach_id)) @business_bp.route('/company//update', methods=['POST']) def update_company(company_id): """Update company""" # Check if user is logged in and is admin if not session.get('is_admin'): flash('You must be logged in as an administrator to access this page.', 'error') return redirect(url_for('admin.admin_login')) try: # Get form data company_name = request.form.get('company_name', '').strip() vat_number = request.form.get('vat_number', '').strip() website = request.form.get('website', '').strip() email = request.form.get('email', '').strip() phone = request.form.get('phone', '').strip() address = request.form.get('address', '').strip() city = request.form.get('city', '').strip() province = request.form.get('province', '').strip() country_id = request.form.get('country_id', '').strip() company_status = request.form.get('company_status', '').strip() # Validate required fields if not company_name: flash('Company name is required.', 'error') return redirect(url_for('business.edit_company', company_id=company_id)) # Database connection DATABASE_URL = ADMIN_CONFIG['DATABASE_URL'] engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() # Update company update_query = """ UPDATE companies SET company_name = :company_name, vat_number = :vat_number, website = :website, email = :email, phone = :phone, address = :address, city = :city, province = :province, country_id = :country_id, company_status = :company_status, update_date = NOW() WHERE company_id = :company_id """ params = { 'company_name': company_name, 'vat_number': vat_number if vat_number else None, 'website': website if website else None, 'email': email if email else None, 'phone': phone if phone else None, 'address': address if address else None, 'city': city if city else None, 'province': province if province else None, 'country_id': country_id if country_id else None, 'company_status': company_status if company_status else None, 'company_id': company_id } db.execute(text(update_query), params) db.commit() db.close() flash('Company updated successfully.', 'success') return redirect(url_for('business.companies')) except Exception as e: flash(f'Error updating company: {str(e)}', 'error') return redirect(url_for('business.edit_company', company_id=company_id)) # More business routes will be added here...