74df76c8-4768-48d5-bb7f-5ee50aa05217

Committed 06 Nov 2023 01:38PM UTC coverage: 67.523% (-0.3%) from 67.801%

Build # 74df76c8-4768-48d5-bb7f-5ee50aa05217

Build Type

Pull #2177

circleci

Committed by

VirginiaDooley

Commit Message

Create TextractResults model

Pull Request Pull Request #2177: Spike: AWS Textract

Run Details

1640 of 2760 branches covered (0.0%)

Branch coverage included in aggregate %.

12 of 62 new or added lines in 3 files covered. (19.35%)

110 existing lines in 10 files now uncovered.

6662 of 9535 relevant lines covered (69.87%)

0.7 hits per line

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

0.0

/ynr/apps/sopn_parsing/management/commands/sopn_tooling_create_official_documents.py

import requests
from candidates.models import Ballot
from django.core.files.base import ContentFile
from django.core.management.base import BaseCommand
from elections.models import Election
from official_documents.models import OfficialDocument


class Command(BaseCommand):
    """This command uses the ballots endpoint to loop over each
    ballot and store each sopn pdf (uploaded_file) locally"""

    def add_arguments(self, parser):
        parser.add_argument(
            "--date",
            "-d",
            action="store",
            help="Election date in ISO format, defaults to 2021-05-06",
            default="2021-05-06",
            type=str,
        )
        parser.add_argument(
            "--site_url",
            "-u",
            action="store",
            help="URL of site to download from",
            default="https://candidates.democracyclub.org.uk/",
            type=str,
        )
        parser.add_argument(
            "--election-count",
            "-c",
            action="store",
            help="URL of site to download from",
            default=50,
            type=int,
        )
        parser.add_argument(
            "--election-slugs", "-s", action="store", required=False
        )

    def handle(self, *args, **options):
        site_url = options.get("site_url")
        election_date = options.get("date")
        election_count = options.get("election_count")

        if options["election_slugs"]:
            election_slugs = options["election_slugs"].split(",")
        else:
            election_slugs = Election.objects.filter(
                election_date=election_date
            ).values_list("slug", flat=True)[:election_count]

        for slug in election_slugs:
            url = f"{site_url}api/next/ballots/?has_sopn=1&page_size=200&election_id={slug}"
            self.create_official_documents(url=url)

    def create_official_documents(self, url):
        data = requests.get(url=url).json()
        next_page = data["next"]
        for ballot_data in data["results"]:
            ballot = Ballot.objects.get(
                ballot_paper_id=ballot_data["ballot_paper_id"]
            )
            sopn_data = ballot_data["sopn"]

            # if we already have the SOPN no need to recreate
            if ballot.officialdocument_set.filter(
                source_url=sopn_data["source_url"]
            ).exists():
                self.stdout.write(
                    f"SOPN already exists for {ballot.ballot_paper_id}"
                )
                continue

            # check if we already have an OfficialDocument with this source
            # downloaded
            official_document = OfficialDocument.objects.filter(
                source_url=sopn_data["source_url"]
            ).first()
            if official_document:
                # if so we dont need to redownload the file, we can create a new
                # object for this ballot with the same file
                self.stdout.write(
                    f"Found SOPN for source {sopn_data['source_url']}"
                )
                OfficialDocument.objects.create(
                    ballot=ballot,
                    source_url=sopn_data["source_url"],
                    uploaded_file=official_document.uploaded_file,
                    document_type=OfficialDocument.NOMINATION_PAPER,
                )
                continue

            # otherwise we dont have this file stored already, so download it as
            # part of creating the OfficialDocument
            self.stdout.write(
                f"Downloading SOPN from {sopn_data['uploaded_file']}"
            )
            file_response = requests.get(sopn_data["uploaded_file"])
            file_object = ContentFile(content=file_response.content)
            official_document = OfficialDocument(
                ballot=ballot,
                source_url=sopn_data["source_url"],
                document_type=OfficialDocument.NOMINATION_PAPER,
            )
            file_extension = sopn_data["uploaded_file"].split(".")[-1]
            filename = f"{ballot.ballot_paper_id}.{file_extension}"
            official_document.uploaded_file.save(
                name=filename, content=file_object
            )

        # this should only be the case where the election object has > 200
        # ballots e.g. parliamentary elections
        if next_page:
            return self.create_official_documents(url=next_page)
        return None

