add vinted-transaction script

.gitignore

@@ -13,3 +13,5 @@ wheels/
 
 **/output/
 **/.cookies/
+
+vinted-transactions.csv

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()