c2872dde-5a84-42ab-ae61-dc2bba04964a

Committed 10 Apr 2024 08:32AM UTC coverage: 68.867% (+0.7%) from 68.145%

Build # c2872dde-5a84-42ab-ae61-dc2bba04964a

Build Type

Pull #2279

circleci

Committed by

symroe

Commit Message

Add initial reports frontend

Pull Request Pull Request #2279: Election dashboards

Run Details

1775 of 2991 branches covered (59.34%)

Branch coverage included in aggregate %.

75 of 156 new or added lines in 5 files covered. (48.08%)

6 existing lines in 1 file now uncovered.

7321 of 10217 relevant lines covered (71.66%)

0.72 hits per line

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

32.88

/ynr/apps/bulk_adding/helpers.py

import csv

from bulk_adding.models import RawPeople
from candidates.models import Ballot, LoggedAction, raise_if_unsafe_to_delete
from candidates.models.db import ActionType, EditType
from candidates.views.version_data import get_change_metadata, get_client_ip
from parties.models import Party, PartyDescription
from people.models import Person
from popolo.models import Membership


def add_person(request, person_data):
    person = Person.objects.create(name=person_data["name"])

    change_metadata = get_change_metadata(request, person_data["source"])

    person.record_version(change_metadata, new_person=True)
    person.save()

    LoggedAction.objects.create(
        user=request.user,
        person=person,
        action_type=ActionType.PERSON_CREATE,
        ip_address=get_client_ip(request),
        popit_person_new_version=change_metadata["version_id"],
        source=change_metadata["information_source"],
        edit_type=EditType.BULK_ADD.name,
    )
    return person


def update_person(
    request=None,
    person=None,
    party=None,
    ballot=None,
    source=None,
    list_position=None,
    party_description=None,
    previous_party_affiliations=None,
    data=None,
):
    election = ballot.election

    person.not_standing.remove(election)

    defaults = {
        "party": party,
        "party_list_position": list_position,
        "elected": None,
        "role": election.candidate_membership_role,
        "party_name": party.name,
    }

    if party_description:
        defaults.update(
            {
                "party_description": party_description,
                "party_description_text": party_description.description,
            }
        )

    membership, _ = Membership.objects.update_or_create(
        post=ballot.post, person=person, ballot=ballot, defaults=defaults
    )

    if previous_party_affiliations:
        for party in previous_party_affiliations:
            membership.previous_party_affiliations.add(party)

    # Now remove other memberships in this election for that
    # person, although we raise an exception if there is any
    # object that has a
    # ForeignKey to the membership, since that would result in
    # losing data.
    old_memberships = (
        Membership.objects.exclude(pk=membership.pk)
        .exclude(ballot__candidates_locked=True)
        .filter(person=person, ballot__election=ballot.election)
    )
    for old_membership in old_memberships:
        raise_if_unsafe_to_delete(old_membership)
        old_membership.delete()

    memberships_for_election = Membership.objects.filter(
        person=person, ballot__election=ballot.election
    )

    if (
        not memberships_for_election.exists()
        or memberships_for_election.count() > 1
    ):
        raise ValueError(
            "Attempt to create invalid memberships for {}".format(person)
        )
    if not data:
        data = {}

    # Writing this as a configable in case we come back to changing this
    # The reason is it's not clear if we want to take the newly entered name as
    # the main name, or keep the previous main name and keep the newly entered name
    # as an "other name".
    # We want to keep both names either way.
    # Set either NAME_MODE to `KEEP_SOPN_NAME` or `KEEP_EXISTING_NAME`
    NAME_MODE = "KEEP_SOPN_NAME"
    sopn_name = data.get("name")
    if sopn_name and person.name != sopn_name:
        if NAME_MODE == "KEEP_SOPN_NAME":
            person.other_names.get_or_create(name=person.name)
            person.name = sopn_name
        if NAME_MODE == "KEEP_EXISTING_NAME":
            person.other_names.get_or_create(name=sopn_name)

    change_metadata = get_change_metadata(request, source)
    person.record_version(change_metadata)
    person.save()

    LoggedAction.objects.create(
        user=request.user,
        person=person,
        action_type=ActionType.CANDIDACY_CREATE,
        ip_address=get_client_ip(request),
        popit_person_new_version=change_metadata["version_id"],
        source=change_metadata["information_source"],
        edit_type=EditType.BULK_ADD.name,
    )


