From d023805bf13963dfbbc63f53e8d0658708ce0139 Mon Sep 17 00:00:00 2001 From: Joseph Montanaro Date: Wed, 28 Oct 2026 10:15:30 -0400 Subject: [PATCH] add vinted-transaction script --- .gitignore | 2 + scripts/vinted.py | 163 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 165 insertions(+) create mode 100644 scripts/vinted.py diff --git a/.gitignore b/.gitignore index 8a4b765..09104ee 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,5 @@ wheels/ **/output/ **/.cookies/ + +vinted-transactions.csv diff --git a/scripts/vinted.py b/scripts/vinted.py new file mode 100644 index 0000000..5c2ba14 --- /dev/null +++ b/scripts/vinted.py @@ -0,0 +1,163 @@ +"""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()