Script to access Linq Connect API

README.md

LinqConnect Transaction Downloader

Download transaction history from LinqConnect (school meal accounts) and save it as JSON or CSV for import into Google Sheets or other tools.

Prerequisites

• Python 3.10+
• Playwright (only needed for browser login mode)

pip install playwright
playwright install chromium

Quick start

From a HAR file (no login needed)

If you've already captured a HAR file from Chrome DevTools:

python3 linqconnect_download.py --har linqconnect.com.har

How to capture a HAR file

1. Open Chrome and navigate to linqconnect.com
2. Open DevTools (F12) and go to the Network tab
3. Check Preserve log
4. Log in and navigate to Dashboard > Transaction History
5. Wait for the transaction list to load
6. Right-click in the Network panel and choose Save all as HAR with content

The HAR file contains the full API response with all your transaction records.

Via browser login

python3 linqconnect_download.py

A Chromium window opens. Log in manually, and the script navigates to the transaction history page and captures the API response automatically.

Headless browser login

LINQCONNECT_USERNAME=you@example.com LINQCONNECT_PASSWORD=yourpassword \
  python3 linqconnect_download.py --headless

Output formats

JSON (default)

python3 linqconnect_download.py --har linqconnect.com.har
# writes linqconnect_transactions.json

Each record looks like:

{
  "Date": "2026-08-27T12:00:00Z",
  "DistrictMemberName": "Sienna Clark",
  "Category": "Prepaid Account",
  "FeeItem": "ES Lunch Meal (1)",
  "Status": "Posted",
  "TransactionId": "DEA05F919CEDEACE",
  "Total": 3.45,
  "Type": "CustomerDocument",
  "PaymentMethod": "Prepaid Account",
  "CurrentMealAccountBalance": 25.4,
  "Items": [...]
}

CSV (for Google Sheets)

python3 linqconnect_download.py --har linqconnect.com.har --format csv
# writes linqconnect_transactions.csv

The CSV flattens the nested Items array into semicolon-separated columns. Open the CSV in Google Sheets or Excel directly.

Filtering by date

Download only transactions on or after a specific date:

python3 linqconnect_download.py --har linqconnect.com.har --since 2026-01-01

Incremental downloads

To download only new records since your last export:

python3 linqconnect_download.py --after-last linqconnect_transactions.json

This reads the existing file, finds the most recent transaction date, fetches all transactions from LinqConnect, keeps only the newer ones, and merges them into the file. Duplicates are detected by TransactionId + Date.

Works with both JSON and CSV files:

python3 linqconnect_download.py --after-last linqconnect_transactions.csv --format csv

You can combine --after-last with --har to merge a new HAR export into an existing file:

python3 linqconnect_download.py --har new_export.har --after-last linqconnect_transactions.json

All options

usage: linqconnect_download.py [-h] [--har HAR] [--since SINCE]
                                [--after-last AFTER_LAST]
                                [--format {json,csv}] [--output OUTPUT]
                                [--headless]

  --har HAR             Path to a Chrome HAR file (skips browser login)
  --since SINCE         Only include transactions on or after this date (YYYY-MM-DD)
  --after-last FILE     Merge new transactions into an existing JSON or CSV file
  --format {json,csv}   Output format (default: json)
  -o, --output OUTPUT   Output file path (default: linqconnect_transactions.{format})
  --headless            Run browser headless; requires LINQCONNECT_USERNAME
                        and LINQCONNECT_PASSWORD environment variables

How it works

LinqConnect uses Auth0 OAuth2 with PKCE for authentication and AWS WAF for bot protection. The API endpoint GET api.linqconnect.com/api/FamilyTransactionSearch?pageSize=0 returns all transaction records in a single response (the pageSize=0 parameter disables pagination).

In browser mode, Playwright launches a real Chromium instance to satisfy the WAF, intercepts the API response after login, and extracts the transaction data. In HAR mode, the script reads the already-captured API response directly from the file.

