BBIT-Kai
153 lines
4.6 KiB
Python
153 lines
4.6 KiB
Python
import json
|
||
import re
|
||
|
||
from langchain_core.messages import HumanMessage
|
||
|
||
from config.llm import *
|
||
|
||
|
||
def get_ticket_response(image_url: str):
|
||
# 构建 prompt
|
||
prompt_text = f"""
|
||
你是一位专业的图像分析AI。你的任务是仔细分析提供的图片内容，并按JSON格式输出结果。
|
||
|
||
## JSON输出结构及字段说明:
|
||
# 蚕茧检测信息数据模型（ImageDescription）
|
||
|
||
该模型用于描述蚕茧检测信息，每条记录包含以下字段：
|
||
|
||
## 1. 含水率 (`moisture_content`)
|
||
- **类型**：浮点数
|
||
- **描述**：茧的含水量，单位为百分比（%）
|
||
|
||
## 2. 下足茧重量 (`cocoon_weight`)
|
||
- **类型**：浮点数
|
||
- **描述**：下足茧的重量，单位为克（可带小数）
|
||
|
||
## 3. 非好蛹粒数 (`defective_pupa_count`)
|
||
- **类型**：整数
|
||
- **描述**：不合格蛹的数量，即非好蛹的个数
|
||
|
||
## 4. 鲜壳量 (`fresh_shell_weight`)
|
||
- **类型**：浮点数
|
||
- **描述**：鲜壳重量，单位为克（可带小数）
|
||
|
||
## 5. 小样粒数 (`sample_count`)
|
||
- **类型**：整数
|
||
- **描述**：检测使用的小样数量，即用于检测的茧粒数
|
||
|
||
## 6. 净重合计 (`net_weight_total`)
|
||
- **类型**：浮点数
|
||
- **描述**：所有样品的净重总和，单位为克
|
||
|
||
## 7. 仪评人姓名 (`evaluator`)
|
||
- **类型**：字符串
|
||
- **描述**：进行仪器检测的人员姓名，可能为空
|
||
|
||
## 8. 复核人员姓名 (`reviewer`)
|
||
- **类型**：字符串
|
||
- **描述**：复核人员姓名，可能为空
|
||
|
||
---
|
||
|
||
### 注意事项
|
||
- 所有字段都是必填的（required），在 JSON 实例中必须提供值
|
||
- 浮点数字段可以包含小数，整数字段只能是整数
|
||
- `evaluator` 和 `reviewer` 可以为空字符串，但字段必须存在
|
||
|
||
最后，只输出严格的 JSON 格式，不要包含其他文字、markdown等内容。
|
||
"""
|
||
messages = [
|
||
HumanMessage(
|
||
content=[
|
||
{"type": "text", "text": prompt_text},
|
||
{"type": "image_url", "image_url": {"url": image_url}},
|
||
]
|
||
)
|
||
]
|
||
# 直接调用模型
|
||
llmRes = llmVision.invoke(messages).content
|
||
# 去掉 ```json 和 ``` 包裹
|
||
cleaned = re.sub(r"^```json\s*|\s*```$", "", llmRes.strip())
|
||
# 解析 JSON
|
||
jsonRes = json.loads(cleaned)
|
||
jsonRes["barcode"] = decode_barcode(image_url)
|
||
return jsonRes
|
||
|
||
|
||
import os
|
||
import base64
|
||
import tempfile
|
||
import requests
|
||
from pyzxing import BarCodeReader
|
||
from PIL import Image
|
||
from io import BytesIO
|
||
from fastapi import UploadFile
|
||
|
||
|
||
def decode_barcode(input_data) -> str:
|
||
"""
|
||
自动识别输入类型并解析条码：
|
||
- URL 字符串
|
||
- Base64 字符串
|
||
- UploadFile / bytes / 文件对象
|
||
返回第一个条码的内容，解析失败返回空字符串
|
||
"""
|
||
# ---------------- 输入处理 ----------------
|
||
img = None
|
||
|
||
# URL
|
||
if isinstance(input_data, str) and (
|
||
input_data.startswith("http://") or input_data.startswith("https://")
|
||
):
|
||
response = requests.get(input_data)
|
||
response.raise_for_status()
|
||
img = Image.open(BytesIO(response.content))
|
||
|
||
# Base64 字符串
|
||
elif isinstance(input_data, str):
|
||
# 过滤 data URI 前缀
|
||
if "," in input_data:
|
||
input_data = input_data.split(",")[1]
|
||
try:
|
||
img_data = base64.b64decode(input_data)
|
||
img = Image.open(BytesIO(img_data))
|
||
except Exception:
|
||
raise ValueError("无法解析 Base64 字符串")
|
||
|
||
# UploadFile / bytes / 文件对象
|
||
elif isinstance(input_data, UploadFile):
|
||
content = input_data.file.read()
|
||
input_data.file.seek(0)
|
||
img = Image.open(BytesIO(content))
|
||
elif isinstance(input_data, bytes):
|
||
img = Image.open(BytesIO(input_data))
|
||
elif hasattr(input_data, "read"): # 文件对象
|
||
content = input_data.read()
|
||
if hasattr(input_data, "seek"):
|
||
input_data.seek(0)
|
||
img = Image.open(BytesIO(content))
|
||
else:
|
||
raise ValueError("不支持的输入类型")
|
||
|
||
# ---------------- 临时文件处理 ----------------
|
||
tmp_fd, tmp_path = tempfile.mkstemp(suffix=".png")
|
||
try:
|
||
with open(tmp_path, "wb") as f:
|
||
img.save(f, format="PNG")
|
||
|
||
# ---------------- 调用 pyzxing ----------------
|
||
reader = BarCodeReader()
|
||
result = reader.decode(tmp_path)
|
||
|
||
if result:
|
||
parsed = result[0].get("parsed", "")
|
||
if isinstance(parsed, bytes):
|
||
parsed = parsed.decode("utf-8")
|
||
return parsed
|
||
return ""
|
||
finally:
|
||
os.close(tmp_fd)
|
||
if os.path.exists(tmp_path):
|
||
os.remove(tmp_path)
|