Python 处理图像并生成 JSONL 元数据文件 - 固定text版本

Python 处理图像并生成 JSONL 元数据文件 - 固定text版本

flyfish

JSONL（JSON Lines）简介

JSONL（JSON Lines，也称为 newline-delimited JSON）是一种轻量级的数据序列化格式，由一系列独立的 JSON 对象组成，每行一个有效的 JSON 对象，行与行之间通过换行符（\n）分隔。
JSONL 是传统 JSON 的“轻量化”变体，通过“每行一个JSON对象”的设计，解决了大规模数据处理时的内存和效率问题，成为日志分析、大数据处理、流式计算等场景的理想选择。
避免传统JSON中数组包裹大量对象时的性能开销（如解析大数组的时间和内存消耗）。
每行可独立验证和处理，某一行出错不影响其他行（而传统JSON中数组内一个元素错误会导致整个文件解析失败）。
每行都是合法的JSON，可被任何JSON解析器处理，只需逐行读取即可。

核心特点

1. 结构简单

   • 每个 JSON 对象独占一行，文件中没有外围的数组括号（如 [ 和 ]）或逗号分隔符。
   • 示例：

     {"name": "Alice", "age": 30}
     {"name": "Bob", "age": 25}
     {"name": "Charlie", "age": 35}
     

2. 流式处理友好

   • 无需一次性加载整个文件到内存，可逐行读取和处理数据，适合处理超大规模数据集（如GB、TB级文件）。
   • 支持实时处理（如日志流、数据流），边生成边解析，内存占用极低。

3. 独立对象

   • 每行都是一个完整、独立的 JSON 对象，行与行之间无依赖关系，便于并行处理（如分布式计算、多线程解析）。

与 JSON 的对比

特性	JSON	JSONL
结构	单个对象/数组（需被 {} 或 [] 包裹）	多行独立对象（每行一个 {}）
内存占用	需整体加载，大文件易导致内存不足	逐行处理，内存占用稳定
随机访问	支持（通过索引或键）	不支持（需按行顺序读取）
适用场景	小规模数据、API交互、配置文件	大规模数据、日志、批量处理、流式数据
解析方式	需要完整解析器（如JSON库）	可简单逐行读取，无需复杂解析逻辑

应用场景

1. 大数据处理

   • 存储机器学习数据集（如样本分行列，便于分布式训练）。
   • 处理日志文件（如Web服务器日志、应用监控日志，实时流式解析）。

2. 批量数据传输

   • ETL（Extract-Transform-Load）流程中，作为不同系统间的数据交换格式，支持增量处理。
   • 数据库导出/导入（如将表数据按行导出为JSONL，便于后续分析）。

3. 并行计算

   • 每行数据可独立处理，适合MapReduce、Spark等分布式框架，提升处理效率。

4. 简单日志格式

   • 替代CSV等格式，支持复杂数据结构（如嵌套对象、数组），同时保持易读性。

生成/解析JSONL

• 生成：按行写入JSON字符串，每行以 \n 结尾（注意避免使用非标准换行符）。
• 解析：逐行读取文件，对每行字符串调用JSON解析器即可。

Python示例（生成JSONL）

import jsondata = [{"name": "Alice", "age": 30},{"name": "Bob", "age": 25},{"name": "Charlie", "age": 35}
]with open("data.jsonl", "w") as f:for item in data:f.write(json.dumps(item) + "\n")

Python示例（解析JSONL）

import jsonwith open("data.jsonl", "r") as f:for line in f:item = json.loads(line.strip())print(item["name"])  # 逐行处理数据

完整代码

