Ubuntu下JSON转TXT实战用Python脚本批量处理活动数据集含中文编码避坑在数据处理工程师的日常工作中JSON与TXT格式的相互转换是一项基础但至关重要的任务。特别是当面对ActivityNet这类包含视频标注信息的大型数据集时如何高效、准确地完成格式转换同时处理好中文编码等细节问题直接影响到后续模型训练和数据分析的效率。本文将手把手带你用Python在Ubuntu环境下实现这一过程避开那些新手常踩的坑。1. 环境准备与基础工具在开始之前确保你的Ubuntu系统已经安装了Python 3.x环境。虽然Ubuntu通常预装了Python但最好确认一下版本python3 --version对于JSON文件的初步查看和验证jq是个非常实用的命令行工具。安装它只需要一条命令sudo apt-get install -y jq安装完成后你可以快速预览JSON文件的结构和内容jq . activity_net.v1-3.min.json | less这个命令会将JSON文件格式化输出并通过less分页显示特别适合快速浏览大型JSON文件。提示在处理大型JSON文件时添加| less可以避免终端被大量输出刷屏按q键退出浏览模式。2. Python处理JSON的核心方法Python的json模块提供了JSON处理的完整解决方案。让我们先来看一个最基本的JSON读取和写入TXT的例子import json # 读取JSON文件 with open(activity_net.v1-3.min.json, r, encodingutf-8) as f: data json.load(f) # 将数据转换为格式化的JSON字符串 json_str json.dumps(data, indent4, ensure_asciiFalse) # 写入TXT文件 with open(output.txt, w, encodingutf-8) as f: f.write(json_str)这段代码虽然简单但有几个关键点需要注意编码指定无论是读取还是写入都明确指定了utf-8编码这是处理中文等非ASCII字符的基础保障。ensure_asciiFalse这个参数让json.dumps()保留非ASCII字符的原貌而不是转换为Unicode转义序列。使用with语句管理文件操作可以自动处理文件的打开和关闭避免资源泄漏。3. 处理大型JSON文件的优化技巧当JSON文件特别大时比如ActivityNet数据集可能达到几百MB甚至GB级别直接加载整个文件到内存可能会引发问题。这时我们可以采用以下几种优化策略3.1 流式处理大文件对于特别大的JSON文件可以使用ijson库进行流式处理import ijson import json output open(large_output.txt, w, encodingutf-8) with open(large_file.json, r, encodingutf-8) as f: # 逐项处理数组元素 items ijson.items(f, item) for item in items: json_str json.dumps(item, ensure_asciiFalse) output.write(json_str \n) output.close()3.2 分批处理与内存管理另一种方法是分批读取和处理数据import json from itertools import islice def batch_process(json_file, batch_size1000): with open(json_file, r, encodingutf-8) as f: while True: batch list(islice(f, batch_size)) if not batch: break data json.loads(.join(batch)) # 处理这批数据...3.3 性能对比下表比较了不同处理方法的性能特点方法内存占用处理速度适用场景全量加载高快小文件(100MB)ijson流式低中等超大文件结构规整分批处理中等中等大型文件需灵活控制4. 中文编码问题深度解析中文编码问题是JSON处理中最常见的痛点之一。让我们深入探讨几个典型场景及解决方案。4.1 编码自动检测有时我们不确定JSON文件的具体编码可以使用chardet库进行检测import chardet def detect_encoding(file_path): with open(file_path, rb) as f: result chardet.detect(f.read()) return result[encoding]4.2 编码转换处理如果遇到编码不一致的情况可以这样处理from codecs import open with open(mixed_encoding.json, r, encodinggbk, errorsreplace) as f: content f.read() # 转换为UTF-8 utf8_content content.encode(gbk).decode(utf-8, errorsignore) data json.loads(utf8_content)4.3 常见中文编码问题解决方案问题1写入TXT后中文显示为Unicode编码如\u4e2d\u6587解决方案在json.dumps()中设置ensure_asciiFalse问题2读取JSON时抛出UnicodeDecodeError解决方案尝试不同的编码utf-8, gbk, gb2312等问题3混合编码文件部分内容乱码解决方案使用errorsreplace参数或用二进制模式读取后手动处理5. 高级应用处理复杂JSON结构ActivityNet等数据集的JSON结构往往比较复杂包含多层嵌套。下面介绍几种处理复杂结构的方法。5.1 提取特定字段假设我们需要提取视频标注信息中的特定字段def extract_fields(data, fields): results [] for item in data[annotations]: result {field: item.get(field) for field in fields} results.append(result) return results # 使用示例 important_fields [video_id, label, segment] extracted_data extract_fields(data, important_fields)5.2 扁平化嵌套结构对于多层嵌套的字典可以使用递归进行扁平化def flatten_dict(d, parent_key, sep_): items [] for k, v in d.items(): new_key f{parent_key}{sep}{k} if parent_key else k if isinstance(v, dict): items.extend(flatten_dict(v, new_key, sepsep).items()) else: items.append((new_key, v)) return dict(items)5.3 自定义JSON序列化当JSON中包含Python特殊对象如datetime时需要自定义序列化from datetime import datetime def custom_serializer(obj): if isinstance(obj, datetime): return obj.isoformat() raise TypeError(fObject of type {type(obj)} is not JSON serializable) json_str json.dumps(data, defaultcustom_serializer, ensure_asciiFalse)6. 实战批量处理ActivityNet数据集现在让我们把这些知识综合起来处理一个真实的ActivityNet数据集转换任务。6.1 完整脚本示例import json import os from tqdm import tqdm def process_activitynet(input_dir, output_dir): # 确保输出目录存在 os.makedirs(output_dir, exist_okTrue) # 遍历输入目录中的所有JSON文件 for filename in tqdm(os.listdir(input_dir)): if filename.endswith(.json): input_path os.path.join(input_dir, filename) output_path os.path.join(output_dir, f{os.path.splitext(filename)[0]}.txt) try: # 读取JSON文件 with open(input_path, r, encodingutf-8) as f: data json.load(f) # 转换为格式化的JSON字符串 json_str json.dumps(data, indent4, ensure_asciiFalse) # 写入TXT文件 with open(output_path, w, encodingutf-8) as f: f.write(json_str) except Exception as e: print(f处理文件 {filename} 时出错: {str(e)}) continue # 使用示例 process_activitynet(/path/to/activitynet/json, /path/to/output/txt)6.2 性能优化版本对于特别大的数据集我们可以进一步优化import ijson import json from concurrent.futures import ThreadPoolExecutor def process_single_file(args): input_path, output_path args try: with open(input_path, r, encodingutf-8) as f: data ijson.items(f, item) with open(output_path, w, encodingutf-8) as out_f: for item in data: json_str json.dumps(item, ensure_asciiFalse) out_f.write(json_str \n) return True except Exception as e: print(fError processing {input_path}: {e}) return False def batch_process_parallel(input_dir, output_dir, workers4): os.makedirs(output_dir, exist_okTrue) tasks [] for filename in os.listdir(input_dir): if filename.endswith(.json): input_path os.path.join(input_dir, filename) output_path os.path.join(output_dir, f{os.path.splitext(filename)[0]}.txt) tasks.append((input_path, output_path)) with ThreadPoolExecutor(max_workersworkers) as executor: results list(executor.map(process_single_file, tasks)) print(f成功处理 {sum(results)}/{len(results)} 个文件)6.3 处理后的验证检查转换完成后建议进行快速验证def validate_conversion(original_json, converted_txt): with open(original_json, r, encodingutf-8) as f1, \ open(converted_txt, r, encodingutf-8) as f2: original_data json.load(f1) try: converted_data json.load(f2) return original_data converted_data except json.JSONDecodeError: # 可能是逐行存储的JSON需要特殊处理 f2.seek(0) converted_lines [json.loads(line) for line in f2] return all(item in original_data for item in converted_lines)在实际项目中我发现使用ujson替代标准json模块可以提升约30%的性能特别是在处理大型文件时。但要注意ujson在某些边缘情况下的行为可能与json模块略有不同。另一个实用的技巧是在处理前先对JSON文件进行预处理移除不必要的空格和换行可以显著减少内存使用。