785eb3cf-f76e-4c4c-8a58-787ccc3990bc

Committed 22 May 2024 10:30AM UTC coverage: 68.125% (-0.1%) from 68.234%

Build # 785eb3cf-f76e-4c4c-8a58-787ccc3990bc

Build Type

push

circleci

Committed by

VirginiaDooley

Commit Message

Add image rotation before facial recognition

Run Details

1528 of 2746 branches covered (55.64%)

Branch coverage included in aggregate %.

0 of 17 new or added lines in 1 file covered. (0.0%)

7 existing lines in 3 files now uncovered.

7314 of 10233 relevant lines covered (71.47%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.35

/ynr/apps/moderation_queue/forms.py

import cgi
from io import BytesIO

import requests
import sorl
from candidates.models.db import ActionType, LoggedAction
from candidates.views.version_data import get_change_metadata, get_client_ip
from django import forms
from django.conf import settings
from django.contrib.sites.models import Site
from django.core.exceptions import ValidationError
from django.core.mail import send_mail
from django.template.loader import render_to_string
from django.urls import reverse
from moderation_queue.helpers import (
    convert_image_to_png,
)
from people.forms.forms import StrippedCharField
from PIL import Image as PILImage

from .models import CopyrightOptions, QueuedImage, SuggestedPostLock


class UploadPersonPhotoImageForm(forms.ModelForm):
    class Meta:
        model = QueuedImage
        fields = [
            "image",
            "why_allowed",
            "justification_for_use",
            "person",
            "decision",
        ]
        widgets = {
            "person": forms.HiddenInput(),
            "decision": forms.HiddenInput(),
            "why_allowed": forms.RadioSelect(),
            "justification_for_use": forms.Textarea(
                attrs={"rows": 1, "columns": 72},
            ),
        }

    def clean(self):
        cleaned_data = super().clean()
        justification_for_use = cleaned_data.get(
            "justification_for_use", ""
        ).strip()
        why_allowed = cleaned_data.get("why_allowed")
        if why_allowed == "other" and not justification_for_use:
            message = (
                "If you checked 'Other' then you must provide a "
                "justification for why we can use it."
            )
            raise ValidationError(message)
        return cleaned_data

    def save(self, commit):
        """
        Before saving, resize and rotate the image as needed
        and convert the image to a PNG. This is done while the
        image is still an InMemoryUpload object.
        """

        original_image = self.instance.image
        png_image = convert_image_to_png(original_image)
        filename = self.instance.image.name
        extension = filename.split(".")[-1]
        filename = filename.replace(extension, "png")
        self.instance.image.save(filename, png_image, save=commit)
        return super().save(commit=commit)


class UploadPersonPhotoURLForm(forms.Form):
    image_url = StrippedCharField(widget=forms.URLInput())
    why_allowed_url = forms.ChoiceField(
        choices=CopyrightOptions.WHY_ALLOWED_CHOICES, widget=forms.RadioSelect()
    )
    justification_for_use_url = StrippedCharField(
        widget=forms.Textarea(attrs={"rows": 1, "columns": 72}), required=False
    )

    def clean_image_url(self):
        image_url = self.cleaned_data["image_url"]
        # At least do a HEAD request to check that the Content-Type
        # looks reasonable:
        response = requests.head(image_url, allow_redirects=True)
        if (400 <= response.status_code < 500) or (
            500 <= response.status_code < 600
        ):
            msg = "That URL produced an HTTP error status code: {0}"
            raise ValidationError(msg.format(response.status_code))
        content_type = response.headers["content-type"]
        main, sub = cgi.parse_header(content_type)
        if not main.startswith("image/"):
            msg = "This URL isn't for an image - it had Content-Type: {0}"
            raise ValidationError(msg.format(main))
        return image_url


class PhotoRotateForm(forms.Form):
    def __init__(self, *args, **kwargs):
        self.queued_image = kwargs.pop("queued_image")
        self.request = kwargs.pop("request")
        super().__init__(*args, **kwargs)

    def clean(self):
        return super().clean()

    def process(self):
        rotation_direction = self.request.POST.get("rotate")
        if not rotation_direction:
            return self.queued_image
        self.rotate_photo(rotation_direction)
        return self.queued_image

    def rotate_photo(self, rotation_direction):
        image = PILImage.open(self.queued_image.image.file)
        if rotation_direction == "left":
            rotated = image.rotate(90, expand=True)
        elif rotation_direction == "right":
            rotated = image.rotate(-90, expand=True)
        buffer = BytesIO()
        rotated.save(buffer, "PNG")
        self.queued_image.image.save(self.queued_image.image.name, buffer)
        self.queued_image.rotation_tried = True
        sorl.thumbnail.delete(self.queued_image.image.name, delete_file=False)
        self.queued_image.save()


class PhotoReviewForm(forms.Form):
    def __init__(self, *args, **kwargs):
        self.queued_image = kwargs.pop("queued_image")
        self.request = kwargs.pop("request")
        super().__init__(*args, **kwargs)

    queued_image_id = forms.IntegerField(
        required=True, widget=forms.HiddenInput()
    )
    x_min = forms.IntegerField(min_value=0)
    x_max = forms.IntegerField(min_value=1)
    y_min = forms.IntegerField(min_value=0)
    y_max = forms.IntegerField(min_value=1)
    decision = forms.ChoiceField(
        choices=QueuedImage.DECISION_CHOICES, widget=forms.widgets.RadioSelect
    )
    rejection_reason = forms.CharField(widget=forms.Textarea(), required=False)
    justification_for_use = forms.CharField(
        widget=forms.Textarea(), required=False
    )
    moderator_why_allowed = forms.ChoiceField(
        choices=CopyrightOptions.WHY_ALLOWED_CHOICES,
        widget=forms.widgets.RadioSelect,
    )

    def process(self):
        action_method = getattr(self, self.cleaned_data["decision"])
        action_method()
        return self.queued_image

    def create_logged_action(self, version_id=""):
        action_types = {
            QueuedImage.APPROVED: ActionType.PHOTO_APPROVE,
            QueuedImage.REJECTED: ActionType.PHOTO_REJECT,
            QueuedImage.IGNORE: ActionType.PHOTO_IGNORE,
        }
        LoggedAction.objects.create(
            user=self.request.user,
            action_type=action_types[self.cleaned_data["decision"]],
            ip_address=get_client_ip(self.request),
            popit_person_new_version=version_id,
            person=self.queued_image.person,
            source=self.update_message,
        )

    @property
    def update_message(self):
        messages = {
            QueuedImage.APPROVED: f'Approved a photo upload from {self.queued_image.uploaded_by} who provided the message: "{self.queued_image.justification_for_use}"',
            QueuedImage.REJECTED: f"Rejected a photo upload from {self.queued_image.uploaded_by}",
            QueuedImage.IGNORE: f"Ignored a photo upload from {self.queued_image.uploaded_by} (This usually means it was a duplicate)",
        }
        return messages[self.cleaned_data["decision"]]

    def approved(self):
        self.queued_image.decision = self.cleaned_data["decision"]
        self.queued_image.crop_min_x = self.cleaned_data["x_min"]
        self.queued_image.crop_min_y = self.cleaned_data["y_min"]
        self.queued_image.crop_max_x = self.cleaned_data["x_max"]
        self.queued_image.crop_max_y = self.cleaned_data["y_max"]
        self.queued_image.save()
        self.queued_image.person.create_person_image(
            queued_image=self.queued_image,
            copyright=self.cleaned_data["moderator_why_allowed"],
        )
        change_metadata = get_change_metadata(self.request, self.update_message)
        self.queued_image.person.record_version(change_metadata)
        self.queued_image.person.save()
        self.create_logged_action(version_id=change_metadata["version_id"])

        candidate_full_url = self.request.build_absolute_uri(
            self.queued_image.person.get_absolute_url(self.request)
        )
        site_name = Site.objects.get_current().name
        subject = f"{site_name} image upload approved"
        self.send_mail(
            subject=subject,
            template_name="moderation_queue/photo_approved_email.txt",
            context={
                "site_name": site_name,
                "candidate_page_url": candidate_full_url,
                "intro": (
                    "Thank you for submitting a photo to "
                    f"{site_name}. It has been uploaded to "
                    "the candidate page here:"
                ),
                "signoff": f"Many thanks from the {site_name} volunteers",
            },
        )

    def rejected(self):
        self.queued_image.decision = self.cleaned_data["decision"]
        self.queued_image.save()
        self.create_logged_action()
        retry_upload_link = self.request.build_absolute_uri(
            reverse(
                "photo-upload",
                kwargs={"person_id": self.queued_image.person.id},
            )
        )
        site_name = Site.objects.get_current().name
        photo_review_url = self.request.build_absolute_uri(
            self.queued_image.get_absolute_url()
        )
        self.send_mail(
            subject=f"{site_name} image moderation results",
            template_name="moderation_queue/photo_rejected_email.txt",
            context={
                "reason": self.cleaned_data["rejection_reason"],
                "retry_upload_link": retry_upload_link,
                "photo_review_url": photo_review_url,
                "intro": (
                    "Thank-you for uploading a photo of "
                    f"{self.queued_image.person.name} to {site_name}, "
                    "but unfortunately we can't use that image because:"
                ),
                "possible_actions": (
                    "You can just reply to this email if you want to "
                    "discuss that further, or you can try uploading a "
                    "photo with a different reason or justification "
                    "for its use using this link:"
                ),
                "signoff": (f"Many thanks from the {site_name} volunteers"),
            },
            email_support_too=True,
        )

    def ignore(self):
        self.queued_image.decision = self.cleaned_data["decision"]
        self.queued_image.save()
        self.create_logged_action()

    def undecided(self):
        self.queued_image.decision = self.cleaned_data["decision"]
        self.queued_image.save()

    def send_mail(
        self, subject, template_name, context, email_support_too=False
    ):
        if not self.queued_image.user:
            # We can't send emails to bots…yet.
            return None

        message = render_to_string(template_name=template_name, context=context)
        recipients = [self.queued_image.user.email]
        if email_support_too:
            recipients.append(settings.SUPPORT_EMAIL)
        return send_mail(
            subject,
            message,
            settings.DEFAULT_FROM_EMAIL,
            recipients,
            fail_silently=False,
        )


class SuggestedPostLockForm(forms.ModelForm):
    class Meta:
        model = SuggestedPostLock
        fields = ["justification", "ballot"]
        widgets = {
            "ballot": forms.HiddenInput(),
            "justification": forms.Textarea(attrs={"rows": 1, "columns": 72}),
        }

    def clean(self):
        ballot = self.cleaned_data["ballot"]
        if ballot.candidates_locked:
            raise ValidationError(
                "Cannot create a lock suggestion for a locked ballot"
            )
        return self.cleaned_data

1	import cgi	1✔
2	from io import BytesIO	1✔
3
4	import requests	1✔
5	import sorl	1✔
6	from candidates.models.db import ActionType, LoggedAction	1✔
7	from candidates.views.version_data import get_change_metadata, get_client_ip	1✔
8	from django import forms	1✔
9	from django.conf import settings	1✔
10	from django.contrib.sites.models import Site	1✔
11	from django.core.exceptions import ValidationError	1✔
12	from django.core.mail import send_mail	1✔
13	from django.template.loader import render_to_string	1✔
14	from django.urls import reverse	1✔
15	from moderation_queue.helpers import (	1✔
16	convert_image_to_png,
17	)
18	from people.forms.forms import StrippedCharField	1✔
19	from PIL import Image as PILImage	1✔
20
21	from .models import CopyrightOptions, QueuedImage, SuggestedPostLock	1✔
22
23
24	class UploadPersonPhotoImageForm(forms.ModelForm):	1✔
25	class Meta:	1✔
26	model = QueuedImage	1✔
27	fields = [	1✔
28	"image",
29	"why_allowed",
30	"justification_for_use",
31	"person",
32	"decision",
33	]
34	widgets = {	1✔
35	"person": forms.HiddenInput(),
36	"decision": forms.HiddenInput(),
37	"why_allowed": forms.RadioSelect(),
38	"justification_for_use": forms.Textarea(
39	attrs={"rows": 1, "columns": 72},
40	),
41	}
42
43	def clean(self):	1✔
44	cleaned_data = super().clean()	1✔
45	justification_for_use = cleaned_data.get(	1✔
46	"justification_for_use", ""
47	).strip()
48	why_allowed = cleaned_data.get("why_allowed")	1✔
49	if why_allowed == "other" and not justification_for_use:	1!
UNCOV 50	message = (	×
51	"If you checked 'Other' then you must provide a "
52	"justification for why we can use it."
53	)
UNCOV 54	raise ValidationError(message)	×
55	return cleaned_data	1✔
56
57	def save(self, commit):	1✔
58	"""
59	Before saving, resize and rotate the image as needed
60	and convert the image to a PNG. This is done while the
61	image is still an InMemoryUpload object.
62	"""
63
64	original_image = self.instance.image	1✔
65	png_image = convert_image_to_png(original_image)	1✔
66	filename = self.instance.image.name	1✔
67	extension = filename.split(".")[-1]	1✔
68	filename = filename.replace(extension, "png")	1✔
69	self.instance.image.save(filename, png_image, save=commit)	1✔
70	return super().save(commit=commit)	1✔
71
72
73	class UploadPersonPhotoURLForm(forms.Form):	1✔
74	image_url = StrippedCharField(widget=forms.URLInput())	1✔
75	why_allowed_url = forms.ChoiceField(	1✔
76	choices=CopyrightOptions.WHY_ALLOWED_CHOICES, widget=forms.RadioSelect()
77	)
78	justification_for_use_url = StrippedCharField(	1✔
79	widget=forms.Textarea(attrs={"rows": 1, "columns": 72}), required=False
80	)
81
82	def clean_image_url(self):	1✔
83	image_url = self.cleaned_data["image_url"]	1✔
84	# At least do a HEAD request to check that the Content-Type
85	# looks reasonable:
86	response = requests.head(image_url, allow_redirects=True)	1✔
87	if (400 <= response.status_code < 500) or (	1✔
88	500 <= response.status_code < 600
89	):
90	msg = "That URL produced an HTTP error status code: {0}"	1✔
91	raise ValidationError(msg.format(response.status_code))	1✔
92	content_type = response.headers["content-type"]	1✔
93	main, sub = cgi.parse_header(content_type)	1✔
94	if not main.startswith("image/"):	1✔
95	msg = "This URL isn't for an image - it had Content-Type: {0}"	1✔
96	raise ValidationError(msg.format(main))	1✔
97	return image_url	1✔
98
99
100	class PhotoRotateForm(forms.Form):	1✔
101	def __init__(self, args, *kwargs):	1✔
102	self.queued_image = kwargs.pop("queued_image")	1✔
103	self.request = kwargs.pop("request")	1✔
104	super().__init__(args, *kwargs)	1✔
105
106	def clean(self):	1✔
107	return super().clean()	1✔
108
109	def process(self):	1✔
110	rotation_direction = self.request.POST.get("rotate")	1✔
111	if not rotation_direction:	1!
UNCOV 112	return self.queued_image	×
113	self.rotate_photo(rotation_direction)	1✔
114	return self.queued_image	1✔
115
116	def rotate_photo(self, rotation_direction):	1✔
117	image = PILImage.open(self.queued_image.image.file)	1✔
118	if rotation_direction == "left":	1!
UNCOV 119	rotated = image.rotate(90, expand=True)	×
120	elif rotation_direction == "right":	1!
121	rotated = image.rotate(-90, expand=True)	1✔
122	buffer = BytesIO()	1✔
123	rotated.save(buffer, "PNG")	1✔
124	self.queued_image.image.save(self.queued_image.image.name, buffer)	1✔
125	self.queued_image.rotation_tried = True	1✔
126	sorl.thumbnail.delete(self.queued_image.image.name, delete_file=False)	1✔
127	self.queued_image.save()	1✔
128
129
130	class PhotoReviewForm(forms.Form):	1✔
131	def __init__(self, args, *kwargs):	1✔
132	self.queued_image = kwargs.pop("queued_image")	1✔
133	self.request = kwargs.pop("request")	1✔
134	super().__init__(args, *kwargs)	1✔
135
136	queued_image_id = forms.IntegerField(	1✔
137	required=True, widget=forms.HiddenInput()
138	)
139	x_min = forms.IntegerField(min_value=0)	1✔
140	x_max = forms.IntegerField(min_value=1)	1✔
141	y_min = forms.IntegerField(min_value=0)	1✔
142	y_max = forms.IntegerField(min_value=1)	1✔
143	decision = forms.ChoiceField(	1✔
144	choices=QueuedImage.DECISION_CHOICES, widget=forms.widgets.RadioSelect
145	)
146	rejection_reason = forms.CharField(widget=forms.Textarea(), required=False)	1✔
147	justification_for_use = forms.CharField(	1✔
148	widget=forms.Textarea(), required=False
149	)
150	moderator_why_allowed = forms.ChoiceField(	1✔
151	choices=CopyrightOptions.WHY_ALLOWED_CHOICES,
152	widget=forms.widgets.RadioSelect,
153	)
154
155	def process(self):	1✔
156	action_method = getattr(self, self.cleaned_data["decision"])	1✔
157	action_method()	1✔
158	return self.queued_image	1✔
159
160	def create_logged_action(self, version_id=""):	1✔
161	action_types = {	1✔
162	QueuedImage.APPROVED: ActionType.PHOTO_APPROVE,
163	QueuedImage.REJECTED: ActionType.PHOTO_REJECT,
164	QueuedImage.IGNORE: ActionType.PHOTO_IGNORE,
165	}
166	LoggedAction.objects.create(	1✔
167	user=self.request.user,
168	action_type=action_types[self.cleaned_data["decision"]],
169	ip_address=get_client_ip(self.request),
170	popit_person_new_version=version_id,
171	person=self.queued_image.person,
172	source=self.update_message,
173	)
174
175	@property	1✔
176	def update_message(self):	1✔
177	messages = {	1✔
178	QueuedImage.APPROVED: f'Approved a photo upload from {self.queued_image.uploaded_by} who provided the message: "{self.queued_image.justification_for_use}"',
179	QueuedImage.REJECTED: f"Rejected a photo upload from {self.queued_image.uploaded_by}",
180	QueuedImage.IGNORE: f"Ignored a photo upload from {self.queued_image.uploaded_by} (This usually means it was a duplicate)",
181	}
182	return messages[self.cleaned_data["decision"]]	1✔
183
184	def approved(self):	1✔
185	self.queued_image.decision = self.cleaned_data["decision"]	1✔
186	self.queued_image.crop_min_x = self.cleaned_data["x_min"]	1✔
187	self.queued_image.crop_min_y = self.cleaned_data["y_min"]	1✔
188	self.queued_image.crop_max_x = self.cleaned_data["x_max"]	1✔
189	self.queued_image.crop_max_y = self.cleaned_data["y_max"]	1✔
190	self.queued_image.save()	1✔
191	self.queued_image.person.create_person_image(	1✔
192	queued_image=self.queued_image,
193	copyright=self.cleaned_data["moderator_why_allowed"],
194	)
195	change_metadata = get_change_metadata(self.request, self.update_message)	1✔
196	self.queued_image.person.record_version(change_metadata)	1✔
197	self.queued_image.person.save()	1✔
198	self.create_logged_action(version_id=change_metadata["version_id"])	1✔
199
200	candidate_full_url = self.request.build_absolute_uri(	1✔
201	self.queued_image.person.get_absolute_url(self.request)
202	)
203	site_name = Site.objects.get_current().name	1✔
204	subject = f"{site_name} image upload approved"	1✔
205	self.send_mail(	1✔
206	subject=subject,
207	template_name="moderation_queue/photo_approved_email.txt",
208	context={
209	"site_name": site_name,
210	"candidate_page_url": candidate_full_url,
211	"intro": (
212	"Thank you for submitting a photo to "
213	f"{site_name}. It has been uploaded to "
214	"the candidate page here:"
215	),
216	"signoff": f"Many thanks from the {site_name} volunteers",
217	},
218	)
219
220	def rejected(self):	1✔
221	self.queued_image.decision = self.cleaned_data["decision"]	1✔
222	self.queued_image.save()	1✔
223	self.create_logged_action()	1✔
224	retry_upload_link = self.request.build_absolute_uri(	1✔
225	reverse(
226	"photo-upload",
227	kwargs={"person_id": self.queued_image.person.id},
228	)
229	)
230	site_name = Site.objects.get_current().name	1✔
231	photo_review_url = self.request.build_absolute_uri(	1✔
232	self.queued_image.get_absolute_url()
233	)
234	self.send_mail(	1✔
235	subject=f"{site_name} image moderation results",
236	template_name="moderation_queue/photo_rejected_email.txt",
237	context={
238	"reason": self.cleaned_data["rejection_reason"],
239	"retry_upload_link": retry_upload_link,
240	"photo_review_url": photo_review_url,
241	"intro": (
242	"Thank-you for uploading a photo of "
243	f"{self.queued_image.person.name} to {site_name}, "
244	"but unfortunately we can't use that image because:"
245	),
246	"possible_actions": (
247	"You can just reply to this email if you want to "
248	"discuss that further, or you can try uploading a "
249	"photo with a different reason or justification "
250	"for its use using this link:"
251	),
252	"signoff": (f"Many thanks from the {site_name} volunteers"),
253	},
254	email_support_too=True,
255	)
256
257	def ignore(self):	1✔
258	self.queued_image.decision = self.cleaned_data["decision"]	1✔
259	self.queued_image.save()	1✔
260	self.create_logged_action()	1✔
261
262	def undecided(self):	1✔
263	self.queued_image.decision = self.cleaned_data["decision"]	1✔
264	self.queued_image.save()	1✔
265
266	def send_mail(	1✔
267	self, subject, template_name, context, email_support_too=False
268	):
269	if not self.queued_image.user:	1✔
270	# We can't send emails to bots…yet.
271	return None	1✔
272
273	message = render_to_string(template_name=template_name, context=context)	1✔
274	recipients = [self.queued_image.user.email]	1✔
275	if email_support_too:	1✔
276	recipients.append(settings.SUPPORT_EMAIL)	1✔
277	return send_mail(	1✔
278	subject,
279	message,
280	settings.DEFAULT_FROM_EMAIL,
281	recipients,
282	fail_silently=False,
283	)
284
285
286	class SuggestedPostLockForm(forms.ModelForm):	1✔
287	class Meta:	1✔
288	model = SuggestedPostLock	1✔
289	fields = ["justification", "ballot"]	1✔
290	widgets = {	1✔
291	"ballot": forms.HiddenInput(),
292	"justification": forms.Textarea(attrs={"rows": 1, "columns": 72}),
293	}
294
295	def clean(self):	1✔
296	ballot = self.cleaned_data["ballot"]	1✔
297	if ballot.candidates_locked:	1✔
298	raise ValidationError(	1✔
299	"Cannot create a lock suggestion for a locked ballot"
300	)
301	return self.cleaned_data	1✔

DemocracyClub / yournextrepresentative / 785eb3cf-f76e-4c4c-8a58-787ccc3990bc

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous