売上CSVを添付したら、いちばん高くついていたのは日付の列でした — 添付形式のトークン単価を測る

#!/usr/bin/env python3
# attach_cost.py — 添付する表データを、形式ごとにトークン単価で比べる
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):
    # 列名を1回だけ書き、値は配列の配列で持つ
    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 — 1列ずつ抜いて、その列が何トークン分を占めているかを差分で測る
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()
    # extrasaction="ignore" を付けないと、抜いた列が残っている行で例外になる
    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"全体: {full} tokens / {len(rows)} 行")
    print(f"{'列名':<22}{'欠くと減る量':>12}{'占有率':>9}{'tok/セル':>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) 日付 -> 起点からの通日。凡例を1行添えて可逆にしておく
rows_day = [dict(r, day=(datetime.date.fromisoformat(r["date"]) - d0).days)
            for r in rows]
legend_day = f"# day: {d0.isoformat()} を 0 とした経過日数\n"
 
# 2) 金額 -> 小数2桁
rows_round = [dict(r, developer_proceeds=f'{float(r["developer_proceeds"]):.2f}')
              for r in rows]
 
# 3) product_name -> 凡例へ外出し（sku から一意に決まるので本体からは外せる）
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 — 表データを「そのまま渡す」代わりに、輪郭だけを渡すための要約を作る
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}: 数値なし"
    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(nums):.6g} "
            f"平均 {statistics.mean(nums):.4g}")
 
def build(cols, rows, unit_map, sample=8, topn=5):
    out = [f"# 表の要約 / 行数 {len(rows)} / 列数 {len(cols)}", "", "## 列ごとの輪郭"]
    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):
            # 日付は上位頻度ではなく範囲で示す（頻度を並べても意味がない）
            out.append(f"- {c}: 日付 / 欠損 {blank} / "
                       f"{min(uniq)} 〜 {max(uniq)} / 実日数 {len(uniq)}")
 
        elif c in unit_map:
            # 単位が混じる列は、単位ごとに分けないと統計が嘘になる
            ukey = unit_map[c]
            out.append(f"- {c}: 数値（単位列 {ukey} により混在 → 単位ごとに集計）"
                       f"/ 欠損 {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)} 種" if len(uniq) > len(top) else ""
            out.append(f"- {c}: カテゴリ / 欠損 {blank} / 一意 {len(uniq)} / "
                       f"上位 {body}{more}")
 
    out += ["", f"## 先頭 {sample} 行（そのままの形）", "```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="値列=単位列")
    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"[全量] {len(ENC.encode(b.getvalue()))} tok / "
                     f"[要約] {len(ENC.encode(d))} tok\n")
    print(d)