Report an Oregon legislator's voting record from the OLIS OData API

getVotingRecord.py

"""Report an Oregon legislator's voting record from the OLIS OData API.

Covers chamber floor votes (MeasureVotes) and committee votes (CommitteeVotes),
annotated with each measure's CatchLine.

Usage:
    python getVotingRecord.py --list
    python getVotingRecord.py --district 19
    python getVotingRecord.py --name Andersen
    python getVotingRecord.py --district 26 --legislator "Rep Neron"
    python getVotingRecord.py --district 19 --sessions 2025R1

A seat that changed hands mid-session needs the predecessor added by hand with
--legislator, because the Legislators roster only lists the current holder.
"""

import argparse
import sys
from collections import Counter

import requests

BASE_URL = "https://api.oregonlegislature.gov/odata/odataservice.svc/"
DEFAULT_SESSIONS = ["2025R1", "2026R1"]
PAGE_SIZE = 1000

CHAMBERS = {"H": "House", "S": "Senate"}

# Value that both MeasureVotes.Vote and CommitteeVotes.Meaning use for a no vote.
NAY = "Nay"


def odata_get(entity, params):
    """GET an entity set, paging through the full result."""
    headers = {"Accept": "application/json"}
    rows = []
    skip = 0

    while True:
        page_params = dict(params, **{"$top": PAGE_SIZE, "$skip": skip,
                                      "$format": "json"})
        response = requests.get(
            f"{BASE_URL}{entity}", headers=headers, params=page_params, timeout=180
        )
        response.raise_for_status()
        page = response.json().get("value", [])
        rows.extend(page)

        if len(page) < PAGE_SIZE:
            return rows
        skip += PAGE_SIZE


def quote(value):
    """Escape a string for use as an OData literal."""
    return "'" + str(value).replace("'", "''") + "'"


def fetch_roster(session_key):
    return odata_get("Legislators", {"$filter": f"SessionKey eq {quote(session_key)}"})


def resolve_legislators(roster, district, name, chamber):
    """Pick roster entries matching the district or name filter."""
    matches = roster

    if chamber:
        matches = [m for m in matches if m.get("Chamber") == chamber]
    if district is not None:
        matches = [m for m in matches
                   if str(m.get("DistrictNumber", "")).strip() == str(district)]
    if name:
        needle = name.lower()
        matches = [m for m in matches
                   if needle in (m.get("LastName") or "").lower()
                   or needle in (m.get("FirstName") or "").lower()
                   or needle in (m.get("LegislatorCode") or "").lower()]

    return matches


def fetch_measure_catchlines(session_key):
    """Map (prefix, number) -> CatchLine for every measure in the session."""
    rows = odata_get("Measures", {
        "$filter": f"SessionKey eq {quote(session_key)}",
        "$select": "MeasurePrefix,MeasureNumber,CatchLine",
    })
    return {(r["MeasurePrefix"], r["MeasureNumber"]): r.get("CatchLine")
            for r in rows}


def fetch_floor_votes(session_key, codes):
    votes = []
    for code in codes:
        votes.extend(odata_get("MeasureVotes", {
            "$filter": (f"SessionKey eq {quote(session_key)} "
                        f"and VoteName eq {quote(code)}"),
            "$orderby": "MeasurePrefix,MeasureNumber,ActionDate",
        }))
    return votes


def fetch_committee_votes(session_key, codes):
    votes = []
    for code in codes:
        votes.extend(odata_get("CommitteeVotes", {
            "$filter": (f"SessionKey eq {quote(session_key)} "
                        f"and VoteName eq {quote(code)}"),
            "$orderby": "MeasurePrefix,MeasureNumber,MeetingDate",
        }))
    return votes


def find_unrostered_voters(session_key, roster):
    """VoteNames that cast floor votes but are absent from the roster.

    These are mid-session departures; the roster keeps only the current holder
    of each seat, so their votes are otherwise invisible to a district lookup.
    """
    rows = odata_get("MeasureVotes", {
        "$filter": f"SessionKey eq {quote(session_key)}",
        "$select": "VoteName",
    })
    codes = {m["LegislatorCode"] for m in roster}
    return sorted({r["VoteName"] for r in rows} - codes)


