""" Beach Terrain Types database service for admin operations """ import os import sys from typing import List, Optional, Dict, Any from sqlalchemy import create_engine, text from sqlalchemy.orm import sessionmaker from werkzeug.utils import secure_filename # Add the backend directory to the Python path sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) try: from app.core.config import settings DATABASE_URL = str(settings.DATABASE_URL) # Convert PostgresDsn to string except ImportError: # Fallback to environment variables if config is not available from dotenv import load_dotenv load_dotenv() # Try to import encryption utilities try: from app.utils.encryption import decrypt_password, is_encrypted except ImportError: def decrypt_password(password, key='babagamma'): return password def is_encrypted(password, key='babagamma'): return False def get_decrypted_env_var(var_name, default=''): """Get an environment variable and decrypt it if necessary""" value = os.getenv(var_name, default) if value and is_encrypted(value): try: return decrypt_password(value, 'babagamma') except Exception: return value return value # Build database URL from components from urllib.parse import quote_plus db_user = os.getenv('DATABASE_USER', 'postgres') db_password = get_decrypted_env_var('DATABASE_PASSWORD') db_host = os.getenv('DATABASE_HOST', 'localhost') db_port = os.getenv('DATABASE_PORT', '5432') db_name = os.getenv('DATABASE_NAME', 'bookbeach') DATABASE_URL = f"postgresql://{db_user}:{quote_plus(db_password)}@{db_host}:{db_port}/{db_name}" # Create database engine and session engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'webp'} UPLOAD_FOLDER = 'static/uploads/terrain_types' def allowed_file(filename): """Check if the file extension is allowed""" return '.' in filename and \ filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS class TerrainTypeService: """Service class for Beach Terrain Types operations""" def __init__(self): self.db = SessionLocal() def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.db.close() def get_all_terrain_types(self) -> List[Dict[str, Any]]: """Get all terrain types""" try: result = self.db.execute(text(""" SELECT terrain_type_id, terrain_name, photo_path, description, CASE WHEN photo_path IS NOT NULL AND photo_path != '' THEN 1 ELSE 0 END as has_photo FROM beach_terrain_types ORDER BY terrain_name """)) terrain_types = [] for row in result: terrain_types.append({ 'terrain_type_id': row.terrain_type_id, 'terrain_name': row.terrain_name, 'photo_path': row.photo_path, 'description': row.description, 'has_photo': bool(row.has_photo) }) return terrain_types except Exception as e: print(f"Error getting terrain types: {e}") return [] def get_terrain_type_by_id(self, terrain_type_id: int) -> Optional[Dict[str, Any]]: """Get a specific terrain type by ID""" try: result = self.db.execute(text(""" SELECT terrain_type_id, terrain_name, photo_path, description FROM beach_terrain_types WHERE terrain_type_id = :terrain_type_id """), {"terrain_type_id": terrain_type_id}) row = result.fetchone() if row: return { 'terrain_type_id': row.terrain_type_id, 'terrain_name': row.terrain_name, 'photo_path': row.photo_path, 'description': row.description } return None except Exception as e: print(f"Error getting terrain type by ID: {e}") return None def create_terrain_type(self, terrain_name: str, description: str = None, photo_path: str = None) -> bool: """Create a new terrain type""" try: self.db.execute(text(""" INSERT INTO beach_terrain_types (terrain_name, description, photo_path) VALUES (:terrain_name, :description, :photo_path) """), { "terrain_name": terrain_name, "description": description, "photo_path": photo_path }) self.db.commit() return True except Exception as e: print(f"Error creating terrain type: {e}") self.db.rollback() return False def update_terrain_type(self, terrain_type_id: int, terrain_name: str, description: str = None, photo_path: str = None) -> bool: """Update an existing terrain type""" try: self.db.execute(text(""" UPDATE beach_terrain_types SET terrain_name = :terrain_name, description = :description, photo_path = :photo_path WHERE terrain_type_id = :terrain_type_id """), { "terrain_type_id": terrain_type_id, "terrain_name": terrain_name, "description": description, "photo_path": photo_path }) self.db.commit() return True except Exception as e: print(f"Error updating terrain type: {e}") self.db.rollback() return False def delete_terrain_type(self, terrain_type_id: int) -> bool: """Delete a terrain type (only if not used by any beach places)""" try: # Check if terrain type is used by any beach places result = self.db.execute(text(""" SELECT COUNT(*) as count FROM beach_places WHERE terrain_type_id = :terrain_type_id """), {"terrain_type_id": terrain_type_id}) count = result.fetchone().count if count > 0: return False # Cannot delete if in use # Get the photo path before deletion to clean up the file terrain_type = self.get_terrain_type_by_id(terrain_type_id) # Delete the terrain type self.db.execute(text(""" DELETE FROM beach_terrain_types WHERE terrain_type_id = :terrain_type_id """), {"terrain_type_id": terrain_type_id}) self.db.commit() # Clean up the photo file if it exists if terrain_type and terrain_type['photo_path']: try: photo_full_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', terrain_type['photo_path']) if os.path.exists(photo_full_path): os.remove(photo_full_path) except Exception as e: print(f"Error deleting photo file: {e}") return True except Exception as e: print(f"Error deleting terrain type: {e}") self.db.rollback() return False def get_terrain_type_usage_count(self, terrain_type_id: int) -> int: """Get the number of beach places using this terrain type""" try: result = self.db.execute(text(""" SELECT COUNT(*) as count FROM beach_places WHERE terrain_type_id = :terrain_type_id """), {"terrain_type_id": terrain_type_id}) return result.fetchone().count except Exception as e: print(f"Error getting terrain type usage count: {e}") return 0 def save_uploaded_file(file, terrain_type_id: int) -> str: """Save uploaded file and return the file path""" if file and file.filename: if allowed_file(file.filename): # Create a secure filename filename = secure_filename(file.filename) # Add terrain type ID to filename to avoid conflicts name, ext = os.path.splitext(filename) filename = f"terrain_{terrain_type_id}_{name}{ext}" # Ensure upload directory exists upload_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', UPLOAD_FOLDER) os.makedirs(upload_dir, exist_ok=True) # Save the file file_path = os.path.join(upload_dir, filename) file.save(file_path) # Return relative path for database storage return f"{UPLOAD_FOLDER}/{filename}" return None