gateway/modules/features/trustee/accounting/connectors/accountingConnectorBexio.py

# Copyright (c) 2025 Patrick Motsch
# All rights reserved.
"""Bexio accounting connector.

API docs: https://docs.bexio.com/
Auth: Personal Access Token (PAT) as Bearer token.
Base URL: https://api.bexio.com/
Note: Bexio uses internal account IDs (int), not account numbers.
The connector caches the chart of accounts to resolve accountNumber -> account_id.
"""

import logging
from typing import List, Dict, Any, Optional

import aiohttp

from ..accountingConnectorBase import (
    BaseAccountingConnector,
    AccountingBooking,
    AccountingChart,
    ConnectorConfigField,
    SyncResult,
)

logger = logging.getLogger(__name__)

_BASE_URL = "https://api.bexio.com"


class AccountingConnectorBexio(BaseAccountingConnector):

    def __init__(self):
        self._chartCache: Dict[str, List[Dict[str, Any]]] = {}

    def getConnectorType(self) -> str:
        return "bexio"

    def getConnectorLabel(self) -> Dict[str, str]:
        return {"en": "Bexio", "de": "Bexio", "fr": "Bexio"}

    def getRequiredConfigFields(self) -> List[ConnectorConfigField]:
        return [
            ConnectorConfigField(
                key="accessToken",
                label={"en": "Personal Access Token", "de": "Persönlicher Zugriffstoken", "fr": "Jeton d'accès personnel"},
                fieldType="password",
                secret=True,
                placeholder="PAT from developer.bexio.com",
            ),
        ]

    def _buildHeaders(self, config: Dict[str, Any]) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {config['accessToken']}",
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

    async def testConnection(self, config: Dict[str, Any]) -> SyncResult:
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(f"{_BASE_URL}/3.0/users/me", headers=self._buildHeaders(config), timeout=aiohttp.ClientTimeout(total=15)) as resp:
                    if resp.status == 200:
                        return SyncResult(success=True)
                    body = await resp.text()
                    return SyncResult(success=False, errorMessage=f"HTTP {resp.status}: {body[:200]}")
        except Exception as e:
            return SyncResult(success=False, errorMessage=str(e))

    async def _loadRawAccounts(self, config: Dict[str, Any]) -> List[Dict[str, Any]]:
        """Load raw account list and cache it for accountNumber -> id mapping."""
        cacheKey = config.get("accessToken", "")[:16]
        if cacheKey in self._chartCache:
            return self._chartCache[cacheKey]

        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(f"{_BASE_URL}/2.0/accounts", headers=self._buildHeaders(config), timeout=aiohttp.ClientTimeout(total=30)) as resp:
                    if resp.status != 200:
                        return []
                    accounts = await resp.json()
                    self._chartCache[cacheKey] = accounts
                    return accounts
        except Exception as e:
            logger.error(f"Bexio loadRawAccounts error: {e}")
            return []

    def _resolveAccountId(self, accounts: List[Dict[str, Any]], accountNumber: str) -> Optional[int]:
        """Resolve an account number string to a Bexio internal account_id."""
        for acc in accounts:
            if str(acc.get("account_no", "")) == accountNumber:
                return acc.get("id")
        return None

    async def getChartOfAccounts(self, config: Dict[str, Any], accountType: Optional[str] = None) -> List[AccountingChart]:
        accounts = await self._loadRawAccounts(config)
        return [
            AccountingChart(
                accountNumber=str(acc.get("account_no", "")),
                label=acc.get("name", ""),
                accountType=acc.get("account_type", None),
            )
            for acc in accounts
        ]

    async def pushBooking(self, config: Dict[str, Any], booking: AccountingBooking) -> SyncResult:
        """Push a manual entry to Bexio.

        Bexio requires account_id (int) rather than account numbers, so we resolve
        via the cached chart of accounts.
        """
        try:
            accounts = await self._loadRawAccounts(config)

            entries = []
            for line in booking.lines:
                debitAccId = self._resolveAccountId(accounts, line.accountNumber) if line.debitAmount > 0 else None
                creditAccId = self._resolveAccountId(accounts, line.accountNumber) if line.creditAmount > 0 else None
                amount = line.debitAmount if line.debitAmount > 0 else line.creditAmount

                if debitAccId is None and creditAccId is None:
                    return SyncResult(success=False, errorMessage=f"Account {line.accountNumber} not found in Bexio chart")

                entry: Dict[str, Any] = {"amount": str(amount), "description": line.description or booking.description}
                if debitAccId:
                    entry["debit_account_id"] = debitAccId
                if creditAccId:
                    entry["credit_account_id"] = creditAccId
                if line.taxCode:
                    entry["tax_id"] = line.taxCode
                entries.append(entry)

            entryType = "manual_single_entry" if len(entries) == 1 else "manual_group_entry"
            payload = {
                "date": booking.bookingDate,
                "reference_nr": booking.reference,
                "type": entryType,
                "entries": entries,
            }

            async with aiohttp.ClientSession() as session:
                url = f"{_BASE_URL}/3.0/accounting/manual-entries"
                async with session.post(url, headers=self._buildHeaders(config), json=payload, timeout=aiohttp.ClientTimeout(total=30)) as resp:
                    body = await resp.json() if resp.content_type == "application/json" else {"raw": await resp.text()}
                    if resp.status in (200, 201):
                        externalId = str(body.get("id", "")) if isinstance(body, dict) else None
                        return SyncResult(success=True, externalId=externalId, rawResponse=body)
                    return SyncResult(success=False, errorMessage=f"HTTP {resp.status}", rawResponse=body)
        except Exception as e:
            return SyncResult(success=False, errorMessage=str(e))

    async def getBookingStatus(self, config: Dict[str, Any], externalId: str) -> SyncResult:
        try:
            async with aiohttp.ClientSession() as session:
                url = f"{_BASE_URL}/3.0/accounting/manual-entries/{externalId}"
                async with session.get(url, headers=self._buildHeaders(config), timeout=aiohttp.ClientTimeout(total=15)) as resp:
                    if resp.status == 200:
                        return SyncResult(success=True, externalId=externalId)
                    return SyncResult(success=False, errorMessage=f"HTTP {resp.status}")
        except Exception as e:
            return SyncResult(success=False, errorMessage=str(e))

    async def getCustomers(self, config: Dict[str, Any]) -> List[Dict[str, Any]]:
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(f"{_BASE_URL}/2.0/contact", headers=self._buildHeaders(config), timeout=aiohttp.ClientTimeout(total=30)) as resp:
                    if resp.status != 200:
                        return []
                    return await resp.json()
        except Exception as e:
            logger.error(f"Bexio getCustomers error: {e}")
            return []