murilo/AttendanceManager
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: Migrate party logos and QR codes to S3-compatible object storage, introducing a new storage service and updating file operations.

Browse source
This commit is contained in:
Murilo Scolari 2026-01-30 23:51:23 -03:00
parent 0b709dd5e7
commit eea1c65ca5
8 changed files with 175 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -6,4 +6,5 @@ __pycache__/
data/
.env*
venv
persistent_storage/
persistent_storage/
os

6
app/config.py
View file

@ -27,3 +27,9 @@ class Config:
SQLALCHEMY_DATABASE_URI = DATABASE_URL
SQLALCHEMY_TRACK_MODIFICATIONS = False
# Backblaze B2 Configuration
B2_KEY_ID = os.environ.get('B2_KEY_ID')
B2_APP_KEY = os.environ.get('B2_APP_KEY')
B2_BUCKET_NAME = os.environ.get('B2_BUCKET_NAME')
B2_ENDPOINT_URL = os.environ.get('B2_ENDPOINT_URL')

7
app/models.py
View file

@ -84,8 +84,11 @@ class Guest(db.Model):
@property
def pix_qr_code_url(self):
if self.pix_qr_code_filename:
# Adjusted for Blueprint 'main'
return url_for('main.serve_persistent_file', filename=f"{Config.PAYMENT_QRCODES_FOLDER_NAME}/{self.pix_qr_code_filename}")
# Generate presigned URL from S3
from app.services.storage import get_storage_service
storage = get_storage_service()
key = f"{Config.PAYMENT_QRCODES_FOLDER_NAME}/{self.pix_qr_code_filename}"
return storage.get_presigned_url(key, expiration=3600)
return None
def get_check_in_time_str(self):

79
app/services/storage.py Normal file
View file

@ -0,0 +1,79 @@
import boto3
from botocore.exceptions import ClientError
from flask import current_app
from app.config import Config
class S3Service:
def __init__(self):
self.s3_client = boto3.client(
's3',
endpoint_url=Config.B2_ENDPOINT_URL,
aws_access_key_id=Config.B2_KEY_ID,
aws_secret_access_key=Config.B2_APP_KEY
)
self.bucket_name = Config.B2_BUCKET_NAME
def upload_file(self, file_obj, filename, folder="", content_type=None):
"""
Uploads a file-like object to B2.
:param file_obj: File-like object (bytes)
:param filename: Desired filename in bucket
:param folder: Optional folder prefix (e.g. 'party_logos')
:param content_type: MIME type of the file
:return: The full key (path) of the uploaded file or None if failed
"""
try:
full_key = f"{folder}/{filename}" if folder else filename
extra_args = {}
if content_type:
extra_args['ContentType'] = content_type
self.s3_client.upload_fileobj(
file_obj,
self.bucket_name,
full_key,
ExtraArgs=extra_args
)
return full_key
except ClientError as e:
current_app.logger.error(f"S3 Upload Error: {e}")
return None
def delete_file(self, filename):
"""
Deletes a file from B2.
:param filename: Full key of the file to delete
"""
try:
self.s3_client.delete_object(Bucket=self.bucket_name, Key=filename)
return True
except ClientError as e:
current_app.logger.error(f"S3 Delete Error: {e}")
return False
def get_presigned_url(self, filename, expiration=3600):
"""
Generates a presigned URL for downloading a file.
:param filename: Full key of the file
:param expiration: Time in seconds for the URL to remain valid
:return: Presigned URL string
"""
try:
response = self.s3_client.generate_presigned_url(
'get_object',
Params={'Bucket': self.bucket_name, 'Key': filename},
ExpiresIn=expiration
)
return response
except ClientError as e:
current_app.logger.error(f"S3 Presigned URL Error: {e}")
return None
# Singleton instance
_s3_service_instance = None
def get_storage_service():
global _s3_service_instance
if _s3_service_instance is None:
_s3_service_instance = S3Service()
return _s3_service_instance

54
app/utils.py
View file