import os
import json
import argparse
from PIL import Imageclass ImageConverter:def __init__(self, src_folder, dest_folder, output_text="a dog", target_size=1024, output_format="JPEG", quality=95, recursive=False, prefix="", suffix="", metadata_file="metadata.jsonl"):"""初始化图像转换器:param src_folder: 源图像文件夹路径:param dest_folder: 目标输出文件夹路径:param output_text: 生成JSON中的text字段内容:param target_size: 输出图像的目标尺寸（默认为1024）:param output_format: 输出图像格式（默认为JPEG）:param quality: 输出图像质量（0-100，仅适用于某些格式）:param recursive: 是否递归处理子文件夹:param prefix: 输出文件名前缀:param suffix: 输出文件名后缀:param metadata_file: 元数据文件名"""self.src_folder = src_folderself.dest_folder = dest_folderself.output_text = output_textself.target_size = target_sizeself.output_format = output_formatself.quality = qualityself.recursive = recursiveself.prefix = prefixself.suffix = suffixself.metadata_file = metadata_fileself.image_list = []def setup_directories(self):"""确保源文件夹存在，创建目标文件夹（如果不存在）"""if not os.path.exists(self.src_folder):raise FileNotFoundError(f"源文件夹 {self.src_folder} 不存在")os.makedirs(self.dest_folder, exist_ok=True)print(f"目标文件夹已准备就绪: {self.dest_folder}")def collect_images(self):"""收集源文件夹中所有支持的图像文件"""supported_formats = ('.png', '.jpg', '.jpeg', '.bmp', '.gif', '.tiff')self.image_list = []if self.recursive:for root, _, files in os.walk(self.src_folder):for f in files:if f.lower().endswith(supported_formats):relative_path = os.path.relpath(root, self.src_folder)target_subfolder = os.path.join(self.dest_folder, relative_path)os.makedirs(target_subfolder, exist_ok=True)self.image_list.append((os.path.join(root, f), target_subfolder))else:files = [f for f in os.listdir(self.src_folder) if f.lower().endswith(supported_formats)]files.sort()  # 按文件名排序self.image_list = [(os.path.join(self.src_folder, f), self.dest_folder) for f in files]print(f"找到 {len(self.image_list)} 张图像")def resize_and_pad_image(self, img):"""等比缩放图像，并将其放入指定大小的白色背景正方形中:param img: PIL 图像对象:return: 处理后的新图像"""original_width, original_height = img.sizeratio = min(self.target_size / original_width, self.target_size / original_height)new_size = (int(original_width * ratio), int(original_height * ratio))resized_img = img.resize(new_size, Image.LANCZOS)# 创建白色背景图像padded_img = Image.new("RGB", (self.target_size, self.target_size), (255, 255, 255))# 居中粘贴position = ((self.target_size - new_size[0]) // 2, (self.target_size - new_size[1]) // 2)padded_img.paste(resized_img, position)return padded_imgdef convert_and_rename_images(self):"""转换图像格式并重命名，同时进行缩放和填充"""results = []for idx, (src_path, dest_subfolder) in enumerate(self.image_list):filename = os.path.basename(src_path)base_name, _ = os.path.splitext(filename)# 构建新文件名new_filename = f"{self.prefix}{idx:02d}{self.suffix}.{self.output_format.lower()}"dest_path = os.path.join(dest_subfolder, new_filename)# 计算相对路径（相对于dest_folder）relative_path = os.path.relpath(dest_path, self.dest_folder)try:with Image.open(src_path) as img:# 转换为RGB模式（如果不是）if img.mode not in ('RGB', 'RGBA'):img = img.convert('RGB')processed_img = self.resize_and_pad_image(img)# 根据输出格式设置保存参数save_args = {}if self.output_format.lower() == 'jpeg':save_args['quality'] = self.qualitysave_args['optimize'] = Trueelif self.output_format.lower() == 'png' and img.mode == 'RGBA':save_args['format'] = 'PNG'processed_img.save(dest_path, **save_args)results.append({"text": self.output_text, "file_name": relative_path})print(f"已保存: {relative_path}")except Exception as e:print(f"处理 {filename} 时出错: {e}")results.append({"text": self.output_text, "file_name": relative_path, "error": str(e)})return resultsdef generate_jsonl(self, results):"""生成 metadata.jsonl 文件，每行一个 JSON 对象，text 在前"""jsonl_path = os.path.join(self.dest_folder, self.metadata_file)with open(jsonl_path, "w", encoding="utf-8") as f:for item in results:f.write(f"{json.dumps(item, ensure_ascii=False)}\n")print(f"JSONL 文件已生成: {jsonl_path}")def run(self):"""执行整个流程"""self.setup_directories()self.collect_images()results = self.convert_and_rename_images()self.generate_jsonl(results)def main():parser = argparse.ArgumentParser(description='图像转换器 - 处理图像并生成JSONL元数据')# 必需参数parser.add_argument('--src', required=True, help='源图像文件夹路径')parser.add_argument('--dest', required=True, help='目标输出文件夹路径')# 可选参数parser.add_argument('--text', default="a dog", help='JSON中的text字段内容')parser.add_argument('--size', type=int, default=1024, help='输出图像的目标尺寸')parser.add_argument('--format', default="JPEG", choices=["JPEG", "PNG", "WEBP"], help='输出图像格式')parser.add_argument('--quality', type=int, default=95, help='输出图像质量（0-100，仅适用于某些格式）')parser.add_argument('--recursive', action='store_true', help='递归处理子文件夹')parser.add_argument('--prefix', default="", help='输出文件名前缀')parser.add_argument('--suffix', default="", help='输出文件名后缀')parser.add_argument('--metadata', default="metadata.jsonl", help='元数据文件名')args = parser.parse_args()# 创建并运行转换器converter = ImageConverter(src_folder=args.src,dest_folder=args.dest,output_text=args.text,target_size=args.size,output_format=args.format,quality=args.quality,recursive=args.recursive,prefix=args.prefix,suffix=args.suffix,metadata_file=args.metadata)converter.run()if __name__ == "__main__":main()

通过命令行直接运脚本

python image_converter.py --src path/to/source --dest path/to/destination --text "a dog" --size 512 --recursive

或者在代码中调用：

# 示例调用
converter = ImageConverter(src_folder="path/to/source/images",dest_folder="path/to/destination",output_text="a dog",target_size=1024,output_format="PNG",recursive=True
)
converter.run()

结果

{"text": "a dog", "file_name": "00.jpg"}
{"text": "a dog", "file_name": "01.jpg"}