# uploads.py – utilitários de armazenamento/limite de HD import os import re from decimal import Decimal from typing import Iterable from django.apps import apps from django.conf import settings from django.core.files.storage import default_storage # ------------------------------------------------------------------------- # Configurações # ------------------------------------------------------------------------- # Limite padrão (pode ser sobrescrito no settings.py com HD_LIMIT_MB = ) DEFAULT_LIMITE_MB = 800 def _limite_mb() -> int: """Obtém o limite (em MB) a partir do settings, com fallback.""" return getattr(settings, "HD_LIMIT_MB", DEFAULT_LIMITE_MB) # ------------------------------------------------------------------------- # Funções de tamanho de pasta (para caminhos legados no disco) # ------------------------------------------------------------------------- def _calcular_tamanho_pasta(path: str) -> int: """Percorre recursivamente *path* (se existir) somando o tamanho (bytes).""" total = 0 if not os.path.exists(path): return 0 for dirpath, _, filenames in os.walk(path): for fname in filenames: fp = os.path.join(dirpath, fname) if os.path.isfile(fp): try: total += os.path.getsize(fp) except (FileNotFoundError, OSError): # Arquivo foi removido enquanto contávamos – ignora continue return total # ------------------------------------------------------------------------- # Cálculo de uso real do usuário # ------------------------------------------------------------------------- def _tamanho_arquivos_queryset(model_label: str, field_name: str, **filters) -> int: """ Soma (em bytes) o tamanho de todos os arquivos `field_name` no queryset `apps.get_model(model_label).objects.filter(**filters)`. Usa default_storage.size() → compatível com S3 e afins. """ Model = apps.get_model(*model_label.split(".", maxsplit=1)) # "app.Model" total = 0 # values_list devolve apenas o nome do arquivo (str) for fname in Model.objects.filter(**filters).values_list(field_name, flat=True): if fname: try: total += default_storage.size(fname) except (OSError, FileNotFoundError): # Arquivo não existe no storage; ignora continue return total def calcular_uso_total_usuario(user, base_media_path: str | None = None) -> int: """ Retorna o total de bytes ocupados por *todos* os arquivos do usuário. Conteúdo incluído: 1. Pasta local `MEDIA_ROOT/usuarios/uid_/` (arquivos legados) 2. Comprovantes em `pagamentos//comprovantes/` 3. Todos os FileFields nos modelos ligados ao usuário: Foto, FotoObra, Pagamento (comprovantes), DocumentoEtapa, EtapaObra (documento_orcamento / contrato_servico) """ base_dir = base_media_path or settings.MEDIA_ROOT total_bytes = 0 # --- 1) pasta "usuarios/uid_" total_bytes += _calcular_tamanho_pasta( os.path.join(base_dir, "usuarios", f"uid_{user.id}") ) # --- 2) comprovantes por tenant if getattr(user, "tenant", None): total_bytes += _calcular_tamanho_pasta( os.path.join( base_dir, "pagamentos", user.tenant.schema_name, "comprovantes" ) ) # --- 3) FileFields em modelos relacionais ----------------------------- # Observação: apps.get_model evita import circular. total_bytes += _tamanho_arquivos_queryset( "obras.Foto", "imagem", obra__usuario=user ) total_bytes += _tamanho_arquivos_queryset( "obras.FotoObra", "imagem", obra__usuario=user ) total_bytes += _tamanho_arquivos_queryset( "obras.Pagamento", "comprovante", obra__usuario=user ) total_bytes += _tamanho_arquivos_queryset( "obras.DocumentoEtapa", "arquivo", etapa__obra__usuario=user ) # Documentos diretos na Etapa total_bytes += _tamanho_arquivos_queryset( "obras.EtapaObra", "documento_orcamento", obra__usuario=user ) total_bytes += _tamanho_arquivos_queryset( "obras.EtapaObra", "contrato_servico", obra__usuario=user ) return total_bytes # ------------------------------------------------------------------------- # Verificação de limite e formatação # ------------------------------------------------------------------------- def atingiu_limite_hd(user, limite_mb: int | None = None) -> bool: """ True se o usuário já ultrapassou (ou alcançou) o limite. """ limite_mb = limite_mb or _limite_mb() return calcular_uso_total_usuario(user) >= limite_mb * 1024 * 1024 def uso_hd_formatado(user, limite_mb: int | None = None) -> dict[str, object]: """ Dados prontos para front-end: { "uso_mb": 42.13, "limite_mb": 800, "percentual": 5, "bloqueado": False } """ limite_mb = limite_mb or _limite_mb() total_bytes = calcular_uso_total_usuario(user) uso_mb = round(total_bytes / (1024 * 1024), 2) percentual = min(100, int((uso_mb / limite_mb) * 100) if limite_mb else 0) return { "uso_mb": uso_mb, "limite_mb": limite_mb, "percentual": percentual, "bloqueado": uso_mb >= limite_mb, } # ------------------------------------------------------------------------- # Caminhos de upload # ------------------------------------------------------------------------- def user_file_path(instance, filename, tipo: str = "outros") -> str: """ Gera `usuarios/uid_//`. """ try: user = instance.obra.usuario # quando o model tem relação via Obra except AttributeError: user = instance.usuario # fallback (upload direto pelo usuário) safe_filename = re.sub(r"[^A-Za-z0-9_.-]", "_", filename) return f"usuarios/uid_{user.id}/{tipo}/{safe_filename}" def pagamento_file_path(instance, filename) -> str: """ `pagamentos//comprovantes/` """ schema = instance.obra.usuario.tenant.schema_name safe_filename = re.sub(r"[^A-Za-z0-9_.-]", "_", filename) return f"pagamentos/{schema}/comprovantes/{safe_filename}" # ------------------------------------------------------------------------- # Salvamento manual respeitando limite # ------------------------------------------------------------------------- def salvar_arquivo_com_limite( user, arquivo_django, destino_relativo: str, limite_mb: int | None = None, ): """ Salva *arquivo_django* em `default_storage` no *destino_relativo*, desde que o usuário não tenha excedido o limite. Exemplo de uso: salvar_arquivo_com_limite( request.user, request.FILES["imagem"], f"usuarios/uid_{request.user.id}/extras/meu_arquivo.png", ) """ limite_mb = limite_mb or _limite_mb() if atingiu_limite_hd(user, limite_mb=limite_mb): raise Exception( f"Limite de {limite_mb} MB atingido. Apague arquivos ou adquira mais espaço." ) # Caso já exista, remove antes de sobrescrever if default_storage.exists(destino_relativo): default_storage.delete(destino_relativo) default_storage.save(destino_relativo, arquivo_django) from PIL import Image from io import BytesIO from django.core.files.uploadedfile import InMemoryUploadedFile import sys def comprimir_imagem(uploaded_file, max_width=600, max_height=600, qualidade=70): try: imagem = Image.open(uploaded_file) imagem_format = imagem.format # Redimensiona mantendo proporção imagem.thumbnail((max_width, max_height)) # Salva em memória buffer = BytesIO() imagem.save(buffer, format='JPEG', quality=qualidade, optimize=True) buffer.seek(0) # Nome original, mas com extensão padronizada para .jpg nome_arquivo = uploaded_file.name.rsplit('.', 1)[0] + '.jpg' return InMemoryUploadedFile( buffer, # conteúdo None, # campo nome_arquivo, # nome 'image/jpeg', # tipo MIME sys.getsizeof(buffer),# tamanho None # charset ) except Exception as e: print(f"Erro ao comprimir imagem: {e}") return uploaded_file # fallback: salva imagem original se der erro