be829fff-cd95-4e07-98d0-ccfbaf011f50

Committed 04 Jun 2024 06:41AM UTC coverage: 67.494% (-0.1%) from 67.603%

Build # be829fff-cd95-4e07-98d0-ccfbaf011f50

Build Type

Pull #2350

circleci

Committed by

symroe

Commit Message

Disable Camelot parsing by default

Pull Request Pull Request #2350: Disable Camelot parsing by default

Run Details

1531 of 2790 branches covered (54.87%)

Branch coverage included in aggregate %.

1 of 5 new or added lines in 2 files covered. (20.0%)

7 existing lines in 2 files now uncovered.

7337 of 10349 relevant lines covered (70.9%)

0.71 hits per line

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

75.0

/ynr/apps/official_documents/models.py

import contextlib
import os
import textwrap
from pathlib import Path
from typing import List

from candidates.models import Ballot
from django.conf import settings
from django.core import mail
from django.core.files.base import ContentFile
from django.core.validators import FileExtensionValidator
from django.db import models
from django.urls import reverse
from django.utils import timezone
from django_extensions.db.models import TimeStampedModel
from sopn_parsing.models import AWSTextractParsedSOPN

DOCUMENT_UPLOADERS_GROUP_NAME = "Document Uploaders"


def document_file_name(instance, filename):
    return os.path.join(
        "official_documents", str(instance.ballot.ballot_paper_id), filename
    )


class OfficialDocument(TimeStampedModel):
    NOMINATION_PAPER = "Nomination paper"

    DOCUMENT_TYPES = (
        (NOMINATION_PAPER, "Nomination paper", "Nomination papers"),
    )

    document_type = models.CharField(
        blank=False,
        choices=[(d[0], d[1]) for d in DOCUMENT_TYPES],
        max_length=100,
    )
    uploaded_file = models.FileField(
        upload_to=document_file_name,
        max_length=800,
        validators=[FileExtensionValidator(allowed_extensions=["pdf", "docx"])],
    )
    ballot = models.ForeignKey(
        "candidates.Ballot", null=False, on_delete=models.CASCADE
    )
    source_url = models.URLField(
        help_text="The URL of this document", max_length=1000
    )
    relevant_pages = models.CharField(
        "The pages containing information about this ballot",
        max_length=50,
        default="",
    )

    class Meta:
        get_latest_by = "modified"

    def __str__(self):
        return "{} ({})".format(self.ballot.ballot_paper_id, self.source_url)

    def get_absolute_url(self):
        return reverse(
            "ballot_paper_sopn",
            kwargs={"ballot_id": self.ballot.ballot_paper_id},
        )

    @property
    def locked(self):
        """
        Is this post election locked?
        """
        return self.ballot.candidates_locked

    @property
    def lock_suggested(self):
        """
        Is there a suggested lock for this document?
        """
        return self.ballot.suggestedpostlock_set.exists()

    def get_pages(self):
        if self.relevant_pages and self.relevant_pages != "all":
            pages = self.relevant_pages.split(",")
            return sorted(int(p) for p in pages)
        return None

    @property
    def get_aws_parsed_sopn(self):
        """given an instance of an official document, return the associated aws parsed sopn"""
        return AWSTextractParsedSOPN.objects.get(sopn=self)

    @property
    def first_page_number(self):
        if self.get_pages():
            return self.get_pages()[0]
        return None

    @property
    def last_page_number(self):
        if self.get_pages():
            return self.get_pages()[-1]
        return None


def election_sopn_file_name(instance: "ElectionSOPN", filename):
    return (
        Path("official_documents")
        / instance.election.slug
        / timezone.now().isoformat()
        / filename
    )


class PageMatchingMethods(models.TextChoices):
    AUTO_MATCHED = "Automatically matched"
    MANUAL_MATCHED = "Manually matched"