class CSVImporter:
    def __init__(self, file_name, election):
        with open(file_name) as f:
            self.csv_data = list(csv.DictReader(f.readlines()))
        self.header_rows = self.csv_data[0].keys()
        self.election = election

        self.validate_posts()

        self.ballots = {}
        for ballot in self.election.ballot_set.all():
            self.ballots[ballot.ballot_paper_id] = {
                "ballot": ballot,
                "data": [],
            }

    @property
    def post_header_name(self):
        if "ballot_paper_id" in self.header_rows:
            return "ballot_paper_id"
        POST_LABELS = [
            "Electoral Area Name",
            "Election Area",
            "AreaName",
            "AREA_NAME",
        ]
        for label in POST_LABELS:
            if label in self.header_rows:
                return label
        raise ValueError("No post header matched")

    @property
    def party_description_header_name(self):
        DESCRIPTION_FIELDS = [
            "Description",
            "Candidates Description",
            "CandidateDescription",
            "CandidateLine5",
            "PARTY_OFFICIAL",
        ]
        for description_field in DESCRIPTION_FIELDS:
            if description_field in self.header_rows:
                return description_field
        raise ValueError("No party description header matched")

    def clean_area_name(self, name):
        name = name.replace(" Ward", "").strip().lower()
        name = name.replace("`", "'")
        return name.replace(" & ", " and ")

    def match_division_to_ballot(self, row):
        if "ballot_paper_id" in row:
            return Ballot.objects.get(ballot_paper_id=row["ballot_paper_id"])
        post_name = self.clean_area_name(row[self.post_header_name])
        for ballot in self.election.ballot_set.all():
            print(self.clean_area_name(ballot.post.label), post_name)
            if self.clean_area_name(ballot.post.label) == post_name:
                return ballot
        raise ValueError("No ballot matched to {}".format(post_name))

    def format_name(self, row):
        FULL_NAME_FIELDS = ["Ballot Paper Name"]
        for name_field in FULL_NAME_FIELDS:
            if name_field in self.header_rows:
                return row[name_field]

        name = []

        FIRST_NAME_FIELDS = [
            "Forename",
            "CandidateForename",
            "CandidateLine2",
            "CAND_CFOR",
        ]
        LAST_NAME_FIELDS = [
            "Surname",
            "CandidateSurname",
            "CandidateLine1",
            "CAND_CSUR_CAPS",
        ]

        for name_part in [FIRST_NAME_FIELDS, LAST_NAME_FIELDS]:
            for name_field in name_part:
                if name_field in self.header_rows:
                    name_part = row[name_field]
                    if name_part.upper() == name_part:
                        # This is all upper case, title case it
                        name_part = name_part.title()
                    name.append(name_part)
        if name:
            return " ".join(name)

        raise ValueError("Blank name")

    def get_register_for_ballot(self, ballot):
        group = ballot.post.group
        return "NI" if group == "Northern Ireland" else "GB"

    def clean_party_name(self, name):
        if name == "Ukip":
            name = "UKIP"
        return name.lower()

    def get_party_description(self, row, ballot):
        register = self.get_register_for_ballot(ballot)
        desc = self.clean_party_name(row[self.party_description_header_name])
        qs = PartyDescription.objects.filter(
            description__iexact=desc, party__register=register
        )
        if qs.exists():
            return qs.first()

        qs = PartyDescription.objects.filter(
            description__istartswith="{}".format(desc), party__register=register
        )
        if qs.exists():
            return qs.first()
        return (
            Party.objects.current()
            .get(name__iexact=desc, register=register)
            .descriptions.first()
        )

    def get_party_id(self, row, ballot_model):
        register = self.get_register_for_ballot(ballot_model)
        desc_model = self.get_party_description(row, ballot_model)
        if desc_model:
            return desc_model.party.ec_id

        party_name = self.clean_party_name(
            row[self.party_description_header_name]
        )

        if party_name in ("", None):
            return "ynmp-party:2"
        print(repr(party_name))

        return Party.objects.register(register).get(name=party_name).ec_id

    def validate_posts(self):
        """
        Figure out if the posts in this CSV look right, raise if not
        """

        unique_posts = {row[self.post_header_name] for row in self.csv_data}
        if len(unique_posts) != self.election.ballot_set.count():
            raise ValueError("Number of posts don't match")

        return True

    def extract(self):
        for row in self.csv_data:
            ballot_model = self.match_division_to_ballot(row)
            ballot_dict = self.ballots[ballot_model.ballot_paper_id]
            print(self.format_name(row))
            print(self.get_party_id(row, ballot_model))
            data = {
                "name": self.format_name(row),
                "party_id": self.get_party_id(row, ballot_model),
            }
            desc = self.get_party_description(row, ballot_model)
            if desc:
                data["description_id"] = str(desc.pk)
            ballot_dict["data"].append(data)

        for ballot_id, ballot_dict in self.ballots.items():
            RawPeople.objects.update_or_create(
                ballot=ballot_dict["ballot"],
                defaults={
                    "data": ballot_dict["data"],
                    "source": "CSV from Council",
                    "source_type": RawPeople.SOURCE_COUNCIL_CSV,
                },
            )

