Claude API实战：如何高效上传图片并处理媒体文件

1次阅读

没有评论

共计 2592 个字符，预计需要花费 7 分钟才能阅读完成。

移动端图片自动旋转问题：开发者上传手机拍摄的图片时，发现图片在 Claude 平台显示方向错误。这是由于移动设备记录的 EXIF 方向信息未被正确处理，而 Claude API 默认不会自动修正方向。
大文件上传超时：当尝试上传超过 20MB 的高清图片时，HTTP 请求经常因超时失败。Claude API 对单次请求有默认 15 秒超时限制，且文档未明确说明分块上传机制。

Claude API 支持以下常见图片 MIME 类型，但需注意：

image/jpeg (必须包含 JFIF 标识)
image/png (建议移除不必要的元数据块)
image/webp (仅支持 Lossy 模式)

异常情况处理：

上传 image/heic 需先转换为 JPEG
SVG 文件必须声明为image/svg+xml
动态 GIF 会被转换为静态首帧

import aiohttp
import os
from tqdm import tqdm

async def chunked_upload(file_path, api_key, chunk_size=5*1024*1024):
    """
    分块上传文件到 Claude API

    参数:
        file_path: 本地文件路径
        api_key: Claude API 密钥
        chunk_size: 分块大小(默认 5MB)
    """
    file_size = os.path.getsize(file_path)
    headers = {'Authorization': f'Bearer {api_key}',
        'Content-Range': f'bytes 0-{min(chunk_size, file_size)-1}/{file_size}'
    }

    with open(file_path, 'rb') as f, tqdm(total=file_size, unit='B', unit_scale=True) as pbar:
        chunk = f.read(chunk_size)
        async with aiohttp.ClientSession() as session:
            while chunk:
                try:
                    async with session.post(
                        'https://api.claude.ai/v1/uploads',
                        headers=headers,
                        data=chunk
                    ) as resp:
                        if resp.status == 308:  # 需要继续上传
                            range_header = resp.headers.get('Range')
                            next_byte = int(range_header.split('-')[-1]) + 1
                            headers['Content-Range'] = f'bytes {next_byte}-{min(next_byte+chunk_size, file_size)-1}/{file_size}'
                            f.seek(next_byte)
                            chunk = f.read(chunk_size)
                            pbar.update(len(chunk))
                        elif resp.status == 201:
                            return await resp.json()
                        else:
                            raise Exception(f"上传失败: {await resp.text()}")
                except Exception as e:
                    # 实现重试逻辑...
                    raise

使用 Pillow 处理 EXIF 方向并优化文件：

from PIL import Image, ImageOps
import io

def process_image(input_bytes):
    """
    处理图片方向并优化格式

    返回:
        优化后的 JPEG 二进制数据
    """
    img = Image.open(io.BytesIO(input_bytes))

    # EXIF 方向修正
    img = ImageOps.exif_transpose(img)

    # 转换为 RGB 模式(处理可能存在的 CMYK 等问题)
    if img.mode != 'RGB':
        img = img.convert('RGB')

    # 输出优化后的 JPEG
    output = io.BytesIO()
    img.save(output, format='JPEG', quality=85, optimize=True)
    return output.getvalue()

记录已上传的字节位置到持久化存储
每次重试前检查存储的进度
从断点处继续上传时需验证文件内容哈希

错误码	处理方案	重试间隔
413	减小分块大小	立即重试
429	指数退避(从 2 秒开始)	2^n 秒
500	更换上传节点	随机 5 -10 秒

移动端开发需特别注意：

iOS 照片通常包含Orientation=6(逆时针 90°)
部分 Android 设备会丢失 EXIF 信息
建议在上传前统一调用ImageOps.exif_transpose

import asyncio
from PIL import Image

async def upload_image(file_path, api_key):
    """完整的图片上传流程示例"""
    try:
        # 1. 预处理图片
        with open(file_path, 'rb') as f:
            processed = process_image(f.read())

        # 2. 分块上传
        temp_path = '/tmp/processed.jpg'
        with open(temp_path, 'wb') as f:
            f.write(processed)

        result = await chunked_upload(temp_path, api_key)
        print(f"上传成功: {result['url']}")

    except aiohttp.ClientError as e:
        print(f"网络错误: {str(e)}")
    except IOError as e:
        print(f"文件处理错误: {str(e)}")
    except Exception as e:
        print(f"未知错误: {str(e)}")

# 使用示例
asyncio.run(upload_image('photo.jpg', 'your_api_key'))