The Date Column Was the Most Expensive Part of My Sales CSV — Measuring What Attachments Really Cost

#!/usr/bin/env python3
# attach_cost.py — price each rendering of the same table in tokens
import csv, io, json, argparse
import tiktoken
 
ENC = tiktoken.get_encoding("o200k_base")
def ntok(s): return len(ENC.encode(s))
 
def load(path):
    with open(path, newline="", encoding="utf-8") as f:
        r = csv.DictReader(f)
        return r.fieldnames, list(r)
 
def as_csv(cols, rows):
    buf = io.StringIO()
    w = csv.DictWriter(buf, fieldnames=cols, lineterminator="\n")
    w.writeheader(); w.writerows(rows)
    return buf.getvalue()
 
def as_tsv(cols, rows):
    out = ["\t".join(cols)]
    out += ["\t".join(str(r[c]) for c in cols) for r in rows]
    return "\n".join(out) + "\n"
 
def as_json_pretty(cols, rows):
    return json.dumps([{c: r[c] for c in cols} for r in rows],
                      ensure_ascii=False, indent=2)
 
def as_json_compact(cols, rows):
    return json.dumps([{c: r[c] for c in cols} for r in rows],
                      ensure_ascii=False, separators=(",", ":"))
 
def as_json_split(cols, rows):
    # column names written once; values live in an array of arrays
    return json.dumps({"columns": cols,
                       "data": [[r[c] for c in cols] for r in rows]},
                      ensure_ascii=False, separators=(",", ":"))
 
def as_md_padded(cols, rows):
    w = [max(len(c), *(len(str(r[c])) for r in rows)) for c in cols]
    line = lambda cells: "| " + " | ".join(
        str(v).ljust(w[i]) for i, v in enumerate(cells)) + " |"
    out = [line(cols), "|" + "|".join("-" * (x + 2) for x in w) + "|"]
    out += [line([r[c] for c in cols]) for r in rows]
    return "\n".join(out) + "\n"
 
def as_md_tight(cols, rows):
    out = ["|" + "|".join(cols) + "|", "|" + "|".join("-" for _ in cols) + "|"]
    out += ["|" + "|".join(str(r[c]) for c in cols) + "|" for r in rows]
    return "\n".join(out) + "\n"
 
def as_yaml(cols, rows):
    out = []
    for r in rows:
        out.append("- " + f"{cols[0]}: {r[cols[0]]}")
        for c in cols[1:]:
            out.append(f"  {c}: {r[c]}")
    return "\n".join(out) + "\n"
 
RENDERERS = [
    ("CSV",                   as_csv),
    ("TSV",                   as_tsv),
    ("Markdown (padded)",     as_md_padded),
    ("Markdown (tight)",      as_md_tight),
    ("JSON records indent=2", as_json_pretty),
    ("JSON records compact",  as_json_compact),
    ("JSON columnar",         as_json_split),
    ("YAML-like",             as_yaml),
]
 
def report(cols, rows):
    base = None
    print(f"{'format':<24}{'bytes':>10}{'tokens':>10}{'ratio':>8}{'tok/row':>9}")
    for name, fn in RENDERERS:
        s = fn(cols, rows)
        t = ntok(s)
        if base is None:
            base = t
        print(f"{name:<24}{len(s.encode()):>10}{t:>10}"
              f"{t / base:>7.2f}x{t / len(rows):>9.1f}")
 
if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("csv_path")
    ap.add_argument("--rows", type=int, default=0)
    a = ap.parse_args()
    cols, rows = load(a.csv_path)
    if a.rows:
        rows = rows[:a.rows]
    report(cols, rows)

#!/usr/bin/env python3
# col_cost.py — price each column by removing it and measuring the delta
import csv, io, sys
import tiktoken
 
ENC = tiktoken.get_encoding("o200k_base")
def ntok(s): return len(ENC.encode(s))
 
def render(cols, rows):
    buf = io.StringIO()
    # without extrasaction="ignore" the dropped key raises on every row
    w = csv.DictWriter(buf, fieldnames=cols,
                       extrasaction="ignore", lineterminator="\n")
    w.writeheader(); w.writerows(rows)
    return buf.getvalue()
 
def main(path):
    with open(path, newline="", encoding="utf-8") as f:
        r = csv.DictReader(f)
        cols, rows = r.fieldnames, list(r)
    full = ntok(render(cols, rows))
    print(f"total: {full} tokens / {len(rows)} rows")
    print(f"{'column':<22}{'delta':>12}{'share':>9}{'tok/cell':>10}")
    for c in cols:
        rest = [x for x in cols if x != c]
        d = full - ntok(render(rest, rows))
        print(f"{c:<22}{d:>12}{d / full * 100:>8.1f}%{d / len(rows):>10.2f}")
 
