Add test mode for manual cloud function runs

- Default to test mode when run without PRODUCTION env var
- Test mode sends email with most recent expenditure from DB
- Includes recipient list, transaction details, and JSON attachment
- Scheduled runs set PRODUCTION=1 to run full flow

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: afriedman412

.github/workflows/deploy.yml

@@ -32,4 +32,5 @@ jobs:
         run: |
           gcloud run jobs update freeway2026 \
             --image $IMAGE \
-            --region us-central1
+            --region us-central1 \
+            --set-env-vars PRODUCTION=1

app/db.py

@@ -1,4 +1,5 @@
-from sqlmodel import SQLModel, create_engine, text
+from sqlmodel import SQLModel, Session, create_engine, text, select
+import pandas as pd
 from app.schemas import Committee, Contribution, Expenditure
 from app.config import POSTGRES_URL
 
@@ -23,3 +24,22 @@ def reset_tables(database_url: str, echo: bool = False) -> None:
         print("CREATING IN DB:", db, "SCHEMA:", schema)
     SQLModel.metadata.drop_all(engine)
     SQLModel.metadata.create_all(engine)
+
+
+def get_latest_expenditure(engine) -> pd.DataFrame:
+    """Get the most recent expenditure from the database, with committee name."""
+    with Session(engine) as session:
+        stmt = select(Expenditure).order_by(
+            Expenditure.expenditure_date.desc()
+        ).limit(1)
+        result = session.exec(stmt).first()
+        if result:
+            data = result.model_dump()
+            # Join committee name
+            if result.committee_id:
+                committee = session.get(Committee, result.committee_id)
+                data["committee_name"] = committee.name if committee else None
+            else:
+                data["committee_name"] = None
+            return pd.DataFrame([data])
+        return pd.DataFrame()

app/helpers.py

@@ -28,6 +28,10 @@ def df_to_csv_bytes(df) -> bytes:
     return buf.getvalue().encode("utf-8")
 
 
+def df_to_json_bytes(df) -> bytes:
+    return df.to_json(orient="records", indent=2).encode("utf-8")
+
+
 def format_results(results: dict) -> str:
     return "\n".join(
         f"{k:25} {results[k]}"

app/mail.py

@@ -1,7 +1,7 @@
 import smtplib
 from email.message import EmailMessage
 import os
-from app.helpers import df_to_csv_bytes, normalize_recipients
+from app.helpers import df_to_csv_bytes, df_to_json_bytes, normalize_recipients
 
 
 def send_email(
@@ -11,6 +11,7 @@ def send_email(
     to: list[str],
     sender: str,
     df=None,
+    attachment_format: str = "csv",
 ):
     msg = EmailMessage()
     msg["From"] = sender
@@ -19,14 +20,22 @@ def send_email(
     msg.set_content(body)
 
     if df is not None and not df.empty:
-        csv_bytes = df_to_csv_bytes(df)
-
-        msg.add_attachment(
-            csv_bytes,
-            maintype="text",
-            subtype="csv",
-            filename="deduped_results.csv",
-        )
+        if attachment_format == "json":
+            data_bytes = df_to_json_bytes(df)
+            msg.add_attachment(
+                data_bytes,
+                maintype="application",
+                subtype="json",
+                filename="results.json",
+            )
+        else:
+            csv_bytes = df_to_csv_bytes(df)
+            msg.add_attachment(
+                csv_bytes,
+                maintype="text",
+                subtype="csv",
+                filename="deduped_results.csv",
+            )
 
     with smtplib.SMTP_SSL("smtp.gmail.com", 465) as smtp:
         smtp.login(

app/main.py

@@ -2,7 +2,7 @@
 from app.ingestion import ingest_jsonl
 from app.mail import send_email
 from app.logger import logger
-from app.db import get_engine, create_tables
+from app.db import get_engine, create_tables, get_latest_expenditure
 from app.helpers import get_now, format_results, get_today
 from app.config import CYCLE, TARGET_EMAILS
 import os
@@ -68,13 +68,57 @@ def run(variant, key, cycle):
         logger.info("No new data! ")
 
 
+def run_test_mode():
+    """Send a test email with the most recent expenditure from the database."""
+    logger.info("Running in TEST MODE...")
+    engine = get_engine()
+
+    latest_df = get_latest_expenditure(engine)
+
+    if latest_df.empty:
+        logger.info("No expenditures found in database!")
+        return
+
+    ts = get_now()
+    row = latest_df.iloc[0]
+    recipients_list = "\n".join(f"  - {email}" for email in TARGET_EMAILS)
+
+    body = f"""[TEST MODE] Most recent expenditure as of {ts}
+
+Recipients:
+{recipients_list}
+
+Latest Record:
+--------------
+Date:          {row['expenditure_date']}
+Amount:        ${row['expenditure_amount']:,.2f}
+Committee:     {row['committee_name']} ({row['committee_id']})
+Payee:         {row['payee_name']}
+Description:   {row['expenditure_description']}
+"""
+
+    send_email(
+        subject=f"[sludgewire] TEST - Last Update Check, {ts}",
+        body=body,
+        to=TARGET_EMAILS,
+        sender="steadynappin@gmail.com",
+        df=latest_df,
+        attachment_format="json"
+    )
+    logger.info("Test email sent!")
+
+
 if __name__ == "__main__":
-    today = get_today()
     create_tables()
-    logger.info("Running...")
-    run(
-        variant="expenditure",
-        key=today,
-        cycle=CYCLE
-    )
-    logger.info("Run complete.")
+
+    if os.environ.get("PRODUCTION"):
+        today = get_today()
+        logger.info("Running production mode...")
+        run(
+            variant="expenditure",
+            key=today,
+            cycle=CYCLE
+        )
+        logger.info("Run complete.")
+    else:
+        run_test_mode()