Move config.py, create_admin.py, manage_media.py, manage_routes.py to app/utils/ for better organization
This commit is contained in:
58
app/utils/config.py
Normal file
58
app/utils/config.py
Normal file
@@ -0,0 +1,58 @@
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
class Config:
|
||||
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key-change-in-production'
|
||||
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///moto_adventure.db'
|
||||
SQLALCHEMY_TRACK_MODIFICATIONS = False
|
||||
|
||||
# Media Configuration - Updated for new media system
|
||||
MEDIA_FOLDER = os.environ.get('MEDIA_FOLDER') or 'app/static/media/posts'
|
||||
MAX_CONTENT_LENGTH = int(os.environ.get('MAX_CONTENT_LENGTH') or 16 * 1024 * 1024) # 16MB
|
||||
|
||||
# Image Processing
|
||||
IMAGE_MAX_WIDTH = int(os.environ.get('IMAGE_MAX_WIDTH') or 1920)
|
||||
IMAGE_MAX_HEIGHT = int(os.environ.get('IMAGE_MAX_HEIGHT') or 1080)
|
||||
IMAGE_QUALITY = int(os.environ.get('IMAGE_QUALITY') or 85)
|
||||
THUMBNAIL_SIZE = int(os.environ.get('THUMBNAIL_SIZE') or 300)
|
||||
|
||||
# Legacy - keeping for backward compatibility
|
||||
UPLOAD_FOLDER = os.environ.get('UPLOAD_FOLDER') or 'app/static/uploads'
|
||||
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'webp', 'gpx'}
|
||||
|
||||
# Email Configuration
|
||||
MAIL_ENABLED = os.environ.get('MAIL_ENABLED', 'true').lower() in ['true', 'on', '1']
|
||||
MAIL_SERVER = os.environ.get('MAIL_SERVER') or 'smtp.gmail.com'
|
||||
MAIL_PORT = int(os.environ.get('MAIL_PORT') or 587)
|
||||
MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in ['true', 'on', '1']
|
||||
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
|
||||
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
|
||||
MAIL_DEFAULT_SENDER = os.environ.get('MAIL_DEFAULT_SENDER') or 'noreply@moto-adv.com'
|
||||
|
||||
# Pensiunea Buongusto Configuration
|
||||
PENSIUNEA_PHONE = os.environ.get('PENSIUNEA_PHONE') or '+40-xxx-xxx-xxx'
|
||||
PENSIUNEA_EMAIL = os.environ.get('PENSIUNEA_EMAIL') or 'info@pensiunebuongusto.ro'
|
||||
PENSIUNEA_WEBSITE = os.environ.get('PENSIUNEA_WEBSITE') or 'https://pensiunebuongusto.ro'
|
||||
|
||||
# Redis Configuration (for Celery)
|
||||
REDIS_URL = os.environ.get('REDIS_URL') or 'redis://localhost:6379/0'
|
||||
|
||||
class DevelopmentConfig(Config):
|
||||
DEBUG = True
|
||||
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or 'sqlite:///moto_adventure_dev.db'
|
||||
|
||||
class ProductionConfig(Config):
|
||||
DEBUG = False
|
||||
|
||||
class TestingConfig(Config):
|
||||
TESTING = True
|
||||
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
|
||||
|
||||
config = {
|
||||
'development': DevelopmentConfig,
|
||||
'production': ProductionConfig,
|
||||
'testing': TestingConfig,
|
||||
'default': DevelopmentConfig
|
||||
}
|
||||
25
app/utils/create_admin.py
Normal file
25
app/utils/create_admin.py
Normal file
@@ -0,0 +1,25 @@
|
||||
import os
|
||||
from app import create_app
|
||||
from app.extensions import db
|
||||
from app.models import User
|
||||
|
||||
def create_admin():
|
||||
app = create_app()
|
||||
with app.app_context():
|
||||
admin_email = os.environ.get('ADMIN_EMAIL')
|
||||
admin_nickname = os.environ.get('ADMIN_NICKNAME')
|
||||
admin_password = os.environ.get('ADMIN_PASSWORD')
|
||||
if not (admin_email and admin_nickname and admin_password):
|
||||
print("Missing ADMIN_EMAIL, ADMIN_NICKNAME, or ADMIN_PASSWORD in environment.")
|
||||
return
|
||||
if User.query.filter_by(email=admin_email).first():
|
||||
print(f"Admin with email {admin_email} already exists.")
|
||||
return
|
||||
user = User(nickname=admin_nickname, email=admin_email, is_admin=True, is_active=True)
|
||||
user.set_password(admin_password)
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
print(f"Admin user {admin_nickname} <{admin_email}> created.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
create_admin()
|
||||
187
app/utils/manage_media.py
Executable file
187
app/utils/manage_media.py
Executable file
@@ -0,0 +1,187 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Media Management Utility for Motorcycle Adventure Community
|
||||
|
||||
This script provides utilities for managing post media files:
|
||||
- Create missing media folders for existing posts
|
||||
- Clean up orphaned media folders
|
||||
- Migrate files from old structure to new structure
|
||||
- Generate thumbnails for images
|
||||
|
||||
Usage:
|
||||
python manage_media.py --help
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import argparse
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
import uuid
|
||||
|
||||
# Add the app directory to the path
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'app'))
|
||||
|
||||
from app import create_app
|
||||
from app.models import Post, PostImage, GPXFile
|
||||
from app.extensions import db
|
||||
|
||||
def create_missing_media_folders():
|
||||
"""Create media folders for existing posts that don't have them"""
|
||||
app = create_app()
|
||||
|
||||
with app.app_context():
|
||||
posts_without_folders = Post.query.filter_by(media_folder=None).all()
|
||||
|
||||
print(f"Found {len(posts_without_folders)} posts without media folders")
|
||||
|
||||
for post in posts_without_folders:
|
||||
# Generate a media folder name
|
||||
folder_name = f"post_{uuid.uuid4().hex[:8]}_{post.created_at.strftime('%Y%m%d')}"
|
||||
post.media_folder = folder_name
|
||||
|
||||
# Create the folder structure
|
||||
media_path = os.path.join(app.root_path, 'static', 'media', 'posts', folder_name)
|
||||
os.makedirs(os.path.join(media_path, 'images'), exist_ok=True)
|
||||
os.makedirs(os.path.join(media_path, 'gpx'), exist_ok=True)
|
||||
|
||||
print(f"Created media folder for post {post.id}: {folder_name}")
|
||||
|
||||
db.session.commit()
|
||||
print("Media folders created successfully!")
|
||||
|
||||
def migrate_old_files():
|
||||
"""Migrate files from old upload structure to new media structure"""
|
||||
app = create_app()
|
||||
|
||||
with app.app_context():
|
||||
# Migrate images
|
||||
old_images_path = os.path.join(app.instance_path, 'uploads', 'images')
|
||||
if os.path.exists(old_images_path):
|
||||
print("Migrating image files...")
|
||||
|
||||
for image in PostImage.query.all():
|
||||
if image.post.media_folder:
|
||||
old_path = os.path.join(old_images_path, image.filename)
|
||||
new_path = os.path.join(app.root_path, 'static', 'media', 'posts',
|
||||
image.post.media_folder, 'images', image.filename)
|
||||
|
||||
if os.path.exists(old_path) and not os.path.exists(new_path):
|
||||
os.makedirs(os.path.dirname(new_path), exist_ok=True)
|
||||
shutil.move(old_path, new_path)
|
||||
print(f"Moved image: {image.filename}")
|
||||
|
||||
# Migrate GPX files
|
||||
old_gpx_path = os.path.join(app.instance_path, 'uploads', 'gpx')
|
||||
if os.path.exists(old_gpx_path):
|
||||
print("Migrating GPX files...")
|
||||
|
||||
for gpx_file in GPXFile.query.all():
|
||||
if gpx_file.post.media_folder:
|
||||
old_path = os.path.join(old_gpx_path, gpx_file.filename)
|
||||
new_path = os.path.join(app.root_path, 'static', 'media', 'posts',
|
||||
gpx_file.post.media_folder, 'gpx', gpx_file.filename)
|
||||
|
||||
if os.path.exists(old_path) and not os.path.exists(new_path):
|
||||
os.makedirs(os.path.dirname(new_path), exist_ok=True)
|
||||
shutil.move(old_path, new_path)
|
||||
print(f"Moved GPX file: {gpx_file.filename}")
|
||||
|
||||
print("File migration completed!")
|
||||
|
||||
def clean_orphaned_folders():
|
||||
"""Remove media folders that don't have corresponding posts"""
|
||||
app = create_app()
|
||||
|
||||
with app.app_context():
|
||||
media_posts_path = os.path.join(app.root_path, 'static', 'media', 'posts')
|
||||
|
||||
if not os.path.exists(media_posts_path):
|
||||
print("No media posts directory found")
|
||||
return
|
||||
|
||||
# Get all folder names from database
|
||||
used_folders = set(post.media_folder for post in Post.query.filter(Post.media_folder.isnot(None)).all())
|
||||
|
||||
# Get all actual folders
|
||||
actual_folders = set(name for name in os.listdir(media_posts_path)
|
||||
if os.path.isdir(os.path.join(media_posts_path, name)))
|
||||
|
||||
# Find orphaned folders
|
||||
orphaned_folders = actual_folders - used_folders
|
||||
|
||||
if orphaned_folders:
|
||||
print(f"Found {len(orphaned_folders)} orphaned folders:")
|
||||
for folder in orphaned_folders:
|
||||
folder_path = os.path.join(media_posts_path, folder)
|
||||
print(f" {folder}")
|
||||
|
||||
# Ask for confirmation before deleting
|
||||
response = input(f"Delete folder {folder}? (y/N): ")
|
||||
if response.lower() == 'y':
|
||||
shutil.rmtree(folder_path)
|
||||
print(f" Deleted: {folder}")
|
||||
else:
|
||||
print(f" Skipped: {folder}")
|
||||
else:
|
||||
print("No orphaned folders found")
|
||||
|
||||
def show_media_stats():
|
||||
"""Show statistics about media storage"""
|
||||
app = create_app()
|
||||
|
||||
with app.app_context():
|
||||
total_posts = Post.query.count()
|
||||
posts_with_media_folders = Post.query.filter(Post.media_folder.isnot(None)).count()
|
||||
total_images = PostImage.query.count()
|
||||
total_gpx_files = GPXFile.query.count()
|
||||
|
||||
print("Media Storage Statistics:")
|
||||
print(f" Total posts: {total_posts}")
|
||||
print(f" Posts with media folders: {posts_with_media_folders}")
|
||||
print(f" Posts without media folders: {total_posts - posts_with_media_folders}")
|
||||
print(f" Total images: {total_images}")
|
||||
print(f" Total GPX files: {total_gpx_files}")
|
||||
|
||||
# Calculate total storage used
|
||||
media_posts_path = os.path.join(app.root_path, 'static', 'media', 'posts')
|
||||
if os.path.exists(media_posts_path):
|
||||
total_size = 0
|
||||
for root, dirs, files in os.walk(media_posts_path):
|
||||
for file in files:
|
||||
file_path = os.path.join(root, file)
|
||||
total_size += os.path.getsize(file_path)
|
||||
|
||||
print(f" Total storage used: {total_size / (1024*1024):.2f} MB")
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description='Media Management Utility')
|
||||
parser.add_argument('--create-folders', action='store_true',
|
||||
help='Create missing media folders for existing posts')
|
||||
parser.add_argument('--migrate-files', action='store_true',
|
||||
help='Migrate files from old structure to new structure')
|
||||
parser.add_argument('--clean-orphaned', action='store_true',
|
||||
help='Clean up orphaned media folders')
|
||||
parser.add_argument('--stats', action='store_true',
|
||||
help='Show media storage statistics')
|
||||
parser.add_argument('--all', action='store_true',
|
||||
help='Run all operations (create folders, migrate files)')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.all:
|
||||
create_missing_media_folders()
|
||||
migrate_old_files()
|
||||
elif args.create_folders:
|
||||
create_missing_media_folders()
|
||||
elif args.migrate_files:
|
||||
migrate_old_files()
|
||||
elif args.clean_orphaned:
|
||||
clean_orphaned_folders()
|
||||
elif args.stats:
|
||||
show_media_stats()
|
||||
else:
|
||||
parser.print_help()
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
228
app/utils/manage_routes.py
Executable file
228
app/utils/manage_routes.py
Executable file
@@ -0,0 +1,228 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Map Route Management Script
|
||||
Utility for managing map routes in the database
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
from flask import Flask
|
||||
|
||||
# Add the project root to Python path
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from app import create_app, db
|
||||
from app.models import Post, GPXFile, MapRoute, User
|
||||
from app.utils.gpx_processor import create_map_route_from_gpx, process_post_approval
|
||||
|
||||
|
||||
def show_help():
|
||||
"""Show available commands"""
|
||||
print("""
|
||||
🗺️ Map Route Management Script
|
||||
|
||||
Available commands:
|
||||
list - List all map routes
|
||||
create <post_id> - Create map route for a specific post
|
||||
recreate <post_id> - Recreate map route for a specific post
|
||||
recreate-all - Recreate all map routes
|
||||
stats - Show database statistics
|
||||
cleanup - Remove map routes for unpublished posts
|
||||
help - Show this help message
|
||||
|
||||
Examples:
|
||||
python manage_routes.py list
|
||||
python manage_routes.py create 1
|
||||
python manage_routes.py recreate-all
|
||||
python manage_routes.py stats
|
||||
""")
|
||||
|
||||
|
||||
def list_routes():
|
||||
"""List all map routes"""
|
||||
app = create_app()
|
||||
with app.app_context():
|
||||
routes = MapRoute.query.join(Post).all()
|
||||
|
||||
if not routes:
|
||||
print("❌ No map routes found")
|
||||
return
|
||||
|
||||
print(f"📍 Found {len(routes)} map routes:\n")
|
||||
print("ID | Post | Title | Author | Published | Points | Distance")
|
||||
print("-" * 70)
|
||||
|
||||
for route in routes:
|
||||
status = "✅" if route.post.published else "❌"
|
||||
print(f"{route.id:2d} | {route.post_id:4d} | {route.post.title[:20]:20s} | {route.post.author.nickname[:10]:10s} | {status:2s} | {route.simplified_points:6d} | {route.total_distance:7.2f} km")
|
||||
|
||||
|
||||
def create_route(post_id):
|
||||
"""Create map route for a specific post"""
|
||||
app = create_app()
|
||||
with app.app_context():
|
||||
post = Post.query.get(post_id)
|
||||
if not post:
|
||||
print(f"❌ Post {post_id} not found")
|
||||
return False
|
||||
|
||||
gpx_files = GPXFile.query.filter_by(post_id=post_id).all()
|
||||
if not gpx_files:
|
||||
print(f"❌ No GPX files found for post {post_id}")
|
||||
return False
|
||||
|
||||
print(f"🔄 Creating map route for post {post_id}: {post.title}")
|
||||
success = create_map_route_from_gpx(gpx_files[0].id)
|
||||
|
||||
if success:
|
||||
print(f"✅ Successfully created map route for post {post_id}")
|
||||
else:
|
||||
print(f"❌ Failed to create map route for post {post_id}")
|
||||
|
||||
return success
|
||||
|
||||
|
||||
def recreate_route(post_id):
|
||||
"""Recreate map route for a specific post"""
|
||||
app = create_app()
|
||||
with app.app_context():
|
||||
# Delete existing route
|
||||
existing = MapRoute.query.filter_by(post_id=post_id).first()
|
||||
if existing:
|
||||
db.session.delete(existing)
|
||||
db.session.commit()
|
||||
print(f"🗑️ Deleted existing map route for post {post_id}")
|
||||
|
||||
# Create new route
|
||||
return create_route(post_id)
|
||||
|
||||
|
||||
def recreate_all_routes():
|
||||
"""Recreate all map routes"""
|
||||
app = create_app()
|
||||
with app.app_context():
|
||||
# Get all posts with GPX files
|
||||
posts_with_gpx = db.session.query(Post).join(GPXFile).distinct().all()
|
||||
|
||||
if not posts_with_gpx:
|
||||
print("❌ No posts with GPX files found")
|
||||
return
|
||||
|
||||
print(f"🔄 Recreating map routes for {len(posts_with_gpx)} posts...")
|
||||
|
||||
# Delete all existing routes
|
||||
MapRoute.query.delete()
|
||||
db.session.commit()
|
||||
print("🗑️ Deleted all existing map routes")
|
||||
|
||||
success_count = 0
|
||||
error_count = 0
|
||||
|
||||
for post in posts_with_gpx:
|
||||
print(f"\n🔄 Processing post {post.id}: {post.title}")
|
||||
success = process_post_approval(post.id)
|
||||
if success:
|
||||
success_count += 1
|
||||
else:
|
||||
error_count += 1
|
||||
|
||||
print(f"\n📊 Results:")
|
||||
print(f"✅ Successfully processed: {success_count}")
|
||||
print(f"❌ Errors: {error_count}")
|
||||
print(f"📍 Total posts: {len(posts_with_gpx)}")
|
||||
|
||||
|
||||
def show_stats():
|
||||
"""Show database statistics"""
|
||||
app = create_app()
|
||||
with app.app_context():
|
||||
total_posts = Post.query.count()
|
||||
published_posts = Post.query.filter_by(published=True).count()
|
||||
posts_with_gpx = db.session.query(Post).join(GPXFile).distinct().count()
|
||||
total_routes = MapRoute.query.count()
|
||||
published_routes = MapRoute.query.join(Post).filter(Post.published == True).count()
|
||||
|
||||
print("📊 Database Statistics:")
|
||||
print(f" 📝 Total posts: {total_posts}")
|
||||
print(f" ✅ Published posts: {published_posts}")
|
||||
print(f" 🗺️ Posts with GPX: {posts_with_gpx}")
|
||||
print(f" 📍 Total map routes: {total_routes}")
|
||||
print(f" 🌍 Published routes: {published_routes}")
|
||||
|
||||
if total_routes > 0:
|
||||
routes = MapRoute.query.all()
|
||||
total_points = sum(r.total_points for r in routes)
|
||||
simplified_points = sum(r.simplified_points for r in routes)
|
||||
total_distance = sum(r.total_distance for r in routes)
|
||||
|
||||
print(f"\n🎯 Route Statistics:")
|
||||
print(f" 📊 Total coordinate points: {total_points:,}")
|
||||
print(f" 🎯 Simplified points: {simplified_points:,}")
|
||||
print(f" 📏 Total distance: {total_distance:.2f} km")
|
||||
print(f" 🗜️ Compression ratio: {simplified_points/total_points*100:.1f}%")
|
||||
|
||||
|
||||
def cleanup_routes():
|
||||
"""Remove map routes for unpublished posts"""
|
||||
app = create_app()
|
||||
with app.app_context():
|
||||
unpublished_routes = MapRoute.query.join(Post).filter(Post.published == False).all()
|
||||
|
||||
if not unpublished_routes:
|
||||
print("✅ No unpublished routes to clean up")
|
||||
return
|
||||
|
||||
print(f"🗑️ Found {len(unpublished_routes)} routes for unpublished posts")
|
||||
|
||||
for route in unpublished_routes:
|
||||
print(f" Removing route for post {route.post_id}: {route.post.title}")
|
||||
db.session.delete(route)
|
||||
|
||||
db.session.commit()
|
||||
print(f"✅ Cleaned up {len(unpublished_routes)} routes")
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
show_help()
|
||||
return
|
||||
|
||||
command = sys.argv[1].lower()
|
||||
|
||||
if command == 'help':
|
||||
show_help()
|
||||
elif command == 'list':
|
||||
list_routes()
|
||||
elif command == 'create':
|
||||
if len(sys.argv) < 3:
|
||||
print("❌ Please provide a post ID")
|
||||
print("Usage: python manage_routes.py create <post_id>")
|
||||
return
|
||||
try:
|
||||
post_id = int(sys.argv[2])
|
||||
create_route(post_id)
|
||||
except ValueError:
|
||||
print("❌ Invalid post ID. Please provide a number.")
|
||||
elif command == 'recreate':
|
||||
if len(sys.argv) < 3:
|
||||
print("❌ Please provide a post ID")
|
||||
print("Usage: python manage_routes.py recreate <post_id>")
|
||||
return
|
||||
try:
|
||||
post_id = int(sys.argv[2])
|
||||
recreate_route(post_id)
|
||||
except ValueError:
|
||||
print("❌ Invalid post ID. Please provide a number.")
|
||||
elif command == 'recreate-all':
|
||||
recreate_all_routes()
|
||||
elif command == 'stats':
|
||||
show_stats()
|
||||
elif command == 'cleanup':
|
||||
cleanup_routes()
|
||||
else:
|
||||
print(f"❌ Unknown command: {command}")
|
||||
show_help()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user