xero_list_bank_transactions.py

import hashlib
import base64
import secrets
import re
from html.parser import HTMLParser
from urllib.parse import urlencode, urlparse, parse_qs
from curl_cffi import requests

# Injected at runtime; fallback for local testing
try:
    BASE_URL
except NameError:
    BASE_URL = "https://go.xero.com"


def run(headers, user_input):
    """
    List all transactions for a specific bank account.

    Fetches and parses the bank transactions page to extract transaction data
    including dates, descriptions, amounts, and reconciliation status.
    """
    # Validate input
    account_id = user_input.get("account_id")
    if not account_id:
        return {'status_code': 400, 'body': {'error': 'account_id is required'}}

    # Normalize account_id format (remove dashes for URL)
    clean_account_id = account_id.replace("-", "").upper()

    # Get fresh Bearer token (needed for authenticated requests)
    access_token, error = _get_bearer_token(headers)
    if not access_token:
        if error and "Session expired" in error:
            return {'status_code': 401, 'body': {'error': 'Session expired - reauthentication required'}}
        return {'status_code': 401, 'body': {'error': f'Failed to obtain access token: {error}'}}

    # Fetch and parse transactions
    try:
        transactions = _fetch_transactions(headers, clean_account_id)
    except AuthenticationError as e:
        return {'status_code': 401, 'body': {'error': str(e)}}
    except TransactionPageError as e:
        return {'status_code': e.status_code, 'body': {'error': str(e)}}
    except Exception as e:
        return {'status_code': 500, 'body': {'error': f'Failed to fetch transactions: {str(e)}'}}

    return {
        'status_code': 200,
        'body': {
            'account_id': account_id,
            'transactions': transactions,
            'total_count': len(transactions)
        }
    }


# === PRIVATE ===

class AuthenticationError(Exception):
    """Raised when authentication fails."""
    pass


class TransactionPageError(Exception):
    """Raised when the transactions page cannot be retrieved."""
    def __init__(self, message, status_code=400):
        super().__init__(message)
        self.status_code = status_code


class TransactionTableParser(HTMLParser):
    """Parse bank transactions from the HTML table."""

    def __init__(self):
        super().__init__()
        self.transactions = []
        self.current_transaction = None
        self.in_tbody = False
        self.in_row = False
        self.in_cell = False
        self.cell_index = 0
        self.cell_content = ""
        self.current_link_href = ""

    def handle_starttag(self, tag, attrs):
        attrs_dict = dict(attrs)

        if tag == "tbody" and attrs_dict.get("id") == "bankTransactionList":
            self.in_tbody = True

        if self.in_tbody and tag == "tr":
            self.in_row = True
            self.current_transaction = {
                "transaction_id": None,
                "date": None,
                "description": None,
                "reference": None,
                "payment_ref": None,
                "spent": None,
                "received": None,
                "balance": None,
                "source": None,
                "status": None
            }
            self.cell_index = 0

        if self.in_row and tag == "td":
            self.in_cell = True
            self.cell_content = ""
            self.current_link_href = ""

        if self.in_cell and tag == "input" and attrs_dict.get("name") == "selectedAccounts":
            # Extract transaction ID from checkbox value
            self.current_transaction["transaction_id"] = attrs_dict.get("value")

        if self.in_cell and tag == "a":
            href = attrs_dict.get("href", "")
            if "bankTransactionID=" in href:
                self.current_link_href = href

        if self.in_cell and tag == "span":
            title = attrs_dict.get("title", "")
            if "reconciled" in title.lower():
                self.current_transaction["status"] = "unreconciled" if "not" in title.lower() else "reconciled"

    def handle_endtag(self, tag):
        if tag == "tbody" and self.in_tbody:
            self.in_tbody = False

        if tag == "td" and self.in_cell:
            self.in_cell = False
            content = self.cell_content.strip()

            # Map cell index to field
            if self.cell_index == 1:  # Date
                self.current_transaction["date"] = content
            elif self.cell_index == 2:  # Description
                self.current_transaction["description"] = content
            elif self.cell_index == 3:  # Reference
                self.current_transaction["reference"] = content if content else None
            elif self.cell_index == 4:  # Payment Ref
                self.current_transaction["payment_ref"] = content if content else None
            elif self.cell_index == 5:  # Spent
                self.current_transaction["spent"] = _parse_amount(content)
            elif self.cell_index == 6:  # Received
                self.current_transaction["received"] = _parse_amount(content)
            elif self.cell_index == 7:  # Balance
                self.current_transaction["balance"] = _parse_amount(content)
            elif self.cell_index == 8:  # Source
                self.current_transaction["source"] = content if content else None
            elif self.cell_index == 9:  # Status (may be set by span title)
                if not self.current_transaction["status"]:
                    if "reconciled" in content.lower():
                        self.current_transaction["status"] = "reconciled" if "unreconciled" not in content.lower() else "unreconciled"

            self.cell_index += 1

        if tag == "tr" and self.in_row:
            self.in_row = False
            if self.current_transaction and self.current_transaction.get("transaction_id"):
                self.transactions.append(self.current_transaction)
            self.current_transaction = None

    def handle_data(self, data):
        if self.in_cell:
            self.cell_content += data