1	import requests	×
2	from candidates.models import Ballot	×
3	from django.core.files.base import ContentFile	×
4	from django.core.management.base import BaseCommand	×
5	from elections.models import Election	×
6	from official_documents.models import OfficialDocument	×
7
8
UNCOV 9	class Command(BaseCommand):	×
10	"""This command uses the ballots endpoint to loop over each
11	ballot and store each sopn pdf (uploaded_file) locally"""
12
UNCOV 13	def add_arguments(self, parser):	×
14	parser.add_argument(	×
15	"--date",
16	"-d",
17	action="store",
18	help="Election date in ISO format, defaults to 2021-05-06",
19	default="2021-05-06",
20	type=str,
21	)
UNCOV 22	parser.add_argument(	×
23	"--site_url",
24	"-u",
25	action="store",
26	help="URL of site to download from",
27	default="https://candidates.democracyclub.org.uk/",
28	type=str,
29	)
UNCOV 30	parser.add_argument(	×
31	"--election-count",
32	"-c",
33	action="store",
34	help="URL of site to download from",
35	default=50,
36	type=int,
37	)
UNCOV 38	parser.add_argument(	×
39	"--election-slugs", "-s", action="store", required=False
40	)
41
UNCOV 42	def handle(self, args, *options):	×
43	site_url = options.get("site_url")	×
44	election_date = options.get("date")	×
45	election_count = options.get("election_count")	×
46
UNCOV 47	if options["election_slugs"]:	×
48	election_slugs = options["election_slugs"].split(",")	×
49	else:
UNCOV 50	election_slugs = Election.objects.filter(	×
51	election_date=election_date
52	).values_list("slug", flat=True)[:election_count]
53
UNCOV 54	for slug in election_slugs:	×
55	url = f"{site_url}api/next/ballots/?has_sopn=1&page_size=200&election_id={slug}"	×
56	self.create_official_documents(url=url)	×
57
UNCOV 58	def create_official_documents(self, url):	×
59	data = requests.get(url=url).json()	×
60	next_page = data["next"]	×
61	for ballot_data in data["results"]:	×
62	ballot = Ballot.objects.get(	×
63	ballot_paper_id=ballot_data["ballot_paper_id"]
64	)
65	sopn_data = ballot_data["sopn"]	×
66
67	# if we already have the SOPN no need to recreate
68	if ballot.officialdocument_set.filter(	×
69	source_url=sopn_data["source_url"]
70	).exists():
71	self.stdout.write(	×
72	f"SOPN already exists for {ballot.ballot_paper_id}"
73	)
74	continue	×
75
76	# check if we already have an OfficialDocument with this source
77	# downloaded
78	official_document = OfficialDocument.objects.filter(	×
79	source_url=sopn_data["source_url"]
80	).first()
81	if official_document:	×
82	# if so we dont need to redownload the file, we can create a new
83	# object for this ballot with the same file
84	self.stdout.write(	×
85	f"Found SOPN for source {sopn_data['source_url']}"
86	)
87	OfficialDocument.objects.create(	×
88	ballot=ballot,
89	source_url=sopn_data["source_url"],
90	uploaded_file=official_document.uploaded_file,
91	document_type=OfficialDocument.NOMINATION_PAPER,
92	)
93	continue	×
94
95	# otherwise we dont have this file stored already, so download it as
96	# part of creating the OfficialDocument
97	self.stdout.write(	×
98	f"Downloading SOPN from {sopn_data['uploaded_file']}"
99	)
100	file_response = requests.get(sopn_data["uploaded_file"])	×
101	file_object = ContentFile(content=file_response.content)	×
UNCOV 102	official_document = OfficialDocument(	×
103	ballot=ballot,
104	source_url=sopn_data["source_url"],
105	document_type=OfficialDocument.NOMINATION_PAPER,
106	)
107	file_extension = sopn_data["uploaded_file"].split(".")[-1]	×
UNCOV 108	filename = f"{ballot.ballot_paper_id}.{file_extension}"	×
UNCOV 109	official_document.uploaded_file.save(	×
110	name=filename, content=file_object
111	)
112
113	# this should only be the case where the election object has > 200
114	# ballots e.g. parliamentary elections
115	if next_page:	×
116	return self.create_official_documents(url=next_page)	×
117	return None	×

DemocracyClub / yournextrepresentative / 74df76c8-4768-48d5-bb7f-5ee50aa05217

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