hack13/Personal-Digital-Asset-Manager
1
0
Fork 0
Code Issues Pull requests Projects Releases 4 Packages 1 Wiki Activity Actions

Decided to keep python based, added S3 support

Browse source
This commit is contained in:
Timothy Rogers 2025-05-24 10:39:31 -04:00
parent c7586d1aa1
commit c752138fb0
9 changed files with 293 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

12
Dockerfile
View file

@ -15,8 +15,20 @@ RUN apt-get update \
&& apt-get install -y --no-install-recommends \
build-essential \
libpq-dev \
imagemagick \
libmagickwand-dev \
libwebp-dev \
webp \
pkg-config \
&& rm -rf /var/lib/apt/lists/*
# Configure ImageMagick policy to allow WebP conversion with higher limits
RUN if [ -f /etc/ImageMagick-6/policy.xml ]; then \
sed -i 's/<policy domain="resource" name="memory" value="256MiB"\/>/<policy domain="resource" name="memory" value="1GiB"\/>/g' /etc/ImageMagick-6/policy.xml && \
sed -i 's/<policy domain="resource" name="disk" value="1GiB"\/>/<policy domain="resource" name="disk" value="4GiB"\/>/g' /etc/ImageMagick-6/policy.xml && \
sed -i 's/<policy domain="coder" rights="none" pattern="WEBP" \/>/<policy domain="coder" rights="read|write" pattern="WEBP" \/>/g' /etc/ImageMagick-6/policy.xml; \
fi
# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

82
app.py
View file

@ -6,6 +6,9 @@ from config import Config
from flask_migrate import Migrate
from extensions import db, migrate
from models import Asset, AssetFile
from storage import StorageBackend
from image_processor import ImageProcessor
from werkzeug.datastructures import FileStorage
def create_app():
app = Flask(__name__)
@ -14,20 +17,14 @@ def create_app():
# Ensure the instance folder exists
os.makedirs(app.instance_path, exist_ok=True)
# Ensure the uploads folder exists
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
# Initialize extensions
db.init_app(app)
migrate.init_app(app, db)
# Initialize database
#with app.app_context():
# db.create_all()
return app
app = create_app()
storage = StorageBackend(app.config['STORAGE_URL'])
def generate_unique_filename(original_filename):
"""Generate a unique filename while preserving the original extension"""
@ -36,16 +33,13 @@ def generate_unique_filename(original_filename):
# Generate a unique filename using UUID
return f"{uuid.uuid4().hex}{ext}"
def allowed_file(filename):
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'pdf', 'zip', 'spp', 'unitypackage', 'fbx', 'blend', 'webp'}
def allowed_file(filename, is_featured_image=False):
if is_featured_image:
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'webp'}
else:
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'pdf', 'zip', 'spp', 'unitypackage', 'fbx', 'blend', 'webp'}
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
def delete_file(filename):
if filename:
file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
if os.path.exists(file_path):
os.remove(file_path)
@app.route('/')
def index():
assets = Asset.query.order_by(Asset.created_at.desc()).all()
@ -60,13 +54,23 @@ def add_asset():
featured_image = request.files.get('featured_image')
additional_files = request.files.getlist('additional_files')
if title and featured_image and allowed_file(featured_image.filename):
if title and featured_image and allowed_file(featured_image.filename, is_featured_image=True):
# Process and convert featured image to WebP
processed_image, ext = ImageProcessor.process_featured_image(featured_image)
# Generate unique filename for featured image
original_featured_filename = secure_filename(featured_image.filename)
unique_featured_filename = generate_unique_filename(original_featured_filename)
unique_featured_filename = f"{uuid.uuid4().hex}{ext}"
# Save featured image with unique filename
featured_image.save(os.path.join(app.config['UPLOAD_FOLDER'], unique_featured_filename))
# Create a FileStorage object from the processed image
processed_file = FileStorage(
stream=processed_image,
filename=unique_featured_filename,
content_type='image/webp'
)
# Save featured image with unique filename using storage backend
storage.save(processed_file, unique_featured_filename)
# Create asset with unique filename
asset = Asset(
@ -84,7 +88,7 @@ def add_asset():
if file and allowed_file(file.filename):
original_filename = secure_filename(file.filename)
unique_filename = generate_unique_filename(original_filename)
file.save(os.path.join(app.config['UPLOAD_FOLDER'], unique_filename))
storage.save(file, unique_filename)
asset_file = AssetFile(
filename=unique_filename,
original_filename=original_filename,
@ -115,16 +119,29 @@ def edit_asset(id):
# Handle featured image update
featured_image = request.files.get('featured_image')
if featured_image and featured_image.filename and allowed_file(featured_image.filename):
if featured_image and featured_image.filename and allowed_file(featured_image.filename, is_featured_image=True):
# Delete old featured image
delete_file(asset.featured_image)
if asset.featured_image:
storage.delete(asset.featured_image)
# Save new featured image
# Process and convert featured image to WebP
processed_image, ext = ImageProcessor.process_featured_image(featured_image)
# Generate unique filename
original_featured_filename = secure_filename(featured_image.filename)
unique_featured_filename = generate_unique_filename(original_featured_filename)
# Save featured image with unique filename
featured_image.save(os.path.join(app.config['UPLOAD_FOLDER'], unique_featured_filename))
unique_featured_filename = f"{uuid.uuid4().hex}{ext}"
# Create a FileStorage object from the processed image
processed_file = FileStorage(
stream=processed_image,
filename=unique_featured_filename,
content_type='image/webp'
)
# Save the processed image
storage.save(processed_file, unique_featured_filename)
asset.featured_image = unique_featured_filename
asset.original_featured_image = original_featured_filename
# Handle additional files
additional_files = request.files.getlist('additional_files')
@ -132,7 +149,7 @@ def edit_asset(id):
if file and allowed_file(file.filename):
original_filename = secure_filename(file.filename)
unique_filename = generate_unique_filename(original_filename)
file.save(os.path.join(app.config['UPLOAD_FOLDER'], unique_filename))
storage.save(file, unique_filename)
asset_file = AssetFile(
filename=unique_filename,
original_filename=original_filename,
@ -151,11 +168,12 @@ def delete_asset(id):
asset = Asset.query.get_or_404(id)
# Delete featured image
delete_file(asset.featured_image)
if asset.featured_image:
storage.delete(asset.featured_image)
# Delete additional files
for file in asset.files:
delete_file(file.filename)
storage.delete(file.filename)
db.session.delete(file)
db.session.delete(asset)
@ -169,8 +187,8 @@ def delete_asset_file(id):
asset_file = AssetFile.query.get_or_404(id)
asset_id = asset_file.asset_id
# Delete the file
delete_file(asset_file.filename)
# Delete the file using storage backend
storage.delete(asset_file.filename)
# Remove from database
db.session.delete(asset_file)
@ -180,4 +198,4 @@ def delete_asset_file(id):
return redirect(url_for('asset_detail', id=asset_id))
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)
app.run(host='0.0.0.0', port=5432, debug=True)

21
config.py
View file

@ -1,4 +1,8 @@
import os
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
BASE_DIR = os.path.abspath(os.path.dirname(__file__))
@ -6,10 +10,23 @@ class Config:
SECRET_KEY = os.environ.get('SECRET_KEY', 'your-secret-key')
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'sqlite:///' + os.path.join(BASE_DIR, 'instance', 'app.db'))
SQLALCHEMY_TRACK_MODIFICATIONS = False
UPLOAD_FOLDER = os.path.join(BASE_DIR, 'static', 'uploads')
# Storage configuration
STORAGE_URL = os.environ.get('STORAGE_URL', 'file://' + os.path.join(BASE_DIR, 'static', 'uploads'))
UPLOAD_FOLDER = os.path.join(BASE_DIR, 'static', 'uploads') # Kept for backwards compatibility
# S3 Configuration (optional)
S3_ACCESS_KEY = os.environ.get('S3_ACCESS_KEY')
S3_SECRET_KEY = os.environ.get('S3_SECRET_KEY')
S3_ENDPOINT_URL = os.environ.get('S3_ENDPOINT_URL')
S3_PUBLIC_URL = os.environ.get('S3_PUBLIC_URL')
@staticmethod
def init_app(app):
# Create necessary directories
os.makedirs(os.path.dirname(app.config['SQLALCHEMY_DATABASE_URI'].replace('sqlite:///', '')), exist_ok=True)
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
# Only create upload folder if using local storage
if app.config['STORAGE_URL'].startswith('file://'):
storage_path = app.config['STORAGE_URL'].replace('file://', '')
os.makedirs(storage_path, exist_ok=True)

95
image_processor.py Normal file
View file

@ -0,0 +1,95 @@
import os
from PIL import Image
from wand.image import Image as WandImage
import io
from typing import BinaryIO, Tuple, Optional
class ImageProcessor:
@staticmethod
def is_animated_gif(file_storage) -> bool:
"""Check if the image is an animated GIF"""
try:
# Save current position
pos = file_storage.tell()
# Go to beginning
file_storage.seek(0)
with Image.open(file_storage) as img:
try:
img.seek(1) # Try to move to the second frame
is_animated = True
except EOFError:
is_animated = False
# Restore position
file_storage.seek(pos)
return is_animated
except Exception:
# Restore position in case of error
file_storage.seek(pos)
return False
@staticmethod
def convert_to_webp(file_storage, quality: int = 90) -> Tuple[BinaryIO, str]:
"""
Convert an image to WebP format.
Returns a tuple of (file_object, extension)
"""
# Save current position
pos = file_storage.tell()
# Go to beginning
file_storage.seek(0)
try:
# Check if it's an animated GIF
if ImageProcessor.is_animated_gif(file_storage):
# Convert animated GIF to animated WebP
file_storage.seek(0)
with WandImage(file=file_storage) as img:
# Configure WebP animation settings
img.format = 'WEBP'
# Higher quality settings for animation
img.options['webp:lossless'] = 'true' # Use lossless for animations
img.options['webp:method'] = '6' # Best compression method
img.options['webp:image-hint'] = 'graph' # Better for animations
img.options['webp:minimize-size'] = 'false' # Prioritize quality
# Animation specific settings
img.options['webp:animation-type'] = 'default'
img.options['webp:loop'] = '0' # Infinite loop
# Save with high quality
webp_bytes = io.BytesIO(img.make_blob(format='webp'))
webp_bytes.seek(0)
return webp_bytes, '.webp'
else:
# Handle static images
file_storage.seek(0)
with Image.open(file_storage) as img:
# Convert RGBA to RGB if necessary
if img.mode in ('RGBA', 'LA'):
background = Image.new('RGB', img.size, (255, 255, 255))
background.paste(img, mask=img.getchannel('A'))
img = background
elif img.mode != 'RGB':
img = img.convert('RGB')
# Save as WebP with high quality
output = io.BytesIO()
img.save(output,
format='WEBP',
quality=quality, # Higher quality
method=6, # Best compression method
lossless=False, # Use lossy for static images
exact=True) # Preserve color exactness
output.seek(0)
return output, '.webp'
finally:
# Restore original position
file_storage.seek(pos)
@staticmethod
def process_featured_image(file_storage) -> Tuple[BinaryIO, str]:
"""Process featured image, converting to WebP format"""
return ImageProcessor.convert_to_webp(file_storage, quality=90)

17
models.py
View file

@ -1,6 +1,7 @@
from datetime import datetime
from extensions import db
import bleach
from flask import current_app
ALLOWED_TAGS = [
'a', 'abbr', 'acronym', 'b', 'blockquote', 'code', 'em', 'i', 'li', 'ol',
@ -48,9 +49,25 @@ class Asset(db.Model):
strip=True
)
return ''
@property
def featured_image_url(self):
"""Get the URL for the featured image"""
from storage import StorageBackend
if self.featured_image:
storage = StorageBackend(current_app.config['STORAGE_URL'])
return storage.url_for(self.featured_image)
return None
class AssetFile(db.Model):
id = db.Column(db.Integer, primary_key=True)
filename = db.Column(db.String(200), nullable=False)
original_filename = db.Column(db.String(200))
asset_id = db.Column(db.Integer, db.ForeignKey('asset.id'), nullable=False)
@property
def file_url(self):
"""Get the URL for the file"""
from storage import StorageBackend
storage = StorageBackend(current_app.config['STORAGE_URL'])
return storage.url_for(self.filename)

5
requirements.txt
View file

@ -5,3 +5,8 @@ Flask-WTF
Pillow
bleach
gunicorn
fsspec>=2023.12.0
s3fs>=2023.12.0
python-dotenv>=1.0.0
pillow-avif-plugin>=1.3.1
Wand>=0.6.13

92
storage.py Normal file
View file

@ -0,0 +1,92 @@
import os
import fsspec
import asyncio
from typing import BinaryIO, Optional, Union
from urllib.parse import urlparse
from flask import current_app
from werkzeug.datastructures import FileStorage
class StorageBackend:
def __init__(self, storage_url: str):
"""
Initialize storage backend with a URL.
Examples:
- file:///path/to/storage (local filesystem)
- s3://bucket-name/path (S3 compatible)
"""
self.storage_url = storage_url
self.parsed_url = urlparse(storage_url)
self.protocol = self.parsed_url.scheme or 'file'
# Configure filesystem
if self.protocol == 's3':
self.fs = fsspec.filesystem(
's3',
key=os.getenv('S3_ACCESS_KEY'),
secret=os.getenv('S3_SECRET_KEY'),
endpoint_url=os.getenv('S3_ENDPOINT_URL'),
client_kwargs={
'endpoint_url': os.getenv('S3_ENDPOINT_URL')
} if os.getenv('S3_ENDPOINT_URL') else None
)
self.bucket = self.parsed_url.netloc
self.base_path = self.parsed_url.path.lstrip('/')
else:
self.fs = fsspec.filesystem('file')
self.base_path = self.parsed_url.path or '/uploads'
def _get_full_path(self, filename: str) -> str:
"""Get full path for a file"""
if self.protocol == 's3':
return os.path.join(self.base_path, filename)
return os.path.join(current_app.root_path, self.base_path, filename)
def save(self, file_storage: FileStorage, filename: str) -> str:
"""Save a file to storage"""
full_path = self._get_full_path(filename)
if self.protocol == 's3':
with self.fs.open(f"{self.bucket}/{full_path}", 'wb') as f:
file_storage.save(f)
return f"s3://{self.bucket}/{full_path}"
else:
os.makedirs(os.path.dirname(full_path), exist_ok=True)
file_storage.save(full_path)
return f"file://{full_path}"
def open(self, filename: str, mode: str = 'rb') -> BinaryIO:
"""Open a file from storage"""
full_path = self._get_full_path(filename)
if self.protocol == 's3':
return self.fs.open(f"{self.bucket}/{full_path}", mode)
return self.fs.open(full_path, mode)
def delete(self, filename: str) -> None:
"""Delete a file from storage"""
full_path = self._get_full_path(filename)
if self.protocol == 's3':
self.fs.delete(f"{self.bucket}/{full_path}")
else:
self.fs.delete(full_path)
def url_for(self, filename: str) -> str:
"""Get URL for a file"""
if self.protocol == 's3':
full_path = self._get_full_path(filename)
if os.getenv('S3_PUBLIC_URL'):
# For public buckets with a custom domain
return f"{os.getenv('S3_PUBLIC_URL')}/{full_path}"
elif os.getenv('S3_ENDPOINT_URL'):
# For MinIO/S3, construct direct URL
endpoint = os.getenv('S3_ENDPOINT_URL').rstrip('/')
return f"{endpoint}/{self.bucket}/{full_path}"
return f"s3://{self.bucket}/{full_path}"
else:
return f"/uploads/{filename}"
def exists(self, filename: str) -> bool:
"""Check if a file exists"""
full_path = self._get_full_path(filename)
if self.protocol == 's3':
return self.fs.exists(f"{self.bucket}/{full_path}")
return self.fs.exists(full_path)

4
templates/asset_detail.html
View file

@ -28,7 +28,7 @@
<div class="asset-content">
<div class="asset-main-image">
<img
src="{{ url_for('static', filename='uploads/' + asset.featured_image) }}"
src="{{ asset.featured_image_url }}"
alt="{{ asset.title }}"
/>
</div>
@ -71,7 +71,7 @@
{% for file in asset.files %}
<li class="file-item">
<a
href="{{ url_for('static', filename='uploads/' + file.filename) }}"
href="{{ file.file_url }}"
target="_blank"
class="file-link"
>

2
templates/index.html
View file

@ -8,7 +8,7 @@
<div class="asset-card">
<div class="asset-card-image">
<img
src="{{ url_for('static', filename='uploads/' + asset.featured_image) }}"
src="{{ asset.featured_image_url }}"
alt="{{ asset.title }}"
loading="lazy"
/>