267 lines
9.1 KiB
Python
267 lines
9.1 KiB
Python
import yaml
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import TypedDict
|
|
|
|
import ynab
|
|
|
|
|
|
class YNABTransaction(TypedDict):
|
|
"""Transaction data from YNAB."""
|
|
|
|
id: str
|
|
date: str
|
|
payee_name: str | None
|
|
category_name: str | None
|
|
amount_milliCurrency: int
|
|
memo: str | None
|
|
account_id: str
|
|
account_name: str
|
|
cleared: str
|
|
import_id: str | None
|
|
|
|
|
|
class YNABConfig(TypedDict):
|
|
"""Configuration loaded from ynab.yml."""
|
|
|
|
token: str
|
|
plan_id: str
|
|
accounts: list[dict[str, str]]
|
|
|
|
|
|
class YNABClient:
|
|
"""Client for fetching transactions from YNAB."""
|
|
|
|
def __init__(self, config_path: str | Path = "ynab.yml") -> None:
|
|
"""Initialize YNAB client from config file."""
|
|
self.config = self._load_config(config_path)
|
|
self.token = self.config["token"]
|
|
self.plan_id = self.config["plan_id"]
|
|
self.account_configs = self.config["accounts"]
|
|
|
|
# Use SDK-native access_token auth for bearer token handling.
|
|
config = ynab.Configuration(access_token=self.token)
|
|
self.api_client = ynab.ApiClient(config)
|
|
self.transactions_api = ynab.TransactionsApi(self.api_client)
|
|
self.accounts_api = ynab.AccountsApi(self.api_client)
|
|
self.categories_api = ynab.CategoriesApi(self.api_client)
|
|
|
|
@staticmethod
|
|
def _load_config(config_path: str | Path) -> YNABConfig:
|
|
"""Load YNAB configuration from YAML file."""
|
|
path = Path(config_path)
|
|
with path.open("r", encoding="utf-8") as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
ynab_config = data.get("ynab", [{}])[0]
|
|
return {
|
|
"token": ynab_config.get("token", ""),
|
|
"plan_id": ynab_config.get("plan_id", ""),
|
|
"accounts": ynab_config.get("accounts", []),
|
|
}
|
|
|
|
@staticmethod
|
|
def _normalize_cleared_status(cleared: object) -> str:
|
|
"""Normalize YNAB cleared status values to lowercase strings."""
|
|
if hasattr(cleared, "value"):
|
|
return str(getattr(cleared, "value")).strip().lower()
|
|
|
|
status = str(cleared).strip()
|
|
if "." in status:
|
|
status = status.split(".")[-1]
|
|
return status.lower()
|
|
|
|
def get_transactions_for_days(
|
|
self,
|
|
days: int = 30,
|
|
exclude_cleared_statuses: list[str] | None = None,
|
|
) -> list[YNABTransaction]:
|
|
"""
|
|
Fetch transactions from configured accounts for the past x days.
|
|
|
|
Args:
|
|
days: Number of days to look back (default: 30)
|
|
exclude_cleared_statuses: Optional list of normalized statuses to skip,
|
|
for example ["reconciled"]
|
|
|
|
Returns:
|
|
List of transactions with transformed data
|
|
"""
|
|
transactions: list[YNABTransaction] = []
|
|
cutoff_date = (datetime.now() - timedelta(days=days)).date()
|
|
excluded_statuses = {
|
|
status.strip().lower() for status in (exclude_cleared_statuses or [])
|
|
}
|
|
|
|
# Fetch all accounts to build name lookup
|
|
accounts_response = self.accounts_api.get_accounts(self.plan_id)
|
|
account_map = {
|
|
str(acc.id): acc.name for acc in accounts_response.data.accounts
|
|
}
|
|
|
|
# Fetch categories for name lookup
|
|
categories_response = self.categories_api.get_categories(self.plan_id)
|
|
category_map = {}
|
|
for category_group in categories_response.data.category_groups:
|
|
for cat in category_group.categories:
|
|
if cat.id:
|
|
category_map[str(cat.id)] = cat.name
|
|
|
|
# If specific accounts are configured, use only those; otherwise use all
|
|
if self.account_configs:
|
|
account_ids = [acc["id"] for acc in self.account_configs]
|
|
else:
|
|
account_ids = list(account_map.keys())
|
|
|
|
# Fetch transactions for each account
|
|
for account_id in account_ids:
|
|
account_name = account_map.get(account_id, account_id)
|
|
|
|
try:
|
|
txn_response = self.transactions_api.get_transactions_by_account(
|
|
plan_id=self.plan_id,
|
|
account_id=account_id,
|
|
since_date=cutoff_date,
|
|
)
|
|
|
|
for txn in txn_response.data.transactions:
|
|
txn_date = txn.var_date
|
|
if txn_date < cutoff_date:
|
|
continue
|
|
|
|
cleared_status = self._normalize_cleared_status(txn.cleared)
|
|
if cleared_status in excluded_statuses:
|
|
continue
|
|
|
|
category_name = None
|
|
if txn.category_id:
|
|
category_name = category_map.get(str(txn.category_id))
|
|
|
|
processed: YNABTransaction = {
|
|
"id": txn.id,
|
|
"date": str(txn.var_date),
|
|
"payee_name": txn.payee_name,
|
|
"category_name": category_name,
|
|
"amount_milliCurrency": txn.amount,
|
|
"memo": txn.memo,
|
|
"account_id": account_id,
|
|
"account_name": account_name,
|
|
"cleared": cleared_status,
|
|
"import_id": txn.import_id,
|
|
}
|
|
transactions.append(processed)
|
|
|
|
except Exception as e:
|
|
print(f"Error fetching transactions for account {account_name}: {e}")
|
|
continue
|
|
|
|
return transactions
|
|
|
|
def get_account_balance(self, account_id: str) -> int:
|
|
"""
|
|
Get current balance for an account (in milliCurrency units).
|
|
|
|
Args:
|
|
account_id: YNAB account ID
|
|
|
|
Returns:
|
|
Account balance in milliCurrency
|
|
"""
|
|
accounts_response = self.accounts_api.get_accounts(self.plan_id)
|
|
for account in accounts_response.data.accounts:
|
|
if str(account.id) == account_id:
|
|
return account.balance
|
|
raise ValueError(f"Account {account_id} not found")
|
|
|
|
def get_last_reconciled_date(
|
|
self,
|
|
account_ids: list[str] | None = None,
|
|
search_days: int = 90,
|
|
) -> datetime | None:
|
|
"""
|
|
Find the date of the most recent reconciled transaction across accounts.
|
|
|
|
Args:
|
|
account_ids: Account IDs to search. If None, uses all configured accounts.
|
|
search_days: How far back to search for reconciled transactions (default 90).
|
|
|
|
Returns:
|
|
The date of the last reconciled transaction, or None if none found.
|
|
"""
|
|
from datetime import date as date_type
|
|
|
|
cutoff_date = (datetime.now() - timedelta(days=search_days)).date()
|
|
|
|
if account_ids is None:
|
|
account_ids = [acc["id"] for acc in self.account_configs]
|
|
|
|
latest_date: date_type | None = None
|
|
|
|
for account_id in account_ids:
|
|
try:
|
|
txn_response = self.transactions_api.get_transactions_by_account(
|
|
plan_id=self.plan_id,
|
|
account_id=account_id,
|
|
since_date=cutoff_date,
|
|
)
|
|
for txn in txn_response.data.transactions:
|
|
status = self._normalize_cleared_status(txn.cleared)
|
|
if status == "reconciled":
|
|
txn_date = txn.var_date
|
|
if latest_date is None or txn_date > latest_date:
|
|
latest_date = txn_date
|
|
except Exception as e:
|
|
print(f"Error searching reconciled transactions for {account_id}: {e}")
|
|
continue
|
|
|
|
return latest_date
|
|
|
|
def create_transactions(
|
|
self,
|
|
transactions: list[ynab.NewTransaction],
|
|
) -> ynab.SaveTransactionsResponseData:
|
|
"""
|
|
Create multiple transactions in YNAB.
|
|
|
|
Args:
|
|
transactions: List of NewTransaction objects to create.
|
|
|
|
Returns:
|
|
SaveTransactionsResponseData with created IDs and duplicate info.
|
|
"""
|
|
wrapper = ynab.PostTransactionsWrapper(transactions=transactions)
|
|
response = self.transactions_api.create_transaction(
|
|
plan_id=self.plan_id,
|
|
data=wrapper,
|
|
)
|
|
return response.data
|
|
|
|
def update_transaction(
|
|
self,
|
|
transaction_id: str,
|
|
cleared: str | None = None,
|
|
flag_color: str | None = None,
|
|
) -> None:
|
|
"""
|
|
Update an existing YNAB transaction.
|
|
|
|
Args:
|
|
transaction_id: The YNAB transaction ID to update.
|
|
cleared: New cleared status ('cleared', 'uncleared', 'reconciled').
|
|
flag_color: New flag color (e.g. 'red', 'blue').
|
|
"""
|
|
fields: dict = {}
|
|
if cleared is not None:
|
|
fields["cleared"] = ynab.TransactionClearedStatus(cleared)
|
|
if flag_color is not None:
|
|
fields["flag_color"] = ynab.TransactionFlagColor(flag_color)
|
|
|
|
existing = ynab.ExistingTransaction(**fields)
|
|
wrapper = ynab.PutTransactionWrapper(transaction=existing)
|
|
self.transactions_api.update_transaction(
|
|
plan_id=self.plan_id,
|
|
transaction_id=transaction_id,
|
|
data=wrapper,
|
|
)
|
|
|