054、CSV 与 JSON 处理：csv 模块高级用法、json 序列化自定义对象

054、CSV 与 JSON 处理csv 模块高级用法、json 序列化自定义对象上周五晚上十一点我盯着屏幕上的报错信息发呆——一个从生产环境导出的CSV文件明明用Excel打开一切正常Python的csv.reader却死活解析出乱码。更诡异的是同一个脚本在测试环境跑得好好的到了线上就崩。排查了半小时发现是某条数据里藏了个肉眼不可见的BOM头加上某个字段里嵌了换行符直接把行解析逻辑干碎了。这种坑踩过一次就忘不掉。今天就把CSV和JSON处理里那些“文档不会告诉你”的细节掰开揉碎。CSV模块别被简单外表骗了很多人觉得csv模块就是reader和writer两个函数的事真遇到复杂场景就抓瞎。先看一个典型的生产环境数据importcsv# 别这样写——默认行为会吞掉引号内的换行withopen(data.csv,r)asf:readercsv.reader(f)forrowinreader:print(row)# 如果某字段包含换行这里就裂开了正确的姿势是显式指定quoting参数。我习惯用csv.QUOTE_ALL或csv.QUOTE_NONNUMERIC但更关键的是处理那些“看起来像CSV但又不是标准CSV”的脏数据。Dialect参数是救命稻草。有一次对接第三方系统对方输出的CSV用分号分隔字段用双引号包裹但引号内居然还有未转义的双引号。标准csv模块直接报错这时候需要自定义DialectclassCustomDialect(csv.Dialect):delimiter;quotechardoublequoteFalse# 这里踩过坑对方没按标准转义escapechar\\lineterminator\nquotingcsv.QUOTE_ALLwithopen(messy.csv,r)asf:readercsv.reader(f,dialectCustomDialect)BOM头处理——这个坑我至少见过三次。Windows生成的CSV文件经常带BOM头\ufeffPython的csv模块不会自动处理。解决方案withopen(data.csv,r,encodingutf-8-sig)asf:# utf-8-sig会自动吃掉BOM头readercsv.DictReader(f)大文件分块写入。有次需要生成500万行的CSV直接一次性写入内存炸了。正确做法是用writerows配合生成器defgenerate_rows():foriinrange(5000000):yield[i,fdata_{i},i*1.5]withopen(big.csv,w,newline)asf:writercsv.writer(f)writer.writerow([id,name,value])writer.writerows(generate_rows())# 生成器不会一次性加载所有数据JSON序列化自定义对象才是硬骨头json.dumps处理字典和列表没问题但遇到自定义类就罢工。最常见的场景是把ORM模型或配置对象转成JSON。基础方案实现__dict__或__slots__。但别直接用obj.__dict__——它会把私有属性也暴露出来而且嵌套对象会递归报错。classUser:def__init__(self,name,age,password):self.namename self.ageage self._passwordpassword# 不想暴露这个defto_dict(self):return{name:self.name,age:self.age}# 然后 json.dumps(user, defaultlambda o: o.to_dict())更优雅的方案自定义JSONEncoder。我项目里常用这个模式importjsonfromdatetimeimportdatetimeclassCustomEncoder(json.JSONEncoder):defdefault(self,obj):ifisinstance(obj,datetime):returnobj.isoformat()ifhasattr(obj,to_dict):returnobj.to_dict()# 这里踩过坑忘记处理可迭代对象ifhasattr(obj,__iter__):returnlist(obj)returnsuper().default(obj)# 使用data{user:user_obj,created_at:datetime.now()}json.dumps(data,clsCustomEncoder,ensure_asciiFalse)反序列化自定义对象——这个更麻烦。json.loads只能返回基本类型要恢复成自定义类需要自己写hookdefdecode_user(dct):ifnameindctandageindct:returnUser(dct[name],dct[age])returndct user_json{name: Alice, age: 30}userjson.loads(user_json,object_hookdecode_user)注意object_hook是递归调用的嵌套字典也会被处理。但有个坑——如果字典里同时包含name和age但又不是User对象这个逻辑会误判。我一般会在类里加个__type__字段做标记。混合场景CSV里藏JSON真实项目里经常遇到CSV的某个字段是JSON字符串。比如导出用户行为日志一列是extra_info里面存着嵌套的JSON。importcsvimportjsonwithopen(logs.csv,r)asf:readercsv.DictReader(f)forrowinreader:try:row[extra_info]json.loads(row[extra_info])exceptjson.JSONDecodeError:# 别直接抛异常记录日志继续处理print(f行{reader.line_num}的extra_info解析失败)row[extra_info]{}反过来写入时要把JSON转成字符串withopen(output.csv,w,newline)asf:writercsv.DictWriter(f,fieldnames[id,info])writer.writeheader()foritemindata:item[info]json.dumps(item[info],ensure_asciiFalse)writer.writerow(item)性能调优别让IO成为瓶颈处理大文件时有几个容易被忽略的点CSV读取用csv.DictReader还是csv.reader如果字段名固定且列数多DictReader更可读但每次访问字段都要做字典查找性能比按索引访问慢20%左右。追求极致性能时用reader配合namedtuplefromcollectionsimportnamedtuple Rownamedtuple(Row,[id,name,value])withopen(data.csv,r)asf:readercsv.reader(f)next(reader)# 跳过表头forrowinreader:rRow(*row)# 用r.name访问比row[1]可读比dict快JSON序列化大对象——如果对象有循环引用json.dumps会直接炸。用json.dumps的default参数处理不了循环引用需要自己写序列化逻辑或者用第三方库如jsonpickle。但我更推荐重构数据结构避免循环引用——这通常是设计问题。个人经验永远不要信任输入数据的编码。CSV文件可能是GBK、UTF-8、UTF-8 with BOM甚至混合编码。我习惯先检测编码chardet.detect()或者直接让用户指定编码参数。CSV的newline参数必须加。不加的话Windows上会多出空行这个坑在官方文档里写了但很多人没注意。JSON序列化自定义对象时别用__dict__。它会把所有属性暴露包括私有属性、方法、甚至__class__。用vars()也一样。老老实实写to_dict方法或者用dataclasses的asdict。生产环境永远不要用json.loads直接解析用户输入。如果数据量不可控用ijson做流式解析防止内存溢出。最后一条CSV和JSON只是数据载体不是数据契约。我见过太多因为字段顺序变化、引号规则不同导致的线上事故。如果可能用csv.DictReader和json.dumps的sort_keysTrue让输出可预测。写这篇笔记的时候我正盯着一个因为CSV字段里包含逗号而导致的解析错误。对方说“Excel能打开啊”我说“Excel能打开不代表Python能正确解析”。这就是CSV的魔幻现实——看起来简单坑起来要命。