简介这是一套Python批量处理JSON转TXT的小工具面向需要在Python中频繁进行数据格式转换的开发者和数据分析人员解决将多个JSON文件合并为单一TXT文件时的重复劳动。脚本基于内置json库实现只需输入文件夹路径即可批量读取多个JSON文件按需提取指定字段后汇总写入同一个TXT文件代码结构清晰修改少数参数就能适配不同需求也可在此基础上增加错误处理、日志输出等能力适合日常数据清洗与归档场景。压缩包容量仅2KB包含2个Python脚本分别承担主功能与JSON加载辅助轻量易用可快速集成到现有项目中。资源已有2494人浏览学习说明其简便方案获得了不少使用者认可对入门Python文件处理或希望简化重复性转换工作的读者来说是一份可直接运行的实用模板便于理解批处理思路并拓展更多数据转换功能。1. 批处理 JSON 转 TXT先想清楚一个 TXT长什么样把几十个 JSON 文件合并成一个 TXT这类需求在数据导出、日志归档、给下游模型或数据库灌数据时非常常见。很多人第一反应是写个循环逐个读取再追加写入但真正决定脚本能不能复用的是转换之外那三个问题源文件是单对象还是 JSON 数组目标 TXT 是每行一条记录还是自由文本字段全部保留还是只抽业务字段。这三个问题不先答完写出来的批处理脚本换一批文件就得重改。本文用 Python 标准库 json 加 glob 完成整套方案不引入第三方依赖从最小脚本讲到嵌套取值、编码兜底和大文件流式解析最后封装成带参数的命令行工具并把输出校验写进流程。新手能照着改路径直接跑熟手可以重点看 4.2 的编码探测和 5.3 的换行折叠两个容易被忽略的边界。2. JSON 结构与目标格式写转换脚本前先做的两个决定2.1 先分清三种源文件形态单对象、JSON 数组、JSONL实际拿到的 JSON 文件很少长得一样。最常见的是三种整个文件是一个对象比如接口的一次响应文件里是一个 JSON 数组比如导出工具给的列表以及 JSONL 格式每行一个独立 JSON 对象整个文件并不是一个合法的 JSON。这三种形态决定了你在哪一层取数据。# 形态1文件顶层是对象 {code:0, data:{id:1}} import json with open(a.json, r, encodingutf-8) as f: data json.load(f) items data.get(data, data) # 形态2文件顶层是数组 [{id:1}, {id:2}] with open(b.json, r, encodingutf-8) as f: data json.load(f) items data if isinstance(data, list) else [data] # 形态3JSONL一行一个对象 items [] with open(c.jsonl, r, encodingutf-8) as f: for line in f: if line.strip(): items.append(json.loads(line))这里有个容易踩的坑用 isinstance(data, list) 判断数组形态时如果源文件顶层对象里某个业务字段本身就是 list直接按数组处理会把整块数据错误展开。稳妥做法是先打印 type(data) 和顶层 key 确认一次再决定解析入口而不是只看某个子字段的类型就下结论。三种形态的分支写进同一个函数靠一个参数切换后续维护成本最低。2.2 目标 TXT 的三种组织方式每行一条、字段拼接、可读文本合并后的 TXT 常见三种目标格式。每行一条完整 JSON 记录的优势是信息不丢、可以用 json.loads 逐行还原排错也直观字段按分隔符拼接适合直接灌数据库或 Excel只提取指定字段写成带标签的文本则适合人工阅读和后续日志分析。三种格式的写出代码差异很小但选错会直接影响下游消费。# 格式A每行一条完整 JSON fout.write(json.dumps(item, ensure_asciiFalse) \n) # 格式B只取字段竖线分隔表头单独写一行 fout.write(f{item.get(id)}|{item.get(name)}|{item.get(time)}\n) # 格式C可读文本带字段标签 fout.write(f[{item.get(time)}] {item.get(title)} {item.get(level)}\n)我一般优先选每行一个 JSON 对象除非需求明确写了要哪些字段因为这种格式后续无论是转 DataFrame、导入 Elasticsearch 还是做 diff还原成本都是最低的。提醒一个细节无论用哪种格式都建议在文件开头写一行表头或格式说明三个月后回看这个 TXT 时能立刻知道每一列是什么而不是重新翻源文件比对。2.3 选型标准库 json 够用什么时候才需要上 ijson处理这类批任务Python 标准库 json 是首选。它内置、零依赖、对绝大多数文件大小都足够快代码写出来别人也能直接看懂。只有当单个 JSON 文件达到几百 MB 甚至上 GB、json.load 一次性把内存打满时才需要换成 ijson 做流式解析或者用 4.3 的 raw_decode 逐段解析。pandas 的 read_json 不是不能做但对转换成一个 TXT这个任务它多引入了 DataFrame 的概念字段顺序和类型会被重排反而容易引入意外。一个判断依据源文件总大小小于内存的十分之一就用 json.load超过这个比例转流式方案。批处理任务里最常见的故障不是解析速度而是单个大文件把进程 OOM 掉后面的文件全部白跑。所以第 3 章的最小脚本里每个文件的读取都放在独立上下文里处理完立即释放引用这是批量任务的基本纪律。3. 用标准库实现 JSON 文件转 TXT 的批处理最小脚本3.1 最小可用版本glob 遍历加逐文件写入先给一个能直接改路径就跑的版本。它的输入是一个目录输出是一个合并后的 TXT每个源文件里的每个 JSON 对象各占一行兼顾单对象文件和 JSON 数组两种形态。import json import glob def batch_json_to_txt(src_dir: str, out_file: str, encoding: str utf-8) - int: 把 src_dir 下所有 .json 文件合并写入 out_file返回写入行数。 files sorted(glob.glob(f{src_dir}/*.json)) written 0 with open(out_file, w, encodingencoding) as fout: for file in files: with open(file, r, encodingencoding) as fin: data json.load(fin) items data if isinstance(data, list) else [data] for item in items: fout.write(json.dumps(item, ensure_asciiFalse) \n) written 1 return written if __name__ __main__: n batch_json_to_txt(./json_files, ./merged.txt) print(f处理完成共写入 {n} 行)这段代码的逻辑可以拆成四层。glob.glob 负责把目录里所有 .json 文件名取出来sorted 保证处理顺序稳定在需要按文件名时间顺序合并时这一点很关键json.load 把单个文件完整解析成 Python 对象isinstance(data, list) 处理顶层是数组还是对象的分叉两种形态都能进同一个写入循环最后 json.dumps 用 ensure_asciiFalse 保证中文直接以可读形式写进 TXT而不是转成 \uXXXX 转义序列。输出文件只打开一次循环写入后统一关闭避免频繁开关文件句柄。注意输出文件的编码参数和输入文件的 encoding 用的是同一个变量实际生产里两者可能不同。建议把输出编码固定为 utf-8输入编码按源文件实际情况单独传参。3.2 关键参数表ensure_ascii、separators、sort_keys 怎么配同样一段数据json.dumps 的参数不同产物差别很大。下面这几个参数是批处理场景里最常用的indent 是类似 json 格式化工具里常见的美化参数但合并成单行 TXT 时千万不能设否则会破坏每行一条记录的结构。参数取值示例作用建议ensure_asciiFalse为 False 时中文原样输出True 时输出 \uXXXX目标给人工看或导入中文系统时设 Falseseparators(,, :)去掉 dumps 默认的空格压缩单行体积每行一条记录时建议压缩能省约 20% 体积sort_keysTrue / False按键名排序输出需要 diff 两个批次结果时设 Trueindent不设或 None美化缩进会引入换行单行记录格式下不要设default自定义函数处理 datetime、Decimal 等 json 不支持的类型的回调数据源带时间字段时必配default 参数最容易翻车。源数据里混入 datetime、Decimal 这类对象时json.dumps 会直接抛 TypeError: Object of type datetime is not JSON serializable批处理跑到第 37 个文件才崩溃前面的输出全部作废。常见做法是加一个兜底转换函数把非 JSON 原生类型统一转成字符串或者用 ISO 格式输出。def json_default(obj): # datetime/date 转成 ISO 字符串 if hasattr(obj, isoformat): return obj.isoformat() # numpy 标量转成 Python 原生类型 if hasattr(obj, item): return obj.item() return str(obj) line json.dumps(item, ensure_asciiFalse, defaultjson_default)3.3 批量场景的三个常见误用第一个误用是把整个合并循环包在一个大 try 里遇到坏文件直接终止整个任务。批处理的价值恰恰在于一个坏文件不能拖垮其余 99 个。正确做法是单个文件失败记到日志里continue 继续处理全部结束后统一汇报失败清单。第二个常见误用是把所有文件读进一个 list 再统一写入小文件没问题但 200 个 10MB 的文件会吃进 2GB 内存正确写法是读一个、写一个、丢引用。第三个误用是每个对象都 open 一次输出文件再 append文件句柄和系统调用开销会让 10 万行数据的任务慢上几十倍。这几个问题改起来都不难但都是线上跑批任务里真实出现过的故障点。4. 嵌套取值、编码兜底与大文件流式解析4.1 深层字段提取与缺键保护如果目标是只抽业务字段写进 TXT那么 data[users][0][profile][name] 这种嵌套取值会很快变成一大串防御代码。更简洁的做法是写一个 deep_get 辅助函数用点分路径表示层级任何一层缺失都返回默认值不抛 KeyError批处理循环因此不用包满 try。def deep_get(obj, dotted_path: str, default): 按 a.b.c 路径取嵌套值key 缺失或类型不匹配时返回 default。 for key in dotted_path.split(.): if isinstance(obj, dict): obj obj.get(key) else: return default if obj is None: return default return obj if obj is not None else default # 使用示例取 item.owner.name缺任何一个 key 都不会抛异常 row f{deep_get(item, id)}|{deep_get(item, owner.name)}\n这段代码的核心是把路径解析和取值分开。路径按 . 切分后逐层向下走obj.get(key) 天然带缺键保护isinstance(obj, dict) 防止在 list 或字符串上继续取属性。这个实现只覆盖 dict 嵌套如果中间层是数组可以扩展支持 items.0.name 这种数字下标把 isinstance 判断改成同时接受 list 并用 int 下标访问。JSON 解析场景里嵌套多数到 dict 为止这个深度已能覆盖大部分导出需求。4.2 编码问题UTF-8 BOM 与 GBK 源文件现实里的 JSON 文件来源很杂。Windows 导出的常带 UTF-8 BOM老业务系统的接口可能直接给 GBK 编码的文件。用固定 encodingutf-8 去读遇到 BOM 会在顶层 key 前多出一个 \ufeff 字符遇到 GBK 直接抛 UnicodeDecodeError。用读字节、探测、解码三步走能在同一份代码里兼容三种编码。from pathlib import Path import json def read_json_auto(path): raw Path(path).read_bytes() # 按字节读入绕过编码猜测 if raw.startswith(b\xef\xbb\xbf): # UTF-8 BOM return json.loads(raw.decode(utf-8-sig)) for enc in (utf-8, gbk): try: return json.loads(raw.decode(enc)) except UnicodeDecodeError: continue raise ValueError(f无法解码文件: {path})判断顺序值得解释。先检查 BOM 是因为带 BOM 的 UTF-8 用 utf-8 解码不会报错但会把 \ufeff 带进 keyutf-8-sig 会剥掉 BOM。之后按 utf-8、gbk 的顺序尝试utf-8 是大多数现代文件的编码把它放前面让成功率最高的路径最先命中避免每次都先用 GBK 试错。decode 失败抛的是 UnicodeDecodeError所以用 try 包住并 continue全部失败才抛 ValueError。在这个函数外面接住异常就能把文件路径记入失败日志而不是中断整个任务。4.3 大文件流式解析raw_decode 与逐行消费单个 JSON 文件大到几百 MB 时json.load 会把整个文件加载进内存同时处理多个这种文件内存很容易吃紧。标准库有一个轻量替代json.JSONDecoder().raw_decode 配合文件流一次只解析一个 JSON 值并把解析位置推进到该值结束。对 JSONL 这种天然逐行的格式尤其适用。import json decoder json.JSONDecoder() buf with open(huge.jsonl, r, encodingutf-8) as fin: for line in fin: stripped line.strip() if stripped and stripped.startswith({): obj, idx decoder.raw_decode(stripped) # obj 是单个字典对象直接进入写入逻辑raw_decode 返回两个值解析出的对象和解析结束的下标。用在 JSONL 场景时由于每行恰好是一个完整 JSON 值raw_decode 一次消费整行idx 恒等于行内容长度所以多数情况只需要取 obj。它比 json.loads 强的地方在于天然的解析一个、丢一个的流式语义不保留整个文件的引用。真正的超大多行数组比如几 MB 的数组压在一行里则建议上 ijson 的 items 接口按 key 迭代那是为流式设计的库标准库不硬撑。5. 把批量脚本封装成命令行工具并验证输出5.1 用 argparse 暴露目录与字段参数固定路径的脚本只能自己用换目录、换输出格式就得改代码。用 argparse 把输入目录、输出文件和抽取字段暴露成命令行参数运维同事不碰 Python 源码也能跑。这里直接在 4.1 的 deep_get 基础上做字段抽取路径里的点号表示嵌套层级。import argparse import json import glob def main(): parser argparse.ArgumentParser(description批量 JSON 文件合并转 TXT) parser.add_argument(src, help存放 JSON 文件的目录) parser.add_argument(--out, defaultmerged.txt, help输出 TXT 路径) parser.add_argument(--fields, help要提取的字段逗号分隔如 id,name,owner.name) parser.add_argument(--sep, default|, help字段分隔符默认竖线) args parser.parse_args() files sorted(glob.glob(f{args.src}/*.json)) written 0 with open(args.out, w, encodingutf-8) as fout: for file in files: try: with open(file, r, encodingutf-8) as fin: data json.load(fin) except (json.JSONDecodeError, UnicodeDecodeError) as exc: print(f跳过 {file}: {exc}) continue items data if isinstance(data, list) else [data] for item in items: if args.fields: values [deep_get(item, f.strip()) for f in args.fields.split(,)] fout.write(args.sep.join(values) \n) else: fout.write(json.dumps(item, ensure_asciiFalse) \n) written 1 print(f完成共写入 {written} 行 - {args.out})运行方式如下第一个命令输出完整 JSON 行第二个命令只抽三个字段并用逗号分隔。python batch_json2txt.py ./json_files --out ./all.txt python batch_json2txt.py ./json_files --out ./names.txt --fields id,name,owner.name --sep ,注意 --fields 与 --sep 是配套的fields 为空时走完整 JSON 输出sep 参数不生效。字段名本身不能含逗号和点号这是命令行传参最容易踩的边界条件。5.2 用行数核对和往返解析做输出校验批处理输出很少有人逐条检查所以校验要可量化。三个动作建议全做统计输出行数并与源文件对象总数对比抽 head 和 tail 各几行确认没有乱码和半截记录对每行 JSON 再做一次 json.loads能还原说明合并没有破坏数据。bad 0 with open(merged.txt, r, encodingutf-8) as f: for i, line in enumerate(f, 1): if line.strip(): try: json.loads(line) except json.JSONDecodeError: bad 1 print(f第 {i} 行无法解析) print(f解析失败行数: {bad})因为目标格式是每行一个合法 JSON这重校验不依赖业务知识就能判断文件完整性——只要有一行被截断或两行粘在一起json.loads 必然抛异常。如果目标格式是字段拼接则把校验换成按分隔符拆列检查列数是否恒定这样能抓到内容里混入分隔符导致的串列问题那是字段拼接格式最隐蔽的坑。5.3 折叠字段内换行保证一行一条记录最后是一个实战斗了很久才总结的细节。字段值里本身含换行符时json.dumps 之后换行会被转成 \n 转义序列所以每行一条 JSON格式下数据不会被拆行但字段拼接格式下字段值里的换行会直接破坏行结构灌库时整表错位。处理方式是在拼接前统一做空白折叠。def clean(value): if not isinstance(value, str): return value return .join(value.split()) # 把 \n \t 连续空格折叠为单个空格把每个字段值先过 clean 再拼接输出的每一行都能保持一行一条记录的结构。这个函数放在字段抽取的列表推导里一行改动就能避免下游导入时的整表错位。批量 JSON 转 TXT 的脚本写到这一步就覆盖了日常导出、归档、灌库三类主要场景的完整链路解析入口按文件形态切换取值路径按业务字段配置编码兜底应对来源杂的旧数据输出校验保证交付物可用。本文还有配套的精品资源点击获取 Python批量JSON转TXT：完整方案与工程实践
简介这是一套Python批量处理JSON转TXT的小工具面向需要在Python中频繁进行数据格式转换的开发者和数据分析人员解决将多个JSON文件合并为单一TXT文件时的重复劳动。脚本基于内置json库实现只需输入文件夹路径即可批量读取多个JSON文件按需提取指定字段后汇总写入同一个TXT文件代码结构清晰修改少数参数就能适配不同需求也可在此基础上增加错误处理、日志输出等能力适合日常数据清洗与归档场景。压缩包容量仅2KB包含2个Python脚本分别承担主功能与JSON加载辅助轻量易用可快速集成到现有项目中。资源已有2494人浏览学习说明其简便方案获得了不少使用者认可对入门Python文件处理或希望简化重复性转换工作的读者来说是一份可直接运行的实用模板便于理解批处理思路并拓展更多数据转换功能。1. 批处理 JSON 转 TXT先想清楚一个 TXT长什么样把几十个 JSON 文件合并成一个 TXT这类需求在数据导出、日志归档、给下游模型或数据库灌数据时非常常见。很多人第一反应是写个循环逐个读取再追加写入但真正决定脚本能不能复用的是转换之外那三个问题源文件是单对象还是 JSON 数组目标 TXT 是每行一条记录还是自由文本字段全部保留还是只抽业务字段。这三个问题不先答完写出来的批处理脚本换一批文件就得重改。本文用 Python 标准库 json 加 glob 完成整套方案不引入第三方依赖从最小脚本讲到嵌套取值、编码兜底和大文件流式解析最后封装成带参数的命令行工具并把输出校验写进流程。新手能照着改路径直接跑熟手可以重点看 4.2 的编码探测和 5.3 的换行折叠两个容易被忽略的边界。2. JSON 结构与目标格式写转换脚本前先做的两个决定2.1 先分清三种源文件形态单对象、JSON 数组、JSONL实际拿到的 JSON 文件很少长得一样。最常见的是三种整个文件是一个对象比如接口的一次响应文件里是一个 JSON 数组比如导出工具给的列表以及 JSONL 格式每行一个独立 JSON 对象整个文件并不是一个合法的 JSON。这三种形态决定了你在哪一层取数据。# 形态1文件顶层是对象 {code:0, data:{id:1}} import json with open(a.json, r, encodingutf-8) as f: data json.load(f) items data.get(data, data) # 形态2文件顶层是数组 [{id:1}, {id:2}] with open(b.json, r, encodingutf-8) as f: data json.load(f) items data if isinstance(data, list) else [data] # 形态3JSONL一行一个对象 items [] with open(c.jsonl, r, encodingutf-8) as f: for line in f: if line.strip(): items.append(json.loads(line))这里有个容易踩的坑用 isinstance(data, list) 判断数组形态时如果源文件顶层对象里某个业务字段本身就是 list直接按数组处理会把整块数据错误展开。稳妥做法是先打印 type(data) 和顶层 key 确认一次再决定解析入口而不是只看某个子字段的类型就下结论。三种形态的分支写进同一个函数靠一个参数切换后续维护成本最低。2.2 目标 TXT 的三种组织方式每行一条、字段拼接、可读文本合并后的 TXT 常见三种目标格式。每行一条完整 JSON 记录的优势是信息不丢、可以用 json.loads 逐行还原排错也直观字段按分隔符拼接适合直接灌数据库或 Excel只提取指定字段写成带标签的文本则适合人工阅读和后续日志分析。三种格式的写出代码差异很小但选错会直接影响下游消费。# 格式A每行一条完整 JSON fout.write(json.dumps(item, ensure_asciiFalse) \n) # 格式B只取字段竖线分隔表头单独写一行 fout.write(f{item.get(id)}|{item.get(name)}|{item.get(time)}\n) # 格式C可读文本带字段标签 fout.write(f[{item.get(time)}] {item.get(title)} {item.get(level)}\n)我一般优先选每行一个 JSON 对象除非需求明确写了要哪些字段因为这种格式后续无论是转 DataFrame、导入 Elasticsearch 还是做 diff还原成本都是最低的。提醒一个细节无论用哪种格式都建议在文件开头写一行表头或格式说明三个月后回看这个 TXT 时能立刻知道每一列是什么而不是重新翻源文件比对。2.3 选型标准库 json 够用什么时候才需要上 ijson处理这类批任务Python 标准库 json 是首选。它内置、零依赖、对绝大多数文件大小都足够快代码写出来别人也能直接看懂。只有当单个 JSON 文件达到几百 MB 甚至上 GB、json.load 一次性把内存打满时才需要换成 ijson 做流式解析或者用 4.3 的 raw_decode 逐段解析。pandas 的 read_json 不是不能做但对转换成一个 TXT这个任务它多引入了 DataFrame 的概念字段顺序和类型会被重排反而容易引入意外。一个判断依据源文件总大小小于内存的十分之一就用 json.load超过这个比例转流式方案。批处理任务里最常见的故障不是解析速度而是单个大文件把进程 OOM 掉后面的文件全部白跑。所以第 3 章的最小脚本里每个文件的读取都放在独立上下文里处理完立即释放引用这是批量任务的基本纪律。3. 用标准库实现 JSON 文件转 TXT 的批处理最小脚本3.1 最小可用版本glob 遍历加逐文件写入先给一个能直接改路径就跑的版本。它的输入是一个目录输出是一个合并后的 TXT每个源文件里的每个 JSON 对象各占一行兼顾单对象文件和 JSON 数组两种形态。import json import glob def batch_json_to_txt(src_dir: str, out_file: str, encoding: str utf-8) - int: 把 src_dir 下所有 .json 文件合并写入 out_file返回写入行数。 files sorted(glob.glob(f{src_dir}/*.json)) written 0 with open(out_file, w, encodingencoding) as fout: for file in files: with open(file, r, encodingencoding) as fin: data json.load(fin) items data if isinstance(data, list) else [data] for item in items: fout.write(json.dumps(item, ensure_asciiFalse) \n) written 1 return written if __name__ __main__: n batch_json_to_txt(./json_files, ./merged.txt) print(f处理完成共写入 {n} 行)这段代码的逻辑可以拆成四层。glob.glob 负责把目录里所有 .json 文件名取出来sorted 保证处理顺序稳定在需要按文件名时间顺序合并时这一点很关键json.load 把单个文件完整解析成 Python 对象isinstance(data, list) 处理顶层是数组还是对象的分叉两种形态都能进同一个写入循环最后 json.dumps 用 ensure_asciiFalse 保证中文直接以可读形式写进 TXT而不是转成 \uXXXX 转义序列。输出文件只打开一次循环写入后统一关闭避免频繁开关文件句柄。注意输出文件的编码参数和输入文件的 encoding 用的是同一个变量实际生产里两者可能不同。建议把输出编码固定为 utf-8输入编码按源文件实际情况单独传参。3.2 关键参数表ensure_ascii、separators、sort_keys 怎么配同样一段数据json.dumps 的参数不同产物差别很大。下面这几个参数是批处理场景里最常用的indent 是类似 json 格式化工具里常见的美化参数但合并成单行 TXT 时千万不能设否则会破坏每行一条记录的结构。参数取值示例作用建议ensure_asciiFalse为 False 时中文原样输出True 时输出 \uXXXX目标给人工看或导入中文系统时设 Falseseparators(,, :)去掉 dumps 默认的空格压缩单行体积每行一条记录时建议压缩能省约 20% 体积sort_keysTrue / False按键名排序输出需要 diff 两个批次结果时设 Trueindent不设或 None美化缩进会引入换行单行记录格式下不要设default自定义函数处理 datetime、Decimal 等 json 不支持的类型的回调数据源带时间字段时必配default 参数最容易翻车。源数据里混入 datetime、Decimal 这类对象时json.dumps 会直接抛 TypeError: Object of type datetime is not JSON serializable批处理跑到第 37 个文件才崩溃前面的输出全部作废。常见做法是加一个兜底转换函数把非 JSON 原生类型统一转成字符串或者用 ISO 格式输出。def json_default(obj): # datetime/date 转成 ISO 字符串 if hasattr(obj, isoformat): return obj.isoformat() # numpy 标量转成 Python 原生类型 if hasattr(obj, item): return obj.item() return str(obj) line json.dumps(item, ensure_asciiFalse, defaultjson_default)3.3 批量场景的三个常见误用第一个误用是把整个合并循环包在一个大 try 里遇到坏文件直接终止整个任务。批处理的价值恰恰在于一个坏文件不能拖垮其余 99 个。正确做法是单个文件失败记到日志里continue 继续处理全部结束后统一汇报失败清单。第二个常见误用是把所有文件读进一个 list 再统一写入小文件没问题但 200 个 10MB 的文件会吃进 2GB 内存正确写法是读一个、写一个、丢引用。第三个误用是每个对象都 open 一次输出文件再 append文件句柄和系统调用开销会让 10 万行数据的任务慢上几十倍。这几个问题改起来都不难但都是线上跑批任务里真实出现过的故障点。4. 嵌套取值、编码兜底与大文件流式解析4.1 深层字段提取与缺键保护如果目标是只抽业务字段写进 TXT那么 data[users][0][profile][name] 这种嵌套取值会很快变成一大串防御代码。更简洁的做法是写一个 deep_get 辅助函数用点分路径表示层级任何一层缺失都返回默认值不抛 KeyError批处理循环因此不用包满 try。def deep_get(obj, dotted_path: str, default): 按 a.b.c 路径取嵌套值key 缺失或类型不匹配时返回 default。 for key in dotted_path.split(.): if isinstance(obj, dict): obj obj.get(key) else: return default if obj is None: return default return obj if obj is not None else default # 使用示例取 item.owner.name缺任何一个 key 都不会抛异常 row f{deep_get(item, id)}|{deep_get(item, owner.name)}\n这段代码的核心是把路径解析和取值分开。路径按 . 切分后逐层向下走obj.get(key) 天然带缺键保护isinstance(obj, dict) 防止在 list 或字符串上继续取属性。这个实现只覆盖 dict 嵌套如果中间层是数组可以扩展支持 items.0.name 这种数字下标把 isinstance 判断改成同时接受 list 并用 int 下标访问。JSON 解析场景里嵌套多数到 dict 为止这个深度已能覆盖大部分导出需求。4.2 编码问题UTF-8 BOM 与 GBK 源文件现实里的 JSON 文件来源很杂。Windows 导出的常带 UTF-8 BOM老业务系统的接口可能直接给 GBK 编码的文件。用固定 encodingutf-8 去读遇到 BOM 会在顶层 key 前多出一个 \ufeff 字符遇到 GBK 直接抛 UnicodeDecodeError。用读字节、探测、解码三步走能在同一份代码里兼容三种编码。from pathlib import Path import json def read_json_auto(path): raw Path(path).read_bytes() # 按字节读入绕过编码猜测 if raw.startswith(b\xef\xbb\xbf): # UTF-8 BOM return json.loads(raw.decode(utf-8-sig)) for enc in (utf-8, gbk): try: return json.loads(raw.decode(enc)) except UnicodeDecodeError: continue raise ValueError(f无法解码文件: {path})判断顺序值得解释。先检查 BOM 是因为带 BOM 的 UTF-8 用 utf-8 解码不会报错但会把 \ufeff 带进 keyutf-8-sig 会剥掉 BOM。之后按 utf-8、gbk 的顺序尝试utf-8 是大多数现代文件的编码把它放前面让成功率最高的路径最先命中避免每次都先用 GBK 试错。decode 失败抛的是 UnicodeDecodeError所以用 try 包住并 continue全部失败才抛 ValueError。在这个函数外面接住异常就能把文件路径记入失败日志而不是中断整个任务。4.3 大文件流式解析raw_decode 与逐行消费单个 JSON 文件大到几百 MB 时json.load 会把整个文件加载进内存同时处理多个这种文件内存很容易吃紧。标准库有一个轻量替代json.JSONDecoder().raw_decode 配合文件流一次只解析一个 JSON 值并把解析位置推进到该值结束。对 JSONL 这种天然逐行的格式尤其适用。import json decoder json.JSONDecoder() buf with open(huge.jsonl, r, encodingutf-8) as fin: for line in fin: stripped line.strip() if stripped and stripped.startswith({): obj, idx decoder.raw_decode(stripped) # obj 是单个字典对象直接进入写入逻辑raw_decode 返回两个值解析出的对象和解析结束的下标。用在 JSONL 场景时由于每行恰好是一个完整 JSON 值raw_decode 一次消费整行idx 恒等于行内容长度所以多数情况只需要取 obj。它比 json.loads 强的地方在于天然的解析一个、丢一个的流式语义不保留整个文件的引用。真正的超大多行数组比如几 MB 的数组压在一行里则建议上 ijson 的 items 接口按 key 迭代那是为流式设计的库标准库不硬撑。5. 把批量脚本封装成命令行工具并验证输出5.1 用 argparse 暴露目录与字段参数固定路径的脚本只能自己用换目录、换输出格式就得改代码。用 argparse 把输入目录、输出文件和抽取字段暴露成命令行参数运维同事不碰 Python 源码也能跑。这里直接在 4.1 的 deep_get 基础上做字段抽取路径里的点号表示嵌套层级。import argparse import json import glob def main(): parser argparse.ArgumentParser(description批量 JSON 文件合并转 TXT) parser.add_argument(src, help存放 JSON 文件的目录) parser.add_argument(--out, defaultmerged.txt, help输出 TXT 路径) parser.add_argument(--fields, help要提取的字段逗号分隔如 id,name,owner.name) parser.add_argument(--sep, default|, help字段分隔符默认竖线) args parser.parse_args() files sorted(glob.glob(f{args.src}/*.json)) written 0 with open(args.out, w, encodingutf-8) as fout: for file in files: try: with open(file, r, encodingutf-8) as fin: data json.load(fin) except (json.JSONDecodeError, UnicodeDecodeError) as exc: print(f跳过 {file}: {exc}) continue items data if isinstance(data, list) else [data] for item in items: if args.fields: values [deep_get(item, f.strip()) for f in args.fields.split(,)] fout.write(args.sep.join(values) \n) else: fout.write(json.dumps(item, ensure_asciiFalse) \n) written 1 print(f完成共写入 {written} 行 - {args.out})运行方式如下第一个命令输出完整 JSON 行第二个命令只抽三个字段并用逗号分隔。python batch_json2txt.py ./json_files --out ./all.txt python batch_json2txt.py ./json_files --out ./names.txt --fields id,name,owner.name --sep ,注意 --fields 与 --sep 是配套的fields 为空时走完整 JSON 输出sep 参数不生效。字段名本身不能含逗号和点号这是命令行传参最容易踩的边界条件。5.2 用行数核对和往返解析做输出校验批处理输出很少有人逐条检查所以校验要可量化。三个动作建议全做统计输出行数并与源文件对象总数对比抽 head 和 tail 各几行确认没有乱码和半截记录对每行 JSON 再做一次 json.loads能还原说明合并没有破坏数据。bad 0 with open(merged.txt, r, encodingutf-8) as f: for i, line in enumerate(f, 1): if line.strip(): try: json.loads(line) except json.JSONDecodeError: bad 1 print(f第 {i} 行无法解析) print(f解析失败行数: {bad})因为目标格式是每行一个合法 JSON这重校验不依赖业务知识就能判断文件完整性——只要有一行被截断或两行粘在一起json.loads 必然抛异常。如果目标格式是字段拼接则把校验换成按分隔符拆列检查列数是否恒定这样能抓到内容里混入分隔符导致的串列问题那是字段拼接格式最隐蔽的坑。5.3 折叠字段内换行保证一行一条记录最后是一个实战斗了很久才总结的细节。字段值里本身含换行符时json.dumps 之后换行会被转成 \n 转义序列所以每行一条 JSON格式下数据不会被拆行但字段拼接格式下字段值里的换行会直接破坏行结构灌库时整表错位。处理方式是在拼接前统一做空白折叠。def clean(value): if not isinstance(value, str): return value return .join(value.split()) # 把 \n \t 连续空格折叠为单个空格把每个字段值先过 clean 再拼接输出的每一行都能保持一行一条记录的结构。这个函数放在字段抽取的列表推导里一行改动就能避免下游导入时的整表错位。批量 JSON 转 TXT 的脚本写到这一步就覆盖了日常导出、归档、灌库三类主要场景的完整链路解析入口按文件形态切换取值路径按业务字段配置编码兜底应对来源杂的旧数据输出校验保证交付物可用。本文还有配套的精品资源点击获取