zydronium/manage.get.gov
1
0
Fork 0
mirror of https://github.com/cisagov/manage.get.gov.git synced 2025-08-14 21:44:08 +02:00
Code Issues Projects Releases Packages Wiki Activity

Update csv_export.py

Browse source
This commit is contained in:
zandercymatics 2024-06-05 15:39:22 -06:00
parent 9898fe16c0
commit d870127eb3
No known key found for this signature in database
GPG key ID: FF4636ABEC9682B7
1 changed files with 130 additions and 96 deletions
Download patch file Download diff file Expand all files Collapse all files

216
src/registrar/utility/csv_export.py
View file

@ -1,7 +1,15 @@
import csv import csv
import logging import logging
from datetime import datetime from datetime import datetime
from registrar.models import Domain, DomainInvitation, DomainRequest, DomainInformation, User, PublicContact, UserDomainRole from registrar.models import (
Domain,
DomainInvitation,
DomainRequest,
DomainInformation,
User,
PublicContact,
UserDomainRole,
)
from django.db.models import QuerySet, Value, Case, When, CharField, Count, Q, F, Value, CharField from django.db.models import QuerySet, Value, Case, When, CharField, Count, Q, F, Value, CharField
from django.utils import timezone from django.utils import timezone
from django.core.paginator import Paginator from django.core.paginator import Paginator
@ -293,6 +301,7 @@ def write_csv_for_domains(
write_header(writer, columns) write_header(writer, columns)
writer.writerows(total_body_rows) writer.writerows(total_body_rows)
def export_data_type_to_csv(csv_file): def export_data_type_to_csv(csv_file):
""" """
All domains report with extra columns. All domains report with extra columns.
@ -740,37 +749,37 @@ class DomainRequestExport:
# create a dictionary of fields which can be included in output # create a dictionary of fields which can be included in output
FIELDS = { FIELDS = {
"Domain request": request.get('requested_domain_name'), "Domain request": request.get("requested_domain_name"),
"Submitted at": request.get('submission_date'), # TODO - different format? "Submitted at": request.get("submission_date"), # TODO - different format?
"Status": request.get('status_display'), "Status": request.get("status_display"),
"Domain type": request.get('domain_type'), "Domain type": request.get("domain_type"),
"Federal type": request.get("human_readable_federal_type"), "Federal type": request.get("human_readable_federal_type"),
"Federal agency": request.get('federal_agency_name'), "Federal agency": request.get("federal_agency_name"),
"Organization name": request.get('organization_name'), "Organization name": request.get("organization_name"),
"Election office": request.get("human_readable_election_board"), "Election office": request.get("human_readable_election_board"),
"City": request.get('city'), "City": request.get("city"),
"State/territory": request.get('state_territory'), "State/territory": request.get("state_territory"),
"Region": None, # TODO - what is this field? "Region": None, # TODO - what is this field?
# Creator # Creator
"Creator first name": request.get('creator__first_name', ""), "Creator first name": request.get("creator__first_name", ""),
"Creator last name": request.get('creator__last_name', ""), "Creator last name": request.get("creator__last_name", ""),
"Creator email": request.get('creator__email', ""), "Creator email": request.get("creator__email", ""),
"Creator approved domains count": request.get("creator_approved_domains_count", 0), "Creator approved domains count": request.get("creator_approved_domains_count", 0),
"Creator active requests count": request.get("creator_active_requests_count", 0), "Creator active requests count": request.get("creator_active_requests_count", 0),
# End of creator # End of creator
"Alternative domains": request.get('all_alternative_domains'), "Alternative domains": request.get("all_alternative_domains"),
# AO # AO
"AO first name": request.get('authorizing_official__first_name', ""), "AO first name": request.get("authorizing_official__first_name", ""),
"AO last name": request.get('authorizing_official__last_name', ""), "AO last name": request.get("authorizing_official__last_name", ""),
"AO email": request.get('authorizing_official__email', ""), "AO email": request.get("authorizing_official__email", ""),
"AO title/role": request.get('authorizing_official__title', ""), "AO title/role": request.get("authorizing_official__title", ""),
# End of AO # End of AO
"Request purpose": request.get('purpose'), "Request purpose": request.get("purpose"),
"Request additional details": request.get('additional_details'), "Request additional details": request.get("additional_details"),
"Other contacts": request.get('all_other_contacts'), "Other contacts": request.get("all_other_contacts"),
"CISA regional representative": request.get('cisa_representative_email'), # TODO - same problem as before "CISA regional representative": request.get("cisa_representative_email"), # TODO - same problem as before
"Current websites": request.get('all_current_websites'), "Current websites": request.get("all_current_websites"),
"Investigator": request.get('investigator_email'), "Investigator": request.get("investigator_email"),
} }
row = [FIELDS.get(column, "") for column in columns] row = [FIELDS.get(column, "") for column in columns]
@ -809,13 +818,8 @@ class DomainRequestExport:
# Convert the request to a querystring for faster processing. Only grab what we need. # Convert the request to a querystring for faster processing. Only grab what we need.
parsed_requests = all_requests.annotate( parsed_requests = all_requests.annotate(
requested_domain_name=DomainRequestExport.get_requested_domain_name_query(), requested_domain_name=DomainRequestExport.get_requested_domain_name_query(),
approved_domain_name=F('approved_domain__name'), approved_domain_name=F("approved_domain__name"),
).values( ).values("requested_domain_name", "generic_org_type", "federal_type", "submission_date")
"requested_domain_name",
"generic_org_type",
"federal_type",
"submission_date"
)
# Set the domain_type field. # Set the domain_type field.
# Relies on a python function, get_organization_type_display, so we have to # Relies on a python function, get_organization_type_display, so we have to
@ -911,74 +915,50 @@ class DomainRequestExport:
parsed_requests = requests_to_convert.annotate( parsed_requests = requests_to_convert.annotate(
requested_domain_name=DomainRequestExport.get_requested_domain_name_query(), requested_domain_name=DomainRequestExport.get_requested_domain_name_query(),
additional_details=DomainRequestExport.get_additional_details_query(), additional_details=DomainRequestExport.get_additional_details_query(),
all_other_contacts=StringAgg( creator_approved_domains_count=DomainRequestExport.get_creator_approved_domains_count_query(),
Concat('other_contacts__first_name', Value(' '), 'other_contacts__last_name', Value(' '), 'other_contacts__email'), creator_active_requests_count=DomainRequestExport.get_creator_active_requests_count_query(),
delimiter=' | ', human_readable_election_board=DomainRequestExport.get_human_readable_election_board_query(),
distinct=True all_other_contacts=DomainRequestExport.get_all_other_contacts_query(),
), all_current_websites=StringAgg("current_websites__website", delimiter=" | ", distinct=True),
all_current_websites=StringAgg('current_websites__website', delimiter=' | ', distinct=True), all_alternative_domains=StringAgg("alternative_domains__website", delimiter=" | ", distinct=True),
all_alternative_domains=StringAgg('alternative_domains__website', delimiter=' | ', distinct=True), federal_agency_name=F("federal_agency__agency"),
investigator_email=F("investigator__email"),
# TODO - this is returning the wrong count
creator_approved_domains_count=Count(
'creator__domain_requests_created__id',
filter=Q(creator__domain_requests_created__status=DomainRequest.DomainRequestStatus.APPROVED),
distinct=True
),
# TODO - this is returning the wrong count
creator_active_requests_count=Count(
'creator__domain_requests_created__id',
filter=Q(creator__domain_requests_created__status__in=[
DomainRequest.DomainRequestStatus.SUBMITTED,
DomainRequest.DomainRequestStatus.IN_REVIEW,
DomainRequest.DomainRequestStatus.ACTION_NEEDED
]),
distinct=True
),
federal_agency_name = F("federal_agency__agency"),
investigator_email = F("investigator__email"),
human_readable_election_board=Case(
When(is_election_board=True, then=Value("Yes")),
When(is_election_board=False, then=Value("No")),
default=Value("N/A"),
output_field=CharField()
)
) )
requests_dict = parsed_requests.values( requests_dict = parsed_requests.values(
# Custom fields # Custom fields
"all_alternative_domains", "all_alternative_domains",
'all_other_contacts', "all_other_contacts",
'all_current_websites', "all_current_websites",
'additional_details', "additional_details",
'requested_domain_name', "requested_domain_name",
"federal_agency_name", "federal_agency_name",
"human_readable_election_board", "human_readable_election_board",
# Creator # Creator
'creator__first_name', "creator__first_name",
'creator__last_name', "creator__last_name",
'creator__email', "creator__email",
"creator_approved_domains_count", "creator_approved_domains_count",
"creator_active_requests_count", "creator_active_requests_count",
# AO # AO
'authorizing_official__first_name', "authorizing_official__first_name",
'authorizing_official__last_name', "authorizing_official__last_name",
'authorizing_official__email', "authorizing_official__email",
'authorizing_official__title', "authorizing_official__title",
# Investigator # Investigator
"investigator_email", "investigator_email",
# Existing fields # Existing fields
"id", "id",
'submission_date', "submission_date",
'status', "status",
'federal_type', "federal_type",
'organization_name', "organization_name",
'is_election_board', "is_election_board",
'city', "city",
'state_territory', "state_territory",
'purpose', "purpose",
'cisa_representative_email', "cisa_representative_email",
'investigator', "investigator",
"generic_org_type", "generic_org_type",
"federal_type", "federal_type",
) )
@ -1018,19 +998,15 @@ class DomainRequestExport:
`requested_domain.name If requested_domain.name is not None else default_message` `requested_domain.name If requested_domain.name is not None else default_message`
""" """
requested_domain_name_query = Case( requested_domain_name_query = Case(
When( When(requested_domain__isnull=False, then=F("requested_domain__name")),
requested_domain__isnull=False,
then=F('requested_domain__name')
),
default=Value(default_message), default=Value(default_message),
output_field=CharField(), output_field=CharField(),
) )
return requested_domain_name_query return requested_domain_name_query
@staticmethod @staticmethod
def get_additional_details_query(default_message=None): def get_additional_details_query(default_message=None, delimiter=" | "):
""" """
A SQL case statement for DomainRequest.anything_else and DomainRequest.cisa_representative_email. A SQL case statement for DomainRequest.anything_else and DomainRequest.cisa_representative_email.
@ -1048,14 +1024,72 @@ class DomainRequestExport:
# TODO - touch up on this # TODO - touch up on this
When( When(
cisa_representative_email__isnull=False, cisa_representative_email__isnull=False,
then=Concat(F('cisa_representative_email'), Value(' | '), F('anything_else')) then=Concat(F("cisa_representative_email"), Value(delimiter), F("anything_else")),
), ),
When( When(
anything_else__isnull=False, anything_else__isnull=False,
then=Concat(F('cisa_representative_email'), Value(' | '), F('anything_else')) then=Concat(F("cisa_representative_email"), Value(delimiter), F("anything_else")),
), ),
default=Value(default_message), default=Value(default_message),
output_field=CharField(), output_field=CharField(),
) )
return additional_details_query return additional_details_query
@staticmethod
def get_all_other_contacts_query(delimiter=" | "):
""" """
all_other_contacts_query = StringAgg(
Concat(
"other_contacts__first_name",
Value(" "),
"other_contacts__last_name",
Value(" "),
"other_contacts__email",
),
delimiter=delimiter,
distinct=True,
)
return all_other_contacts_query
@staticmethod
def get_creator_approved_domains_count_query():
""" """
query = Count(
"creator__domain_requests_created__id",
filter=Q(creator__domain_requests_created__status=DomainRequest.DomainRequestStatus.APPROVED),
distinct=True,
)
return query
@staticmethod
def get_creator_active_requests_count_query():
""" """
query = Count(
"creator__domain_requests_created__id",
filter=Q(
creator__domain_requests_created__status__in=[
DomainRequest.DomainRequestStatus.SUBMITTED,
DomainRequest.DomainRequestStatus.IN_REVIEW,
DomainRequest.DomainRequestStatus.ACTION_NEEDED,
]
),
distinct=True,
)
return query
@staticmethod
def get_human_readable_election_board_query():
""" """
query = Case(
When(is_election_board=True, then=Value("Yes")),
When(is_election_board=False, then=Value("No")),
default=Value("N/A"),
output_field=CharField(),
)
return query