catnab/scripts/vinted.py

"""Categorize Vinted transactions in YNAB from an undated, reverse-chronological item list.

Usage: uv run scripts/vinted.py vinted-transactions.csv [--dry-run] [--days 90]

The CSV must have `item`, `price` and `category` columns, ordered newest first. Since there are
no dates, each YNAB transaction (newest first) is matched to the first remaining item with the
same amount, and that item is removed from the pool.
"""
import csv
from dataclasses import dataclass
from decimal import Decimal
import os
import sys

import click
import ynab
from ynab.models.transaction_detail import TransactionDetail

from catnab.ynab_client import YnabClient


@dataclass
class VintedItem:
    name: str
    # milliunits, negative for outflows, to match YNAB
    amount: int
    category: str


def load_items(path: str) -> list[VintedItem]:
    with open(path, newline='') as f:
        return [
            VintedItem(
                name=row['item'].strip(),
                amount=-int(Decimal(row['price']) * 1000),
                category=row['category'].strip(),
            )
            for row in csv.DictReader(f)
        ]


def is_vinted(tx: TransactionDetail) -> bool:
    return tx.payee_name is not None and 'vinted' in tx.payee_name.lower() and not tx.deleted


def is_updatable(tx: TransactionDetail) -> bool:
    # same guards as the Amazon classifier: don't touch anything already handled by us or the user
    return (
        tx.flag_color != 'purple'
        and not tx.approved
        and len(tx.subtransactions) == 0
    )


def match_items(
    txs: list[TransactionDetail],
    items: list[VintedItem],
) -> tuple[list[tuple[TransactionDetail, VintedItem]], list[TransactionDetail], list[VintedItem]]:
    """Returns (matches, unmatched transactions, leftover items)."""
    remaining = list(items)
    matches = []
    unmatched = []
    for tx in sorted(txs, key=lambda t: t.var_date, reverse=True):
        for i, item in enumerate(remaining):
            if item.amount == tx.amount:
                matches.append((tx, remaining.pop(i)))
                break
        else:
            unmatched.append(tx)
    return matches, unmatched, remaining


def resolve_category(name: str, category_ids: dict[str, str]) -> str | None:
    """Find a YNAB category id by `Group: Name` or by bare name (case-insensitive)."""
    name = name.lower()
    candidates = [
        cid for full, cid in category_ids.items()
        if full.lower() == name or full.split(': ', 1)[-1].lower() == name
    ]
    return candidates[0] if len(candidates) == 1 else None


def fmt_amount(milliunits: int) -> str:
    return f'${abs(milliunits) / 1000:.2f}'


@click.command
@click.argument('csv_path', type=click.Path(exists=True, dir_okay=False))
@click.option('--days', type=int, default=90, help='How far back to look for YNAB transactions.')
@click.option('--dry-run', is_flag=True, help="Show what would change without updating YNAB.")
def main(csv_path: str, days: int, dry_run: bool):
    """Categorize Vinted transactions in YNAB from a list of purchased items."""
    client = YnabClient(
        budget_id=os.environ['YNAB_BUDGET_ID'],
        ynab_token=os.environ['YNAB_TOKEN'],
        sync_days=days,
    )
    items = load_items(csv_path)
    print(f'Loaded {len(items)} items from {csv_path}')

    print('Fetching YNAB transactions...')
    # Match against every Vinted transaction, including ones we won't update, so that an
    # already-approved transaction still consumes its item and doesn't shift later matches.
    txs = [tx for tx in client.history if is_vinted(tx)]
    print(f'Found {len(txs)} Vinted transactions in the last {days} days')

    matches, unmatched, leftover = match_items(txs, items)

    updates = []
    bad_categories = set()
    print('\nMatches:')
    for tx, item in matches:
        line = f'  {tx.var_date}  {fmt_amount(tx.amount):>8}  {item.name}'
        if not is_updatable(tx):
            print(f'{line}  (skipped: already approved/flagged/split)')
            continue

        category_id = None
        if item.category:
            category_id = resolve_category(item.category, client.category_ids)
            if category_id is None:
                bad_categories.add(item.category)
                print(f'{line}  -> ??? "{item.category}" not found or ambiguous')
                continue
        print(f'{line}  -> {item.category or "(no category, memo only)"}')

        updates.append(ynab.SaveTransactionWithIdOrImportId(
            id=tx.id,
            category_id=category_id,
            memo=item.name,
            flag_color=ynab.TransactionFlagColor.PURPLE,
        ))

    if unmatched:
        print('\nYNAB transactions with no matching item:')
        for tx in unmatched:
            print(f'  {tx.var_date}  {fmt_amount(tx.amount):>8}  {tx.payee_name}')

    if leftover:
        print('\nItems with no matching YNAB transaction:')
        for item in leftover:
            print(f'  {fmt_amount(item.amount):>8}  {item.name}')

    if bad_categories:
        print('\nUnresolved categories (fix the CSV, or use the full "Group: Name"):')
        for c in sorted(bad_categories):
            print(f'  {c}')

    print(f'\n{len(updates)} transactions to update.')
    if dry_run:
        print('Dry run, not updating YNAB.')
        return
    if bad_categories:
        print('Refusing to update while categories are unresolved.')
        sys.exit(1)
    if updates:
        print('Updating YNAB...')
        client.update_transactions(updates)
        print('Done.')


if __name__ == '__main__':
    main()