From b8090fd46fef767daaba75fcb4289ced68bbaa98 Mon Sep 17 00:00:00 2001 From: Dave Date: Sat, 16 Dec 2023 09:48:48 -0800 Subject: [PATCH] Add transaction detail CSV summary output --- fec.py | 109 ++++++++++++++++++++++++++++++++++++--- requirements.txt | 1 + server/data/summaries.py | 16 ++++++ 3 files changed, 118 insertions(+), 8 deletions(-) diff --git a/fec.py b/fec.py index 0bedba9..32e935d 100755 --- a/fec.py +++ b/fec.py @@ -1,7 +1,9 @@ #!/usr/bin/env python3 # ruff: noqa: E501 +import csv import json +import typing as t import click import sqlalchemy.orm as sao @@ -20,6 +22,7 @@ ) from server.data.nicknames import NicknamesManager from server.data.search import ContactContributionSearcher +from server.data.summaries import ContributionSummary @click.group() @@ -262,7 +265,14 @@ def contributions(): required=False, default=None, ) -def search( +@click.option( + "--emit", + type=str, + required=False, + default="human", + help="Output format. One of: human, csv-overview, csv-contributions", +) +def search( # noqa: C901 first_name: str | None = None, last_name: str | None = None, zip_code: str | None = None, @@ -273,6 +283,7 @@ def search( contact_zip: str | None = None, google: str | None = None, linkedin: str | None = None, + emit: str = "human", ): """Search summarized FEC contributions data.""" data_manager = DataManager(data) if data is not None else DataManager.default() @@ -304,13 +315,95 @@ def search( "You must provide a contact dir, zip file, or explicit name & zip." ) - for contact, summary in searcher.search_and_summarize_contacts(contact_provider): - assert contact.city - assert contact.state - print( - f"{contact.first_name.title()} {contact.last_name.title()} ({contact.city.title()}, {contact.state})" - ) - print(str(summary)) + def _emit_overview_csv(summaries: t.Iterable[tuple[Contact, ContributionSummary]]): + """Emit a CSV overview of the search results.""" + fieldnames = [ + "last_name", + "first_name", + "city", + "state", + "zip", + "total_usd", + "dem_usd", + "rep_usd", + "other_usd", + "donated_to", + ] + writer = csv.DictWriter(click.get_text_stream("stdout"), fieldnames=fieldnames) + writer.writeheader() + for contact, summary in summaries: + writer.writerow( + { + "first_name": contact.first_name.title(), + "last_name": contact.last_name.title(), + "city": (contact.city or "").title(), + "state": contact.state, + "zip": contact.zip_code, + "total_usd": summary.total_cents / 100, + "dem_usd": summary.party_total_cents("DEM") / 100, + "rep_usd": summary.party_total_cents("REP") / 100, + "other_usd": summary.party_total_cents("OTH") / 100, + "donated_to": "/".join( + sorted(c.name for c in summary.committees()) + ), + } + ) + + def _emit_contributions_csv( + summaries: t.Iterable[tuple[Contact, ContributionSummary]] + ): + """Emit a detailed CSV with line-by-line transactions.""" + fieldnames = [ + "last_name", + "first_name", + "city", + "state", + "zip", + "fec_contribution_id", + "fec_committee_id", + "committee", + "party", + "amount_usd", + ] + writer = csv.DictWriter(click.get_text_stream("stdout"), fieldnames=fieldnames) + writer.writeheader() + for contact, summary in summaries: + for contribution in summary.contributions: + writer.writerow( + { + "first_name": contact.first_name.title(), + "last_name": contact.last_name.title(), + "city": (contact.city or "").title(), + "state": contact.state, + "zip": contact.zip_code, + "fec_contribution_id": contribution.id, + "fec_committee_id": contribution.committee_id, + "committee": contribution.committee.name, + "party": contribution.committee.party, + "amount_usd": contribution.amount_cents / 100, + } + ) + + def _emit_human(summaries: t.Iterable[tuple[Contact, ContributionSummary]]): + for contact, summary in summaries: + print( + f"{contact.first_name.title()} {contact.last_name.title()} ({(contact.city or '').title()}, {contact.state})" + ) + print(str(summary)) + + summaries = searcher.search_and_summarize_contacts(contact_provider) + sorted_summaries = sorted( + summaries, + key=lambda contact_summary: contact_summary[0].last_name.upper(), + ) + if emit == "human": + _emit_human(sorted_summaries) + elif emit == "csv-overview": + _emit_overview_csv(sorted_summaries) + elif emit == "csv-contributions": + _emit_contributions_csv(sorted_summaries) + else: + raise click.UsageError(f"Unknown emit format: {emit}") if __name__ == "__main__": diff --git a/requirements.txt b/requirements.txt index 71794b7..4ff0f4c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,4 @@ phonenumbers~=8.13.26 pydantic-settings~=2.1.0 pydantic~=2.5.1 sqlalchemy~=2.0.23 +tqdm~=4.66.1 diff --git a/server/data/summaries.py b/server/data/summaries.py index 26611d0..0fac982 100644 --- a/server/data/summaries.py +++ b/server/data/summaries.py @@ -24,6 +24,11 @@ def __init__(self, contributions: t.Iterable[Contribution]): if contribution.amount_cents > 0 ] + @property + def contributions(self) -> t.Iterable[Contribution]: + """The contributions that make up the summary.""" + return self._contributions + @property def total_cents(self) -> int: """The total amount of all contributions, in cents.""" @@ -74,6 +79,17 @@ def party_total_cents(self, party: str) -> int: if contribution.committee.adjusted_party == party ) + def party_total_cents_anything_but(self, parties: set[str]) -> int: + """ + Return the total amount of contributions for everything + but the named parties. + """ + return sum( + contribution.amount_cents + for contribution in self._contributions + if contribution.committee.adjusted_party not in parties + ) + def party_total_fmt(self, party: str) -> str: """Return the total amount of contributions for a party, formatted.""" return fmt_usd(self.party_total_cents(party))