Python处理图像并生成JSONL元数据文件

目录

• JSONL（JSON Lines）简介
• 核心特点
• 与 JSON 的对比
• 应用场景
• 生成/解析JSONL
• 完整代码

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

1.大数据处理

存储机器学习数据集（如样本分行列，便于分布式训练）。

处理日志文件（如Web服务器日志、应用监控日志，实时流式解析）。

2.批量数据传输

ETL（Extract-Transform-Load）流程中，作为不同系统间的数据交换格式，支持增量处理。

数据库导出/导入（如将表数据按行导出为JSONL，便于后续分析）。

3.并行计算

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

4.简单日志格式

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

生成：按行写入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"}

Python 处理图像并生成 JSONL 元数据文件 – 灵活text版本

完整代码

import os
import json
import argparse
from PIL import Image
from xpinyin import Pinyin

class ImageConverter:
    def __init__(self, src_folder, dest_folder, target_size=1024, 
                 output_format="JPEG", quality=95, recursive=False, 
                 prefix="", suffix="", metadata_file="metadata.jsonl"):
        """
        初始化图像转换器
        
        :param src_folder: 源图像文件夹路径
        :param dest_folder: 目标输出文件夹路径
        :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.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 = []
        self.pinyin_converter = Pinyin()

    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, f))
        else:
            files = [f for f in os.listdir(self.src_folder) if f.lower().endswith(supported_formats)]
            files.sort()  # 按文件名排序
            # 修复此处：添加for循环遍历文件
            for f in files:
                self.image_list.append((os.path.join(self.src_folder, f), self.dest_folder, f))
                
        print(f"找到 {len(self.image_list)} 张图像")


    def filename_to_pinyin(self, filename):
        """将文件名中的中文部分转换为带音调的拼音"""
        base_name, ext = os.path.splitext(filename)
        chinese_chars = ''.join([c for c in base_name if 'u4e00' <= c <= 'u9fff'])
        
        if not chinese_chars:
            return ""  # 无中文字符时返回空（可根据需求调整）
            
        pinyin_list = []
        for char in chinese_chars:
            py = self.pinyin_converter.get_pinyin(char, tone_marks='numbers')
            # 确保音调存在，默认补1
            if not any(c.isdigit() for c in py):
                py += '1'
            pinyin_list.append(py.lower())  # 转换为小写
            
        return '_'.join(pinyin_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, filename) in enumerate(self.image_list):
            base_name, ext = os.path.splitext(filename)
            pinyin_text = self.filename_to_pinyin(base_name)
            
            # 构建新文件名
            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": pinyin_text,  # 使用转换后的拼音作为text
                        "file_name": relative_path
                    })
                    print(f"已保存: {relative_path}  (text={pinyin_text})")
            except Exception as e:
                print(f"处理 {filename} 时出错: {e}")
                results.append({
                    "text": pinyin_text,
                    "file_name": relative_path,
                    "error": str(e)
                })
                
        return results

    def generate_jsonl(self, results):
        """生成 metadata.jsonl 文件，每行一个 JSON 对象"""
        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('--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,
        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()

不再使用固定的output_text，而是从图像文件名中提取中文部分转换为拼音.

新增拼音转换函数filename_to_pinyin方法专门处理文件名，只保留中文字符并转换为带音调的拼音（如天.jpg→tian1）

使用方法

python converter.py --src ./images --dest ./processed_images --size 512 --format PNG

到此这篇关于Python处理图像并生成JSONL元数据文件的文章就介绍到这了,更多相关Python处理图像内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

您可能感兴趣的文章:

• Python数据清洗之抽取jsonl文件数据字段并合并
• Python中json文件和jsonl文件的区别小结
• python json jsonl 的用法详解
• 使用python读写txt和json(jsonl)大文件的方法步骤
• JSONLINT：python的json数据验证库实例解析