def clean(value):
    if value is None:
        return ""
    return " ".join(str(value).split())


def day(timestamp):
    return clean(timestamp)[:10]


def measure_label(row):
    return f"{clean(row.get('MeasurePrefix'))} {clean(row.get('MeasureNumber'))}".strip()


def write_report(path, session_key, members, floor, committee, catchlines, warnings,
                 nay_only=False):
    """Write the report. Tallies always cover every vote; when nay_only is set
    the itemized listings are narrowed to Nay votes."""
    with open(path, "w", encoding="utf-8") as out:
        out.write(f"Voting Record - Session {session_key}\n")
        out.write("=" * 78 + "\n\n")

        for member in members:
            out.write(f"{clean(member.get('Title'))} {clean(member.get('FirstName'))} "
                      f"{clean(member.get('LastName'))}\n")
            out.write(f"  LegislatorCode: {clean(member.get('LegislatorCode'))}\n")
            out.write(f"  Chamber:        "
                      f"{CHAMBERS.get(member.get('Chamber'), clean(member.get('Chamber')))}\n")
            out.write(f"  District:       {clean(member.get('DistrictNumber'))}\n")
            out.write(f"  Party:          {clean(member.get('Party'))}\n\n")

        for note in warnings:
            out.write(f"NOTE: {note}\n")
        if warnings:
            out.write("\n")

        out.write(f"Floor votes: {len(floor)}\n")
        for value, count in Counter(v.get("Vote") for v in floor).most_common():
            out.write(f"  {clean(value)}: {count}\n")

        out.write(f"Committee votes: {len(committee)}\n")
        for value, count in Counter(v.get("Meaning") for v in committee).most_common():
            out.write(f"  {clean(value)}: {count}\n")

        listed_floor = [v for v in floor if v.get("Vote") == NAY] if nay_only else floor
        listed_committee = ([v for v in committee if v.get("Meaning") == NAY]
                            if nay_only else committee)

        if nay_only:
            out.write(f"\nListing Nay votes only "
                      f"({len(listed_floor)} floor, {len(listed_committee)} committee). "
                      f"Tallies above cover all votes.\n")

        out.write("\n" + "=" * 78 + "\n")

        out.write(f"\nFLOOR VOTES{' (Nay only)' if nay_only else ''}\n")
        out.write("-" * 78 + "\n\n")
        for vote in listed_floor:
            key = (vote.get("MeasurePrefix"), vote.get("MeasureNumber"))
            out.write(f"{measure_label(vote)}  [{clean(vote.get('Vote'))}]\n")
            out.write(f"  Date:      {day(vote.get('ActionDate'))}\n")
            out.write(f"  Chamber:   "
                      f"{CHAMBERS.get(vote.get('Chamber'), clean(vote.get('Chamber')))}\n")
            out.write(f"  Voter:     {clean(vote.get('VoteName'))}\n")
            out.write(f"  CatchLine: {clean(catchlines.get(key))}\n")
            out.write(f"  Action:    {clean(vote.get('ActionText'))}\n\n")

        out.write(f"\nCOMMITTEE VOTES{' (Nay only)' if nay_only else ''}\n")
        out.write("-" * 78 + "\n\n")
        for vote in listed_committee:
            key = (vote.get("MeasurePrefix"), vote.get("MeasureNumber"))
            out.write(f"{measure_label(vote)}  [{clean(vote.get('Meaning'))}]\n")
            out.write(f"  Date:      {day(vote.get('MeetingDate'))}\n")
            out.write(f"  Committee: {clean(vote.get('CommitteeCode'))}\n")
            out.write(f"  Voter:     {clean(vote.get('VoteName'))}\n")
            out.write(f"  CatchLine: {clean(catchlines.get(key))}\n\n")


