整套工具跑一遍
下面是组装好的 dtool 精简版:读、清洗、汇总、写 CSV。运行它,看报告和结果:
import csv, io
SALES = ("date,store,item,qty,price\n2026-09-01,东门,包子,10,2\n2026-09-01,东门,豆浆, 8 ,3\n2026-09-01,西门,油条,,2.5\n2026-09-01,西门,包子,10,2\n2026-09-01,东门,包子,10,2\n2026-09-02,北门,豆浆,five,3\n2026-09-02,北门,油条,4,2.5\n2026-09-02,西门,豆浆,6,3\n")
def clean(rows):
out, bad, seen = [], [], set()
for i, r in enumerate(rows, 2):
r = {k: v.strip() for k, v in r.items()}
key = tuple(r.values())
if key in seen:
continue
seen.add(key)
try:
r["qty"] = int(r["qty"] or 0)
r["price"] = float(r["price"])
except ValueError:
bad.append(i)
continue
out.append(r)
return out, bad
def by_store(rows):
t = {}
for r in rows:
t[r["store"]] = t.get(r["store"], 0) + r["qty"] * r["price"]
return t
rows = list(csv.DictReader(io.StringIO(SALES)))
good, bad = clean(rows)
t = by_store(good)
buf = io.StringIO()
w = csv.DictWriter(buf, fieldnames=["store", "total"], lineterminator="\n")
w.writeheader()
w.writerows([{"store": k, "total": int(v)} for k, v in t.items()])
print("处理" + str(len(good)) + "行,跳过" + str(len(bad)) + "行:" + ",".join(str(b) for b in bad) + "|" + str(len(buf.getvalue().strip().split("\n"))))
全部评论