__init__.py
185 lines
| 5.9 KiB
| text/x-python
|
PythonLexer
|
r1824 | from itertools import zip_longest | ||
|
r1590 | import boards | ||
from boards.models import STATUS_ARCHIVE | ||||
from django.core.files.images import get_image_dimensions | ||||
|
r1368 | from django.db import models | ||
|
r1273 | |||
|
r1305 | from boards import utils | ||
|
r1590 | from boards.models.attachment.viewers import get_viewers, AbstractViewer, \ | ||
FILE_TYPES_IMAGE | ||||
|
r1866 | from boards.utils import get_upload_filename, get_extension, cached_result, \ | ||
get_file_mimetype | ||||
|
r1273 | |||
class AttachmentManager(models.Manager): | ||||
def create_with_hash(self, file): | ||||
|
r1305 | file_hash = utils.get_file_hash(file) | ||
|
r1855 | attachment = self.get_existing_duplicate(file_hash, file) | ||
|
r1824 | if not attachment: | ||
|
r1866 | file_type = get_file_mimetype(file) | ||
|
r1371 | attachment = self.create(file=file, mimetype=file_type, | ||
hash=file_hash) | ||||
|
r1273 | |||
return attachment | ||||
|
r1660 | def create_from_url(self, url): | ||
existing = self.filter(url=url) | ||||
if len(existing) > 0: | ||||
attachment = existing[0] | ||||
else: | ||||
attachment = self.create(url=url) | ||||
return attachment | ||||
|
r1590 | def get_random_images(self, count, tags=None): | ||
images = self.filter(mimetype__in=FILE_TYPES_IMAGE).exclude( | ||||
|
r1592 | attachment_posts__thread__status=STATUS_ARCHIVE) | ||
|
r1590 | if tags is not None: | ||
|
r1592 | images = images.filter(attachment_posts__threads__tags__in=tags) | ||
|
r1590 | return images.order_by('?')[:count] | ||
|
r1855 | def get_existing_duplicate(self, file_hash, file): | ||
|
r1824 | """ | ||
Gets an attachment with the same file if one exists. | ||||
""" | ||||
existing = self.filter(hash=file_hash) | ||||
attachment = None | ||||
for existing_attachment in existing: | ||||
existing_file = existing_attachment.file | ||||
|
r1857 | |||
file_chunks = file.chunks() | ||||
existing_file_chunks = existing_file.chunks() | ||||
if self._compare_chunks(file_chunks, existing_file_chunks): | ||||
|
r1856 | attachment = existing_attachment | ||
return attachment | ||||
|
r1824 | |||
|
r1937 | def get_by_alias(self, name): | ||
|
r1951 | pack_name, sticker_name = name.split('/') | ||
|
r1938 | try: | ||
|
r1951 | return AttachmentSticker.objects.get(name=sticker_name, stickerpack__name=pack_name).attachment | ||
|
r1938 | except AttachmentSticker.DoesNotExist: | ||
return None | ||||
|
r1937 | |||
|
r1857 | def _compare_chunks(self, chunks1, chunks2): | ||
""" | ||||
Compares 2 chunks of different sizes (e.g. first chunk array contains | ||||
all data in 1 chunk, and other one -- in a multiple of smaller ones. | ||||
""" | ||||
equal = True | ||||
position1 = 0 | ||||
position2 = 0 | ||||
chunk1 = None | ||||
chunk2 = None | ||||
chunk1ended = False | ||||
chunk2ended = False | ||||
while True: | ||||
if not chunk1 or len(chunk1) <= position1: | ||||
try: | ||||
chunk1 = chunks1.__next__() | ||||
position1 = 0 | ||||
except StopIteration: | ||||
chunk1ended = True | ||||
if not chunk2 or len(chunk2) <= position2: | ||||
try: | ||||
chunk2 = chunks2.__next__() | ||||
position2 = 0 | ||||
except StopIteration: | ||||
chunk2ended = True | ||||
if chunk1ended and chunk2ended: | ||||
|
r1858 | # Same size chunksm checked for equality previously | ||
break | ||||
elif chunk1ended or chunk2ended: | ||||
# Different size chunks, not equal | ||||
equal = False | ||||
|
r1857 | break | ||
elif chunk1[position1] != chunk2[position2]: | ||||
|
r1858 | # Different bytes, not equal | ||
|
r1857 | equal = False | ||
break | ||||
else: | ||||
position1 += 1 | ||||
position2 += 1 | ||||
return equal | ||||
|
r1273 | |||
class Attachment(models.Model): | ||||
objects = AttachmentManager() | ||||
|
r1759 | class Meta: | ||
app_label = 'boards' | ||||
ordering = ('id',) | ||||
|
r1660 | file = models.FileField(upload_to=get_upload_filename, null=True) | ||
|
r1866 | mimetype = models.CharField(max_length=200, null=True) | ||
|
r1660 | hash = models.CharField(max_length=36, null=True) | ||
|
r1750 | url = models.TextField(blank=True, default='') | ||
|
r1273 | |||
def get_view(self): | ||||
file_viewer = None | ||||
|
r1286 | for viewer in get_viewers(): | ||
|
r1273 | if viewer.supports(self.mimetype): | ||
|
r1306 | file_viewer = viewer | ||
|
r1273 | break | ||
if file_viewer is None: | ||||
|
r1306 | file_viewer = AbstractViewer | ||
|
r1273 | |||
|
r1940 | return file_viewer(self.file, self.mimetype, self.id, self.url).get_view() | ||
|
r1273 | |||
|
r1389 | def __str__(self): | ||
|
r1660 | return self.url or self.file.url | ||
|
r1590 | |||
def get_random_associated_post(self): | ||||
posts = boards.models.Post.objects.filter(attachments__in=[self]) | ||||
return posts.order_by('?').first() | ||||
@cached_result() | ||||
def get_size(self): | ||||
|
r1660 | if self.file: | ||
if self.mimetype in FILE_TYPES_IMAGE: | ||||
return get_image_dimensions(self.file) | ||||
else: | ||||
return 200, 150 | ||||
|
r1590 | |||
def get_thumb_url(self): | ||||
split = self.file.url.rsplit('.', 1) | ||||
w, h = 200, 150 | ||||
return '%s.%sx%s.%s' % (split[0], w, h, split[1]) | ||||
@cached_result() | ||||
def get_preview_size(self): | ||||
|
r1866 | size = 200, 150 | ||
|
r1590 | if self.mimetype in FILE_TYPES_IMAGE: | ||
preview_path = self.file.path.replace('.', '.200x150.') | ||||
|
r1866 | try: | ||
size = get_image_dimensions(preview_path) | ||||
except Exception: | ||||
pass | ||||
return size | ||||
|
r1706 | |||
def is_internal(self): | ||||
return self.url is None or len(self.url) == 0 | ||||
|
r1937 | |||
|
r1951 | class StickerPack(models.Model): | ||
name = models.TextField(unique=True) | ||||
tripcode = models.TextField(blank=True) | ||||
def __str__(self): | ||||
return self.name | ||||
|
r1937 | class AttachmentSticker(models.Model): | ||
|
r1986 | attachment = models.ForeignKey('Attachment', on_delete=models.CASCADE) | ||
|
r1937 | name = models.TextField(unique=True) | ||
|
r1986 | stickerpack = models.ForeignKey('StickerPack', on_delete=models.CASCADE) | ||
|
r1937 | |||
def __str__(self): | ||||
|
r1951 | # Local stickers do not have a sticker pack | ||
if hasattr(self, 'stickerpack'): | ||||
return '{}/{}'.format(str(self.stickerpack), self.name) | ||||
else: | ||||
return self.name | ||||