if __name__ == "__main__":
    main(sys.argv[1])

import csv, datetime, json
 
d0 = min(datetime.date.fromisoformat(r["date"]) for r in rows)
 
# 1) date -> day offset, with a legend line so it stays reversible
rows_day = [dict(r, day=(datetime.date.fromisoformat(r["date"]) - d0).days)
            for r in rows]
legend_day = f"# day: days elapsed since {d0.isoformat()}\n"
 
# 2) money -> two decimals
rows_round = [dict(r, developer_proceeds=f'{float(r["developer_proceeds"]):.2f}')
              for r in rows]
 
# 3) product_name -> legend (it is determined by the sku, so drop it from the body)
legend_sku = {r["product_sku"]: r["product_name"] for r in rows}
legend_line = "# product_sku: " + json.dumps(legend_sku, ensure_ascii=False) + "\n"

#!/usr/bin/env python3
# digest.py — describe a table instead of shipping it
import csv, io, re, sys, argparse, statistics
from collections import Counter, defaultdict
import tiktoken
 
ENC = tiktoken.get_encoding("o200k_base")
DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
 
def is_num(v):
    try:
        float(v); return True
    except (TypeError, ValueError):
        return False
 
def qs(nums):
    nums = sorted(nums)
    q = lambda p: nums[min(len(nums) - 1, int(len(nums) * p))]
    return nums[0], q(.5), q(.9), nums[-1]
 
def describe_num(name, vals, unit=None):
    nums = [float(v) for v in vals if is_num(v)]
    if not nums:
        return f"- {name}: no numeric values"
    lo, p50, p90, hi = qs(nums)
    u = f" [{unit}]" if unit else ""
    return (f"- {name}{u}: n {len(nums)} / min {lo:g} p50 {p50:g} "
            f"p90 {p90:g} max {hi:g} / sum {sum(nums):.6g} "
            f"mean {statistics.mean(nums):.4g}")
 
def build(cols, rows, unit_map, sample=8, topn=5):
    out = [f"# table digest / rows {len(rows)} / cols {len(cols)}", "", "## columns"]
    for c in cols:
        vals = [r[c] for r in rows]
        uniq = sorted(set(vals))
        blank = sum(1 for v in vals if v == "")
 
        if vals and all(DATE.match(v) for v in vals if v):
            # dates deserve a range, not a frequency ranking
            out.append(f"- {c}: date / blank {blank} / "
                       f"{min(uniq)} .. {max(uniq)} / distinct days {len(uniq)}")
 
        elif c in unit_map:
            # a column with mixed units produces nonsense statistics unless split
            ukey = unit_map[c]
            out.append(f"- {c}: numeric (units vary by {ukey} -> grouped) "
                       f"/ blank {blank}")
            groups = defaultdict(list)
            for r in rows:
                groups[r[ukey]].append(r[c])
            for k in sorted(groups, key=lambda k: -len(groups[k])):
                out.append("  " + describe_num(c, groups[k], unit=k))
 
        elif len(uniq) > 12 and all(is_num(v) for v in vals if v):
            out.append(describe_num(c, vals))
 
        else:
            top = Counter(vals).most_common(topn)
            body = " ".join(f"{k}({v})" for k, v in top)
            more = f" ... {len(uniq) - len(top)} more" if len(uniq) > len(top) else ""
            out.append(f"- {c}: categorical / blank {blank} / distinct {len(uniq)} / "
                       f"top {body}{more}")
 
    out += ["", f"## first {sample} rows (verbatim)", "```csv"]
    b = io.StringIO()
    w = csv.DictWriter(b, fieldnames=cols, lineterminator="\n")
    w.writeheader(); w.writerows(rows[:sample])
    out += [b.getvalue().rstrip(), "```"]
    return "\n".join(out) + "\n"
 
if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("csv_path")
    ap.add_argument("--unit-of", action="append", default=[], metavar="VALUECOL=UNITCOL")
    a = ap.parse_args()
    unit_map = dict(x.split("=", 1) for x in a.unit_of)
 
    with open(a.csv_path, newline="", encoding="utf-8") as f:
        r = csv.DictReader(f)
        cols, rows = r.fieldnames, list(r)
 
    d = build(cols, rows, unit_map)
    b = io.StringIO()
    w = csv.DictWriter(b, fieldnames=cols, lineterminator="\n")
    w.writeheader(); w.writerows(rows)
    sys.stderr.write(f"[full] {len(ENC.encode(b.getvalue()))} tok / "
                     f"[digest] {len(ENC.encode(d))} tok\n")
    print(d)