""" Google Civic Information API Integration The Google Civic Information API is the gold standard for: - Address-to-representative mapping - Elected officials contact information - Election data and polling locations - Voter information API Docs: https://developers.google.com/civic-information Free Tier: 25,000 requests/day Cost: Free for non-commercial use SETUP: 1. Get API key: https://console.cloud.google.com/ 2. Enable "Google Civic Information API" 3. Add to .env: GOOGLE_CIVIC_API_KEY=your-key USAGE: from discovery.google_civic_integration import GoogleCivicAPI api = GoogleCivicAPI() # Get representatives for an address reps = await api.get_representatives("1600 Pennsylvania Ave NW, Washington DC") # Get upcoming elections elections = await api.get_elections() # Get voter info voter_info = await api.get_voter_info("123 Main St, Tuscaloosa, AL") """ import asyncio from typing import Dict, List, Optional from datetime import datetime from pathlib import Path import httpx from loguru import logger try: from pyspark.sql import SparkSession from config.settings import settings SPARK_AVAILABLE = True except ImportError: SPARK_AVAILABLE = False settings = None logger.warning("Running without Spark/settings - limited functionality") class GoogleCivicAPI: """ Integration with Google Civic Information API. Best for: - "Who represents this address?" queries - Finding all elected officials for a location - Election information - Polling locations """ BASE_URL = "https://www.googleapis.com/civicinfo/v2" def __init__(self, api_key: Optional[str] = None): """ Initialize Google Civic API client. Args: api_key: Google Civic Information API key If not provided, will try to get from settings.google_civic_api_key """ if api_key: self.api_key = api_key elif SPARK_AVAILABLE and hasattr(settings, 'google_civic_api_key'): self.api_key = settings.google_civic_api_key else: self.api_key = None logger.warning("⚠️ GOOGLE_CIVIC_API_KEY not found") logger.warning(" Get one at: https://console.cloud.google.com/") logger.warning(" Add to .env: GOOGLE_CIVIC_API_KEY=your-key") self.cache_dir = Path("data/cache/google_civic") self.cache_dir.mkdir(parents=True, exist_ok=True) async def get_representatives( self, address: str, levels: Optional[List[str]] = None, roles: Optional[List[str]] = None ) -> Dict: """ Get elected officials for a given address. Args: address: Street address (e.g., "1600 Pennsylvania Ave NW, Washington DC") levels: Filter by government level: ['country', 'administrativeArea1' (state), 'administrativeArea2' (county), 'locality' (city), 'subLocality1' (neighborhood)] roles: Filter by role: ['legislatorUpperBody', 'legislatorLowerBody', 'deputyHeadOfGovernment', 'headOfGovernment', 'executiveCouncil', etc.] Returns: Dict with 'offices' and 'officials' keys """ if not self.api_key: raise ValueError("Google Civic API key required. Set GOOGLE_CIVIC_API_KEY in .env") params = { "address": address, "key": self.api_key } if levels: params["levels"] = levels if roles: params["roles"] = roles logger.info(f"Fetching representatives for: {address}") async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.get( f"{self.BASE_URL}/representatives", params=params ) response.raise_for_status() data = response.json() # Extract and format data officials_data = { "address": address, "normalized_address": data.get("normalizedInput", {}).get("line1", address), "officials": [], "source": "google_civic_api", "fetched_at": datetime.utcnow().isoformat() } # Parse offices and officials offices = data.get("offices", []) officials = data.get("officials", []) for office in offices: office_name = office.get("name") office_level = office.get("levels", ["unknown"])[0] office_roles = office.get("roles", []) # Get official indices for this office official_indices = office.get("officialIndices", []) for idx in official_indices: if idx < len(officials): official = officials[idx] officials_data["officials"].append({ "name": official.get("name"), "office": office_name, "level": office_level, "roles": office_roles, "party": official.get("party"), "phones": official.get("phones", []), "urls": official.get("urls", []), "emails": official.get("emails", []), "photo_url": official.get("photoUrl"), "address": official.get("address", [{}])[0] if official.get("address") else {}, "channels": official.get("channels", []) # Social media }) logger.info(f"✅ Found {len(officials_data['officials'])} officials for {address}") return officials_data except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}") raise except Exception as e: logger.error(f"Error fetching representatives: {e}") raise async def get_elections(self) -> Dict: """ Get information about upcoming elections. Returns: Dict with 'elections' list """ if not self.api_key: raise ValueError("Google Civic API key required") logger.info("Fetching upcoming elections") async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.get( f"{self.BASE_URL}/elections", params={"key": self.api_key} ) response.raise_for_status() data = response.json() elections_data = { "elections": data.get("elections", []), "source": "google_civic_api", "fetched_at": datetime.utcnow().isoformat() } logger.info(f"✅ Found {len(elections_data['elections'])} upcoming elections") return elections_data except Exception as e: logger.error(f"Error fetching elections: {e}") raise async def get_voter_info( self, address: str, election_id: Optional[str] = None ) -> Dict: """ Get voter information for an address. Args: address: Voter's address election_id: Specific election ID (default: next election) Returns: Dict with polling location, ballot info, etc. """ if not self.api_key: raise ValueError("Google Civic API key required") params = { "address": address, "key": self.api_key } if election_id: params["electionId"] = election_id logger.info(f"Fetching voter info for: {address}") async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.get( f"{self.BASE_URL}/voterinfo", params=params ) response.raise_for_status() data = response.json() voter_info = { "address": address, "normalized_address": data.get("normalizedInput", {}).get("line1", address), "election": data.get("election"), "polling_locations": data.get("pollingLocations", []), "early_vote_sites": data.get("earlyVoteSites", []), "contests": data.get("contests", []), "state": data.get("state", []), "source": "google_civic_api", "fetched_at": datetime.utcnow().isoformat() } logger.info(f"✅ Found voter info for {address}") return voter_info except Exception as e: logger.error(f"Error fetching voter info: {e}") raise async def get_representatives_by_division(self, ocd_id: str) -> Dict: """ Get representatives for an Open Civic Data division ID. Args: ocd_id: OCD ID (e.g., "ocd-division/country:us/state:al/county:tuscaloosa") Returns: Dict with officials for that division """ if not self.api_key: raise ValueError("Google Civic API key required") logger.info(f"Fetching representatives for OCD ID: {ocd_id}") async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.get( f"{self.BASE_URL}/representatives/{ocd_id}", params={"key": self.api_key} ) response.raise_for_status() return response.json() except Exception as e: logger.error(f"Error fetching division representatives: {e}") raise def save_to_json(self, data: Dict, filename: str): """Save data to JSON cache.""" import json filepath = self.cache_dir / filename with open(filepath, 'w') as f: json.dump(data, f, indent=2) logger.info(f"💾 Saved to {filepath}") # ============================================================================ # Example Usage # ============================================================================ async def example_usage(): """Example usage of Google Civic API.""" # Initialize (will get key from settings or environment) api = GoogleCivicAPI() if not api.api_key: logger.error("❌ API key not found. Please set GOOGLE_CIVIC_API_KEY in .env") return # Example 1: Get representatives for Tuscaloosa City Hall logger.info("\n" + "="*80) logger.info("Example 1: Get representatives for Tuscaloosa, AL") logger.info("="*80) try: reps = await api.get_representatives("2201 University Blvd, Tuscaloosa, AL 35401") print(f"\n✅ Found {len(reps['officials'])} officials:") for official in reps['officials'][:10]: # Show first 10 print(f"\n • {official['name']}") print(f" Office: {official['office']}") print(f" Level: {official['level']}") print(f" Party: {official.get('party', 'N/A')}") if official.get('phones'): print(f" Phone: {official['phones'][0]}") if official.get('urls'): print(f" Website: {official['urls'][0]}") # Save to cache api.save_to_json(reps, "tuscaloosa_representatives.json") except Exception as e: logger.error(f"Error: {e}") # Example 2: Get upcoming elections logger.info("\n" + "="*80) logger.info("Example 2: Get upcoming elections") logger.info("="*80) try: elections = await api.get_elections() print(f"\n✅ Found {len(elections['elections'])} upcoming elections:") for election in elections['elections']: print(f"\n • {election['name']}") print(f" Date: {election['electionDay']}") print(f" ID: {election['id']}") api.save_to_json(elections, "upcoming_elections.json") except Exception as e: logger.error(f"Error: {e}") # Example 3: Get voter info logger.info("\n" + "="*80) logger.info("Example 3: Get voter information") logger.info("="*80) try: voter_info = await api.get_voter_info("2201 University Blvd, Tuscaloosa, AL 35401") print(f"\n✅ Voter Information:") print(f" Election: {voter_info['election']['name'] if voter_info.get('election') else 'None upcoming'}") if voter_info.get('polling_locations'): print(f" Polling Locations: {len(voter_info['polling_locations'])}") if voter_info.get('contests'): print(f" Ballot Contests: {len(voter_info['contests'])}") api.save_to_json(voter_info, "tuscaloosa_voter_info.json") except Exception as e: logger.error(f"Error: {e}") logger.info("\n✅ Examples complete!") if __name__ == "__main__": # Run examples asyncio.run(example_usage())