当前位置：首页 > article >正文

【Python】使用Python实现调用API获取图片存储到本地

article 2025/9/10 8:03:33

使用Python实现调用API获取图片存储到本地

在这里插入图片描述

1、项目概述

开发一个自动化工具，用于从JSON数据源中提取图像ID，通过调用指定API获取未经压缩的原始图像文件，并确保下载结果与Postman等工具直接调用API获取的文件完全一致。

2、核心功能

自动解析含嵌套结构的JSON输入文件

API请求模拟引擎

精确复现Postman请求行为

支持自动重定向跟踪

禁用自动解压缩（保留原始二进制流）

3、环境准备

# 安装依赖
pip install requests python-magic-bin==0.4.24  # Windows
pip install requests python-magic           # Linux/macOS

编辑input.json文件格式：

[{"postimage": "{...}","preimage": "{...}","original_image_id": "6cd6187a-a20e-f011-998a-000d3ac8927d"},// 更多数据条目...
]

4、代码实现

import json
import requests
import base64
import os
import mimetypes# 配置参数
INPUT_JSON = "input.json"  # 输入的JSON文件路径
OUTPUT_DIR = "raw_images"   # 图片输出目录（修改目录以示区别）
API_URL = "https://prod-53.southeastasia.logic.azure.com:443/workflows/8de920a489c7490a9e2207bd538ce964/triggers/manual/paths/invoke?api-version=2016-06-01&sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=qO01WQl5tzDZeGpeVid43SpndovXbq4G7QO0NWcz7B8"def determine_image_type(base64_str):"""通过数据URI或魔术数字检测图片类型"""try:# 处理数据URI格式if base64_str.startswith("data:image/"):mime_type = base64_str.split(";")[0].split(":")[1]return mimetypes.guess_extension(mime_type) or ".png"# 通过魔术数字检测原始二进制格式decoded = base64.b64decode(base64_str)if decoded.startswith(b'\xff\xd8\xff'):return ".jpg"elif decoded.startswith(b'\x89PNG\r\n\x1a\n'):return ".png"elif decoded.startswith(b'GIF87a') or decoded.startswith(b'GIF89a'):return ".gif"elif decoded.startswith(b'RIFF') and decoded[8:12] == b'WEBP':return ".webp"except Exception:passreturn ".bin"  # 未知类型def process_images():# 创建输出目录os.makedirs(OUTPUT_DIR, exist_ok=True)# 读取JSON文件try:with open(INPUT_JSON, 'r', encoding='utf-8') as f:data = json.load(f)except Exception as e:print(f"读取JSON文件失败: {e}")return# 遍历处理每个条目for idx, item in enumerate(data, 1):image_id = item.get("original_image_id")if not image_id:print(f"条目 {idx} 缺少original_image_id，已跳过")continueprint(f"正在处理 ({idx}/{len(data)}): {image_id}")# 调用APItry:response = requests.post(API_URL,json={"EntityGuid": image_id},headers={"Content-Type": "application/json"},timeout=60  # 延长超时时间)print(f"响应状态码: {response.status_code}")response.raise_for_status()except requests.exceptions.RequestException as e:print(f"API调用失败: {str(e)[:200]}")  # 显示部分错误信息continue# 处理base64数据try:base64_str = response.text# 检测并分割数据URIif "," in base64_str:header, payload = base64_str.split(",", 1)file_ext = determine_image_type(header)base64_str = payloadelse:decoded = base64.b64decode(base64_str)file_ext = determine_image_type(base64_str)# 解码base64image_data = base64.b64decode(base64_str)# 生成文件名（保留原始格式）output_path = os.path.join(OUTPUT_DIR, f"{image_id}{file_ext}")# 保存原始二进制数据with open(output_path, "wb") as f:f.write(image_data)print(f"原始图片已保存至: {output_path} ({len(image_data):,} bytes)")except Exception as e:print(f"处理失败: {str(e)[:200]}")if __name__ == "__main__":process_images()