Skip to content

Commit

Permalink
Slightly improve the API factoring. It's still weird, but good enough…
Browse files Browse the repository at this point in the history
… for now
  • Loading branch information
davepeck committed Nov 30, 2023
1 parent ddf1eb3 commit 98dcff1
Show file tree
Hide file tree
Showing 2 changed files with 66 additions and 31 deletions.
40 changes: 9 additions & 31 deletions fec.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,7 @@
get_engine,
)
from server.data.nicknames import NicknamesManager
from server.data.npa import AreaCodeManager
from server.data.summaries import AlternativeContactsHelper, ContributionSummaryManager
from server.data.usps import ZipCodeManager
from server.data.search import ContactContributionSearcher


@click.group()
Expand Down Expand Up @@ -233,10 +231,7 @@ def search(
):
"""Search summarized FEC contributions data."""
data_manager = DataManager(data) if data is not None else DataManager.default()
nicknames_manager = NicknamesManager.from_data_manager(data_manager)
area_code_manager = AreaCodeManager.from_data_manager(data_manager)
zip_code_manager = ZipCodeManager.from_data_manager(data_manager)
alternatives_helper = AlternativeContactsHelper(zip_code_manager, area_code_manager)
searcher = ContactContributionSearcher(data_manager)

contact_provider: IContactProvider | None = None

Expand All @@ -260,30 +255,13 @@ def search(
"You must provide a contact dir, zip file, or explicit name & zip."
)

state_to_manager = {}
seen_contacts = set()

for contact in contact_provider.get_contacts():
for alternative in alternatives_helper.get_alternatives(contact):
if alternative.duplicate_key in seen_contacts:
continue
seen_contacts.add(alternative.duplicate_key)

state = alternative.state
assert state

manager = state_to_manager.get(state)
if manager is None:
manager = ContributionSummaryManager(
get_engine(data_manager, state),
nicknames_manager,
)
state_to_manager[state] = manager
summary = manager.preferred_summary_for_contact(alternative)
if summary is None:
continue
print(alternative.first_name.title(), alternative.last_name.title())
print(str(summary))
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))


if __name__ == "__main__":
Expand Down
57 changes: 57 additions & 0 deletions server/data/search.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
import typing as t

from server.data.contacts import Contact, IContactProvider
from server.data.manager import DataManager
from server.data.models import get_engine
from server.data.nicknames import NicknamesManager
from server.data.npa import AreaCodeManager
from server.data.summaries import (
AlternativeContactsHelper,
ContributionSummary,
ContributionSummaryManager,
)
from server.data.usps import ZipCodeManager


class ContactContributionSearcher:
def __init__(self, data_manager: DataManager):
self._data_manager = data_manager
self._nicknames_manager = NicknamesManager.from_data_manager(data_manager)
self._area_code_manager = AreaCodeManager.from_data_manager(data_manager)
self._zip_code_manager = ZipCodeManager.from_data_manager(data_manager)
self._alternatives_helper = AlternativeContactsHelper(
self._zip_code_manager, self._area_code_manager
)
self._seen = set()
self._state_to_manager = {}

def search_and_summarize(
self, contact: Contact
) -> tuple[Contact, ContributionSummary] | None:
"""Search for a contact and summarize their contributions."""
for alternative in self._alternatives_helper.get_alternatives(contact):
if alternative.duplicate_key in self._seen:
continue
state = alternative.state
assert state

manager = self._state_to_manager.get(state)
if manager is None:
manager = ContributionSummaryManager(
get_engine(self._data_manager, state), self._nicknames_manager
)
self._state_to_manager[state] = manager

summary = manager.preferred_summary_for_contact(alternative)
if summary is not None:
return (alternative, summary)
return None

def search_and_summarize_contacts(
self, contacts: IContactProvider
) -> t.Iterable[tuple[Contact, ContributionSummary]]:
"""Search for contacts and summarize their contributions."""
for contact in contacts.get_contacts():
result = self.search_and_summarize(contact)
if result is not None:
yield result

0 comments on commit 98dcff1

Please sign in to comment.