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

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

flyfish

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