from __future__ import annotations import logging from datetime import datetime, timedelta, timezone from typing import Any import httpx logger = logging.getLogger("uvicorn.error") TOKEN_URL = "https://identity.xero.com/connect/token" CONNECTIONS_URL = "https://api.xero.com/connections" ACCOUNTING_BASE_URL = "https://api.xero.com/api.xro/2.0" class XeroApiError(RuntimeError): pass class XeroClient: def __init__(self, client_id: str, client_secret: str, redirect_uri: str): self.client_id = client_id self.client_secret = client_secret self.redirect_uri = redirect_uri async def exchange_code(self, code: str, code_verifier: str | None = None) -> dict[str, Any]: data = { "grant_type": "authorization_code", "code": code, "redirect_uri": self.redirect_uri, } if code_verifier: data["code_verifier"] = code_verifier async with httpx.AsyncClient(timeout=60) as client: response = await client.post( TOKEN_URL, data=data, auth=(self.client_id, self.client_secret), ) return self._json_or_raise(response) async def refresh(self, refresh_token: str) -> dict[str, Any]: async with httpx.AsyncClient(timeout=60) as client: response = await client.post( TOKEN_URL, data={"grant_type": "refresh_token", "refresh_token": refresh_token}, auth=(self.client_id, self.client_secret), ) return self._json_or_raise(response) async def connections(self, access_token: str) -> list[dict[str, Any]]: return await self._get(access_token, None, CONNECTIONS_URL) async def contacts(self, access_token: str, xero_tenant_id: str, modified_after: datetime | None) -> list[dict[str, Any]]: return await self._paged_get(access_token, xero_tenant_id, f"{ACCOUNTING_BASE_URL}/Contacts", "Contacts", modified_after) async def items(self, access_token: str, xero_tenant_id: str, modified_after: datetime | None) -> list[dict[str, Any]]: logger.info("[xero_client] items: 开始请求 Xero Items API, modified_after=%s", modified_after) data = await self._get(access_token, xero_tenant_id, f"{ACCOUNTING_BASE_URL}/Items", modified_after=modified_after) items_list = data.get("Items", []) logger.info("[xero_client] items: 获取到 %d 条 items", len(items_list)) return items_list async def tax_rates(self, access_token: str, xero_tenant_id: str) -> list[dict[str, Any]]: data = await self._get(access_token, xero_tenant_id, f"{ACCOUNTING_BASE_URL}/TaxRates") return data.get("TaxRates", []) async def create_invoice(self, access_token: str, xero_tenant_id: str, invoice_payload: dict[str, Any]) -> dict[str, Any]: data = await self._post(access_token, xero_tenant_id, f"{ACCOUNTING_BASE_URL}/Invoices", {"Invoices": [invoice_payload]}) invoices = data.get("Invoices", []) if not invoices: raise XeroApiError("Xero returned empty Invoices array") return invoices[0] async def get_invoice(self, access_token: str, xero_tenant_id: str, invoice_id: str) -> dict[str, Any]: data = await self._get(access_token, xero_tenant_id, f"{ACCOUNTING_BASE_URL}/Invoices/{invoice_id}") invoices = data.get("Invoices", []) return invoices[0] if invoices else {} async def _paged_get(self, access_token: str, xero_tenant_id: str, url: str, key: str, modified_after: datetime | None) -> list[dict[str, Any]]: page = 1 rows: list[dict[str, Any]] = [] while True: data = await self._get(access_token, xero_tenant_id, url, params={"page": page}, modified_after=modified_after) batch = data.get(key, []) if not batch: break rows.extend(batch) if len(batch) < 100: break page += 1 return rows async def _post(self, access_token: str, xero_tenant_id: str, url: str, json_body: dict[str, Any]): headers = { "Authorization": f"Bearer {access_token}", "Accept": "application/json", "Content-Type": "application/json", "xero-tenant-id": xero_tenant_id, } import json as _json logger.info("[xero_client] _post REQUEST: url=%s, body=%s", url, _json.dumps(json_body, default=str)[:3000]) async with httpx.AsyncClient(timeout=120) as client: response = await client.post(url, headers=headers, json=json_body) logger.info("[xero_client] _post RESPONSE: url=%s, status=%d, body=%s", url, response.status_code, response.text[:5000]) return self._json_or_raise(response) async def _get(self, access_token: str, xero_tenant_id: str | None, url: str, params: dict | None = None, modified_after: datetime | None = None): headers = {"Authorization": f"Bearer {access_token}", "Accept": "application/json"} if xero_tenant_id: headers["xero-tenant-id"] = xero_tenant_id if modified_after: headers["If-Modified-Since"] = modified_after.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") # 详细日志：请求信息（隐藏 token） safe_headers = {k: (v[:20] + "..." if k == "Authorization" else v) for k, v in headers.items()} logger.info("[xero_client] _get REQUEST: url=%s, params=%s, headers=%s", url, params, safe_headers) async with httpx.AsyncClient(timeout=120) as client: response = await client.get(url, headers=headers, params=params) # 详细日志：响应信息（截取前 2000 字符） body_preview = response.text[:2000] logger.info("[xero_client] _get RESPONSE: url=%s, status=%d, bytes=%d, body=%s", url, response.status_code, len(response.content), body_preview) return self._json_or_raise(response) def expires_at_from_token(self, token_data: dict[str, Any]) -> datetime: return datetime.now(timezone.utc) + timedelta(seconds=int(token_data.get("expires_in", 1800)) - 120) def _json_or_raise(self, response: httpx.Response): if response.status_code >= 400: # 解析出真正的 ValidationErrors，而非截断原始 JSON detail = "" try: data = response.json() errors = [] for elem in data.get("Elements", []): for ve in elem.get("ValidationErrors", []): errors.append(ve.get("Message", "")) for li in elem.get("LineItems", []): for ve in li.get("ValidationErrors", []): errors.append(f"[{li.get('ItemCode', '?')}] {ve.get('Message', '')}") if errors: detail = "; ".join(errors) else: detail = data.get("Message", response.text[:500]) except Exception: detail = response.text[:500] raise XeroApiError(f"Xero API error {response.status_code}: {detail}") return response.json()