class ElectionSOPN(TimeStampedModel):
    """
    Stores SOPNs that contain candidate information for each ballot in an election.

    We don't use this directly, rather we use this as a donor document for populating BallotSOPN instances.

    """

    election = models.OneToOneField(
        "elections.Election", on_delete=models.CASCADE, blank=True
    )

    uploaded_file = models.FileField(
        upload_to=election_sopn_file_name,
        max_length=1200,
        validators=[FileExtensionValidator(allowed_extensions=["pdf"])],
    )

    source_url = models.URLField(
        help_text="The URL of this document", max_length=1200
    )

    page_matching_method = models.CharField(
        max_length=255, choices=PageMatchingMethods.choices, null=True
    )

    replacement_reason = models.CharField(
        verbose_name="Reason for replacement",
        help_text="Please be as descriptive as possible to explain why the replacement SOPN is needed",
        max_length=255,
        blank=True,
    )

    class Meta:
        get_latest_by = "modified"

    def __str__(self):
        return "{} ({})".format(self.election.slug, self.source_url)

    def get_absolute_url(self):
        return reverse(
            "election_sopn",
            kwargs={"election_id": self.election.slug},
        )

    @property
    def pages_matched(self):
        return (
            bool(self.page_matching_method)
            and not self.election.ballot_set.filter(
                sopn__relevant_pages__isnull=True
            ).exists()
        )


def ballot_sopn_file_name(instance: "BaseBallotSOPN", filename):
    return (
        Path("official_documents")
        / instance.ballot.ballot_paper_id
        / "sopn"
        / timezone.now().isoformat()
        / filename
    )


class BaseBallotSOPN(TimeStampedModel):
    """
    An abstract class used by BallotSOPN and BallotSOPNHistory. Used to keep both models in sync.

    """

    ballot = models.OneToOneField(
        "candidates.Ballot", on_delete=models.CASCADE, related_name="sopn"
    )

    # Don't add `ballot` here as we want different related names for each model
    uploaded_file = models.FileField(
        upload_to=ballot_sopn_file_name,
        max_length=1200,
        validators=[
            FileExtensionValidator(
                allowed_extensions=["pdf", "jpeg", "jpg", "png"]
            )
        ],
    )

    source_url = models.URLField(
        help_text="The URL of this document", max_length=1200
    )

    relevant_pages = models.CharField(
        max_length=20,
        help_text="The pages in the ElectionSOPN that relate to this Ballot",
        default="all",
    )

    replacement_reason = models.CharField(
        verbose_name="Reason for replacement",
        help_text="Please be as descriptive as possible to explain why the replacement SOPN is needed",
        max_length=255,
        blank=True,
    )

    class Meta:
        get_latest_by = "modified"
        abstract = True

    def __str__(self):
        return "{} ({})".format(self.ballot.ballot_paper_id, self.source_url)

    @property
    def first_page_int(self):
        return min(self.page_number_list)

    @property
    def page_number_list(self) -> List[int]:
        return [int(i) for i in self.relevant_pages.split(",") if i != "all"]

    @property
    def one_based_relevant_pages(self):
        return ", ".join([str(i + 1) for i in self.page_number_list])


class BallotSOPN(BaseBallotSOPN):
    def get_absolute_url(self):
        return reverse(
            "ballot_paper_sopn",
            kwargs={"ballot_id": self.ballot.ballot_paper_id},
        )

    def parse(self):
        """
        Runs code that is needed to parse this SOPN.

        When run, this should kick off the process that ends up in a set of models
        that can be used by the bulk adding application.

        The actual implementation of the parsing (e.g the backends that are used,
        and async invocations) shouldn't matter here, the key point is that this is
        the front door.

        """

        from sopn_parsing.helpers.extract_tables import extract_ballot_table
        from sopn_parsing.helpers.textract_helpers import (
            NotUsingAWSException,
            TextractSOPNHelper,
        )

        # Pull out the tables from PDFs

        # AWS Textract
        # Don't break if we're not on AWS
        with contextlib.suppress(NotUsingAWSException):
            textract_helper = TextractSOPNHelper(self)
            # Start detection isn't going to populate anything for a while.
            # There's a cron job that should pick up the result and carry on parsing later.
            textract_helper.start_detection()

        if getattr(
            settings, "CAMELOT_ENABLED", False
        ) and self.uploaded_file.name.endswith(".pdf"):
            # Camelot
            extract_ballot_table(self.ballot)


class BallotSOPNHistory(BaseBallotSOPN):
    ballot = models.ForeignKey(
        "candidates.Ballot",
        on_delete=models.CASCADE,
        related_name="sopn_history",
    )


