从一个JSONL文件中逐行读取文件,并将这些问题保存到一个新的JSONL文件中
import json
import argparse
import os # 导入os模块用于检查文件是否存在def read_questions_from_jsonl(file_path, limit):"""从JSONL文件中读取指定数量的question部分的内容参数:file_path: JSONL文件的路径limit: 要读取的问题数量返回:一个包含最多`limit`个question的列表"""questions = []count = 0try:with open(file_path, 'r', encoding='utf-8') as file:for line in file:if count >= limit:break # 达到限制,停止读取# 解析每行JSONjson_data = json.loads(line)# 提取questionif 'question' in json_data:questions.append(json_data['question'])count += 1except json.JSONDecodeError:print("错误: 文件格式无效或损坏")return questionsdef save_questions_to_jsonl(questions, output_file_path):"""将问题列表保存到JSONL文件中参数:questions: 要保存的问题列表output_file_path: 输出文件的路径"""with open(output_file_path, 'w', encoding='utf-8') as file: for question in questions:json_line = json.dumps({"question": question}, ensure_ascii=False)file.write(json_line + '\n')def main():# 设置命令行参数解析parser = argparse.ArgumentParser(description='从JSONL文件中提取问题并保存到新的JSONL文件中')parser.add_argument('input_path', help='输入JSONL文件的路径')parser.add_argument('output_path', help='输出JSONL文件的路径')parser.add_argument('--limit', type=int, default=100, help='要读取的问题数量,默认为100') # 设置默认值# 解析命令行参数args = parser.parse_args()# 检查输入文件是否存在if not os.path.exists(args.input_path):raise FileNotFoundError(f"错误: 输入文件 {args.input_path} 不存在")# 读取指定数量的问题questions = read_questions_from_jsonl(args.input_path, args.limit)# 将问题保存到JSONL文件中save_questions_to_jsonl(questions, args.output_path)print(f"已将 {len(questions)} 个问题保存到 {args.output_path}")if __name__ == "__main__":main()
检查输入文件是否存在
import os# 检查输入文件是否存在
if not os.path.exists(args.input_path):raise FileNotFoundError(f"错误: 输入文件 {args.input_path} 不存在")
检查要输出的文件是否存在
使用open函数以写入模式(‘w’)打开文件时,如果指定的文件路径不存在,Python会自动创建一个新的文件。
如果存在在现有文件后追加,如果不存在,创建新的输出文件,可修改函数为
def save_questions_to_jsonl(questions, output_file_path):"""将问题列表保存到JSONL文件中参数:questions: 要保存的问题列表output_file_path: 输出文件的路径"""# 检查输出文件是否存在if os.path.exists(output_file_path):# 文件存在,以追加模式打开mode = 'a'else:# 文件不存在,以写入模式打开mode = 'w'# 根据mode变量动态选择文件打开模式with open(output_file_path, mode, encoding='utf-8') as file:for question in questions:json_line = json.dumps({"question": question}, ensure_ascii=False)file.write(json_line + '\n')