Coverage for website/photos/tasks.py: 35.19%

1import logging 

2import os 

3 

4from django.db import transaction 

5from django.dispatch import Signal 

6from django.utils import timezone 

7 

8from celery import shared_task 

9from django_drf_filepond.models import TemporaryUpload, TemporaryUploadChunked 

10from django_drf_filepond.models import storage as filepond_storage 

11 

12from mailinglists.services import get_member_email_addresses 

13from members.models.member import Member 

14from photos.models import Album 

15from utils.snippets import send_email 

16 

17from .services import extract_archive 

18 

19logger = logging.getLogger(__name__) 

20album_uploaded = Signal() 

21 

22 

23@shared_task 

24def process_album_upload( 

25 archive_upload_id: str, album_id: int, uploader_id: int | None = None 

26): 

27 upload = TemporaryUpload.objects.get(upload_id=archive_upload_id) 

28 

29 try: 

30 album = Album.objects.get(id=album_id) 

31 except Album.DoesNotExist: 

32 logger.exception("Album %s does not exist", album_id) 

33 upload.file.delete() 

34 upload.delete() 

35 

36 uploader = ( 

37 Member.objects.filter(id=uploader_id).first() 

38 if uploader_id is not None 

39 else None 

40 ) 

41 

42 try: 

43 with transaction.atomic(): 

44 # We make the upload atomic separately, so we can keep using the db if it fails. 

45 # See https://docs.djangoproject.com/en/4.2/topics/db/transactions/#handling-exceptions-within-postgresql-transactions. 

46 warnings, count = extract_archive(album, upload.file) 

47 album.is_processing = False 

48 album.save() 

49 

50 # Send signal to notify that an album has been uploaded. This is used 

51 # by facedetection, and possibly in the future to notify the uploader. 

52 album_uploaded.send(sender=None, album=album) 

53 

54 if uploader is not None: 

55 # Notify uploader of the upload result. 

56 send_email( 

57 to=get_member_email_addresses(uploader), 

58 subject=("Album upload processed completed."), 

59 txt_template="photos/email/upload-processed.txt", 

60 context={ 

61 "name": uploader.first_name, 

62 "album": album, 

63 "upload_name": upload.upload_name, 

64 "warnings": warnings, 

65 "num_processed": count, 

66 }, 

67 ) 

68 except Exception as e: 

69 logger.exception(f"Failed to process album upload: {e}", exc_info=e) 

70 

71 finally: 

72 upload.file.delete() 

73 upload.delete() 

74 

75 

76@shared_task 

77def clean_broken_uploads(): 

78 # Cancel and remove completed uploads that are older than 12 hours. 

79 for upload in TemporaryUpload.objects.filter( 

80 uploaded__lte=timezone.now() - timezone.timedelta(hours=12) 

81 ): 

82 logger.info(f"Removing old upload {upload.upload_id}") 

83 upload.file.delete() 

84 upload.delete() 

85 

86 # Cancel and remove uploads that have not received new chunks in the last hour. 

87 for tuc in TemporaryUploadChunked.objects.filter( 

88 last_upload_time__lt=timezone.now() - timezone.timedelta(hours=1), 

89 ).exclude(upload_id__in=TemporaryUpload.objects.values("upload_id")): 

90 logger.info(f"Removing incomplete chunked upload {tuc.upload_id}") 

91 # Store the chunk and check if we've now completed the upload. 

92 upload_file = os.path.join( 

93 tuc.upload_dir, f"{tuc.file_id}_{tuc.last_chunk + 1}" 

94 ) 

95 filepond_storage.delete(upload_file) 

96 tuc.delete()