Python處理圖像并生成JSONL元數(shù)據(jù)文件
JSONL(JSON Lines)簡(jiǎn)介
JSONL(JSON Lines,也稱(chēng)為 newline-delimited JSON)是一種輕量級(jí)的數(shù)據(jù)序列化格式,由一系列獨(dú)立的 JSON 對(duì)象組成,每行一個(gè)有效的 JSON 對(duì)象,行與行之間通過(guò)換行符(\n)分隔。
JSONL 是傳統(tǒng) JSON 的“輕量化”變體,通過(guò)“每行一個(gè)JSON對(duì)象”的設(shè)計(jì),解決了大規(guī)模數(shù)據(jù)處理時(shí)的內(nèi)存和效率問(wèn)題,成為日志分析、大數(shù)據(jù)處理、流式計(jì)算等場(chǎng)景的理想選擇。
避免傳統(tǒng)JSON中數(shù)組包裹大量對(duì)象時(shí)的性能開(kāi)銷(xiāo)(如解析大數(shù)組的時(shí)間和內(nèi)存消耗)。
每行可獨(dú)立驗(yàn)證和處理,某一行出錯(cuò)不影響其他行(而傳統(tǒng)JSON中數(shù)組內(nèi)一個(gè)元素錯(cuò)誤會(huì)導(dǎo)致整個(gè)文件解析失敗)。
每行都是合法的JSON,可被任何JSON解析器處理,只需逐行讀取即可。
核心特點(diǎn)
1.結(jié)構(gòu)簡(jiǎn)單
每個(gè) JSON 對(duì)象獨(dú)占一行,文件中沒(méi)有外圍的數(shù)組括號(hào)(如 [ 和 ])或逗號(hào)分隔符。
示例:
{"name": "Alice", "age": 30}
{"name": "Bob", "age": 25}
{"name": "Charlie", "age": 35}
2.流式處理友好
無(wú)需一次性加載整個(gè)文件到內(nèi)存,可逐行讀取和處理數(shù)據(jù),適合處理超大規(guī)模數(shù)據(jù)集(如GB、TB級(jí)文件)。
支持實(shí)時(shí)處理(如日志流、數(shù)據(jù)流),邊生成邊解析,內(nèi)存占用極低。
3.獨(dú)立對(duì)象
每行都是一個(gè)完整、獨(dú)立的 JSON 對(duì)象,行與行之間無(wú)依賴(lài)關(guān)系,便于并行處理(如分布式計(jì)算、多線程解析)。
與 JSON 的對(duì)比
| 特性 | JSON | JSONL |
|---|---|---|
| 結(jié)構(gòu) | 單個(gè)對(duì)象/數(shù)組(需被 {} 或 [] 包裹) | 多行獨(dú)立對(duì)象(每行一個(gè) {}) |
| 內(nèi)存占用 | 需整體加載,大文件易導(dǎo)致內(nèi)存不足 | 逐行處理,內(nèi)存占用穩(wěn)定 |
| 隨機(jī)訪問(wèn) | 支持(通過(guò)索引或鍵) | 不支持(需按行順序讀取) |
| 適用場(chǎng)景 | 小規(guī)模數(shù)據(jù)、API交互、配置文件 | 大規(guī)模數(shù)據(jù)、日志、批量處理、流式數(shù)據(jù) |
| 解析方式 | 需要完整解析器(如JSON庫(kù)) | 可簡(jiǎn)單逐行讀取,無(wú)需復(fù)雜解析邏輯 |
應(yīng)用場(chǎng)景
1.大數(shù)據(jù)處理
存儲(chǔ)機(jī)器學(xué)習(xí)數(shù)據(jù)集(如樣本分行列,便于分布式訓(xùn)練)。
處理日志文件(如Web服務(wù)器日志、應(yīng)用監(jiān)控日志,實(shí)時(shí)流式解析)。
2.批量數(shù)據(jù)傳輸
ETL(Extract-Transform-Load)流程中,作為不同系統(tǒng)間的數(shù)據(jù)交換格式,支持增量處理。
數(shù)據(jù)庫(kù)導(dǎo)出/導(dǎo)入(如將表數(shù)據(jù)按行導(dǎo)出為JSONL,便于后續(xù)分析)。
3.并行計(jì)算
每行數(shù)據(jù)可獨(dú)立處理,適合MapReduce、Spark等分布式框架,提升處理效率。
4.簡(jiǎn)單日志格式
替代CSV等格式,支持復(fù)雜數(shù)據(jù)結(jié)構(gòu)(如嵌套對(duì)象、數(shù)組),同時(shí)保持易讀性。
生成/解析JSONL
生成:按行寫(xiě)入JSON字符串,每行以 \n 結(jié)尾(注意避免使用非標(biāo)準(zhǔn)換行符)。
解析:逐行讀取文件,對(duì)每行字符串調(diào)用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"]) # 逐行處理數(shù)據(jù)
完整代碼
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"):
"""
初始化圖像轉(zhuǎn)換器
:param src_folder: 源圖像文件夾路徑
:param dest_folder: 目標(biāo)輸出文件夾路徑
:param output_text: 生成JSON中的text字段內(nèi)容
:param target_size: 輸出圖像的目標(biāo)尺寸(默認(rèn)為1024)
:param output_format: 輸出圖像格式(默認(rèn)為JPEG)
:param quality: 輸出圖像質(zhì)量(0-100,僅適用于某些格式)
:param recursive: 是否遞歸處理子文件夾
:param prefix: 輸出文件名前綴
:param suffix: 輸出文件名后綴
:param metadata_file: 元數(shù)據(jù)文件名
"""
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):
"""確保源文件夾存在,創(chuàng)建目標(biāo)文件夾(如果不存在)"""
if not os.path.exists(self.src_folder):
raise FileNotFoundError(f"源文件夾 {self.src_folder} 不存在")
os.makedirs(self.dest_folder, exist_ok=True)
print(f"目標(biāo)文件夾已準(zhǔn)備就緒: {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 圖像對(duì)象
: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)
# 創(chuàng)建白色背景圖像
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):
"""轉(zhuǎn)換圖像格式并重命名,同時(shí)進(jìn)行縮放和填充"""
results = []
for idx, (src_path, dest_subfolder) in enumerate(self.image_list):
filename = os.path.basename(src_path)
base_name, _ = os.path.splitext(filename)
# 構(gòu)建新文件名
new_filename = f"{self.prefix}{idx:02d}{self.suffix}.{self.output_format.lower()}"
dest_path = os.path.join(dest_subfolder, new_filename)
# 計(jì)算相對(duì)路徑(相對(duì)于dest_folder)
relative_path = os.path.relpath(dest_path, self.dest_folder)
try:
with Image.open(src_path) as img:
# 轉(zhuǎn)換為RGB模式(如果不是)
if img.mode not in ('RGB', 'RGBA'):
img = img.convert('RGB')
processed_img = self.resize_and_pad_image(img)
# 根據(jù)輸出格式設(shè)置保存參數(shù)
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} 時(shí)出錯(cuò): {e}")
results.append({"text": self.output_text, "file_name": relative_path, "error": str(e)})
return results
def generate_jsonl(self, results):
"""生成 metadata.jsonl 文件,每行一個(gè) JSON 對(duì)象,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):
"""執(zhí)行整個(gè)流程"""
self.setup_directories()
self.collect_images()
results = self.convert_and_rename_images()
self.generate_jsonl(results)
def main():
parser = argparse.ArgumentParser(description='圖像轉(zhuǎn)換器 - 處理圖像并生成JSONL元數(shù)據(jù)')
# 必需參數(shù)
parser.add_argument('--src', required=True, help='源圖像文件夾路徑')
parser.add_argument('--dest', required=True, help='目標(biāo)輸出文件夾路徑')
# 可選參數(shù)
parser.add_argument('--text', default="a dog", help='JSON中的text字段內(nèi)容')
parser.add_argument('--size', type=int, default=1024, help='輸出圖像的目標(biāo)尺寸')
parser.add_argument('--format', default="JPEG", choices=["JPEG", "PNG", "WEBP"], help='輸出圖像格式')
parser.add_argument('--quality', type=int, default=95, help='輸出圖像質(zhì)量(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='元數(shù)據(jù)文件名')
args = parser.parse_args()
# 創(chuàng)建并運(yùn)行轉(zhuǎn)換器
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()通過(guò)命令行直接運(yùn)腳本
python image_converter.py --src path/to/source --dest path/to/destination --text "a dog" --size 512 --recursive
或者在代碼中調(diào)用:
# 示例調(diào)用
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()
結(jié)果
{"text": "a dog", "file_name": "00.jpg"}
{"text": "a dog", "file_name": "01.jpg"}
Python 處理圖像并生成 JSONL 元數(shù)據(jù)文件 - 靈活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"):
"""
初始化圖像轉(zhuǎn)換器
:param src_folder: 源圖像文件夾路徑
:param dest_folder: 目標(biāo)輸出文件夾路徑
:param target_size: 輸出圖像的目標(biāo)尺寸(默認(rèn)為1024)
:param output_format: 輸出圖像格式(默認(rèn)為JPEG)
:param quality: 輸出圖像質(zhì)量(0-100,僅適用于某些格式)
:param recursive: 是否遞歸處理子文件夾
:param prefix: 輸出文件名前綴
:param suffix: 輸出文件名后綴
:param metadata_file: 元數(shù)據(jù)文件名
"""
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):
"""確保源文件夾存在,創(chuàng)建目標(biāo)文件夾(如果不存在)"""
if not os.path.exists(self.src_folder):
raise FileNotFoundError(f"源文件夾 {self.src_folder} 不存在")
os.makedirs(self.dest_folder, exist_ok=True)
print(f"目標(biāo)文件夾已準(zhǔn)備就緒: {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() # 按文件名排序
# 修復(fù)此處:添加for循環(huán)遍歷文件
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):
"""將文件名中的中文部分轉(zhuǎn)換為帶音調(diào)的拼音"""
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 "" # 無(wú)中文字符時(shí)返回空(可根據(jù)需求調(diào)整)
pinyin_list = []
for char in chinese_chars:
py = self.pinyin_converter.get_pinyin(char, tone_marks='numbers')
# 確保音調(diào)存在,默認(rèn)補(bǔ)1
if not any(c.isdigit() for c in py):
py += '1'
pinyin_list.append(py.lower()) # 轉(zhuǎn)換為小寫(xiě)
return '_'.join(pinyin_list)
def resize_and_pad_image(self, img):
"""
等比縮放圖像,并將其放入指定大小的白色背景正方形中
:param img: PIL 圖像對(duì)象
: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)
# 創(chuàng)建白色背景圖像
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):
"""轉(zhuǎn)換圖像格式并重命名,同時(shí)進(jìn)行縮放和填充"""
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)
# 構(gòu)建新文件名
new_filename = f"{self.prefix}{idx:02d}{self.suffix}.{self.output_format.lower()}"
dest_path = os.path.join(dest_subfolder, new_filename)
# 計(jì)算相對(duì)路徑(相對(duì)于dest_folder)
relative_path = os.path.relpath(dest_path, self.dest_folder)
try:
with Image.open(src_path) as img:
# 轉(zhuǎn)換為RGB模式(如果不是)
if img.mode not in ('RGB', 'RGBA'):
img = img.convert('RGB')
processed_img = self.resize_and_pad_image(img)
# 根據(jù)輸出格式設(shè)置保存參數(shù)
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, # 使用轉(zhuǎn)換后的拼音作為text
"file_name": relative_path
})
print(f"已保存: {relative_path} (text={pinyin_text})")
except Exception as e:
print(f"處理 {filename} 時(shí)出錯(cuò): {e}")
results.append({
"text": pinyin_text,
"file_name": relative_path,
"error": str(e)
})
return results
def generate_jsonl(self, results):
"""生成 metadata.jsonl 文件,每行一個(gè) JSON 對(duì)象"""
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):
"""執(zhí)行整個(gè)流程"""
self.setup_directories()
self.collect_images()
results = self.convert_and_rename_images()
self.generate_jsonl(results)
def main():
parser = argparse.ArgumentParser(description='圖像轉(zhuǎn)換器 - 處理圖像并生成JSONL元數(shù)據(jù)')
# 必需參數(shù)
parser.add_argument('--src', required=True, help='源圖像文件夾路徑')
parser.add_argument('--dest', required=True, help='目標(biāo)輸出文件夾路徑')
# 可選參數(shù)
parser.add_argument('--size', type=int, default=1024, help='輸出圖像的目標(biāo)尺寸')
parser.add_argument('--format', default="JPEG", choices=["JPEG", "PNG", "WEBP"], help='輸出圖像格式')
parser.add_argument('--quality', type=int, default=95, help='輸出圖像質(zhì)量(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='元數(shù)據(jù)文件名')
args = parser.parse_args()
# 創(chuàng)建并運(yùn)行轉(zhuǎn)換器
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,而是從圖像文件名中提取中文部分轉(zhuǎn)換為拼音.
新增拼音轉(zhuǎn)換函數(shù)filename_to_pinyin方法專(zhuān)門(mén)處理文件名,只保留中文字符并轉(zhuǎn)換為帶音調(diào)的拼音(如天.jpg→tian1)
使用方法
python converter.py --src ./images --dest ./processed_images --size 512 --format PNG
到此這篇關(guān)于Python處理圖像并生成JSONL元數(shù)據(jù)文件的文章就介紹到這了,更多相關(guān)Python處理圖像內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
關(guān)于torch.scatter與torch_scatter庫(kù)的使用整理
這篇文章主要介紹了關(guān)于torch.scatter與torch_scatter庫(kù)的使用整理,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-09-09
python+selenium定時(shí)爬取丁香園的新型冠狀病毒數(shù)據(jù)并制作出類(lèi)似的地圖(部署到云服務(wù)器)
這篇文章主要介紹了python+selenium定時(shí)爬取丁香園的新冠病毒每天的數(shù)據(jù)并制作出類(lèi)似的地圖(部署到云服務(wù)器),本文給大家介紹的非常詳細(xì),具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2020-02-02
解決python中顯示圖片的plt.imshow plt.show()內(nèi)存泄漏問(wèn)題
這篇文章主要介紹了解決python中顯示圖片的plt.imshow plt.show()內(nèi)存泄漏問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-04-04
python使用requests實(shí)現(xiàn)發(fā)送帶文件請(qǐng)求功能
這篇文章主要介紹了python使用requests實(shí)現(xiàn)發(fā)送帶文件請(qǐng)求,本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2022-12-12
Python實(shí)現(xiàn)PPT幻燈片的添加、刪除或隱藏操作
PowerPoint文檔是商務(wù)、教育、創(chuàng)意等各領(lǐng)域常見(jiàn)的用于展示、教育和傳達(dá)信息的格式,在制作PPT演示文稿時(shí),靈活地操作幻燈片是提高演示效果、優(yōu)化內(nèi)容組織的關(guān)鍵步驟,本文給大家介紹了Python 操作PPT幻燈片- 添加、刪除、或隱藏幻燈片,需要的朋友可以參考下2024-08-08
python實(shí)現(xiàn)馬丁策略回測(cè)3000只股票的實(shí)例代碼
這篇文章主要介紹了python實(shí)現(xiàn)馬丁策略回測(cè)3000只股票,本文給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2021-01-01
python內(nèi)置函數(shù)之eval函數(shù)詳解
這篇文章主要為大家介紹了python內(nèi)置函數(shù)之eval函數(shù),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下,希望能夠給你帶來(lái)幫助2022-01-01
Pytho樹(shù)的直徑的計(jì)算實(shí)現(xiàn)
樹(shù)的直徑是樹(shù)中任意兩個(gè)節(jié)點(diǎn)之間最長(zhǎng)路徑的長(zhǎng)度,本文主要介紹了Pytho樹(shù)的直徑的計(jì)算實(shí)現(xiàn),具有一定的參考價(jià)值,感興趣的可以了解一下2023-11-11

