74b2179c83
The /import response returns activities=[input echo with errors annotated]
— its length equals input size regardless of actual persistence. The
summary{total,imported,skipped,duplicates} block is the authoritative
signal. When imported<total, raise with errorMessage + skipped + duplicates
counts.
262 lines
10 KiB
Python
262 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Iterable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from broker_sync.models import Account, Activity
|
|
|
|
_LOGIN_PATH = "/api/v1/auth/login"
|
|
_ACCOUNTS_PATH = "/api/v1/accounts"
|
|
_IMPORT_CHECK = "/api/v1/activities/import/check"
|
|
_IMPORT_REAL = "/api/v1/activities/import"
|
|
|
|
|
|
class WealthfolioError(Exception):
|
|
pass
|
|
|
|
|
|
class WealthfolioUnauthorizedError(WealthfolioError):
|
|
"""Raised when login itself fails (bad creds or Wealthfolio down).
|
|
|
|
Distinct from a 401 on a random endpoint — those trigger an
|
|
automatic re-login attempt.
|
|
"""
|
|
|
|
|
|
class ImportValidationError(WealthfolioError):
|
|
"""`/activities/import/check` returned a non-2xx. We never reach the real import."""
|
|
|
|
|
|
class WealthfolioSink:
|
|
"""Push canonical Activities to Wealthfolio via its CSV import endpoint.
|
|
|
|
Auth is JWT-cookie via POST /api/v1/auth/login. Cookies are persisted
|
|
to disk so CronJob pods can reuse them across runs (Wealthfolio's
|
|
/auth/login is 5-req/min rate-limited).
|
|
|
|
Not multi-process safe — file locking is added in Phase 1 when we
|
|
fan out to multiple CronJobs.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
base_url: str,
|
|
username: str,
|
|
password: str,
|
|
session_path: Path | str,
|
|
transport: httpx.AsyncBaseTransport | None = None,
|
|
) -> None:
|
|
self._username = username
|
|
self._password = password
|
|
self._session_path = Path(session_path)
|
|
self._client = httpx.AsyncClient(
|
|
base_url=base_url.rstrip("/"),
|
|
timeout=30.0,
|
|
transport=transport,
|
|
)
|
|
|
|
async def close(self) -> None:
|
|
await self._client.aclose()
|
|
|
|
# -- session --
|
|
|
|
def _load_cookies(self) -> dict[str, str]:
|
|
if not self._session_path.exists():
|
|
return {}
|
|
raw = json.loads(self._session_path.read_text())
|
|
got = raw.get("cookies", {})
|
|
assert isinstance(got, dict)
|
|
return got
|
|
|
|
def _save_cookies(self, cookies: dict[str, str]) -> None:
|
|
self._session_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._session_path.write_text(json.dumps({"cookies": cookies}))
|
|
|
|
async def login(self) -> None:
|
|
# Wealthfolio 3.2's LoginRequest is `{ password: String }` only — a
|
|
# username key is rejected as an unknown field (HTTP 400). The
|
|
# `username` constructor arg is kept for a future Wealthfolio
|
|
# release that may add multi-user support.
|
|
resp = await self._client.post(
|
|
_LOGIN_PATH,
|
|
json={"password": self._password},
|
|
)
|
|
if resp.status_code == 401:
|
|
raise WealthfolioUnauthorizedError("Wealthfolio /auth/login returned 401")
|
|
resp.raise_for_status()
|
|
cookies = dict(resp.cookies.items())
|
|
if not cookies:
|
|
raise WealthfolioError("/auth/login returned 2xx but no Set-Cookie")
|
|
self._save_cookies(cookies)
|
|
|
|
@staticmethod
|
|
def _cookie_header(cookies: dict[str, str]) -> str:
|
|
return "; ".join(f"{k}={v}" for k, v in cookies.items())
|
|
|
|
async def _request(self, method: str, path: str, **kw: Any) -> httpx.Response:
|
|
cookies = self._load_cookies()
|
|
headers = dict(kw.pop("headers", {}) or {})
|
|
if cookies:
|
|
headers["Cookie"] = self._cookie_header(cookies)
|
|
resp = await self._client.request(method, path, headers=headers, **kw)
|
|
if resp.status_code == 401 and path != _LOGIN_PATH:
|
|
await self.login()
|
|
cookies = self._load_cookies()
|
|
headers["Cookie"] = self._cookie_header(cookies)
|
|
resp = await self._client.request(method, path, headers=headers, **kw)
|
|
return resp
|
|
|
|
# -- accounts --
|
|
|
|
async def list_accounts(self) -> list[dict[str, Any]]:
|
|
resp = await self._request("GET", _ACCOUNTS_PATH)
|
|
resp.raise_for_status()
|
|
raw = resp.json()
|
|
assert isinstance(raw, list)
|
|
return raw
|
|
|
|
async def ensure_account(self, account: Account) -> str:
|
|
"""Idempotently create the account and return Wealthfolio's UUID for it.
|
|
|
|
Wealthfolio generates its own UUIDs on POST /accounts, ignoring any
|
|
`id` we supply. We identify accounts by (provider, providerAccountId)
|
|
which Wealthfolio DOES preserve verbatim. Our own Account.id is
|
|
used as the providerAccountId.
|
|
"""
|
|
existing = await self.list_accounts()
|
|
for a in existing:
|
|
if (a.get("provider") == account.provider and a.get("providerAccountId") == account.id):
|
|
wf_id = a.get("id")
|
|
assert isinstance(wf_id, str)
|
|
return wf_id
|
|
|
|
# NewAccount is camelCase with required booleans.
|
|
# See apps/server/src/models.rs#NewAccount.
|
|
resp = await self._request(
|
|
"POST",
|
|
_ACCOUNTS_PATH,
|
|
json={
|
|
"name": account.name,
|
|
"accountType": str(account.account_type),
|
|
"currency": account.currency,
|
|
"isDefault": False,
|
|
"isActive": True,
|
|
"isArchived": False,
|
|
"trackingMode": "TRANSACTIONS",
|
|
"provider": account.provider,
|
|
"providerAccountId": account.id,
|
|
},
|
|
)
|
|
resp.raise_for_status()
|
|
created = resp.json()
|
|
wf_id = created.get("id")
|
|
if not isinstance(wf_id, str):
|
|
raise WealthfolioError(f"POST /accounts returned no id: {created}")
|
|
return wf_id
|
|
|
|
# -- activity import --
|
|
|
|
@staticmethod
|
|
def _activity_to_import_row(a: Activity) -> dict[str, Any]:
|
|
"""Match Wealthfolio's ActivityImport struct (camelCase JSON)."""
|
|
row: dict[str, Any] = {
|
|
"date": a.date.isoformat(),
|
|
"symbol": a.symbol or "$CASH",
|
|
"activityType": str(a.activity_type),
|
|
"currency": a.currency,
|
|
"accountId": a.account_id,
|
|
# Required booleans on ActivityImport (no defaults in Rust struct).
|
|
"isDraft": False,
|
|
"isValid": True,
|
|
}
|
|
if a.quantity is not None:
|
|
row["quantity"] = format(a.quantity, "f")
|
|
if a.unit_price is not None:
|
|
row["unitPrice"] = format(a.unit_price, "f")
|
|
if a.amount is not None:
|
|
row["amount"] = format(a.amount, "f")
|
|
if a.fee:
|
|
row["fee"] = format(a.fee, "f")
|
|
if a.notes:
|
|
row["comment"] = a.notes
|
|
return row
|
|
|
|
async def import_activities(self, activities: Iterable[Activity]) -> list[dict[str, Any]]:
|
|
rows = [self._activity_to_import_row(a) for a in activities]
|
|
if not rows:
|
|
return []
|
|
|
|
# Step 1 — /import/check hydrates each row with resolved asset_id,
|
|
# exchange_mic, quote_ccy, instrument_type, quote_mode (and flags
|
|
# errors). The /import endpoint on Wealthfolio 3.2+ DOES NOT
|
|
# re-resolve — if we send the un-enriched row the activity is
|
|
# silently dropped (import returns 200 OK with activities=[] in
|
|
# the payload). We must feed check's output into import.
|
|
check = await self._request("POST", _IMPORT_CHECK, json={"activities": rows})
|
|
if check.status_code >= 400:
|
|
try:
|
|
payload = check.json()
|
|
except Exception:
|
|
payload = {"raw": check.text}
|
|
raise ImportValidationError(f"Wealthfolio /import/check rejected: {payload}")
|
|
|
|
checked = check.json()
|
|
if not isinstance(checked, list):
|
|
raise ImportValidationError(
|
|
f"Wealthfolio /import/check returned non-list: {type(checked).__name__}")
|
|
|
|
invalid = [r for r in checked if isinstance(r, dict) and r.get("errors")]
|
|
if invalid:
|
|
raise ImportValidationError(f"Wealthfolio /import/check flagged {len(invalid)} row(s); "
|
|
f"first: {invalid[0]}")
|
|
# Drop any row the server marked is_valid=false (shouldn't happen
|
|
# without errors, but defensive).
|
|
valid_rows = [r for r in checked if isinstance(r, dict) and r.get("isValid")]
|
|
|
|
real = await self._request("POST", _IMPORT_REAL, json={"activities": valid_rows})
|
|
real.raise_for_status()
|
|
raw = real.json()
|
|
# Two observed response shapes:
|
|
# - {activities:[...], importRunId:"...", summary:{total,imported,skipped,...}}
|
|
# - bare list (older builds)
|
|
if isinstance(raw, dict) and "activities" in raw:
|
|
got = raw["activities"]
|
|
summary = raw.get("summary") if isinstance(raw.get("summary"), dict) else None
|
|
elif isinstance(raw, list):
|
|
got = raw
|
|
summary = None
|
|
else:
|
|
got = []
|
|
summary = None
|
|
# Summary.imported is THE truth. The `activities` field echoes input
|
|
# with errors annotated — its length equals input even when zero
|
|
# actually persisted.
|
|
if summary is not None:
|
|
imported_n = int(summary.get("imported", 0))
|
|
total_n = int(summary.get("total", len(valid_rows)))
|
|
if imported_n < total_n:
|
|
err_msg = summary.get("errorMessage") or "no errorMessage"
|
|
skipped = int(summary.get("skipped", 0))
|
|
dupes = int(summary.get("duplicates", 0))
|
|
raise ImportValidationError(
|
|
f"Wealthfolio /import persisted {imported_n}/{total_n} "
|
|
f"(skipped={skipped} duplicates={dupes}). "
|
|
f"errorMessage: {err_msg}"
|
|
)
|
|
# Legacy silent-drop guard for no-summary responses.
|
|
elif valid_rows and not got:
|
|
first_warn = next(
|
|
(r.get("warnings") for r in checked if isinstance(r, dict) and r.get("warnings")),
|
|
None,
|
|
)
|
|
raise ImportValidationError(
|
|
f"Wealthfolio /import silently dropped all {len(valid_rows)} rows. "
|
|
f"First checked row: {checked[0] if checked else 'none'}. "
|
|
f"First warning: {first_warn}"
|
|
)
|
|
return got
|