def add_ballot_sopn(
    ballot: Ballot,
    pdf_content: ContentFile,
    source_url: str,
    relevant_pages: str = "all",
    replacement_reason=None,
    parse=True,
):
    """
    Manage creating BallotSOPNs with history
    """
    if not replacement_reason:
        replacement_reason = ""

    BallotSOPNHistory.objects.create(
        ballot=ballot,
        relevant_pages=relevant_pages,
        uploaded_file=pdf_content,
        source_url=source_url,
        replacement_reason=replacement_reason,
    )

    BallotSOPN.objects.filter(ballot=ballot).delete()
    ballot_sopn: BallotSOPN = BallotSOPN.objects.create(
        ballot=ballot,
        relevant_pages=relevant_pages,
        uploaded_file=pdf_content,
        source_url=source_url,
        replacement_reason=replacement_reason,
    )
    if parse:
        ballot_sopn.parse()
    return ballot_sopn


def send_ballot_sopn_update_notification(ballot_sopn: BallotSOPN, request):
    message = textwrap.dedent(
        f"""\
    Hello,
    
    The user {request.user.username} has updated the SOPN for the ballot with ID {ballot_sopn.ballot.ballot_paper_id}.
    
    You can see this newly uploaded SOPN here:
    
    {request.build_absolute_uri(ballot_sopn.get_absolute_url())}
    
    Their reason for the new upload was:
    
    > "{ballot_sopn.replacement_reason}"
    
    """
    )

    mail.send_mail(
        f"SOPN for {ballot_sopn.ballot.ballot_paper_id} updated",
        message,
        "hello@democracyclub.org.uk",
        settings.SOPN_UPDATE_NOTIFICATION_EMAILS,
        fail_silently=True,
    )