linqconnect_download.py

#!/usr/bin/env python3
"""
Download transaction history from LinqConnect.

Supports two data sources:
  - Browser login via Playwright (handles Auth0 OAuth2 + AWS WAF)
  - HAR file captured from Chrome DevTools

Usage:
    # Download all via browser login
    python3 linqconnect_download.py

    # Extract from a HAR file (no login needed)
    python3 linqconnect_download.py --har linqconnect.com.har

    # Only transactions after a date
    python3 linqconnect_download.py --since 2026-01-01

    # Incremental: download only new records since last saved file
    python3 linqconnect_download.py --after-last linqconnect_transactions.json

    # Save as CSV instead of JSON
    python3 linqconnect_download.py --format csv

    # Headless browser mode (reads credentials from env)
    python3 linqconnect_download.py --headless
"""

import argparse
import csv
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

API_BASE = "https://api.linqconnect.com/api"
LOGIN_URL = "https://linqconnect.com"
TRANSACTION_ENDPOINT = f"{API_BASE}/FamilyTransactionSearch?pageSize=0"


def parse_date(date_str: str) -> datetime:
    """Parse ISO date strings from the API."""
    date_str = date_str.rstrip("Z")
    if "." in date_str:
        date_str = date_str.split(".")[0]
    return datetime.fromisoformat(date_str).replace(tzinfo=timezone.utc)


def get_last_date_from_file(filepath: str) -> datetime | None:
    """Read existing data file and return the most recent transaction date."""
    path = Path(filepath)
    if not path.exists():
        print(f"File {filepath} not found, downloading all transactions.")
        return None

    if filepath.endswith(".json"):
        with open(filepath) as f:
            data = json.load(f)
        if not data:
            return None
        dates = [parse_date(r["Date"]) for r in data]
        latest = max(dates)
        print(f"Last transaction in {filepath}: {latest.isoformat()}")
        return latest
    elif filepath.endswith(".csv"):
        with open(filepath) as f:
            reader = csv.DictReader(f)
            dates = [parse_date(row["Date"]) for row in reader]
        if not dates:
            return None
        latest = max(dates)
        print(f"Last transaction in {filepath}: {latest.isoformat()}")
        return latest
    else:
        print(f"Unsupported file format: {filepath}", file=sys.stderr)
        sys.exit(1)


def extract_from_har(har_path: str) -> list[dict]:
    """Extract transaction records from a Chrome HAR file."""
    path = Path(har_path)
    if not path.exists():
        print(f"HAR file not found: {har_path}", file=sys.stderr)
        sys.exit(1)

    print(f"Reading HAR file: {har_path}")
    with open(path) as f:
        har = json.load(f)

    for entry in har["log"]["entries"]:
        url = entry["request"]["url"]
        if "FamilyTransactionSearch" in url:
            resp_text = entry["response"]["content"].get("text", "")
            if not resp_text:
                continue
            data = json.loads(resp_text)
            records = data.get("Data", [])
            print(f"Extracted {len(records)} transactions from HAR file.")
            return records

    print(
        "No FamilyTransactionSearch response found in the HAR file.\n"
        "Make sure you navigated to the transaction history page before "
        "saving the HAR.",
        file=sys.stderr,
    )
    sys.exit(1)


