from types import SimpleNamespace from django.core.files.base import ContentFile from django.core.files.storage import default_storage from django.db import models, IntegrityError from django.db.models import Model from authentication.models import ToolshedUser def hash_upload(instance, filename): return f"{instance.hash[:2]}/{instance.hash[2:4]}/{instance.hash[4:6]}/{instance.hash[6:]}" class FileManager(models.Manager): def get_or_create(self, **kwargs): if 'data' in kwargs and type(kwargs['data']) == str: import base64 from hashlib import sha256 content = base64.b64decode(kwargs['data'], validate=True) kwargs.pop('data') content_hash = sha256(content).hexdigest() kwargs['file'] = ContentFile(content, content_hash) kwargs['hash'] = content_hash else: raise ValueError('data must be a base64 encoded string or file and hash must be provided') try: return self.get(hash=kwargs['hash']), False except self.model.DoesNotExist: return self.create(**kwargs), True def create(self, **kwargs): if 'data' in kwargs and type(kwargs['data']) == str: import base64 from hashlib import sha256 content = base64.b64decode(kwargs['data'], validate=True) kwargs.pop('data') content_hash = sha256(content).hexdigest() kwargs['file'] = ContentFile(content, content_hash) kwargs['hash'] = content_hash elif 'file' in kwargs and 'hash' in kwargs and type(kwargs['file']) == ContentFile: pass else: raise ValueError('data must be a base64 encoded string or file and hash must be provided') if not self.filter(hash=kwargs['hash']).exists(): # The upload path is derived entirely from the hash (hash_upload, above), and hash # is DB-unique - so if no File row owns this hash yet, anything already sitting at # its computed path is necessarily a stale orphan (e.g. left behind by a bug in # cleanup code that deleted a File row without removing its stored bytes, or a # crashed upload). Clear it before saving instead of letting Django's storage layer # invent an alternate filename to avoid the "collision" - a suffixed name would # silently break every part of the app that derives this file's URL purely from its # hash (media serving, thumbnail generation, item/avatar attachment), and no future # caller could ever discover it again. expected_path = hash_upload(SimpleNamespace(hash=kwargs['hash']), '') if default_storage.exists(expected_path): default_storage.delete(expected_path) return super().create(**kwargs) else: raise IntegrityError('File with this hash already exists') class File(Model): file = models.FileField(upload_to=hash_upload, null=False, blank=False, unique=True) mime_type = models.CharField(max_length=255, null=False, blank=False) hash = models.CharField(max_length=64, null=False, blank=False, unique=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) objects = FileManager()