Converting Amazon item CSV to Alfred JSON

f.py

"""
Total pile of general hacks to take a CSV from stdin and turn it into some
Alfred compatible JSON.

Copyright 2022-present Joshua Kehn
License: MIT

Don't @ me with problems.

Field orders (if you were curious to expand):

0: Order Date
1: Order ID
2: Title
3: Category
4: ASIN/ISBN
5: UNSPSC Code
6: Website
7: Release Date
8: Condition
9: Seller
10: Seller Credentials
11: List Price Per Unit
12: Purchase Price Per Unit
13: Quantity
14: Payment Instrument Type
15: Purchase Order Number
16: PO Line Number
17: Ordering Customer Email
18: Shipment Date
19: Shipping Address Name
20: Shipping Address Street 1
21: Shipping Address Street 2
22: Shipping Address City
23: Shipping Address State
24: Shipping Address Zip
25: Order Status
26: Carrier Name & Tracking Number
27: Item Subtotal
28: Item Subtotal Tax
29: Item Total
30: Tax Exemption Applied
31: Tax Exemption Type
32: Exemption Opt-Out
33: Buyer Name
34: Currency
35: Group Name

"""


import csv
import sys
from datetime import datetime
from typing import Dict
import json


class Record:
    def __init__(
        self,
        order_date,
        order_id,
        title,
        category,
        asin,
        unspsc,
        per_unit_price,
        quantity,
        item_total,
    ) -> None:
        self.order_date = datetime.strptime(order_date, "%m/%d/%y")
        self.order_id = order_id
        self.title = title
        self.category = category
        self.asin = asin
        self.unspsc = unspsc
        self.quantity = quantity
        self.per_unit_price = per_unit_price
        self.item_total = item_total

    def nice_date(self) -> str:
        return self.order_date.strftime("%Y-%m-%d")

    def nice_price(self) -> str:
        if self.quantity != "1":
            return f"{self.per_unit_price} ea ({self.item_total} for {self.quantity})"
        return self.item_total
        return f"${self.item_total:.2f}"


reader = csv.reader(sys.stdin)
discovered_items = {}

total = 0
for row in reader:
    items = len(row)
    if items != 36:
        sys.stderr.write(f"ERR: {items}: {row}\n")
        continue

    if row[0] == "Order Date":
        continue

    try:
        total += float(row[29][1:])
    except Exception as e:
        sys.stderr.write(f"{e}\n")
        pass

    asin = row[4]
    if not asin:
        sys.stderr.write(f"ERR: No ASIN/ISBN found for {row}\n")
        continue

    rec = Record(
        row[0],  # Order Date
        row[1],  # Order ID
        row[2],  # Title
        row[3],  # Category
        row[4],  # ASIN/ISBN
        row[5],  # UNSPSC Code
        # row[11], # List price per unit
        row[12],  # Purchase price per unit
        row[13],  # Quantity
        # row[27], # Item subtotal
        row[29],  # Item Total
    )

    history = discovered_items.get(asin)
    if history is None:
        history = [rec]
        discovered_items[asin] = history
    else:
        history.append(rec)

sys.stderr.write(f"LIFETIME TOTAL: ${total:,.2f}\n")


def records_to_json(recs) -> Dict:
    last = recs[-1]
    item = {
        "uid": last.asin,
        "title": last.title,
        "subtitle": f"Last purchased {last.nice_date()} for {last.nice_price()}",
        "arg": f"https://www.amazon.com/gp/product/{last.asin}",
        "mods": {
            "cmd": {
                "valid": True,
                "arg": f"https://www.amazon.com/gp/your-account/order-details/?ie=UTF8&orderID={last.order_id}",
                "subtitle": f"Open order {last.order_id}",
            }
        },
        "text": {
            "copy": f"https://www.amazon.com/gp/product/{last.asin}",
        },
    }
    return item


print(
    json.dumps(
        {
            "items": [
                records_to_json(recs)
                for recs in sorted(
                    discovered_items.values(),
                    key=lambda r: r[-1].order_date,
                    reverse=True,
                )
            ]
        }
    )
)