def parse_args(argv):
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--district", help="district number, e.g. 19")
    parser.add_argument("--name", help="legislator first or last name, e.g. Andersen")
    parser.add_argument("--legislator", action="append", default=[], metavar="CODE",
                        help="extra LegislatorCode to include, e.g. 'Rep Neron'; "
                             "repeatable")
    parser.add_argument("--chamber", choices=["H", "S"], default="H",
                        help="H for House (default), S for Senate")
    parser.add_argument("--sessions", nargs="+", default=DEFAULT_SESSIONS,
                        metavar="KEY", help="session keys (default: 2025R1 2026R1)")
    parser.add_argument("--nay-only", action="store_true",
                        help="list only Nay votes; tallies still cover every vote")
    parser.add_argument("--list", action="store_true",
                        help="list the chamber roster and exit")

    args = parser.parse_args(argv)
    if not (args.district or args.name or args.legislator or args.list):
        parser.error("give --district, --name, --legislator, or --list")
    return args


def list_roster(session_key, roster, chamber):
    members = [m for m in roster if m.get("Chamber") == chamber]
    members.sort(key=lambda m: int(m.get("DistrictNumber") or 0))
    print(f"\n{session_key} - {CHAMBERS[chamber]} ({len(members)})")
    for member in members:
        print(f"  District {str(member.get('DistrictNumber')):>3}  "
              f"{clean(member.get('LegislatorCode')):<20} "
              f"{clean(member.get('FirstName'))} {clean(member.get('LastName'))} "
              f"({clean(member.get('Party'))})")


def run(args):
    for session_key in args.sessions:
        roster = fetch_roster(session_key)

        if args.list:
            list_roster(session_key, roster, args.chamber)
            continue

        members = resolve_legislators(roster, args.district, args.name, args.chamber)
        by_code = {m["LegislatorCode"]: m for m in members}

        for code in args.legislator:
            if code in by_code:
                continue
            match = next((m for m in roster if m["LegislatorCode"] == code), None)
            by_code[code] = match or {"LegislatorCode": code,
                                      "Title": "(not on session roster)",
                                      "FirstName": "", "LastName": "",
                                      "Chamber": "", "DistrictNumber": "", "Party": ""}

        if not by_code:
            print(f"{session_key}: no legislator matched; "
                  f"run with --list to see the roster")
            continue

        codes = sorted(by_code)
        floor = fetch_floor_votes(session_key, codes)
        committee = fetch_committee_votes(session_key, codes)
        catchlines = fetch_measure_catchlines(session_key)

        warnings = []
        if args.district and not args.legislator:
            missing = find_unrostered_voters(session_key, roster)
            if missing:
                warnings.append(
                    "These voters cast floor votes but are not on the session "
                    "roster, meaning they left office mid-session: "
                    + ", ".join(missing)
                    + ". If one of them held this district, re-run adding "
                      '--legislator "<code>" to include their votes.')

        if args.district:
            label = f"{'HD' if args.chamber == 'H' else 'SD'}{args.district}"
        else:
            label = clean(by_code[codes[0]].get("LastName")) or codes[0]
        suffix = "_nay" if args.nay_only else ""
        path = f"votes_{label.replace(' ', '_')}_{session_key}{suffix}.txt"
        write_report(path, session_key, [by_code[c] for c in codes],
                     floor, committee, catchlines, warnings,
                     nay_only=args.nay_only)

        if args.nay_only:
            nays = sum(1 for v in floor if v.get("Vote") == NAY)
            committee_nays = sum(1 for v in committee if v.get("Meaning") == NAY)
            print(f"{session_key}: {', '.join(codes)} - {nays} floor Nay votes "
                  f"(of {len(floor)}), {committee_nays} committee Nay votes "
                  f"(of {len(committee)}) -> {path}")
        else:
            print(f"{session_key}: {', '.join(codes)} - {len(floor)} floor votes, "
                  f"{len(committee)} committee votes -> {path}")
        for note in warnings:
            print(f"  NOTE: {note}")


def main(argv=None):
    args = parse_args(argv if argv is not None else sys.argv[1:])
    try:
        run(args)
    except requests.exceptions.RequestException as e:
        print(f"Error accessing OLIS API: {e}")
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())