@ -18,13 +18,14 @@ from fpdf import FPDF, XPos, YPos
from app.config import Config
from app.extensions import db
from app.services.storage import get_storage_service
# Import models inside functions or use strings to avoid circular imports?
# Usually strictly typing requires models. But Python is dynamic.
# I will import models inside functions where necessary or use argument passing style.
def save_base64_as_png(base64_string, charge_id):
"""Decodifica uma string base64 e a salva como um arquivo PNG."""
"""Decodifica uma string base64 e a envia para o S3."""
if not base64_string or not charge_id:
return None
try:
@ -33,25 +34,39 @@ def save_base64_as_png(base64_string, charge_id):
else:
clean_base64_string = base64_string
img_data = base64.b64decode(clean_base64_string)
img = Image.open(io.BytesIO(img_data))
# Cria objeto em memória para upload
img_io = io.BytesIO(img_data)
filename = f"{charge_id}.png"
save_path = os.path.join(Config.PAYMENT_QRCODES_SAVE_PATH, filename)
img.save(save_path, 'PNG')
return filename
storage = get_storage_service()
# Salva na pasta payment_qrcodes dentro do bucket
full_key = storage.upload_file(
img_io,
filename,
folder=Config.PAYMENT_QRCODES_FOLDER_NAME,
content_type="image/png"
)
if full_key:
return filename # Retornamos apenas o nome do arquivo para manter compatibilidade com o banco
return None
except Exception as e:
current_app.logger.error(f"Erro ao salvar imagem base64 para charge_id {charge_id}: {e}")
current_app.logger.error(f"Erro ao salvar imagem base64 no S3 para charge_id {charge_id}: {e}")
return None
def delete_pix_qr_code_file(guest):
"""Deleta o arquivo de imagem do QR Code do PIX associado a um convidado."""
"""Deleta o arquivo de imagem do QR Code do PIX associado a um convidado do S3."""
if guest and guest.pix_qr_code_filename:
try:
file_path = os.path.join(Config.PAYMENT_QRCODES_SAVE_PATH, guest.pix_qr_code_filename)
if os.path.exists(file_path):
os.remove(file_path)
current_app.logger.info(f"Arquivo QR Code PIX {guest.pix_qr_code_filename} deletado com sucesso.")
storage = get_storage_service()
folder = Config.PAYMENT_QRCODES_FOLDER_NAME
# O banco guarda 'filename.png', mas no S3 está em 'payment_qrcodes/filename.png'
key = f"{folder}/{guest.pix_qr_code_filename}"
if storage.delete_file(key):
current_app.logger.info(f"Arquivo QR Code PIX {key} deletado do S3 com sucesso.")
else:
current_app.logger.warning(f"Arquivo QR Code PIX {guest.pix_qr_code_filename} não encontrado para deleção.")
current_app.logger.warning(f"Falha ao deletar {key} do S3.")
except Exception as e:
current_app.logger.error(f"Erro ao deletar o arquivo QR Code PIX {guest.pix_qr_code_filename}: {e}")
@ -97,10 +112,17 @@ def generate_qr_code_image(qr_data, guest_name, party, output_format='PNG', font
logo_img_raw, logo_height = None, 0
if party.logo_filename:
logo_path = os.path.join(Config.PARTY_LOGOS_SAVE_PATH, party.logo_filename)
if os.path.exists(logo_path):
try:
# Load logo from S3
storage = get_storage_service()
key = f"{Config.PARTY_LOGOS_FOLDER_NAME}/{party.logo_filename}"
# Download image bytes from S3
response = storage.s3_client.get_object(Bucket=storage.bucket_name, Key=key)
logo_bytes = response['Body'].read()
logo_height = int(CARD_WIDTH / LOGO_ASPECT_RATIO)
logo_img_raw = Image.open(logo_path).convert("RGBA")
logo_img_raw = Image.open(io.BytesIO(logo_bytes)).convert("RGBA")
try:
vibrant_colors = get_vibrant_colors(logo_img_raw, num_colors=2)
if len(vibrant_colors) >= 1:
@ -109,6 +131,8 @@ def generate_qr_code_image(qr_data, guest_name, party, output_format='PNG', font
GRADIENT_START = tuple(int(c * 0.5) for c in main_color)
except Exception as e:
current_app.logger.warning(f"Não foi possível extrair cores: {e}.")
except Exception as e:
current_app.logger.warning(f"Falha ao carregar logo do S3: {e}")
qr_instance = qrcode.QRCode(version=1, error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=8, border=2)
qr_instance.add_data(qr_data)

11
app/views/main.py
View file

@ -4,6 +4,7 @@ from flask_login import login_required, current_user
from app.models import User, Guest, Party
from app.extensions import db
from app.config import Config
from app.services.storage import get_storage_service
bp = Blueprint('main', __name__)
@ -67,10 +68,16 @@ def complete_profile():
@bp.route('/persistent/<path:filename>')
def serve_persistent_file(filename):
"""
Serve arquivos da pasta de armazenamento persistente.
Redireciona para URL pré-assinada do S3 para servir arquivos.
"""
try:
return send_from_directory(Config.STORAGE_BASE_PATH, filename)
storage = get_storage_service()
presigned_url = storage.get_presigned_url(filename, expiration=3600)
if presigned_url:
return redirect(presigned_url)
else:
current_app.logger.error(f"Falha ao gerar URL pré-assinada para {filename}")
abort(404)
except Exception as e:
current_app.logger.error(f"Erro ao servir arquivo persistente {filename}: {e}")
abort(404)

44
app/views/party.py
View file

@ -17,6 +17,7 @@ from app.utils import (
PDF, get_all_guests_for_export, delete_pix_qr_code_file
)
from app.views import check_collaboration_permission
from app.services.storage import get_storage_service
bp = Blueprint('party', __name__)
@ -72,9 +73,14 @@ def delete_party(party_id):
flash("Apenas o dono da festa pode deletá-la.", "danger")
return redirect(url_for('main.dashboard'))
# Delete logo from S3 if exists
if party.logo_filename:
try: os.remove(os.path.join(Config.PARTY_LOGOS_SAVE_PATH, party.logo_filename))
except OSError: pass
try:
storage = get_storage_service()
key = f"{Config.PARTY_LOGOS_FOLDER_NAME}/{party.logo_filename}"
storage.delete_file(key)
except Exception as e:
current_app.logger.warning(f"Failed to delete logo from S3: {e}")
db.session.delete(party)
db.session.commit()
@ -93,20 +99,38 @@ def upload_logo(party_id):
if file.filename == '':
return jsonify({'success': False, 'message': 'Nenhum arquivo selecionado.'}), 400
if file and allowed_file(file.filename):
storage = get_storage_service()
# Delete old logo from S3 if exists
if party.logo_filename:
try:
os.remove(os.path.join(Config.PARTY_LOGOS_SAVE_PATH, party.logo_filename))
except OSError:
pass
old_key = f"{Config.PARTY_LOGOS_FOLDER_NAME}/{party.logo_filename}"
storage.delete_file(old_key)
except Exception as e:
current_app.logger.warning(f"Failed to delete old logo: {e}")
filename = secure_filename(file.filename)
unique_id = uuid.uuid4().hex
ext = filename.rsplit('.', 1)[1].lower()
new_filename = f"{party_id}_{unique_id}.{ext}"
file.save(os.path.join(Config.PARTY_LOGOS_SAVE_PATH, new_filename))
party.logo_filename = new_filename
db.session.commit()
logo_url = url_for('main.serve_persistent_file', filename=f'{Config.PARTY_LOGOS_FOLDER_NAME}/{new_filename}')
return jsonify({'success': True, 'message': 'Logo da festa atualizado!', 'logo_url': logo_url})
# Upload to S3
content_type = f"image/{ext}" if ext in ['png', 'jpg', 'jpeg', 'gif'] else 'application/octet-stream'
full_key = storage.upload_file(
file.stream,
new_filename,
folder=Config.PARTY_LOGOS_FOLDER_NAME,
content_type=content_type
)
if full_key:
party.logo_filename = new_filename
db.session.commit()
# Generate presigned URL for immediate display
logo_url = storage.get_presigned_url(full_key, expiration=3600)
return jsonify({'success': True, 'message': 'Logo da festa atualizado!', 'logo_url': logo_url})
else:
return jsonify({'success': False, 'message': 'Erro ao fazer upload do logo.'}), 500
else:
return jsonify({'success': False, 'message': 'Tipo de arquivo inválido.'}), 400

1
requirements.txt
View file

@ -43,4 +43,5 @@ typing_extensions
tzdata
urllib3
Werkzeug
boto3
WTForms