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

export updates

Browse source
This commit is contained in:
CocoByte 2024-11-13 01:16:55 -07:00
parent c481c3c08d
commit 32232c5d56
No known key found for this signature in database
GPG key ID: BBFAA2526384C97F
6 changed files with 236 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -21,6 +21,11 @@ from registrar.utility.constants import BranchChoices
from registrar.utility.enums import DefaultEmail
# ---Logger
import logging
from venv import logger
from registrar.management.commands.utility.terminal_helper import TerminalColors, TerminalHelper
logger = logging.getLogger(__name__)
@ -197,6 +202,8 @@ class BaseExport(ABC):
All domain metadata:
Exports domains of all statuses plus domain managers.
"""
TerminalHelper.colorful_logger(logger.info, TerminalColors.OKGREEN, f"Exporting data")
writer = csv.writer(csv_file)
columns = cls.get_columns()
sort_fields = cls.get_sort_fields()
@ -226,6 +233,8 @@ class BaseExport(ABC):
)
models_dict = convert_queryset_to_dict(annotated_queryset, is_model=False)
TerminalHelper.colorful_logger(logger.info, TerminalColors.OKGREEN, f"COLUMNS: {columns}")
# Write to csv file before the write_csv
cls.write_csv_before(writer, **export_kwargs)
@ -374,8 +383,8 @@ class DomainExport(BaseExport):
if first_ready_on is None:
first_ready_on = "(blank)"
# organization_type has generic_org_type AND is_election
domain_org_type = model.get("organization_type")
# organization_type has organization_type AND is_election
domain_org_type = model.get("converted_generic_org_type") or model.get("organization_type")
human_readable_domain_org_type = DomainRequest.OrgChoicesElectionOffice.get_org_label(domain_org_type)
domain_federal_type = model.get("federal_type")
human_readable_domain_federal_type = BranchChoices.get_branch_label(domain_federal_type)
@ -392,6 +401,7 @@ class DomainExport(BaseExport):
):
security_contact_email = "(blank)"
# create a dictionary of fields which can be included in output.
# "extra_fields" are precomputed fields (generated in the DB or parsed).
FIELDS = {
@ -400,12 +410,12 @@ class DomainExport(BaseExport):
"First ready on": first_ready_on,
"Expiration date": expiration_date,
"Domain type": domain_type,
"Agency": model.get("federal_agency__agency"),
"Organization name": model.get("organization_name"),
"City": model.get("city"),
"State": model.get("state_territory"),
"Agency": model.get("converted_federal_agency__agency"),
"Organization name": model.get("converted_organization_name"),
"City": model.get("converted_city"),
"State": model.get("converted_state_territory"),
"SO": model.get("so_name"),
"SO email": model.get("senior_official__email"),
"SO email": model.get("converted_senior_official__email"),
"Security contact email": security_contact_email,
"Created at": model.get("domain__created_at"),
"Deleted": model.get("domain__deleted"),
@ -414,8 +424,20 @@ class DomainExport(BaseExport):
}
row = [FIELDS.get(column, "") for column in columns]
TerminalHelper.colorful_logger(logger.info, TerminalColors.YELLOW, f"PARSING ROW: {row}")
return row
def get_filtered_domain_infos_by_org(domain_infos_to_filter, org_to_filter_by):
"""Returns a list of Domain Requests that has been filtered by the given organization value."""
return domain_infos_to_filter.filter(
# Filter based on the generic org value returned by converted_generic_org_type
id__in=[
domainInfos.id for domainInfos in domain_infos_to_filter if domainInfos.converted_generic_org_type and domainInfos.converted_generic_org_type == org_to_filter_by
]
)
@classmethod
def get_sliced_domains(cls, filter_condition):
"""Get filtered domains counts sliced by org type and election office.
@ -423,23 +445,23 @@ class DomainExport(BaseExport):
when a domain has more that one manager.
"""
domains = DomainInformation.objects.all().filter(**filter_condition).distinct()
domains_count = domains.count()
federal = domains.filter(generic_org_type=DomainRequest.OrganizationChoices.FEDERAL).distinct().count()
interstate = domains.filter(generic_org_type=DomainRequest.OrganizationChoices.INTERSTATE).count()
domain_informations = DomainInformation.objects.all().filter(**filter_condition).distinct()
domains_count = domain_informations.count()
federal = cls.get_filtered_domain_infos_by_org(domain_informations, DomainRequest.OrganizationChoices.FEDERAL).distinct().count()
interstate = cls.get_filtered_domain_infos_by_org(domain_informations, DomainRequest.OrganizationChoices.INTERSTATE).count()
state_or_territory = (
domains.filter(generic_org_type=DomainRequest.OrganizationChoices.STATE_OR_TERRITORY).distinct().count()
cls.get_filtered_domain_infos_by_org(domain_informations, DomainRequest.OrganizationChoices.STATE_OR_TERRITORY).distinct().count()
)
tribal = domains.filter(generic_org_type=DomainRequest.OrganizationChoices.TRIBAL).distinct().count()
county = domains.filter(generic_org_type=DomainRequest.OrganizationChoices.COUNTY).distinct().count()
city = domains.filter(generic_org_type=DomainRequest.OrganizationChoices.CITY).distinct().count()
tribal = cls.get_filtered_domain_infos_by_org(domain_informations, DomainRequest.OrganizationChoices.TRIBAL).distinct().count()
county = cls.get_filtered_domain_infos_by_org(domain_informations, DomainRequest.OrganizationChoices.COUNTY).distinct().count()
city = cls.get_filtered_domain_infos_by_org(domain_informations, DomainRequest.OrganizationChoices.CITY).distinct().count()
special_district = (
domains.filter(generic_org_type=DomainRequest.OrganizationChoices.SPECIAL_DISTRICT).distinct().count()
cls.get_filtered_domain_infos_by_org(domain_informations, DomainRequest.OrganizationChoices.SPECIAL_DISTRICT).distinct().count()
)
school_district = (
domains.filter(generic_org_type=DomainRequest.OrganizationChoices.SCHOOL_DISTRICT).distinct().count()
cls.get_filtered_domain_infos_by_org(domain_informations, DomainRequest.OrganizationChoices.SCHOOL_DISTRICT).distinct().count()
)
election_board = domains.filter(is_election_board=True).distinct().count()
election_board = domain_informations.filter(is_election_board=True).distinct().count()
return [
domains_count,
@ -461,11 +483,15 @@ class DomainDataType(DomainExport):
Inherits from BaseExport -> DomainExport
"""
TerminalHelper.colorful_logger(logger.info, TerminalColors.YELLOW, f"DomainDataType!!")
@classmethod
def get_columns(cls):
"""
Overrides the columns for CSV export specific to DomainExport.
"""
TerminalHelper.colorful_logger(logger.info, TerminalColors.YELLOW, f"...getting columns")
return [
"Domain name",
"Status",
@ -524,7 +550,7 @@ class DomainDataType(DomainExport):
"""
Get a list of tables to pass to select_related when building queryset.
"""
return ["domain", "senior_official"]
return ["domain", "converted_senior_official"]
@classmethod
def get_prefetch_related(cls):
@ -660,11 +686,11 @@ class DomainRequestsDataType:
cls.safe_get(getattr(request, "all_alternative_domains", None)),
cls.safe_get(getattr(request, "all_other_contacts", None)),
cls.safe_get(getattr(request, "all_current_websites", None)),
cls.safe_get(getattr(request, "converted_federal_agency", None)),
cls.safe_get(getattr(request.converted_senior_official, "first_name", None)),
cls.safe_get(getattr(request.converted_senior_official, "last_name", None)),
cls.safe_get(getattr(request.converted_senior_official, "email", None)),
cls.safe_get(getattr(request.converted_senior_official, "title", None)),
cls.safe_get(getattr(request, "federal_agency", None)),
cls.safe_get(getattr(request.senior_official, "first_name", None)),
cls.safe_get(getattr(request.senior_official, "last_name", None)),
cls.safe_get(getattr(request.senior_official, "email", None)),
cls.safe_get(getattr(request.senior_official, "title", None)),
cls.safe_get(getattr(request.creator, "first_name", None)),
cls.safe_get(getattr(request.creator, "last_name", None)),
cls.safe_get(getattr(request.creator, "email", None)),
@ -1223,25 +1249,35 @@ class DomainRequestExport(BaseExport):
def model(cls):
# Return the model class that this export handles
return DomainRequest
def get_filtered_domain_requests_by_org(domain_requests_to_filter, org_to_filter_by):
"""Returns a list of Domain Requests that has been filtered by the given organization value"""
return domain_requests_to_filter.filter(
# Filter based on the generic org value returned by converted_generic_org_type
id__in=[
domainRequest.id for domainRequest in domain_requests_to_filter if domainRequest.converted_generic_org_type and domainRequest.converted_generic_org_type == org_to_filter_by
]
)
@classmethod
def get_sliced_requests(cls, filter_condition):
"""Get filtered requests counts sliced by org type and election office."""
requests = DomainRequest.objects.all().filter(**filter_condition).distinct()
requests_count = requests.count()
federal = requests.filter(generic_org_type=DomainRequest.OrganizationChoices.FEDERAL).distinct().count()
interstate = requests.filter(generic_org_type=DomainRequest.OrganizationChoices.INTERSTATE).distinct().count()
federal = cls.get_filtered_domain_requests_by_org(requests, DomainRequest.OrganizationChoices.FEDERAL).distinct().count()
interstate = cls.get_filtered_domain_requests_by_org(requests, DomainRequest.OrganizationChoices.INTERSTATE).distinct().count()
state_or_territory = (
requests.filter(generic_org_type=DomainRequest.OrganizationChoices.STATE_OR_TERRITORY).distinct().count()
cls.get_filtered_domain_requests_by_org(requests, DomainRequest.OrganizationChoices.STATE_OR_TERRITORY).distinct().count()
)
tribal = requests.filter(generic_org_type=DomainRequest.OrganizationChoices.TRIBAL).distinct().count()
county = requests.filter(generic_org_type=DomainRequest.OrganizationChoices.COUNTY).distinct().count()
city = requests.filter(generic_org_type=DomainRequest.OrganizationChoices.CITY).distinct().count()
tribal = cls.get_filtered_domain_requests_by_org(requests, DomainRequest.OrganizationChoices.TRIBAL).distinct().count()
county = cls.get_filtered_domain_requests_by_org(requests, DomainRequest.OrganizationChoices.COUNTY).distinct().count()
city = cls.get_filtered_domain_requests_by_org(requests, DomainRequest.OrganizationChoices.CITY).distinct().count()
special_district = (
requests.filter(generic_org_type=DomainRequest.OrganizationChoices.SPECIAL_DISTRICT).distinct().count()
cls.get_filtered_domain_requests_by_org(requests, DomainRequest.OrganizationChoices.SPECIAL_DISTRICT).distinct().count()
)
school_district = (
requests.filter(generic_org_type=DomainRequest.OrganizationChoices.SCHOOL_DISTRICT).distinct().count()
cls.get_filtered_domain_requests_by_org(requests, DomainRequest.OrganizationChoices.SCHOOL_DISTRICT).distinct().count()
)
election_board = requests.filter(is_election_board=True).distinct().count()
@ -1269,7 +1305,7 @@ class DomainRequestExport(BaseExport):
human_readable_federal_type = BranchChoices.get_branch_label(federal_type) if federal_type else None
# Handle the org_type field
org_type = model.get("generic_org_type") or model.get("organization_type")
org_type = model.get("converted_generic_org_type") or model.get("organization_type")
human_readable_org_type = DomainRequest.OrganizationChoices.get_org_label(org_type) if org_type else None
# Handle the status field. Defaults to the wrong format.
@ -1327,9 +1363,9 @@ class DomainRequestExport(BaseExport):
"Creator email": model.get("creator__email"),
"Investigator": model.get("investigator__email"),
# Untouched fields
"Organization name": model.get("organization_name"),
"City": model.get("city"),
"State/territory": model.get("state_territory"),
"Organization name": model.get("converted_organization_name"),
"City": model.get("converted_city"),
"State/territory": model.get("converted_state_territory"),
"Request purpose": model.get("purpose"),
"CISA regional representative": model.get("cisa_representative_email"),
"Last submitted date": model.get("last_submitted_date"),