Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
| import os | |
| import sys | |
| import json | |
| import time | |
| import logging | |
| import base64 | |
| from io import BytesIO | |
| from datetime import datetime, timezone | |
| from dotenv import load_dotenv | |
| from bs4 import BeautifulSoup | |
| import PyPDF2 | |
| # Load environment variables from .env file | |
| load_dotenv() | |
| # Pipeline status tracking (no-op when running standalone) | |
| _PIPELINE_SCRIPT = os.environ.get("PIPELINE_CURRENT_SCRIPT") | |
| _pipeline = None | |
| _last_status_write = 0.0 | |
| if _PIPELINE_SCRIPT: | |
| try: | |
| from pipeline_status import PipelineStatus | |
| _pipeline = PipelineStatus() | |
| except Exception: | |
| pass | |
| def _update_pipeline_progress(current, total, unit="bills", message=""): | |
| global _last_status_write | |
| if not _pipeline: | |
| return | |
| now = time.time() | |
| if now - _last_status_write < 3.0: | |
| return | |
| _last_status_write = now | |
| try: | |
| _pipeline.update_progress(_PIPELINE_SCRIPT, current, total, unit, message) | |
| except Exception: | |
| pass | |
| # Import shared LegiScan API helper | |
| from legiscan_api import legi_request, API_KEY, RATE_LIMIT as _DEFAULT_RATE_LIMIT, get_api_call_count | |
| if not API_KEY: | |
| print("Error: Please set LEGISCAN_API_KEY in your .env file.") | |
| sys.exit(1) | |
| # Modes for testing (can also be enabled via PIPELINE_TEST_MODE env var) | |
| # Quick test: pulls only TEST_MAX_BILLS bills | |
| TESTING_MODE = os.getenv("PIPELINE_TEST_MODE", "").lower() in ("1", "true", "yes") | |
| # Full test: pulls all bills for TEST_STATE and TEST_YEAR without bill count cap | |
| FULL_TESTING_MODE = False | |
| TEST_STATE = 'CA' | |
| TEST_YEAR = 2026 | |
| TEST_MAX_BILLS = int(os.getenv("PIPELINE_TEST_MAX_BILLS", "3")) | |
| # Output files | |
| CACHE_FILE = "data/bill_cache.json" # Stores bill_id -> change_hash | |
| OUTPUT_FILE = "data/known_bills.json" # Final bills data | |
| # Query settings | |
| QUERY = "artificial intelligence" | |
| START_YEAR = 2023 | |
| END_YEAR = datetime.now(timezone.utc).year | |
| # Include all state legislatures plus U.S. Congress (both chambers) | |
| STATES = [ | |
| "AL","AK","AZ","AR","CA","CO","CT","DE","FL","GA", | |
| "HI","ID","IL","IN","IA","KS","KY","LA","ME","MD", | |
| "MA","MI","MN","MS","MO","MT","NE","NV","NH","NJ", | |
| "NM","NY","NC","ND","OH","OK","OR","PA","RI","SC", | |
| "SD","TN","TX","UT","VT","VA","WA","WV","WI","WY", | |
| "US" # U.S. Congress | |
| ] | |
| # Rate limiting (seconds between requests) | |
| RATE_LIMIT = _DEFAULT_RATE_LIMIT | |
| # Create logs directory if it doesn't exist | |
| os.makedirs("data_updating_scripts/logs", exist_ok=True) | |
| # Logging configuration | |
| LOG_FILE = "data_updating_scripts/logs/fetch_ai_bills.log" | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s [%(levelname)s] %(message)s", | |
| handlers=[ | |
| logging.StreamHandler(sys.stdout), | |
| logging.FileHandler(LOG_FILE) | |
| ] | |
| ) | |
| logger = logging.getLogger(__name__) | |
| # Apply testing overrides | |
| if TESTING_MODE: | |
| logger.info(f"*** TESTING MODE: fetching only {TEST_MAX_BILLS} bills from {TEST_STATE} ({TEST_YEAR}) ***") | |
| STATES = [TEST_STATE] | |
| if FULL_TESTING_MODE: | |
| logger.info(f"*** FULL TESTING MODE: fetching all bills from {TEST_STATE} ({TEST_YEAR}) ***") | |
| STATES = [TEST_STATE] | |
| def load_json(path, default): | |
| try: | |
| with open(path, 'r') as f: | |
| return json.load(f) | |
| except (FileNotFoundError, json.JSONDecodeError): | |
| return default | |
| def save_json(path, data): | |
| # Create directory if it doesn't exist | |
| os.makedirs(os.path.dirname(path), exist_ok=True) | |
| with open(path, 'w') as f: | |
| json.dump(data, f, indent=2) | |
| logger.info(f"Saved JSON to {path}") | |
| def extract_plain_text(html_content: str) -> str: | |
| soup = BeautifulSoup(html_content, "html.parser") | |
| return soup.get_text(separator="\n", strip=True) | |
| def main(): | |
| cache = load_json(CACHE_FILE, {}) | |
| existing = load_json(OUTPUT_FILE, []) | |
| existing_map = {b.get("bill_id"): b for b in existing} | |
| logger.info(f"Loaded cache entries: {len(cache)}, existing bills: {len(existing)}") | |
| collected = [] | |
| total_fetched = 0 | |
| skipped_unchanged = 0 | |
| new_or_updated = 0 | |
| years = [TEST_YEAR] if (TESTING_MODE or FULL_TESTING_MODE) else list(range(START_YEAR, END_YEAR + 1)) | |
| for state in STATES: | |
| for year in years: | |
| page = 1 | |
| while True: | |
| if TESTING_MODE and total_fetched >= TEST_MAX_BILLS: | |
| logger.info("Reached TEST_MAX_BILLS limit, stopping early.") | |
| break | |
| params = {"state": state, "year": year, "query": QUERY, "page": page} | |
| logger.info(f"Searching {state} for {year}, page {page}") | |
| data = legi_request("getSearch", params) | |
| if not data: | |
| logger.warning(f"Search failed for {state} {year} page {page} after retries — skipping remaining pages") | |
| break | |
| results = data.get("searchresult", {}) | |
| summary = results.get("summary", {}) | |
| bills = [v for k, v in results.items() if k != "summary"] | |
| if not bills: | |
| logger.info(f"No bills on page {page} for {state} {year}") | |
| break | |
| logger.info(f"Found {len(bills)} bills on {state} {year} page {page}") | |
| for bill in bills: | |
| if TESTING_MODE and total_fetched >= TEST_MAX_BILLS: | |
| break | |
| bill_id = str(bill.get("bill_id")) | |
| state_code = bill.get("state") | |
| bill_num = bill.get("bill_number") | |
| # Check change_hash from search result BEFORE calling getBill | |
| # This avoids thousands of unnecessary API calls for unchanged bills | |
| search_hash = bill.get("change_hash") | |
| old_hash = cache.get(bill_id) | |
| if search_hash and search_hash == old_hash and bill_id in existing_map: | |
| existing_entry = existing_map[bill_id] | |
| # Re-fetch if actions are empty (one-time backfill for history field fix) | |
| if existing_entry.get("actions") in ([], None): | |
| logger.info(f"Re-fetching {state_code}_{bill_num} (empty actions, need history backfill)") | |
| else: | |
| # Bill unchanged since last run - reuse existing entry entirely | |
| entry = existing_entry | |
| entry["lastUpdatedAt"] = datetime.now(timezone.utc).isoformat() | |
| collected.append(entry) | |
| total_fetched += 1 | |
| skipped_unchanged += 1 | |
| logger.info(f"Unchanged {state_code}_{bill_num} (ID: {bill_id}) - skipped API call") | |
| continue | |
| # Bill is new or changed - need full details from API | |
| logger.info(f"Fetching details for {state_code}_{bill_num} (ID: {bill_id})") | |
| details_resp = legi_request("getBill", {"id": bill_id}) | |
| if not details_resp: | |
| continue | |
| details = details_resp.get("bill", {}) | |
| sess_year = details.get("session", {}).get("year_start", 0) | |
| if sess_year < START_YEAR: | |
| continue | |
| new_hash = details.get("change_hash") | |
| now_iso = datetime.now(timezone.utc).isoformat() | |
| # Extract all relevant dates | |
| explicit = details.get("last_action_date") | |
| status_date = details.get("status_date") | |
| last_vote_date = details.get("last_vote_date") | |
| last_amendment_date = details.get("last_amendment_date") | |
| actions = details.get("history", details.get("actions", [])) | |
| action_dates = [a.get("date") or a.get("action_date") for a in actions if a.get("date") or a.get("action_date")] | |
| most_recent_action = max(action_dates) if action_dates else None | |
| candidates = [d for d in [explicit, status_date, last_vote_date, last_amendment_date, most_recent_action] if d] | |
| last_action_date = max(candidates) if candidates else None | |
| bill_url = details.get("url") # Bill detail page URL | |
| # Fetch bill text for new or changed bills | |
| plain_text = None | |
| texts = details.get("texts", []) | |
| if texts: | |
| doc_id = texts[0].get("doc_id") | |
| text_resp = legi_request("getBillText", {"id": doc_id}) | |
| if text_resp and "text" in text_resp: | |
| raw_b64 = text_resp["text"].get("doc", "") | |
| try: | |
| decoded = base64.b64decode(raw_b64) | |
| if decoded[:4] == b'%PDF': | |
| # PDF document - extract with PyPDF2 | |
| reader = PyPDF2.PdfReader(BytesIO(decoded)) | |
| pages = [p.extract_text() for p in reader.pages if p.extract_text()] | |
| plain_text = "\n".join(pages).strip() if pages else None | |
| if plain_text: | |
| logger.info(f"Extracted {len(plain_text)} chars from PDF for {bill_id}") | |
| else: | |
| logger.warning(f"PDF extraction returned no text for {bill_id}") | |
| else: | |
| # HTML document | |
| html = decoded.decode("utf-8", errors="ignore") | |
| plain_text = extract_plain_text(html) | |
| except Exception as e: | |
| logger.error(f"Failed decoding text for {bill_id}: {e}") | |
| # If bill existed before, preserve its text if we failed to fetch new text | |
| if plain_text is None and bill_id in existing_map: | |
| plain_text = existing_map[bill_id].get("text") | |
| entry = { | |
| "bill_id": bill_id, | |
| "state": state_code, | |
| "bill_number": bill_num, | |
| "session_year": f"{details.get('session', {}).get('year_start', '')}-{details.get('session', {}).get('year_end', '')}", | |
| "title": details.get("title"), | |
| "description": details.get("description"), | |
| "status": details.get("status"), | |
| "sponsors": [s.get("name") for s in details.get("sponsors", [])], | |
| "text": plain_text, | |
| "last_action_date": last_action_date, | |
| "status_date": status_date, | |
| "last_vote_date": last_vote_date, | |
| "last_amendment_date": last_amendment_date, | |
| "actions": actions, | |
| "bill_url": bill_url, | |
| "change_hash": new_hash, | |
| "lastUpdatedAt": now_iso | |
| } | |
| cache[bill_id] = new_hash | |
| new_or_updated += 1 | |
| logger.info( | |
| f"{'Updated' if bill_id in existing_map else 'New'} bill: title='{entry['title']}', " | |
| f"sponsors={len(entry['sponsors'])}, status={entry['status']}, last_action_date={entry['last_action_date']}" | |
| ) | |
| collected.append(entry) | |
| total_fetched += 1 | |
| _update_pipeline_progress(total_fetched, total_fetched, "bills", | |
| f"Fetched {total_fetched} bills ({new_or_updated} new, {skipped_unchanged} unchanged)") | |
| time.sleep(RATE_LIMIT) | |
| if page >= summary.get("page_total", 1): | |
| break | |
| page += 1 | |
| time.sleep(RATE_LIMIT) | |
| if TESTING_MODE and total_fetched >= TEST_MAX_BILLS: | |
| break | |
| if TESTING_MODE and total_fetched >= TEST_MAX_BILLS: | |
| break | |
| dedup = {e["bill_id"]: e for e in collected} | |
| # Preserve existing bills not found in this run's search results. | |
| # Bills can be missed if search pages time out or LegiScan changes indexing. | |
| preserved = 0 | |
| for bill_id, bill in existing_map.items(): | |
| if bill_id not in dedup: | |
| dedup[bill_id] = bill | |
| preserved += 1 | |
| if preserved: | |
| logger.info(f"Preserved {preserved} existing bills not in current search results") | |
| all_bills = list(dedup.values()) | |
| save_json(OUTPUT_FILE, all_bills) | |
| save_json(CACHE_FILE, cache) | |
| _update_pipeline_progress(len(all_bills), len(all_bills), "bills", | |
| f"Done: {len(all_bills)} bills ({new_or_updated} new, {skipped_unchanged} unchanged)") | |
| api_calls = get_api_call_count() | |
| logger.info(f"Completed run, saved {len(all_bills)} bills to {OUTPUT_FILE}") | |
| logger.info(f" Unchanged (skipped API calls): {skipped_unchanged}") | |
| logger.info(f" New or updated (fetched from API): {new_or_updated}") | |
| logger.info(f" Total LegiScan API calls this run: {api_calls}") | |
| if skipped_unchanged > 0: | |
| logger.info(f" Saved ~{skipped_unchanged * 2} API calls by skipping unchanged bills") | |
| if __name__ == "__main__": | |
| main() | |