def login_and_fetch(headless: bool = False) -> list[dict]:
    """Log in via Playwright and fetch all transactions."""
    try:
        from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout
    except ImportError:
        print(
            "Playwright is required for browser login.\n"
            "Install it with: pip install playwright && playwright install chromium\n"
            "Or use --har to extract from a HAR file instead.",
            file=sys.stderr,
        )
        sys.exit(1)

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=headless)
        context = browser.new_context()
        page = context.new_page()

        captured_transactions = []

        def handle_response(response):
            if "FamilyTransactionSearch" in response.url:
                try:
                    data = response.json()
                    captured_transactions.append(data)
                except Exception:
                    pass

        page.on("response", handle_response)

        print("Navigating to LinqConnect...")
        page.goto(LOGIN_URL, wait_until="networkidle")

        if headless:
            username = os.environ.get("LINQCONNECT_USERNAME", "")
            password = os.environ.get("LINQCONNECT_PASSWORD", "")
            if not username or not password:
                print(
                    "Error: --headless requires LINQCONNECT_USERNAME and "
                    "LINQCONNECT_PASSWORD environment variables.",
                    file=sys.stderr,
                )
                browser.close()
                sys.exit(1)

            page.wait_for_selector('input[name="username"]', timeout=15000)
            page.fill('input[name="username"]', username)
            page.click('button[type="submit"]')

            page.wait_for_selector('input[name="password"]', timeout=15000)
            page.fill('input[name="password"]', password)
            page.click('button[type="submit"]')
        else:
            print("Please log in manually in the browser window...")

        print("Waiting for login to complete...")
        try:
            page.wait_for_url("**/dashboard**", timeout=120000)
        except PlaywrightTimeout:
            try:
                page.wait_for_url("**/home**", timeout=10000)
            except PlaywrightTimeout:
                if "linqconnect.com" in page.url and "login" not in page.url:
                    pass
                else:
                    print(
                        "Login timed out. Current URL: " + page.url,
                        file=sys.stderr,
                    )
                    browser.close()
                    sys.exit(1)

        print(f"Logged in. Current URL: {page.url}")

        print("Navigating to transaction history...")
        page.goto(
            f"{LOGIN_URL}/dashboard/transaction-history",
            wait_until="networkidle",
        )

        print("Waiting for transaction data to load...")
        page.wait_for_timeout(5000)

        if captured_transactions:
            data = captured_transactions[0]
            records = data.get("Data", [])
            print(f"Captured {len(records)} transactions from API response.")
            browser.close()
            return records

        print("API intercept missed, fetching via page context...")
        result = page.evaluate(
            """async () => {
                const resp = await fetch(
                    'https://api.linqconnect.com/api/FamilyTransactionSearch?pageSize=0',
                    { credentials: 'include' }
                );
                return await resp.json();
            }"""
        )
        records = result.get("Data", [])
        print(f"Fetched {len(records)} transactions via page context.")
        browser.close()
        return records


def flatten_record(record: dict) -> dict:
    """Flatten a transaction record for CSV output."""
    items = record.get("Items", [])
    item_names = "; ".join(i.get("ItemName", "") for i in items)
    item_types = "; ".join(i.get("TransactionType", "") for i in items)
    item_amounts = "; ".join(str(i.get("Amount", "")) for i in items)
    item_quantities = "; ".join(str(i.get("Quantity", "")) for i in items)
    person_names = "; ".join(
        i.get("PersonName", "") for i in items if i.get("PersonName")
    )
    account_balance = items[0].get("AccountBalance", "") if items else ""

    return {
        "Date": record.get("Date", ""),
        "DistrictMemberName": record.get("DistrictMemberName", ""),
        "Category": record.get("Category", ""),
        "FeeItem": record.get("FeeItem", ""),
        "Status": record.get("Status", ""),
        "TransactionId": record.get("TransactionId", ""),
        "Total": record.get("Total", ""),
        "Type": record.get("Type", ""),
        "PaymentMethod": record.get("PaymentMethod", ""),
        "CurrentMealAccountBalance": record.get(
            "CurrentMealAccountBalance", ""
        ),
        "ItemNames": item_names,
        "ItemTransactionTypes": item_types,
        "ItemAmounts": item_amounts,
        "ItemQuantities": item_quantities,
        "PersonNames": person_names,
        "AccountBalance": account_balance,
    }


def save_json(records: list[dict], filepath: str):
    with open(filepath, "w") as f:
        json.dump(records, f, indent=2)
    print(f"Saved {len(records)} transactions to {filepath}")


