Files
moto-adv-website/app/routes/community.py
ske087 187254beca feat: Add interactive map functionality with Leaflet.js
- Implemented interactive map card with expand functionality
- Added Leaflet.js integration with OpenStreetMap tiles
- Created expandable map modal (80% screen coverage)
- Fixed cover image display on community page
- Enhanced post detail page with interactive route visualization
- Added proper error handling and fallback content
- Cleaned up JavaScript structure and removed duplicate code
- Updated community index template to use cover images
- Added GPX file processing utilities
- Fixed indentation error in run.py

Map features:
- Country-level positioning (Romania default)
- Zoom controls and interactive navigation
- Test marker with popup functionality
- Expandable full-screen view with X button
- Clean console logging for debugging
- Responsive design with Tailwind CSS styling
2025-07-24 21:36:42 +03:00

768 lines
32 KiB
Python

from flask import Blueprint, render_template, request, redirect, url_for, flash, current_app, jsonify
from flask_login import login_required, current_user
from app.models import Post, PostImage, GPXFile, User, Comment, Like
from app.extensions import db
from app.forms import PostForm, CommentForm
from app.media_config import MediaConfig
from werkzeug.utils import secure_filename
from werkzeug.exceptions import RequestEntityTooLarge
import os
import uuid
import shutil
from datetime import datetime
from PIL import Image
import gpxpy
community = Blueprint('community', __name__)
@community.route('/')
def index():
"""Community main page with map and posts"""
page = request.args.get('page', 1, type=int)
posts = Post.query.filter_by(published=True).order_by(Post.created_at.desc()).paginate(
page=page, per_page=12, error_out=False
)
# Get posts with GPX files for map display
posts_with_routes = Post.query.filter_by(published=True).join(GPXFile).all()
return render_template('community/index.html', posts=posts, posts_with_routes=posts_with_routes)
@community.route('/post/<int:id>')
def post_detail(id):
"""Individual post detail page"""
post = Post.query.get_or_404(id)
if not post.published and (not current_user.is_authenticated or
(current_user.id != post.author_id and not current_user.is_admin)):
flash('Post not found.', 'error')
return redirect(url_for('community.index'))
form = CommentForm()
comments = Comment.query.filter_by(post_id=id).order_by(Comment.created_at.asc()).all()
return render_template('community/post_detail.html', post=post, form=form, comments=comments)
@community.route('/profile')
@login_required
def profile():
"""User profile page with their posts"""
page = request.args.get('page', 1, type=int)
posts = Post.query.filter_by(author_id=current_user.id).order_by(Post.created_at.desc()).paginate(
page=page, per_page=10, error_out=False
)
# Count posts by status
published_count = Post.query.filter_by(author_id=current_user.id, published=True).count()
pending_count = Post.query.filter_by(author_id=current_user.id, published=False).count()
return render_template('community/profile.html',
posts=posts,
published_count=published_count,
pending_count=pending_count)
@community.route('/edit-post/<int:id>', methods=['GET', 'POST'])
@login_required
def edit_post(id):
"""Edit existing post"""
post = Post.query.get_or_404(id)
# Check if user owns the post
if current_user.id != post.author_id:
flash('You can only edit your own posts.', 'error')
return redirect(url_for('community.profile'))
if request.method == 'POST':
try:
# Get form data
title = request.form.get('title', '').strip()
subtitle = request.form.get('subtitle', '').strip()
content = request.form.get('content', '').strip()
difficulty = request.form.get('difficulty', type=int)
# Validation
if not title:
return jsonify({'success': False, 'error': 'Title is required'})
if not difficulty or difficulty < 1 or difficulty > 5:
return jsonify({'success': False, 'error': 'Valid difficulty level is required'})
if not content:
return jsonify({'success': False, 'error': 'Content is required'})
# Update post
post.title = title
post.subtitle = subtitle
post.content = content
post.difficulty = difficulty
post.published = False # Reset to pending review
post.updated_at = datetime.now()
current_app.logger.info(f'Updating post {post.id}: {post.title} by user {current_user.id}')
# Handle new cover picture upload (if provided)
if 'cover_picture' in request.files:
cover_file = request.files['cover_picture']
if cover_file and cover_file.filename:
try:
result = save_image(cover_file, post.id)
if result['success']:
# Remove old cover image if exists
old_cover = PostImage.query.filter_by(post_id=post.id, is_cover=True).first()
if old_cover:
db.session.delete(old_cover)
# Save new cover image
cover_image = PostImage(
filename=result['filename'],
original_name=cover_file.filename,
size=result['size'],
mime_type=result['mime_type'],
post_id=post.id,
is_cover=True
)
db.session.add(cover_image)
except Exception as e:
current_app.logger.warning(f'Error processing new cover image: {str(e)}')
# Handle new GPX file upload (if provided)
if 'gpx_file' in request.files:
gpx_file = request.files['gpx_file']
if gpx_file and gpx_file.filename:
try:
result = save_gpx_file(gpx_file, post.id)
if result['success']:
# Remove old GPX files
old_gpx_files = GPXFile.query.filter_by(post_id=post.id).all()
for old_gpx in old_gpx_files:
db.session.delete(old_gpx)
# Save new GPX file
gpx_file_record = GPXFile(
filename=result['filename'],
original_name=gpx_file.filename,
size=result['size'],
post_id=post.id
)
db.session.add(gpx_file_record)
except Exception as e:
current_app.logger.warning(f'Error processing new GPX file: {str(e)}')
db.session.commit()
current_app.logger.info(f'Post {post.id} updated successfully')
return jsonify({
'success': True,
'message': 'Post updated successfully! It has been resubmitted for admin review.',
'redirect_url': url_for('community.profile')
})
except Exception as e:
db.session.rollback()
current_app.logger.error(f'Error updating post: {str(e)}')
return jsonify({'success': False, 'error': f'An error occurred while updating your post: {str(e)}'})
# GET request - show edit form
return render_template('community/edit_post.html', post=post)
@community.route('/delete-post/<int:id>', methods=['DELETE', 'POST'])
@login_required
def delete_post(id):
"""Delete a post and all its associated media"""
post = Post.query.get_or_404(id)
# Check if user owns the post
if current_user.id != post.author_id:
return jsonify({'success': False, 'error': 'You can only delete your own posts'})
try:
# Delete associated media files from filesystem
if post.media_folder:
media_path = os.path.join(current_app.root_path, 'static', 'media', 'posts', post.media_folder)
if os.path.exists(media_path):
import shutil
shutil.rmtree(media_path)
current_app.logger.info(f'Deleted media folder: {media_path}')
# Delete database records (cascading should handle related records)
db.session.delete(post)
db.session.commit()
current_app.logger.info(f'Post {post.id} "{post.title}" deleted by user {current_user.id}')
return jsonify({
'success': True,
'message': f'Adventure "{post.title}" has been deleted successfully.'
})
except Exception as e:
db.session.rollback()
current_app.logger.error(f'Error deleting post {id}: {str(e)}')
return jsonify({
'success': False,
'error': f'An error occurred while deleting the post: {str(e)}'
})
@community.route('/new-post', methods=['GET', 'POST'])
@login_required
def new_post():
"""Create new post page"""
if request.method == 'POST':
try:
# Debug: Print all form data and files
current_app.logger.info(f'POST request received. Form data: {dict(request.form)}')
current_app.logger.info(f'Files received: {list(request.files.keys())}')
# Get form data
title = request.form.get('title', '').strip()
subtitle = request.form.get('subtitle', '').strip()
content = request.form.get('content', '').strip()
difficulty = request.form.get('difficulty', type=int)
current_app.logger.info(f'Parsed form data - Title: "{title}", Subtitle: "{subtitle}", Content: "{content}", Difficulty: {difficulty}')
# Validation
if not title:
current_app.logger.warning('Validation failed: Title is required')
return jsonify({'success': False, 'error': 'Title is required'})
if not difficulty or difficulty < 1 or difficulty > 5:
current_app.logger.warning(f'Validation failed: Invalid difficulty level: {difficulty}')
return jsonify({'success': False, 'error': 'Valid difficulty level is required'})
# Allow empty content for now to simplify testing
if not content:
content = f"Adventure details for {title}"
current_app.logger.info(f'Using default content: "{content}"')
# Create post
post = Post(
title=title,
subtitle=subtitle,
content=content,
difficulty=difficulty,
media_folder=f"post_{uuid.uuid4().hex[:8]}_{datetime.now().strftime('%Y%m%d')}",
published=False, # Set to pending review by default
author_id=current_user.id
)
current_app.logger.info(f'Creating post: {post.title} by user {current_user.id}')
db.session.add(post)
db.session.flush() # Get the post ID
current_app.logger.info(f'Post created with ID: {post.id}')
# Create media folder for this post
media_folder_path = os.path.join(current_app.root_path, 'static', 'media', 'posts', post.media_folder)
os.makedirs(media_folder_path, exist_ok=True)
# Create subfolders for different media types
os.makedirs(os.path.join(media_folder_path, 'images'), exist_ok=True)
os.makedirs(os.path.join(media_folder_path, 'gpx'), exist_ok=True)
current_app.logger.info(f'Created media folder: {media_folder_path}')
# Handle cover picture upload (if provided)
if 'cover_picture' in request.files:
cover_file = request.files['cover_picture']
if cover_file and cover_file.filename:
try:
current_app.logger.info(f'Processing cover picture: {cover_file.filename}')
result = save_image(cover_file, post.id)
if result['success']:
# Save as cover image
cover_image = PostImage(
filename=result['filename'],
original_name=cover_file.filename,
size=result['size'],
mime_type=result['mime_type'],
post_id=post.id,
is_cover=True # Mark as cover image
)
db.session.add(cover_image)
current_app.logger.info(f'Cover image saved: {result["filename"]}')
except Exception as e:
current_app.logger.warning(f'Error processing cover image: {str(e)}')
# Handle GPX file upload (if provided)
if 'gpx_file' in request.files:
gpx_file = request.files['gpx_file']
if gpx_file and gpx_file.filename:
try:
current_app.logger.info(f'Processing GPX file: {gpx_file.filename}')
result = save_gpx_file(gpx_file, post.id)
if result['success']:
gpx_file_record = GPXFile(
filename=result['filename'],
original_name=gpx_file.filename,
size=result['size'],
post_id=post.id
)
db.session.add(gpx_file_record)
db.session.flush() # Get the GPX file ID
# Extract GPX statistics
from app.utils.gpx_processor import process_gpx_file
if process_gpx_file(gpx_file_record):
current_app.logger.info(f'GPX statistics extracted for: {result["filename"]}')
else:
current_app.logger.warning(f'Failed to extract GPX statistics for: {result["filename"]}')
current_app.logger.info(f'GPX file saved: {result["filename"]}')
except Exception as e:
current_app.logger.warning(f'Error processing GPX file: {str(e)}')
# Handle section images (multiple images from content sections)
section_images_processed = 0
for key in request.files.keys():
if key.startswith('section_image_') and not key.endswith('_section'):
try:
image_file = request.files[key]
if image_file and image_file.filename:
current_app.logger.info(f'Processing section image: {image_file.filename}')
result = save_image(image_file, post.id)
if result['success']:
section_image = PostImage(
filename=result['filename'],
original_name=image_file.filename,
size=result['size'],
mime_type=result['mime_type'],
post_id=post.id,
is_cover=False # Section images are not cover images
)
db.session.add(section_image)
section_images_processed += 1
current_app.logger.info(f'Section image saved: {result["filename"]}')
except Exception as e:
current_app.logger.warning(f'Error processing section image {key}: {str(e)}')
current_app.logger.info(f'Total section images processed: {section_images_processed}')
db.session.commit()
current_app.logger.info(f'Post {post.id} committed to database successfully')
return jsonify({
'success': True,
'message': 'Adventure created successfully! It will be reviewed by admin before publishing.',
'redirect_url': url_for('community.index')
})
except Exception as e:
db.session.rollback()
current_app.logger.error(f'Error creating post: {str(e)}')
return jsonify({'success': False, 'error': f'An error occurred while creating your post: {str(e)}'})
# GET request - show the form
return render_template('community/new_post.html')
@community.route('/migrate-db')
@login_required
def migrate_database():
"""Run database migration - for admin use"""
if not current_user.is_admin:
flash('Unauthorized access.', 'error')
return redirect(url_for('community.index'))
try:
from sqlalchemy import text
# Check if is_cover column exists in post_images table
result = db.session.execute(text('PRAGMA table_info(post_images)'))
columns = [row[1] for row in result.fetchall()]
if 'is_cover' not in columns:
# Add the is_cover column
db.session.execute(text('ALTER TABLE post_images ADD COLUMN is_cover BOOLEAN DEFAULT 0'))
db.session.commit()
flash('Successfully added is_cover column to post_images table', 'success')
else:
flash('is_cover column already exists', 'info')
return redirect(url_for('admin.index'))
except Exception as e:
db.session.rollback()
current_app.logger.error(f'Migration error: {str(e)}')
flash(f'Migration error: {str(e)}', 'error')
return redirect(url_for('admin.index'))
@community.route('/simple-form', methods=['GET'])
@login_required
def simple_form():
"""Simple form for testing post creation"""
return render_template('community/simple_form.html')
@community.route('/simple-test', methods=['GET', 'POST'])
@login_required
def simple_test():
"""Very simple test post creation"""
if request.method == 'POST':
try:
# Simple form submission with minimal data
title = request.form.get('title', 'Test Post').strip()
post = Post(
title=title,
subtitle="Simple test post",
content="This is a simple test post to verify the system.",
difficulty=3,
media_folder=f"test_{uuid.uuid4().hex[:8]}",
published=False,
author_id=current_user.id
)
db.session.add(post)
db.session.commit()
flash('Test post created successfully!', 'success')
return redirect(url_for('community.index'))
except Exception as e:
db.session.rollback()
flash(f'Error: {str(e)}', 'error')
return '''
<form method="POST">
<input type="text" name="title" placeholder="Post title" required>
<button type="submit">Create Test Post</button>
</form>
'''
@login_required
def test_post():
"""Simple test post creation for debugging"""
if request.method == 'POST':
try:
# Create a simple test post
post = Post(
title="Test Post - " + datetime.now().strftime('%H:%M:%S'),
subtitle="Test subtitle",
content="This is a test post created to verify the system is working.",
difficulty=3,
published=False, # Pending review
author_id=current_user.id
)
db.session.add(post)
db.session.commit()
flash('Test post created successfully!', 'success')
return redirect(url_for('community.index'))
except Exception as e:
db.session.rollback()
flash(f'Error creating test post: {str(e)}', 'error')
return redirect(url_for('community.test_post'))
# Simple test form
return '''
<h2>Test Post Creation</h2>
<form method="POST">
<button type="submit">Create Test Post</button>
</form>
<a href="/community/">Back to Community</a>
'''
@community.route('/post/<int:id>/comment', methods=['POST'])
@login_required
def add_comment(id):
"""Add comment to post"""
post = Post.query.get_or_404(id)
form = CommentForm()
if form.validate_on_submit():
comment = Comment(
content=form.content.data,
author_id=current_user.id,
post_id=post.id
)
db.session.add(comment)
db.session.commit()
flash('Your comment has been added.', 'success')
return redirect(url_for('community.post_detail', id=id))
@community.route('/post/<int:id>/add-images', methods=['POST'])
@login_required
def add_images_to_post(id):
"""Add additional images to an existing post"""
post = Post.query.get_or_404(id)
# Check if user owns the post or is admin
if current_user.id != post.author_id and not current_user.is_admin:
return jsonify({'success': False, 'error': 'Unauthorized'})
try:
uploaded_images = []
# Handle multiple image uploads
if 'images' in request.files:
files = request.files.getlist('images')
for image_file in files:
if image_file and image_file.filename and MediaConfig.is_allowed_file(image_file.filename, 'images'):
result = save_image(image_file, post.id)
if result['success']:
# Create PostImage record
post_image = PostImage(
filename=result['filename'],
original_name=image_file.filename,
size=result['size'],
mime_type=result['mime_type'],
post_id=post.id,
is_cover=False,
description=request.form.get(f'description_{image_file.filename}', '')
)
db.session.add(post_image)
uploaded_images.append({
'filename': result['filename'],
'original_name': image_file.filename,
'url': post_image.get_url(),
'thumbnail_url': post_image.get_thumbnail_url()
})
db.session.commit()
return jsonify({
'success': True,
'message': f'Successfully uploaded {len(uploaded_images)} images',
'images': uploaded_images
})
except Exception as e:
db.session.rollback()
current_app.logger.error(f'Error adding images to post: {str(e)}')
return jsonify({'success': False, 'error': 'An error occurred while uploading images'})
@community.route('/post/<int:id>/like', methods=['POST'])
@login_required
def toggle_like(id):
"""Toggle like on post"""
post = Post.query.get_or_404(id)
existing_like = Like.query.filter_by(user_id=current_user.id, post_id=post.id).first()
if existing_like:
db.session.delete(existing_like)
liked = False
else:
like = Like(user_id=current_user.id, post_id=post.id)
db.session.add(like)
liked = True
db.session.commit()
return jsonify({'liked': liked, 'count': post.get_like_count()})
def save_image(image_file, post_id):
"""Save uploaded image file with thumbnails"""
try:
# Get post to access media folder
post = Post.query.get(post_id)
if not post or not post.media_folder:
raise ValueError("Post or media folder not found")
# Validate file type and MIME type
if not MediaConfig.is_allowed_file(image_file.filename, 'images'):
raise ValueError("File type not allowed")
# Create upload directory for this post
upload_dir = MediaConfig.get_media_path(current_app, post.media_folder, 'images')
os.makedirs(upload_dir, exist_ok=True)
# Create thumbnails directory
thumbnails_dir = os.path.join(upload_dir, 'thumbnails')
os.makedirs(thumbnails_dir, exist_ok=True)
# Generate unique filename
file_ext = image_file.filename.rsplit('.', 1)[1].lower()
filename = f"{uuid.uuid4().hex}.{file_ext}"
filepath = os.path.join(upload_dir, filename)
thumbnail_path = os.path.join(thumbnails_dir, filename)
# Open and process image
image = Image.open(image_file)
if image.mode in ('RGBA', 'LA', 'P'):
image = image.convert('RGB')
# Resize main image if too large
image.thumbnail(MediaConfig.IMAGE_MAX_SIZE, Image.Resampling.LANCZOS)
image.save(filepath, 'JPEG', quality=MediaConfig.IMAGE_QUALITY, optimize=True)
# Create thumbnail
thumbnail = image.copy()
thumbnail.thumbnail(MediaConfig.THUMBNAIL_SIZE, Image.Resampling.LANCZOS)
thumbnail.save(thumbnail_path, 'JPEG', quality=MediaConfig.IMAGE_QUALITY, optimize=True)
file_size = os.path.getsize(filepath)
return {
'success': True,
'filename': filename,
'size': file_size,
'mime_type': 'image/jpeg',
'has_thumbnail': True
}
except Exception as e:
current_app.logger.error(f'Error saving image: {str(e)}')
return {'success': False, 'error': str(e)}
def save_gpx_file(gpx_file, post_id):
"""Save uploaded GPX file with validation"""
try:
# Get post to access media folder
post = Post.query.get(post_id)
if not post or not post.media_folder:
raise ValueError("Post or media folder not found")
# Validate file type
if not MediaConfig.is_allowed_file(gpx_file.filename, 'gpx'):
raise ValueError("File type not allowed")
# Create upload directory for this post
upload_dir = MediaConfig.get_media_path(current_app, post.media_folder, 'gpx')
os.makedirs(upload_dir, exist_ok=True)
# Generate unique filename
file_ext = gpx_file.filename.rsplit('.', 1)[1].lower()
filename = f"{uuid.uuid4().hex}.{file_ext}"
filepath = os.path.join(upload_dir, filename)
# Validate GPX file content
gpx_content = gpx_file.read()
try:
gpx = gpxpy.parse(gpx_content.decode('utf-8'))
# Check if GPX has any tracks or waypoints
if not gpx.tracks and not gpx.waypoints and not gpx.routes:
raise ValueError("GPX file appears to be empty or invalid")
except Exception as e:
raise ValueError(f"Invalid GPX file: {str(e)}")
# Save file
with open(filepath, 'wb') as f:
f.write(gpx_content)
file_size = len(gpx_content)
# Extract basic statistics
total_distance = 0
total_points = 0
for track in gpx.tracks:
total_distance += track.length_2d() or 0
for segment in track.segments:
total_points += len(segment.points)
return {
'success': True,
'filename': filename,
'size': file_size,
'stats': {
'total_distance': round(total_distance / 1000, 2) if total_distance else 0, # Convert to km
'total_points': total_points,
'tracks_count': len(gpx.tracks),
'waypoints_count': len(gpx.waypoints)
}
}
except Exception as e:
current_app.logger.error(f'Error saving GPX file: {str(e)}')
return {
'success': False,
'error': str(e)
}
@community.route('/api/routes')
def api_routes():
"""API endpoint to get all routes for map display"""
posts_with_routes = Post.query.filter_by(published=True).join(GPXFile).all()
routes_data = []
for post in posts_with_routes:
for gpx_file in post.gpx_files:
try:
# Read and parse GPX file using new folder structure
if post.media_folder:
gpx_path = os.path.join(current_app.root_path, 'static', 'media', 'posts',
post.media_folder, 'gpx', gpx_file.filename)
else:
# Fallback to old path for existing files
gpx_path = os.path.join(current_app.instance_path, 'uploads', 'gpx', gpx_file.filename)
if os.path.exists(gpx_path):
with open(gpx_path, 'r') as f:
gpx_content = f.read()
gpx = gpxpy.parse(gpx_content)
# Extract coordinates
coordinates = []
for track in gpx.tracks:
for segment in track.segments:
for point in segment.points:
coordinates.append([point.latitude, point.longitude])
if coordinates:
routes_data.append({
'id': post.id,
'title': post.title,
'author': post.author.nickname,
'coordinates': coordinates,
'url': url_for('community.post_detail', id=post.id)
})
except Exception as e:
current_app.logger.error(f'Error processing GPX file {gpx_file.filename}: {str(e)}')
continue
return jsonify(routes_data)
@community.route('/media/posts/<post_folder>/images/<filename>')
def serve_image(post_folder, filename):
"""Serve post images with caching headers"""
try:
image_path = MediaConfig.get_media_path(current_app, post_folder, 'images')
file_path = os.path.join(image_path, filename)
if not os.path.exists(file_path):
flash('Image not found.', 'error')
return redirect(url_for('community.index'))
from flask import send_file
return send_file(file_path, as_attachment=False,
cache_timeout=86400) # Cache for 24 hours
except Exception as e:
current_app.logger.error(f'Error serving image: {str(e)}')
flash('Error loading image.', 'error')
return redirect(url_for('community.index'))
@community.route('/media/posts/<post_folder>/images/thumbnails/<filename>')
def serve_thumbnail(post_folder, filename):
"""Serve post image thumbnails with caching headers"""
try:
thumbnail_path = MediaConfig.get_media_path(current_app, post_folder, 'images')
thumbnail_path = os.path.join(thumbnail_path, 'thumbnails', filename)
if not os.path.exists(thumbnail_path):
# Fallback to main image
return serve_image(post_folder, filename)
from flask import send_file
return send_file(thumbnail_path, as_attachment=False,
cache_timeout=86400) # Cache for 24 hours
except Exception as e:
current_app.logger.error(f'Error serving thumbnail: {str(e)}')
return serve_image(post_folder, filename)
@community.route('/media/posts/<post_folder>/gpx/<filename>')
@login_required
def serve_gpx(post_folder, filename):
"""Serve GPX files for download - requires login"""
try:
gpx_path = MediaConfig.get_media_path(current_app, post_folder, 'gpx')
file_path = os.path.join(gpx_path, filename)
if not os.path.exists(file_path):
flash('GPX file not found.', 'error')
return redirect(url_for('community.index'))
from flask import send_file
return send_file(file_path, as_attachment=True,
download_name=f'route_{post_folder}.gpx',
mimetype='application/gpx+xml')
except Exception as e:
current_app.logger.error(f'Error serving GPX file: {str(e)}')
flash('Error downloading GPX file.', 'error')
return redirect(url_for('community.index'))