把不同兼容接口响应归一成图片二进制。
| 8 | |
| 9 | |
| 10 | class ImageResponseExtractor: |
| 11 | """把不同兼容接口响应归一成图片二进制。""" |
| 12 | |
| 13 | def __init__(self, download_image: Callable[[str], bytes]): |
| 14 | self.download_image = download_image |
| 15 | |
| 16 | def extract_from_images_response(self, result: Dict[str, Any]) -> bytes: |
| 17 | data = result.get("data") |
| 18 | if not isinstance(data, list) or not data: |
| 19 | raise ValueError( |
| 20 | "图片接口未返回 data 图片数据。\n" |
| 21 | f"响应内容: {str(result)[:500]}" |
| 22 | ) |
| 23 | |
| 24 | item = data[0] |
| 25 | if not isinstance(item, dict): |
| 26 | raise ValueError(f"图片接口 data 格式异常: {str(item)[:300]}") |
| 27 | |
| 28 | if item.get("b64_json"): |
| 29 | image_data = self._decode_base64_image(item["b64_json"]) |
| 30 | logger.info(f"图片响应解析成功: b64_json, {len(image_data)} bytes") |
| 31 | return image_data |
| 32 | |
| 33 | if item.get("url"): |
| 34 | logger.info("图片响应返回 URL，开始下载图片") |
| 35 | return self.download_image(item["url"]) |
| 36 | |
| 37 | raise ValueError( |
| 38 | "无法从图片接口响应中提取图片数据：未找到 b64_json 或 url。\n" |
| 39 | f"响应片段: {str(item)[:500]}" |
| 40 | ) |
| 41 | |
| 42 | def extract_from_chat_response(self, result: Dict[str, Any]) -> bytes: |
| 43 | choices = result.get("choices") |
| 44 | if not isinstance(choices, list) or not choices: |
| 45 | raise ValueError( |
| 46 | "Chat 图片接口未返回 choices 数据。\n" |
| 47 | f"响应内容: {str(result)[:500]}" |
| 48 | ) |
| 49 | |
| 50 | content = choices[0].get("message", {}).get("content") |
| 51 | if not isinstance(content, str): |
| 52 | raise ValueError( |
| 53 | "Chat 图片接口响应 content 格式异常。\n" |
| 54 | f"响应内容: {str(result)[:500]}" |
| 55 | ) |
| 56 | |
| 57 | data_urls = re.findall(r"!\[.*?\]\((data:image/[^;]+;base64,[^\s\)]+)\)", content) |
| 58 | if data_urls: |
| 59 | logger.info("从 Markdown 中提取到 Base64 图片数据") |
| 60 | return self._decode_base64_image(data_urls[0]) |
| 61 | |
| 62 | urls = re.findall(r"!\[.*?\]\((https?://[^\s\)]+)\)", content) |
| 63 | if urls: |
| 64 | logger.info(f"从 Markdown 中提取到 {len(urls)} 个图片 URL") |
| 65 | return self.download_image(urls[0]) |
| 66 | |
| 67 | stripped = content.strip() |
no outgoing calls