def save_csv(records: list[dict], filepath: str):
    if not records:
        print("No records to save.")
        return
    flat = [flatten_record(r) for r in records]
    fieldnames = list(flat[0].keys())
    with open(filepath, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(flat)
    print(f"Saved {len(records)} transactions to {filepath}")


def merge_records(
    existing: list[dict], new: list[dict]
) -> list[dict]:
    """Merge new records into existing, deduplicating by TransactionId+Date."""
    seen = set()
    for r in existing:
        key = (r.get("TransactionId", ""), r.get("Date", ""))
        seen.add(key)

    added = 0
    for r in new:
        key = (r.get("TransactionId", ""), r.get("Date", ""))
        if key not in seen:
            existing.append(r)
            seen.add(key)
            added += 1

    existing.sort(key=lambda r: r.get("Date", ""), reverse=True)
    print(f"Merged {added} new transactions (total: {len(existing)}).")
    return existing


def main():
    parser = argparse.ArgumentParser(
        description="Download LinqConnect transaction history."
    )
    parser.add_argument(
        "--har",
        help="Path to a Chrome HAR file to extract transactions from "
        "(skips browser login).",
    )
    parser.add_argument(
        "--since",
        help="Only download transactions on or after this date (YYYY-MM-DD).",
    )
    parser.add_argument(
        "--after-last",
        help="Path to existing data file; download only newer transactions.",
    )
    parser.add_argument(
        "--format",
        choices=["json", "csv"],
        default="json",
        help="Output format (default: json).",
    )
    parser.add_argument(
        "--output",
        "-o",
        help="Output file path (default: linqconnect_transactions.{format}).",
    )
    parser.add_argument(
        "--headless",
        action="store_true",
        help="Run headless; requires LINQCONNECT_USERNAME and "
        "LINQCONNECT_PASSWORD env vars.",
    )
    args = parser.parse_args()

    output = args.output or f"linqconnect_transactions.{args.format}"

    since_date = None
    if args.since:
        since_date = datetime.fromisoformat(args.since).replace(
            tzinfo=timezone.utc
        )
        print(f"Filtering transactions since {since_date.date()}")
    elif args.after_last:
        since_date = get_last_date_from_file(args.after_last)
        if since_date:
            print(f"Will download transactions after {since_date.isoformat()}")

    if args.har:
        records = extract_from_har(args.har)
    else:
        records = login_and_fetch(headless=args.headless)

    if since_date:
        before = len(records)
        records = [r for r in records if parse_date(r["Date"]) > since_date]
        print(f"Filtered to {len(records)} transactions (from {before} total).")

    if args.after_last and Path(args.after_last).exists():
        if args.after_last.endswith(".json"):
            with open(args.after_last) as f:
                existing = json.load(f)
            records = merge_records(existing, records)
            output = args.output or args.after_last
        elif args.after_last.endswith(".csv"):
            with open(args.after_last) as f:
                reader = csv.DictReader(f)
                existing_flat = list(reader)
            new_flat = [flatten_record(r) for r in records]
            seen = set()
            for row in existing_flat:
                seen.add((row.get("TransactionId", ""), row.get("Date", "")))
            added = 0
            for row in new_flat:
                key = (row.get("TransactionId", ""), row.get("Date", ""))
                if key not in seen:
                    existing_flat.append(row)
                    seen.add(key)
                    added += 1
            existing_flat.sort(key=lambda r: r.get("Date", ""), reverse=True)
            print(f"Merged {added} new rows (total: {len(existing_flat)}).")
            output_path = args.output or args.after_last
            if existing_flat:
                fieldnames = list(existing_flat[0].keys())
                with open(output_path, "w", newline="") as f:
                    writer = csv.DictWriter(f, fieldnames=fieldnames)
                    writer.writeheader()
                    writer.writerows(existing_flat)
                print(f"Saved {len(existing_flat)} transactions to {output_path}")
            return

    if args.format == "csv":
        save_csv(records, output)
    else:
        save_json(records, output)


if __name__ == "__main__":
    main()