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 的对比

应用场景

1. 大数据处理

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

2. 批量数据传输

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

3. 并行计算

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

4. 简单日志格式

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

生成/解析JSONL

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

Python示例（生成JSONL）

import json

data = [
    {"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 json

with 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 Image

class 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_folder
        self.dest_folder = dest_folder
        self.output_text = output_text
        self.target_size = target_size
        self.output_format = output_format
        self.quality = quality
        self.recursive = recursive
        self.prefix = prefix
        self.suffix = suffix
        self.metadata_file = metadata_file
        self.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.size
        ratio = 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_img

    def 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.quality
                        save_args['optimize'] = True
                    elif 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 results

    def 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"}