'commit'
This commit is contained in:
HuangHai
@@ -48,39 +48,6 @@ class ReadImageKit:
|
||||
"productName": "generic"
|
||||
}
|
||||
|
||||
_prompt = (
|
||||
"仅输出JSON数组（不含任何说明文字），按从左到右、从上到下的顺序识别图片中由【绿色方框】标识的充电站区域。识别规则如下：\n"
|
||||
"1. 必须是图中用绿色实线方框圈出的区域。\n"
|
||||
"2. 每一个卡片区域必须同时具备以下所有要素，否则严禁识别：\n"
|
||||
" - 场站名称 (station_name)；\n"
|
||||
" - 距离信息 (distance, 例如 '2.82km' 或 '90m')，通常位于卡片右侧蓝色胶囊区域内；\n"
|
||||
" - 金额/电费 (price，例如 '1.2500')，通常以红色字体显示；\n"
|
||||
" - **充电枪信息 (piles)**: 必须包含类型('快充'、'慢充'、'超充')、总枪数和空闲枪数。识别规则如下：\n"
|
||||
" - 如果明确标有'快'、'慢'或'超'，则对应为'快充'、'慢充'、'超充'。\n"
|
||||
" - 如果图中只显示数字（如 '34/42'）而没有明确标明类型，请统一识别为 `type: \"慢充\"`。\n"
|
||||
" - 必须以 JSON 对象列表形式输出，例如：`[{\"type\": \"快充\", \"free\": 4, \"total\": 4}, {\"type\": \"慢充\", \"free\": 1, \"total\": 2}]`。\n"
|
||||
"3. 如果绿色方框内缺少上述任何一项要素，说明它不是真正的场站卡片，请直接跳过。\n"
|
||||
"\n"
|
||||
"JSON对象字段要求：\n"
|
||||
"1. b_use: 状态标识（1或0）。如果场站名称为灰色或带有“暂停使用”等标签，则为0，否则为1。\n"
|
||||
"2. station_name: 场站名称；\n"
|
||||
"3. price: 一度电的价格（数字，如 1.2500）；\n"
|
||||
"4. piles: 充电枪列表 [{type: '快'|'慢'|'超', free: 4, total: 4}]；\n"
|
||||
"5. parking: 停车费用描述（通常在蓝色'P'图标后，例如 '免费停车三小时'）；\n"
|
||||
"6. distance: 距离信息字符串（例如 '2.82km' 或 '90m'）；\n"
|
||||
"7. bounds: {x1,y1,x2,y2} 区域像素坐标（0-1000）；\n"
|
||||
"8. bounds_norm: {left,top,right,bottom} 归一化坐标(0-1)；\n"
|
||||
"9. station_name_bounds: 场站名称文字区域坐标 {x1,y1,x2,y2}（0-1000）；\n"
|
||||
"10. station_name_bounds_norm: 场站名称文字归一化坐标(0-1)。\n"
|
||||
"\n"
|
||||
"重要约束：\n"
|
||||
"A. 严禁识别未被绿色方框圈出的区域。如顶部的“长春市”选择、顶部的搜索框、以及中间的“推荐站点”等标签。\n"
|
||||
"B. 真正的场站卡片在绿色方框内包含：场站名称、金额（红色文字）、距离（右侧蓝色背景内）、充电枪状态（绿色或蓝色徽章，格式为 闲x/x 或 x/x）。\n"
|
||||
"C. 严禁将顶部的功能图标（如我的订单、收藏站点等）误认为场站卡片。\n"
|
||||
"\n"
|
||||
"严格返回纯JSON格式。"
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _extract_json(text: str) -> str:
|
||||
if not text:
|
||||
@@ -126,6 +93,79 @@ class ReadImageKit:
|
||||
"- 严禁输出 Markdown 代码块标签，严格返回纯JSON对象。"
|
||||
)
|
||||
|
||||
@classmethod
|
||||
async def detect_ad_popup(cls, image_path: str, device_info=None):
|
||||
"""
|
||||
使用 VL 模型检测是否存在弹窗广告，并返回关闭按钮坐标
|
||||
"""
|
||||
if not os.path.exists(image_path):
|
||||
return None
|
||||
|
||||
with open(image_path, "rb") as image_file:
|
||||
encoded_image = base64.b64encode(image_file.read()).decode("utf-8")
|
||||
img_input = {"url": f"data:image/jpeg;base64,{encoded_image}"}
|
||||
|
||||
# 构建提示词
|
||||
prompt = (
|
||||
"请仔细检查这张图片中是否存在**弹窗广告**或**悬浮广告**。\n"
|
||||
"广告可能有以下几种形式：\n"
|
||||
"1. **屏幕中央的大型弹窗广告**：通常遮挡了页面内容，内容多为优惠券、活动推广、或“免费停车”类提示。\n"
|
||||
"2. **悬浮广告**：一个小型的可点击图标，带有明显的关闭按钮。\n"
|
||||
"3. **底部的横幅广告**：带有明显的关闭按钮。\n\n"
|
||||
"请返回关闭按钮的中心坐标。\n\n"
|
||||
"请以纯 JSON 格式输出：\n"
|
||||
"{\n"
|
||||
" \"has_ad\": true/false,\n"
|
||||
" \"ad_type\": \"center\" | \"bottom\" | \"other\",\n"
|
||||
" \"close_point\": [x, y] // 0-1000 归一化坐标\n"
|
||||
"}\n"
|
||||
"如果没有广告，请返回 `{\"has_ad\": false}`。"
|
||||
)
|
||||
|
||||
try:
|
||||
resp = await cls._client.chat.completions.create(
|
||||
model=VL_MODEL_NAME,
|
||||
timeout=30,
|
||||
messages=[
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "image_url", "image_url": img_input},
|
||||
{"type": "text", "text": prompt},
|
||||
],
|
||||
}
|
||||
],
|
||||
)
|
||||
content = resp.choices[0].message.content.strip()
|
||||
raw_json = cls._extract_json(content)
|
||||
result = json.loads(raw_json)
|
||||
|
||||
if result.get("has_ad") and result.get("close_point"):
|
||||
norm_point = result["close_point"]
|
||||
|
||||
# 转换坐标
|
||||
if len(norm_point) == 2:
|
||||
if not device_info:
|
||||
device_info = cls._FALLBACK_DEVICE_INFO
|
||||
|
||||
w = device_info.get("displayWidth", FALLBACK_WIDTH)
|
||||
h = device_info.get("displayHeight", FALLBACK_HEIGHT)
|
||||
|
||||
x = int(norm_point[0] / 1000 * w)
|
||||
y = int(norm_point[1] / 1000 * h)
|
||||
|
||||
return {
|
||||
"x": x,
|
||||
"y": y,
|
||||
"ad_type": result.get("ad_type", "unknown")
|
||||
}
|
||||
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to detect ad popup: {e}")
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _to_minutes(t_str: str) -> int:
|
||||
"""HH:MM -> 分钟数"""
|
||||
@@ -309,202 +349,116 @@ class ReadImageKit:
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
async def recognize_card_with_vl(image_path: str):
|
||||
async def parse_vl_image(image_path: str, json_metadata, device_info=None):
|
||||
"""
|
||||
对卡片切片进行 VL 识别
|
||||
基于 _vl.jpg (带绿框) 和 JSON 元数据进行整页识别 (节约 TOKEN)
|
||||
"""
|
||||
if not os.path.exists(image_path):
|
||||
return None
|
||||
|
||||
with open(image_path, "rb") as f:
|
||||
encoded = base64.b64encode(f.read()).decode("utf-8")
|
||||
img_input = {"url": f"data:image/jpeg;base64,{encoded}"}
|
||||
|
||||
prompt = (
|
||||
"你是一个专业的充电桩信息提取助手。请识别图片中充电站卡片的核心信息。\n"
|
||||
"注意以下识别规则：\n"
|
||||
"1. **场站名称 (station_name)**: 场站名称通常位于卡片的第一行（左上角），字体最大且加粗。严禁提取'距离优先'、'空闲较多'、'价格最低'等筛选标签。严禁将卡片中下部的描述信息（如'导航到...'、'停车说明...'、'距您...'等）识别为名称。名称通常是一个具体的地点或品牌名。\n"
|
||||
"2. **价格 (price)**: 提取数字部分，通常以'元/度'结尾，字体颜色通常为红色或加粗。\n"
|
||||
"3. **桩数 (piles)**: 识别所有类型的充电桩。请提取'快'、'慢'或'超'字样及其对应的'空闲/总数'（如 '快 4/4'）。\n"
|
||||
" - 如果明确标有'快'、'慢'或'超'，则对应为'快充'、'慢充'、'超充'。\n"
|
||||
" - 如果图中只显示数字（如 '34/42'）而没有明确标明类型，请统一识别为 `type: \"慢充\"`。\n"
|
||||
" - 必须以 JSON 对象列表形式输出，例如：`[{\"type\": \"快充\", \"free\": 4, \"total\": 4}]`。\n"
|
||||
"4. **距离 (distance)**: 提取如 '2.5km' 格式，通常在蓝色胶囊背景内。\n"
|
||||
"5. **停车 (parking)**: 简要描述停车收费信息，通常在蓝色'P'图标后。\n\n"
|
||||
"请以 JSON 格式输出：\n"
|
||||
"{\n"
|
||||
" \"station_name\": \"场站名称\",\n"
|
||||
" \"price\": \"价格\",\n"
|
||||
" \"piles\": [\n"
|
||||
" {\"type\": \"快\", \"free\": 4, \"total\": 4}\n"
|
||||
" ],\n"
|
||||
" \"distance\": \"距离\",\n"
|
||||
" \"parking\": \"停车信息\"\n"
|
||||
"}\n"
|
||||
"只输出 JSON，不要有任何解释文字。"
|
||||
)
|
||||
|
||||
try:
|
||||
resp = await ReadImageKit._client.chat.completions.create(
|
||||
model=VL_MODEL_NAME,
|
||||
timeout=30,
|
||||
messages=[
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": prompt},
|
||||
{"type": "image_url", "image_url": img_input}
|
||||
]
|
||||
}
|
||||
],
|
||||
max_tokens=300,
|
||||
temperature=0.01
|
||||
)
|
||||
content = resp.choices[0].message.content.strip()
|
||||
raw = ReadImageKit._extract_json(content)
|
||||
return json.loads(raw)
|
||||
except Exception as e:
|
||||
logger.error(f"VL recognize_card_with_vl error: {e}")
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
async def get_stations_hybrid(image_path, device_info=None):
|
||||
"""
|
||||
混合识别模式：图形学切片 + VL 模型识别
|
||||
"""
|
||||
if device_info is None:
|
||||
device_info = ReadImageKit._FALLBACK_DEVICE_INFO
|
||||
|
||||
# 1. 使用 Kit 中的图形学算法识别卡片区域
|
||||
json_data = Kit.crop_cards_from_image(image_path, save_debug=True)
|
||||
if not json_data or not json_data.get("cards"):
|
||||
logger.warning("No cards detected by graphical slicing.")
|
||||
if not json_metadata or "cards" not in json_metadata:
|
||||
return []
|
||||
|
||||
# 2. 对每个卡片区域进行 VL 识别
|
||||
tasks = []
|
||||
card_metas = []
|
||||
|
||||
# 为了使用 VL，我们需要将每个卡片切出来保存到临时文件
|
||||
temp_dir = os.path.join(os.path.dirname(image_path), "Patches")
|
||||
os.makedirs(temp_dir, exist_ok=True)
|
||||
|
||||
img = Image.open(image_path).convert("RGB")
|
||||
|
||||
for card in json_data["cards"]:
|
||||
rect = card["rect"] # [x1, y1, x2, y2]
|
||||
patch_path = os.path.join(temp_dir, f"patch_{card['id']}_{uuid.uuid4().hex[:8]}.jpg")
|
||||
|
||||
# 裁剪并保存
|
||||
patch = img.crop((rect[0], rect[1], rect[2], rect[3]))
|
||||
patch.save(patch_path)
|
||||
|
||||
card_metas.append({
|
||||
"meta": card,
|
||||
"patch_path": patch_path
|
||||
})
|
||||
tasks.append(ReadImageKit.recognize_card_with_vl(patch_path))
|
||||
|
||||
# 并行执行 VL 识别
|
||||
results = await asyncio.gather(*tasks)
|
||||
|
||||
final_stations = []
|
||||
for i, res in enumerate(results):
|
||||
meta = card_metas[i]["meta"]
|
||||
patch_path = card_metas[i]["patch_path"]
|
||||
|
||||
if res and isinstance(res, dict) and res.get("station_name"):
|
||||
# 注入点击坐标和原始区域信息
|
||||
res["uia_center_x"] = meta["click_point"][0]
|
||||
res["uia_center_y"] = meta["click_point"][1]
|
||||
res["rect"] = meta["rect"]
|
||||
|
||||
# 转换 bounds 到 0-1000 空间
|
||||
w, h = img.size
|
||||
rect = meta["rect"]
|
||||
res["bounds"] = [
|
||||
int(rect[0] * 1000 / w),
|
||||
int(rect[1] * 1000 / h),
|
||||
int(rect[2] * 1000 / w),
|
||||
int(rect[3] * 1000 / h)
|
||||
]
|
||||
|
||||
final_stations.append(res)
|
||||
logger.info(f"卡片 {meta['id']} 识别成功: {res['station_name']}")
|
||||
else:
|
||||
logger.warning(f"卡片 {meta['id']} 识别失败或无名称")
|
||||
|
||||
# 清理临时切片文件
|
||||
try:
|
||||
if os.path.exists(patch_path):
|
||||
os.remove(patch_path)
|
||||
except:
|
||||
pass
|
||||
|
||||
# 3. 后置过滤
|
||||
if final_stations:
|
||||
processed_excluded_titles = {str(x).replace(" ", "").strip() for x in STATION_EXCLUDED_TITLES}
|
||||
filtered = []
|
||||
for item in final_stations:
|
||||
name = item.get("station_name")
|
||||
normalized_name = str(name).replace(" ", "").strip()
|
||||
if normalized_name in processed_excluded_titles:
|
||||
continue
|
||||
if any(kw in normalized_name for kw in STATION_BLACKLIST_KEYWORDS):
|
||||
continue
|
||||
filtered.append(item)
|
||||
final_stations = filtered
|
||||
|
||||
return final_stations
|
||||
|
||||
@classmethod
|
||||
async def get_stations_from_image(cls, image_path: str, device_info=None):
|
||||
"""
|
||||
使用 Qwen-VL 模型从截图中识别充电站列表
|
||||
"""
|
||||
if device_info is None:
|
||||
device_info = cls._FALLBACK_DEVICE_INFO
|
||||
|
||||
if not os.path.exists(image_path):
|
||||
logger.error(f"Image not found: {image_path}")
|
||||
return []
|
||||
|
||||
# 将图片转换为 Base64
|
||||
with open(image_path, "rb") as image_file:
|
||||
encoded_image = base64.b64encode(image_file.read()).decode("utf-8")
|
||||
with open(image_path, "rb") as f:
|
||||
encoded = base64.b64encode(f.read()).decode("utf-8")
|
||||
img_input = {"url": f"data:image/jpeg;base64,{encoded}"}
|
||||
|
||||
cards_meta = json_metadata["cards"]
|
||||
# 按 id 或 Y 坐标排序，确保顺序一致
|
||||
cards_meta.sort(key=lambda x: x["rect"][1])
|
||||
|
||||
prompt = (
|
||||
"图片中用绿色矩形框标记了若干个充电站卡片区域。\n"
|
||||
"请按从上到下的顺序，依次识别每个绿色框内的场站信息，并返回一个JSON数组。\n"
|
||||
"数组中元素的顺序必须与图片中绿色框从上到下的顺序严格一致。\n"
|
||||
"如果某个框内不是有效的场站卡片（例如是广告或推荐标签），请返回null或空对象，不要跳过顺序。\n"
|
||||
"\n"
|
||||
"每个JSON对象包含以下字段：\n"
|
||||
"1. b_use: 状态标识（1或0）。如果场站名称为灰色或带有“暂停使用”等标签，则为0，否则为1。\n"
|
||||
"2. station_name: 场站名称；\n"
|
||||
"3. price: 价格（数字，如 1.2500）；\n"
|
||||
"4. piles: 充电枪列表 [{type: '快充'|'慢充'|'超充', free: 4, total: 4}]。如果图中只显示数字（如 '34/42'）而没有明确标明类型，请统一识别为 `type: \"慢充\"`；\n"
|
||||
"5. parking: 停车费用描述（通常在蓝色'P'图标后，例如 '免费停车三小时'）；\n"
|
||||
"6. distance: 距离信息字符串（例如 '2.82km' 或 '90m'）。\n"
|
||||
"\n"
|
||||
"严格返回纯JSON格式。"
|
||||
)
|
||||
|
||||
try:
|
||||
response = await cls._client.chat.completions.create(
|
||||
resp = await ReadImageKit._client.chat.completions.create(
|
||||
model=VL_MODEL_NAME,
|
||||
timeout=30,
|
||||
messages=[
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": cls._prompt},
|
||||
{
|
||||
"type": "image_url",
|
||||
"image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"},
|
||||
},
|
||||
{"type": "image_url", "image_url": img_input},
|
||||
{"type": "text", "text": prompt},
|
||||
],
|
||||
}
|
||||
},
|
||||
],
|
||||
max_tokens=2000,
|
||||
temperature=0.01
|
||||
)
|
||||
|
||||
content = response.choices[0].message.content
|
||||
logger.info(f"VL Model Response: {content}")
|
||||
content = resp.choices[0].message.content or ""
|
||||
raw = ReadImageKit._extract_json(content)
|
||||
llm_data = json.loads(raw)
|
||||
|
||||
json_str = cls._extract_json(content)
|
||||
stations = json.loads(json_str)
|
||||
final_stations = []
|
||||
if isinstance(llm_data, list):
|
||||
# 尝试与 cards_meta 对齐
|
||||
count = min(len(llm_data), len(cards_meta))
|
||||
for i in range(count):
|
||||
item = llm_data[i]
|
||||
meta = cards_meta[i]
|
||||
|
||||
# 后处理：如果 bounds 是归一化的，则转换为像素坐标（如果需要）
|
||||
# 或者如果 bounds 是 0-1000 的，则保持原样或按需转换
|
||||
return stations
|
||||
if not item or not isinstance(item, dict):
|
||||
continue
|
||||
|
||||
name = item.get("station_name")
|
||||
if not name or name == "未知":
|
||||
continue
|
||||
|
||||
# 注入元数据中的点击坐标
|
||||
click_pt = meta.get("click_point", [0, 0])
|
||||
item["uia_center_x"] = click_pt[0]
|
||||
item["uia_center_y"] = click_pt[1]
|
||||
item["rect"] = meta.get("rect")
|
||||
|
||||
# 转换 bounds 到 0-1000 空间 (用于匹配)
|
||||
img_w = 1080
|
||||
img_h = 2400
|
||||
if device_info:
|
||||
img_w = device_info.get("displayWidth") or device_info.get("width") or 1080
|
||||
img_h = device_info.get("displayHeight") or device_info.get("height") or 2400
|
||||
|
||||
rect = meta["rect"]
|
||||
item["bounds"] = [
|
||||
int(rect[0] * 1000 / img_w),
|
||||
int(rect[1] * 1000 / img_h),
|
||||
int(rect[2] * 1000 / img_w),
|
||||
int(rect[3] * 1000 / img_h)
|
||||
]
|
||||
|
||||
final_stations.append(item)
|
||||
|
||||
# 后置过滤
|
||||
if final_stations:
|
||||
processed_excluded_titles = {str(x).replace(" ", "").strip() for x in STATION_EXCLUDED_TITLES}
|
||||
filtered = []
|
||||
for item in final_stations:
|
||||
name = item.get("station_name")
|
||||
normalized_name = str(name).replace(" ", "").strip()
|
||||
if normalized_name in processed_excluded_titles:
|
||||
continue
|
||||
if any(kw in normalized_name for kw in STATION_BLACKLIST_KEYWORDS):
|
||||
continue
|
||||
filtered.append(item)
|
||||
final_stations = filtered
|
||||
|
||||
return final_stations
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error calling VL model: {e}")
|
||||
logger.error(f"Error calling VL model for page recognition: {e}")
|
||||
return []
|
||||
|
||||
@classmethod
|
||||
|
||||
Reference in New Issue
Block a user