1	import csv	1✔
2
3	from bulk_adding.models import RawPeople	1✔
4	from candidates.models import Ballot, LoggedAction, raise_if_unsafe_to_delete	1✔
5	from candidates.models.db import ActionType, EditType	1✔
6	from candidates.views.version_data import get_change_metadata, get_client_ip	1✔
7	from parties.models import Party, PartyDescription	1✔
8	from people.models import Person	1✔
9	from popolo.models import Membership	1✔
10
11
12	def add_person(request, person_data):	1✔
13	person = Person.objects.create(name=person_data["name"])	1✔
14
15	change_metadata = get_change_metadata(request, person_data["source"])	1✔
16
17	person.record_version(change_metadata, new_person=True)	1✔
18	person.save()	1✔
19
20	LoggedAction.objects.create(	1✔
21	user=request.user,
22	person=person,
23	action_type=ActionType.PERSON_CREATE,
24	ip_address=get_client_ip(request),
25	popit_person_new_version=change_metadata["version_id"],
26	source=change_metadata["information_source"],
27	edit_type=EditType.BULK_ADD.name,
28	)
29	return person	1✔
30
31
32	def update_person(	1✔
33	request=None,
34	person=None,
35	party=None,
36	ballot=None,
37	source=None,
38	list_position=None,
39	party_description=None,
40	previous_party_affiliations=None,
41	data=None,
42	):
43	election = ballot.election	1✔
44
45	person.not_standing.remove(election)	1✔
46
47	defaults = {	1✔
48	"party": party,
49	"party_list_position": list_position,
50	"elected": None,
51	"role": election.candidate_membership_role,
52	"party_name": party.name,
53	}
54
55	if party_description:	1✔
56	defaults.update(	1✔
57	{
58	"party_description": party_description,
59	"party_description_text": party_description.description,
60	}
61	)
62
63	membership, _ = Membership.objects.update_or_create(	1✔
64	post=ballot.post, person=person, ballot=ballot, defaults=defaults
65	)
66
67	if previous_party_affiliations:	1✔
68	for party in previous_party_affiliations:	1✔
69	membership.previous_party_affiliations.add(party)	1✔
70
71	# Now remove other memberships in this election for that
72	# person, although we raise an exception if there is any
73	# object that has a
74	# ForeignKey to the membership, since that would result in
75	# losing data.
76	old_memberships = (	1✔
77	Membership.objects.exclude(pk=membership.pk)
78	.exclude(ballot__candidates_locked=True)
79	.filter(person=person, ballot__election=ballot.election)
80	)
81	for old_membership in old_memberships:	1!
82	raise_if_unsafe_to_delete(old_membership)	×
83	old_membership.delete()	×
84
85	memberships_for_election = Membership.objects.filter(	1✔
86	person=person, ballot__election=ballot.election
87	)
88
89	if (	1!
90	not memberships_for_election.exists()
91	or memberships_for_election.count() > 1
92	):
93	raise ValueError(	×
94	"Attempt to create invalid memberships for {}".format(person)
95	)
96	if not data:	1✔
97	data = {}	1✔
98
99	# Writing this as a configable in case we come back to changing this
100	# The reason is it's not clear if we want to take the newly entered name as
101	# the main name, or keep the previous main name and keep the newly entered name
102	# as an "other name".
103	# We want to keep both names either way.
104	# Set either NAME_MODE to `KEEP_SOPN_NAME` or `KEEP_EXISTING_NAME`
105	NAME_MODE = "KEEP_SOPN_NAME"	1✔
106	sopn_name = data.get("name")	1✔
107	if sopn_name and person.name != sopn_name:	1✔
108	if NAME_MODE == "KEEP_SOPN_NAME":	1!
109	person.other_names.get_or_create(name=person.name)	1✔
110	person.name = sopn_name	1✔
111	if NAME_MODE == "KEEP_EXISTING_NAME":	1!
112	person.other_names.get_or_create(name=sopn_name)	×
113
114	change_metadata = get_change_metadata(request, source)	1✔
115	person.record_version(change_metadata)	1✔
116	person.save()	1✔
117
118	LoggedAction.objects.create(	1✔
119	user=request.user,
120	person=person,
121	action_type=ActionType.CANDIDACY_CREATE,
122	ip_address=get_client_ip(request),
123	popit_person_new_version=change_metadata["version_id"],
124	source=change_metadata["information_source"],
125	edit_type=EditType.BULK_ADD.name,
126	)
127
128
129	class CSVImporter:	1✔
130	def __init__(self, file_name, election):	1✔
131	with open(file_name) as f:	×
132	self.csv_data = list(csv.DictReader(f.readlines()))	×
133	self.header_rows = self.csv_data[0].keys()	×
134	self.election = election	×
135
136	self.validate_posts()	×
137
138	self.ballots = {}	×
139	for ballot in self.election.ballot_set.all():	×
140	self.ballots[ballot.ballot_paper_id] = {	×
141	"ballot": ballot,
142	"data": [],
143	}
144
145	@property	1✔
146	def post_header_name(self):	1✔
147	if "ballot_paper_id" in self.header_rows:	×
148	return "ballot_paper_id"	×
149	POST_LABELS = [	×
150	"Electoral Area Name",
151	"Election Area",
152	"AreaName",
153	"AREA_NAME",
154	]
155	for label in POST_LABELS:	×
156	if label in self.header_rows:	×
157	return label	×
158	raise ValueError("No post header matched")	×
159
160	@property	1✔
161	def party_description_header_name(self):	1✔
162	DESCRIPTION_FIELDS = [	×
163	"Description",
164	"Candidates Description",
165	"CandidateDescription",
166	"CandidateLine5",
167	"PARTY_OFFICIAL",
168	]
169	for description_field in DESCRIPTION_FIELDS:	×
170	if description_field in self.header_rows:	×
171	return description_field	×
172	raise ValueError("No party description header matched")	×
173
174	def clean_area_name(self, name):	1✔
175	name = name.replace(" Ward", "").strip().lower()	×
176	name = name.replace("`", "'")	×
177	return name.replace(" & ", " and ")	×
178
179	def match_division_to_ballot(self, row):	1✔
180	if "ballot_paper_id" in row:	×
181	return Ballot.objects.get(ballot_paper_id=row["ballot_paper_id"])	×
182	post_name = self.clean_area_name(row[self.post_header_name])	×
183	for ballot in self.election.ballot_set.all():	×
184	print(self.clean_area_name(ballot.post.label), post_name)	×
185	if self.clean_area_name(ballot.post.label) == post_name:	×
186	return ballot	×
187	raise ValueError("No ballot matched to {}".format(post_name))	×
188
189	def format_name(self, row):	1✔
190	FULL_NAME_FIELDS = ["Ballot Paper Name"]	×
191	for name_field in FULL_NAME_FIELDS:	×
192	if name_field in self.header_rows:	×
193	return row[name_field]	×
194
195	name = []	×
196
197	FIRST_NAME_FIELDS = [	×
198	"Forename",
199	"CandidateForename",
200	"CandidateLine2",
201	"CAND_CFOR",
202	]
203	LAST_NAME_FIELDS = [	×
204	"Surname",
205	"CandidateSurname",
206	"CandidateLine1",
207	"CAND_CSUR_CAPS",
208	]
209
210	for name_part in [FIRST_NAME_FIELDS, LAST_NAME_FIELDS]:	×
211	for name_field in name_part:	×
212	if name_field in self.header_rows:	×
213	name_part = row[name_field]	×
214	if name_part.upper() == name_part:	×
215	# This is all upper case, title case it
216	name_part = name_part.title()	×
217	name.append(name_part)	×
218	if name:	×
219	return " ".join(name)	×
220
221	raise ValueError("Blank name")	×
222
223	def get_register_for_ballot(self, ballot):	1✔
224	group = ballot.post.group	×
225	return "NI" if group == "Northern Ireland" else "GB"	×
226
227	def clean_party_name(self, name):	1✔
228	if name == "Ukip":	×
229	name = "UKIP"	×
230	return name.lower()	×
231
232	def get_party_description(self, row, ballot):	1✔
233	register = self.get_register_for_ballot(ballot)	×
234	desc = self.clean_party_name(row[self.party_description_header_name])	×
235	qs = PartyDescription.objects.filter(	×
236	description__iexact=desc, party__register=register
237	)
238	if qs.exists():	×
239	return qs.first()	×
240
241	qs = PartyDescription.objects.filter(	×
242	description__istartswith="{}".format(desc), party__register=register
243	)
244	if qs.exists():	×
245	return qs.first()	×
246	return (	×
247	Party.objects.current()
248	.get(name__iexact=desc, register=register)
249	.descriptions.first()
250	)
251
252	def get_party_id(self, row, ballot_model):	1✔
253	register = self.get_register_for_ballot(ballot_model)	×
254	desc_model = self.get_party_description(row, ballot_model)	×
255	if desc_model:	×
256	return desc_model.party.ec_id	×
257
258	party_name = self.clean_party_name(	×
259	row[self.party_description_header_name]
260	)
261
262	if party_name in ("", None):	×
263	return "ynmp-party:2"	×
264	print(repr(party_name))	×
265
UNCOV 266	return Party.objects.register(register).get(name=party_name).ec_id	×
267
268	def validate_posts(self):	1✔
269	"""
270	Figure out if the posts in this CSV look right, raise if not
271	"""
272
UNCOV 273	unique_posts = {row[self.post_header_name] for row in self.csv_data}	×
274	if len(unique_posts) != self.election.ballot_set.count():	×
275	raise ValueError("Number of posts don't match")	×
276
UNCOV 277	return True	×
278
279	def extract(self):	1✔
UNCOV 280	for row in self.csv_data:	×
281	ballot_model = self.match_division_to_ballot(row)	×
282	ballot_dict = self.ballots[ballot_model.ballot_paper_id]	×
283	print(self.format_name(row))	×
284	print(self.get_party_id(row, ballot_model))	×
285	data = {	×
286	"name": self.format_name(row),
287	"party_id": self.get_party_id(row, ballot_model),
288	}
UNCOV 289	desc = self.get_party_description(row, ballot_model)	×
290	if desc:	×
291	data["description_id"] = str(desc.pk)	×
292	ballot_dict["data"].append(data)	×
293
UNCOV 294	for ballot_id, ballot_dict in self.ballots.items():	×
295	RawPeople.objects.update_or_create(	×
296	ballot=ballot_dict["ballot"],
297	defaults={
298	"data": ballot_dict["data"],
299	"source": "CSV from Council",
300	"source_type": RawPeople.SOURCE_COUNCIL_CSV,
301	},
302	)

DemocracyClub / yournextrepresentative / c2872dde-5a84-42ab-ae61-dc2bba04964a

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