def _parse_amount(text):
    """Parse amount string to float, handling commas and empty values."""
    if not text or not text.strip():
        return None
    # Remove commas and whitespace
    clean = text.replace(",", "").replace(" ", "").strip()
    if not clean:
        return None
    try:
        return float(clean)
    except ValueError:
        return None


def _get_bearer_token(headers):
    """
    Perform silent OIDC flow to obtain a fresh Bearer token.
    Uses PKCE (Proof Key for Code Exchange) for secure auth code exchange.
    """
    cookies = headers.get("Cookie", "")

    # Generate PKCE code verifier and challenge
    code_verifier = secrets.token_hex(32) + secrets.token_hex(32)
    code_challenge = base64.urlsafe_b64encode(
        hashlib.sha256(code_verifier.encode()).digest()
    ).decode().rstrip('=')

    # Build authorization URL for silent auth
    state = secrets.token_hex(16)
    auth_params = {
        'client_id': 'xero_business_go',
        'redirect_uri': 'https://go.xero.com/oidc/silent.html',
        'response_type': 'code',
        'scope': 'openid profile email xero_frontend-apis xero_frontend-platform-apis',
        'state': state,
        'code_challenge': code_challenge,
        'code_challenge_method': 'S256',
        'prompt': 'none'
    }

    auth_url = f"https://login.xero.com/identity/connect/authorize?{urlencode(auth_params)}"

    response = requests.get(
        auth_url,
        headers={"Cookie": cookies},
        allow_redirects=True,
        impersonate="chrome131",
        timeout=30
    )

    final_url = str(response.url)
    parsed = urlparse(final_url)
    query_params = parse_qs(parsed.query)

    if 'code' not in query_params:
        if 'login' in final_url.lower() or response.status_code == 401:
            return None, "Session expired - login required"
        return None, f"No auth code in response URL: {final_url}"

    auth_code = query_params['code'][0]

    token_data = {
        'client_id': 'xero_business_go',
        'code': auth_code,
        'redirect_uri': 'https://go.xero.com/oidc/silent.html',
        'code_verifier': code_verifier,
        'grant_type': 'authorization_code'
    }

    token_response = requests.post(
        'https://identity.xero.com/connect/token',
        data=token_data,
        headers={
            'Content-Type': 'application/x-www-form-urlencoded',
            'Origin': 'https://go.xero.com',
            'Referer': 'https://go.xero.com/'
        },
        impersonate="chrome131",
        timeout=30
    )

    if token_response.status_code != 200:
        return None, f"Token exchange failed: {token_response.status_code}"

    token_json = token_response.json()
    return token_json.get('access_token'), None


def _fetch_transactions(headers, clean_account_id):
    """Fetch and parse the transactions page."""
    base_url = BASE_URL
    cookies = headers.get("Cookie", "")

    page_url = f"{base_url}/Bank/BankTransactions.aspx?accountId={clean_account_id}"

    response = requests.get(
        page_url,
        headers={
            "Cookie": cookies,
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
        },
        impersonate="chrome131",
        timeout=60
    )

    # Check for authentication issues
    if response.status_code == 401 or response.status_code == 403:
        raise AuthenticationError('Authentication failed')

    # Check for redirects to login page
    if 'login' in str(response.url).lower() and 'identity' in str(response.url).lower():
        raise AuthenticationError('Session expired - redirected to login')

    if response.status_code != 200:
        raise TransactionPageError(
            f'Failed to fetch transactions page: HTTP {response.status_code}',
            status_code=response.status_code
        )

    html_content = response.text

    # Check if we got a valid transactions page
    if 'bankTransactionList' not in html_content:
        # Check for error messages or login redirects
        if 'login' in html_content.lower() and 'password' in html_content.lower():
            raise AuthenticationError('Session expired - login required')
        raise TransactionPageError(
            'Invalid response - bank transactions table not found. Account may not exist or access denied.',
            status_code=400
        )

    # Parse the HTML to extract transactions
    parser = TransactionTableParser()
    parser.feed(html_content)

    return parser.transactions