1	import contextlib	1✔
2	import os	1✔
3	import textwrap	1✔
4	from pathlib import Path	1✔
5	from typing import List	1✔
6
7	from candidates.models import Ballot	1✔
8	from django.conf import settings	1✔
9	from django.core import mail	1✔
10	from django.core.files.base import ContentFile	1✔
11	from django.core.validators import FileExtensionValidator	1✔
12	from django.db import models	1✔
13	from django.urls import reverse	1✔
14	from django.utils import timezone	1✔
15	from django_extensions.db.models import TimeStampedModel	1✔
16	from sopn_parsing.models import AWSTextractParsedSOPN	1✔
17
18	DOCUMENT_UPLOADERS_GROUP_NAME = "Document Uploaders"	1✔
19
20
21	def document_file_name(instance, filename):	1✔
22	return os.path.join(	×
23	"official_documents", str(instance.ballot.ballot_paper_id), filename
24	)
25
26
27	class OfficialDocument(TimeStampedModel):	1✔
28	NOMINATION_PAPER = "Nomination paper"	1✔
29
30	DOCUMENT_TYPES = (	1✔
31	(NOMINATION_PAPER, "Nomination paper", "Nomination papers"),
32	)
33
34	document_type = models.CharField(	1✔
35	blank=False,
36	choices=[(d[0], d[1]) for d in DOCUMENT_TYPES],
37	max_length=100,
38	)
39	uploaded_file = models.FileField(	1✔
40	upload_to=document_file_name,
41	max_length=800,
42	validators=[FileExtensionValidator(allowed_extensions=["pdf", "docx"])],
43	)
44	ballot = models.ForeignKey(	1✔
45	"candidates.Ballot", null=False, on_delete=models.CASCADE
46	)
47	source_url = models.URLField(	1✔
48	help_text="The URL of this document", max_length=1000
49	)
50	relevant_pages = models.CharField(	1✔
51	"The pages containing information about this ballot",
52	max_length=50,
53	default="",
54	)
55
56	class Meta:	1✔
57	get_latest_by = "modified"	1✔
58
59	def __str__(self):	1✔
60	return "{} ({})".format(self.ballot.ballot_paper_id, self.source_url)	×
61
62	def get_absolute_url(self):	1✔
63	return reverse(	×
64	"ballot_paper_sopn",
65	kwargs={"ballot_id": self.ballot.ballot_paper_id},
66	)
67
68	@property	1✔
69	def locked(self):	1✔
70	"""
71	Is this post election locked?
72	"""
73	return self.ballot.candidates_locked	×
74
75	@property	1✔
76	def lock_suggested(self):	1✔
77	"""
78	Is there a suggested lock for this document?
79	"""
80	return self.ballot.suggestedpostlock_set.exists()	×
81
82	def get_pages(self):	1✔
83	if self.relevant_pages and self.relevant_pages != "all":	×
84	pages = self.relevant_pages.split(",")	×
85	return sorted(int(p) for p in pages)	×
86	return None	×
87
88	@property	1✔
89	def get_aws_parsed_sopn(self):	1✔
90	"""given an instance of an official document, return the associated aws parsed sopn"""
91	return AWSTextractParsedSOPN.objects.get(sopn=self)	×
92
93	@property	1✔
94	def first_page_number(self):	1✔
95	if self.get_pages():	×
96	return self.get_pages()[0]	×
97	return None	×
98
99	@property	1✔
100	def last_page_number(self):	1✔
101	if self.get_pages():	×
102	return self.get_pages()[-1]	×
103	return None	×
104
105
106	def election_sopn_file_name(instance: "ElectionSOPN", filename):	1✔
107	return (	1✔
108	Path("official_documents")
109	/ instance.election.slug
110	/ timezone.now().isoformat()
111	/ filename
112	)
113
114
115	class PageMatchingMethods(models.TextChoices):	1✔
116	AUTO_MATCHED = "Automatically matched"	1✔
117	MANUAL_MATCHED = "Manually matched"	1✔
118
119
120	class ElectionSOPN(TimeStampedModel):	1✔
121	"""
122	Stores SOPNs that contain candidate information for each ballot in an election.
123
124	We don't use this directly, rather we use this as a donor document for populating BallotSOPN instances.
125
126	"""
127
128	election = models.OneToOneField(	1✔
129	"elections.Election", on_delete=models.CASCADE, blank=True
130	)
131
132	uploaded_file = models.FileField(	1✔
133	upload_to=election_sopn_file_name,
134	max_length=1200,
135	validators=[FileExtensionValidator(allowed_extensions=["pdf"])],
136	)
137
138	source_url = models.URLField(	1✔
139	help_text="The URL of this document", max_length=1200
140	)
141
142	page_matching_method = models.CharField(	1✔
143	max_length=255, choices=PageMatchingMethods.choices, null=True
144	)
145
146	replacement_reason = models.CharField(	1✔
147	verbose_name="Reason for replacement",
148	help_text="Please be as descriptive as possible to explain why the replacement SOPN is needed",
149	max_length=255,
150	blank=True,
151	)
152
153	class Meta:	1✔
154	get_latest_by = "modified"	1✔
155
156	def __str__(self):	1✔
157	return "{} ({})".format(self.election.slug, self.source_url)	×
158
159	def get_absolute_url(self):	1✔
160	return reverse(	1✔
161	"election_sopn",
162	kwargs={"election_id": self.election.slug},
163	)
164
165	@property	1✔
166	def pages_matched(self):	1✔
167	return (	1✔
168	bool(self.page_matching_method)
169	and not self.election.ballot_set.filter(
170	sopn__relevant_pages__isnull=True
171	).exists()
172	)
173
174
175	def ballot_sopn_file_name(instance: "BaseBallotSOPN", filename):	1✔
176	return (	1✔
177	Path("official_documents")
178	/ instance.ballot.ballot_paper_id
179	/ "sopn"
180	/ timezone.now().isoformat()
181	/ filename
182	)
183
184
185	class BaseBallotSOPN(TimeStampedModel):	1✔
186	"""
187	An abstract class used by BallotSOPN and BallotSOPNHistory. Used to keep both models in sync.
188
189	"""
190
191	ballot = models.OneToOneField(	1✔
192	"candidates.Ballot", on_delete=models.CASCADE, related_name="sopn"
193	)
194
195	# Don't add `ballot` here as we want different related names for each model
196	uploaded_file = models.FileField(	1✔
197	upload_to=ballot_sopn_file_name,
198	max_length=1200,
199	validators=[
200	FileExtensionValidator(
201	allowed_extensions=["pdf", "jpeg", "jpg", "png"]
202	)
203	],
204	)
205
206	source_url = models.URLField(	1✔
207	help_text="The URL of this document", max_length=1200
208	)
209
210	relevant_pages = models.CharField(	1✔
211	max_length=20,
212	help_text="The pages in the ElectionSOPN that relate to this Ballot",
213	default="all",
214	)
215
216	replacement_reason = models.CharField(	1✔
217	verbose_name="Reason for replacement",
218	help_text="Please be as descriptive as possible to explain why the replacement SOPN is needed",
219	max_length=255,
220	blank=True,
221	)
222
223	class Meta:	1✔
224	get_latest_by = "modified"	1✔
225	abstract = True	1✔
226
227	def __str__(self):	1✔
228	return "{} ({})".format(self.ballot.ballot_paper_id, self.source_url)	1✔
229
230	@property	1✔
231	def first_page_int(self):	1✔
232	return min(self.page_number_list)	×
233
234	@property	1✔
235	def page_number_list(self) -> List[int]:	1✔
236	return [int(i) for i in self.relevant_pages.split(",") if i != "all"]	×
237
238	@property	1✔
239	def one_based_relevant_pages(self):	1✔
240	return ", ".join([str(i + 1) for i in self.page_number_list])	×
241
242
243	class BallotSOPN(BaseBallotSOPN):	1✔
244	def get_absolute_url(self):	1✔
245	return reverse(	1✔
246	"ballot_paper_sopn",
247	kwargs={"ballot_id": self.ballot.ballot_paper_id},
248	)
249
250	def parse(self):	1✔
251	"""
252	Runs code that is needed to parse this SOPN.
253
254	When run, this should kick off the process that ends up in a set of models
255	that can be used by the bulk adding application.
256
257	The actual implementation of the parsing (e.g the backends that are used,
258	and async invocations) shouldn't matter here, the key point is that this is
259	the front door.
260
261	"""
262
263	from sopn_parsing.helpers.extract_tables import extract_ballot_table	1✔
264	from sopn_parsing.helpers.textract_helpers import (	1✔
265	NotUsingAWSException,
266	TextractSOPNHelper,
267	)
268
269	# Pull out the tables from PDFs
270
271	# AWS Textract
272	# Don't break if we're not on AWS
273	with contextlib.suppress(NotUsingAWSException):	1✔
274	textract_helper = TextractSOPNHelper(self)	1✔
275	# Start detection isn't going to populate anything for a while.
276	# There's a cron job that should pick up the result and carry on parsing later.
277	textract_helper.start_detection()	×
278
279	if getattr(	1!
280	settings, "CAMELOT_ENABLED", False
281	) and self.uploaded_file.name.endswith(".pdf"):
282	# Camelot
UNCOV 283	extract_ballot_table(self.ballot)	×
284
285
286	class BallotSOPNHistory(BaseBallotSOPN):	1✔
287	ballot = models.ForeignKey(	1✔
288	"candidates.Ballot",
289	on_delete=models.CASCADE,
290	related_name="sopn_history",
291	)
292
293
294	def add_ballot_sopn(	1✔
295	ballot: Ballot,
296	pdf_content: ContentFile,
297	source_url: str,
298	relevant_pages: str = "all",
299	replacement_reason=None,
300	parse=True,
301	):
302	"""
303	Manage creating BallotSOPNs with history
304	"""
305	if not replacement_reason:	1✔
306	replacement_reason = ""	1✔
307
308	BallotSOPNHistory.objects.create(	1✔
309	ballot=ballot,
310	relevant_pages=relevant_pages,
311	uploaded_file=pdf_content,
312	source_url=source_url,
313	replacement_reason=replacement_reason,
314	)
315
316	BallotSOPN.objects.filter(ballot=ballot).delete()	1✔
317	ballot_sopn: BallotSOPN = BallotSOPN.objects.create(	1✔
318	ballot=ballot,
319	relevant_pages=relevant_pages,
320	uploaded_file=pdf_content,
321	source_url=source_url,
322	replacement_reason=replacement_reason,
323	)
324	if parse:	1!
325	ballot_sopn.parse()	1✔
326	return ballot_sopn	1✔
327
328
329	def send_ballot_sopn_update_notification(ballot_sopn: BallotSOPN, request):	1✔
330	message = textwrap.dedent(	1✔
331	f"""\
332	Hello,
333
334	The user {request.user.username} has updated the SOPN for the ballot with ID {ballot_sopn.ballot.ballot_paper_id}.
335
336	You can see this newly uploaded SOPN here:
337
338	{request.build_absolute_uri(ballot_sopn.get_absolute_url())}
339
340	Their reason for the new upload was:
341
342	> "{ballot_sopn.replacement_reason}"
343
344	"""
345	)
346
347	mail.send_mail(	1✔
348	f"SOPN for {ballot_sopn.ballot.ballot_paper_id} updated",
349	message,
350	"hello@democracyclub.org.uk",
351	settings.SOPN_UPDATE_NOTIFICATION_EMAILS,
352	fail_silently=True,
353	)

DemocracyClub / yournextrepresentative / be829fff-cd95-4e07-98d0-ccfbaf011f50

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