580 lines
25 KiB
Python
580 lines
25 KiB
Python
# coding=utf-8
|
||
import logging
|
||
import os
|
||
import sys
|
||
import json
|
||
|
||
# Ensure sys path includes root for imports if not already
|
||
project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
if project_root not in sys.path:
|
||
sys.path.append(project_root)
|
||
|
||
from Util.VLMKit import VLMKit
|
||
from Apps.TeLaiDian.Kit import draw_rectangles, setup_logger, read_image, detect_price_info_container_cv
|
||
from Apps.TeLaiDian.Config.Setting import SAFE_EXCLUDE_RATIO, BOTTOM_SAFE_EXCLUDE_RATIO, MIN_CARD_HEIGHT
|
||
|
||
# 初始化日志
|
||
logger = setup_logger("ReadImageKit")
|
||
|
||
class ReadImageKit:
|
||
def __init__(self):
|
||
self.vlm = VLMKit()
|
||
|
||
async def find_price_tab_vlm(self, image_path):
|
||
"""
|
||
使用 VLM 在详情页顶部标签栏中寻找“价格”标签的点击位置
|
||
"""
|
||
prompt = """
|
||
分析这张特来电充电站详情页截图，找到顶部标签栏中“价格”两个字所在的点击区域中心。
|
||
要求：
|
||
1. 仅在页面最上方的标签栏里查找，该标签栏通常包含“价格 / 终端 / 电站 / 评论 / 周边”等文字。
|
||
2. 不要选择下面“价格信息”模块中的数字（例如 1.0689 元/度）或其它文本。
|
||
3. 不要选择最顶端系统状态栏或返回按钮等区域。
|
||
|
||
输出格式为 JSON：
|
||
{
|
||
"found": true/false,
|
||
"reason": "为什么认为这个位置是顶部“价格”标签",
|
||
"point": [x, y] // 归一化坐标，范围 [0-1000]
|
||
}
|
||
"""
|
||
try:
|
||
res_text = await self.vlm.analyze_image(image_path, prompt)
|
||
json_str = self.vlm.extract_json(res_text)
|
||
data = json.loads(json_str)
|
||
|
||
if data.get("found") and data.get("point"):
|
||
p = data["point"]
|
||
img = read_image(image_path)
|
||
if img is not None:
|
||
h, w = img.shape[:2]
|
||
actual_p = [int(p[0] * w / 1000), int(p[1] * h / 1000)]
|
||
bbox = [actual_p[0]-60, actual_p[1]-30, actual_p[0]+60, actual_p[1]+30]
|
||
draw_rectangles(image_path, bboxes=[bbox], click_points=[actual_p])
|
||
logger.info(f"已生成价格标签诊断图片: {image_path.replace('.jpg', '_tab_vl.jpg')}")
|
||
|
||
return data
|
||
except Exception as e:
|
||
logger.error(f"VLM 寻找价格标签失败: {e}")
|
||
return {"found": False}
|
||
|
||
async def find_price_entrance_vlm(self, image_path):
|
||
"""
|
||
使用 VLM 在详情页寻找价格入口
|
||
重点识别：上方是大字价格（如 1.3435元/度），下方紧跟“全部时段 >”文字的区域
|
||
"""
|
||
# 1. 尝试使用 CV 检测“价格信息”容器矩形
|
||
container_norm = detect_price_info_container_cv(image_path)
|
||
container_desc = ""
|
||
vlm_image_path = image_path
|
||
|
||
if container_norm:
|
||
# 如果找到了容器，绘制绿框诊断图
|
||
img = read_image(image_path)
|
||
if img is not None:
|
||
h, w = img.shape[:2]
|
||
x1 = int(container_norm[0] * w / 1000)
|
||
y1 = int(container_norm[1] * h / 1000)
|
||
x2 = int(container_norm[2] * w / 1000)
|
||
y2 = int(container_norm[3] * h / 1000)
|
||
|
||
# 在左侧划分出一个子区域（当前价区域）
|
||
# 价格信息容器通常左右平分，左侧是当前价，右侧是会员价
|
||
current_price_x2 = x1 + (x2 - x1) // 2
|
||
|
||
# 绘制绿框
|
||
# 我们把整个容器标绿，并特别说明左侧是目标
|
||
diag_path = image_path.replace(".jpg", "_price_box.jpg")
|
||
import cv2
|
||
diag_img = img.copy()
|
||
cv2.rectangle(diag_img, (x1, y1), (x2, y2), (0, 255, 0), 3)
|
||
cv2.line(diag_img, (current_price_x2, y1), (current_price_x2, y2), (0, 255, 0), 2)
|
||
from Apps.TeLaiDian.Kit import save_image
|
||
save_image(diag_path, diag_img)
|
||
logger.info(f"[CV] 已检测到价格容器并保存绿框诊断图: {diag_path}")
|
||
|
||
# 更新 VLM 使用的图片为带绿框的图片，或者在提示词中说明
|
||
vlm_image_path = diag_path
|
||
container_desc = f"\n**视觉辅助**：图中已用【绿色矩形框】标出了“价格信息”区域。该区域被中间竖线分为左右两块。请重点分析【左侧半块】（即“当前价”所在的灰色背景区域）。"
|
||
|
||
prompt = f"""
|
||
分析这张特来电充电站详情页截图，找到进入“电价详情”的点击入口。
|
||
{container_desc}
|
||
|
||
**核心识别目标**：
|
||
1. 在页面中上部区域（或绿框内的左侧部分）寻找价格显示模块。
|
||
2. 该模块包含**红色、加粗的大字价格数字**（例如：1.3435元/度）。
|
||
3. **绝对必要标识**：在价格数字的正下方，**必须能够清晰看到**文本“全部时段 >”或“全部时段”。
|
||
4. 点击目标是“全部时段 >”文字所在的中心位置。
|
||
|
||
**严格限制与排除**：
|
||
1. **严禁选择 y > 850 的区域**（即屏幕最底部的 15% 区域）。
|
||
2. **严禁选择底部的浮动条**：浮动条通常包含“扫码充电”按钮，严禁返回其坐标。
|
||
3. 如果绿框存在，请优先在绿框的【左侧区域】寻找。
|
||
|
||
输出格式为 JSON：
|
||
{{
|
||
"found": true/false,
|
||
"reason": "解释你看到了什么，是否在绿框内找到了‘全部时段 >’",
|
||
"point": [x, y] // 归一化坐标 [0-1000]
|
||
}}
|
||
"""
|
||
try:
|
||
res_text = await self.vlm.analyze_image(vlm_image_path, prompt)
|
||
json_str = self.vlm.extract_json(res_text)
|
||
data = json.loads(json_str)
|
||
|
||
# 保存诊断图片
|
||
if data.get("found") and data.get("point"):
|
||
p = data["point"]
|
||
img = read_image(image_path)
|
||
if img is not None:
|
||
h, w = img.shape[:2]
|
||
actual_p = [int(p[0] * w / 1000), int(p[1] * h / 1000)]
|
||
bbox = [actual_p[0]-60, actual_p[1]-40, actual_p[0]+60, actual_p[1]+40]
|
||
draw_rectangles(image_path, bboxes=[bbox], click_points=[actual_p])
|
||
logger.info(f"已生成价格入口诊断图片: {image_path.replace('.jpg', '_vl.jpg')}")
|
||
|
||
return data
|
||
except Exception as e:
|
||
logger.error(f"VLM 寻找价格入口失败: {e}")
|
||
return {"found": False}
|
||
|
||
async def find_price_tab_ocr(self, image_path):
|
||
"""
|
||
使用 OCR 在详情页顶部标签栏中寻找“价格”标签
|
||
不再使用 VLM
|
||
"""
|
||
from Apps.TeLaiDian.Kit import draw_rectangles
|
||
img = read_image(image_path)
|
||
if img is None:
|
||
return {"found": False}
|
||
|
||
h, w = img.shape[:2]
|
||
reader = get_easyocr_reader(gpu=True)
|
||
|
||
# 限制只在顶部 30% 区域查找标签
|
||
roi = img[0:int(h*0.3), :]
|
||
results = reader.read_text(roi)
|
||
|
||
for (quad, text, prob) in results:
|
||
if '价格' in text and prob > 0.3:
|
||
# 转换坐标 (加上 ROI 的偏移，这里 y 偏移是 0)
|
||
res = reader.get_normalized_rect(quad, w, h)
|
||
center_x = (res[0] + res[2]) // 2
|
||
center_y = (res[1] + res[3]) // 2
|
||
|
||
# 绘制诊断图
|
||
actual_p = [int(center_x * w / 1000), int(center_y * h / 1000)]
|
||
bbox = [int(res[0] * w / 1000), int(res[1] * h / 1000), int(res[2] * w / 1000), int(res[3] * h / 1000)]
|
||
draw_rectangles(image_path, bboxes=[bbox], click_points=[actual_p])
|
||
logger.info(f"[OCR] 已定位顶部价格标签并保存诊断图: {image_path.replace('.jpg', '_tab_vl.jpg')}")
|
||
|
||
return {
|
||
"found": True,
|
||
"reason": f"OCR 定位到‘{text}’",
|
||
"point": [center_x, center_y]
|
||
}
|
||
|
||
logger.warning(f"[OCR] 未能在顶部区域定位到‘价格’标签")
|
||
return {"found": False}
|
||
|
||
async def find_price_tab_ocr(self, image_path):
|
||
"""
|
||
使用 OCR 在详情页寻找顶部“价格”标签
|
||
"""
|
||
img = read_image(image_path)
|
||
if img is None:
|
||
return {"found": False}
|
||
h, w = img.shape[:2]
|
||
|
||
from Util.EasyOcrKit import get_easyocr_reader
|
||
reader = get_easyocr_reader(gpu=True)
|
||
|
||
# 顶部标签栏通常在屏幕上部 10%-30% 区域
|
||
# 我们先识别所有文字，然后过滤出“价格”
|
||
results = reader.read_text(img)
|
||
for (quad, text, prob) in results:
|
||
if "价格" in text and prob > 0.3:
|
||
# 获取中心点
|
||
rect = reader.get_normalized_rect(quad, w, h)
|
||
center_x = (rect[0] + rect[2]) // 2
|
||
center_y = (rect[1] + rect[3]) // 2
|
||
|
||
# 校验位置：必须在屏幕上半部分
|
||
if center_y < 500:
|
||
logger.info(f"[OCR] 找到顶部价格标签: '{text}', 置信度: {prob:.4f}, 坐标: [{center_x}, {center_y}]")
|
||
return {
|
||
"found": True,
|
||
"point": [center_x, center_y]
|
||
}
|
||
|
||
return {"found": False}
|
||
|
||
async def find_price_entrance_ocr(self, image_path):
|
||
"""
|
||
使用 OCR 在详情页寻找价格入口 (全部时段 >)
|
||
不再使用 VLM 和 掩码 CV
|
||
"""
|
||
container_norm = detect_price_info_container_cv(image_path)
|
||
if container_norm:
|
||
# 找到矩形框 [x1, y1, x2, y2]
|
||
x1, y1, x2, y2 = container_norm
|
||
# 计算中心点
|
||
center_x = (x1 + x2) // 2
|
||
center_y = (y1 + y2) // 2
|
||
|
||
# 绘制诊断图 (复用之前的逻辑)
|
||
img = read_image(image_path)
|
||
if img is not None:
|
||
h, w = img.shape[:2]
|
||
actual_p = [int(center_x * w / 1000), int(center_y * h / 1000)]
|
||
bbox = [int(x1 * w / 1000), int(y1 * h / 1000), int(x2 * w / 1000), int(y2 * h / 1000)]
|
||
draw_rectangles(image_path, bboxes=[bbox], click_points=[actual_p])
|
||
logger.info(f"[OCR] 已定位价格入口并保存诊断图: {image_path.replace('.jpg', '_vl.jpg')}")
|
||
|
||
return {
|
||
"found": True,
|
||
"reason": "OCR 精准定位到‘全部时段’文字",
|
||
"point": [center_x, center_y]
|
||
}
|
||
|
||
logger.warning(f"[OCR] 未能在页面中定位到‘全部时段’")
|
||
return {"found": False}
|
||
|
||
async def find_close_button_vlm(self, image_path):
|
||
"""
|
||
使用 VLM 在页面上寻找关闭按钮（用于清理广告弹窗）
|
||
"""
|
||
prompt = """
|
||
分析这张截图，判断是否存在弹窗广告或遮罩层。
|
||
如果存在，请找到关闭按钮（通常是圆圈里的 X，或者写着“跳过”、“关闭”的按钮）。
|
||
|
||
**重要警告**：
|
||
1. 严禁选择屏幕最右上角的按钮（微信小程序的“胶囊按钮”，包含三个点和圆圈）。
|
||
2. 广告关闭按钮通常在弹窗的边缘，或者是页面中心大弹窗的某个角落。
|
||
|
||
输出格式为 JSON：
|
||
{
|
||
"has_ad": true/false,
|
||
"reason": "简单描述发现的弹窗",
|
||
"close_point": [x, y] // 归一化坐标 [0-1000]，如果不存在则为 null
|
||
}
|
||
"""
|
||
try:
|
||
res_text = await self.vlm.analyze_image(image_path, prompt)
|
||
json_str = self.vlm.extract_json(res_text)
|
||
data = json.loads(json_str)
|
||
|
||
# 保存诊断图片
|
||
if data.get("has_ad") and data.get("close_point"):
|
||
p = data["close_point"]
|
||
# 转换坐标
|
||
img = read_image(image_path)
|
||
if img is not None:
|
||
h, w = img.shape[:2]
|
||
actual_p = [int(p[0] * w / 1000), int(p[1] * h / 1000)]
|
||
# 生成虚拟 bbox
|
||
bbox = [actual_p[0]-50, actual_p[1]-30, actual_p[0]+50, actual_p[1]+30]
|
||
draw_rectangles(image_path, bboxes=[bbox], click_points=[actual_p])
|
||
logger.info(f"已生成广告关闭诊断图片: {image_path.replace('.jpg', '_vl.jpg')}")
|
||
|
||
return data
|
||
except Exception as e:
|
||
logger.error(f"VLM 寻找关闭按钮失败: {e}")
|
||
return {"has_ad": False}
|
||
|
||
async def check_warm_popup_vlm(self, image_path):
|
||
"""
|
||
检测详情页是否存在“温馨提示/下次再说”弹窗
|
||
"""
|
||
prompt = """
|
||
分析这张特来电充电站详情页截图，判断是否存在带有“下次再说”或“下次现说”文案的温馨提示弹窗。
|
||
要求：
|
||
1. 只关注覆盖在详情页上方的弹窗或遮罩，其上包含“下次再说”“下次现说”等文字按钮。
|
||
2. 不要将正常页面中的列表项、价格卡片、终端状态等区域误判为弹窗。
|
||
3. 如果存在该弹窗，请给出“下次再说”按钮的大致点击中心位置。
|
||
|
||
输出格式为 JSON：
|
||
{
|
||
"has_popup": true/false,
|
||
"reason": "为什么认为有或没有温馨提示弹窗",
|
||
"button_point": [x, y] // 归一化坐标，范围 [0-1000]，没有则为 null
|
||
}
|
||
"""
|
||
try:
|
||
res_text = await self.vlm.analyze_image(image_path, prompt)
|
||
json_str = self.vlm.extract_json(res_text)
|
||
data = json.loads(json_str)
|
||
|
||
if data.get("has_popup") and data.get("button_point"):
|
||
p = data["button_point"]
|
||
img = read_image(image_path)
|
||
if img is not None:
|
||
h, w = img.shape[:2]
|
||
actual_p = [int(p[0] * w / 1000), int(p[1] * h / 1000)]
|
||
bbox = [actual_p[0]-80, actual_p[1]-40, actual_p[0]+80, actual_p[1]+40]
|
||
draw_rectangles(image_path, bboxes=[bbox], click_points=[actual_p])
|
||
logger.info(f"已生成温馨提示弹窗诊断图片: {image_path.replace('.jpg', '_warm_vl.jpg')}")
|
||
|
||
return data
|
||
except Exception as e:
|
||
logger.error(f"VLM 检测温馨提示弹窗失败: {e}")
|
||
return {"has_popup": False}
|
||
|
||
async def check_wrong_page_vlm(self, image_path):
|
||
"""
|
||
检查是否误触进入了错误的页面（如：新人福利、我的卡券、活动页等）
|
||
"""
|
||
prompt = """
|
||
分析这张截图，判断这是否是一个真实的“充电站详情页”。
|
||
|
||
**识别准则**：
|
||
1. 真正的“详情页”必须包含：充电站的具体名称、电价列表、终端状态（空闲/占用）等信息。
|
||
2. 如果页面标题是“新人福利专区”、“活动规则”、“我的卡券”、“领券中心”或类似的营销活动页，则判定为错误页面。
|
||
|
||
输出格式为 JSON：
|
||
{
|
||
"is_detail_page": true/false,
|
||
"page_type": "detail" / "marketing" / "coupons" / "other",
|
||
"reason": "判断依据"
|
||
}
|
||
"""
|
||
try:
|
||
res_text = await self.vlm.analyze_image(image_path, prompt)
|
||
json_str = self.vlm.extract_json(res_text)
|
||
data = json.loads(json_str)
|
||
return data
|
||
except Exception as e:
|
||
logger.error(f"VLM 检查页面类型失败: {e}")
|
||
return {"is_detail_page": True} # 默认认为是详情页，避免死循环
|
||
|
||
async def analyze_detail_price(self, image_path):
|
||
"""
|
||
分析详情页截图，提取电价信息，包括优惠价、PLUS价和挂牌价
|
||
"""
|
||
prompt = """
|
||
分析这张充电站价格详情页截图，提取**分时电价表**。
|
||
对于每个时段，请识别并提取以下所有价格信息（如果存在）：
|
||
1. 优惠价 (通常是红色或加粗的大字，作为默认 price)
|
||
2. PLUS会员价 (标有 "PLUS" 标签的价格)
|
||
3. 挂牌价 (标有 "挂牌价" 标签的价格)
|
||
4. 电费 (Base electricity price)
|
||
5. 服务费 (Service fee)
|
||
|
||
请提取每个时段的：
|
||
- start: 开始时间 (HH:MM)
|
||
- end: 结束时间 (HH:MM)
|
||
- price: 优惠价 (元/度)
|
||
- plus_price: PLUS会员价 (元/度)
|
||
- market_price: 挂牌价 (元/度)
|
||
- elec_price: 电费 (元/度)
|
||
- service_price: 服务费 (元/度)
|
||
|
||
输出格式为 JSON 数组：
|
||
[
|
||
{
|
||
"start": "16:00",
|
||
"end": "21:00",
|
||
"price": 1.3435,
|
||
"plus_price": 1.3035,
|
||
"market_price": 1.4435,
|
||
"elec_price": 0.9435,
|
||
"service_price": 0.4000
|
||
},
|
||
...
|
||
]
|
||
注意：
|
||
- 如果某个字段缺失，请设为 null。
|
||
- 确保 price 包含电费和服务费的总和。
|
||
- 如果无法识别任何价格信息，请返回空数组 []。
|
||
"""
|
||
try:
|
||
res_text = await self.vlm.analyze_image(image_path, prompt)
|
||
logger.info(f"VLM Price Analysis Result for {os.path.basename(image_path)}: {res_text[:200]}...")
|
||
|
||
json_str = self.vlm.extract_json(res_text)
|
||
prices = json.loads(json_str)
|
||
|
||
normalized_prices = []
|
||
if isinstance(prices, list):
|
||
for p in prices:
|
||
new_p = p.copy()
|
||
if 'time_range' in p and ('start' not in p or 'end' not in p):
|
||
tr = p['time_range'].replace('~', '-').replace(' ', '')
|
||
parts = tr.split('-')
|
||
if len(parts) >= 2:
|
||
new_p['start'] = parts[0]
|
||
new_p['end'] = parts[1]
|
||
|
||
if 'price' not in p:
|
||
if 'total_price' in p:
|
||
new_p['price'] = p['total_price']
|
||
elif 'elec_price' in p and 'service_price' in p:
|
||
try:
|
||
new_p['price'] = float(p['elec_price']) + float(p['service_price'])
|
||
except:
|
||
pass
|
||
normalized_prices.append(new_p)
|
||
return normalized_prices
|
||
return []
|
||
except Exception as e:
|
||
logger.error(f"分析电价详情失败: {e}")
|
||
return []
|
||
|
||
async def analyze_detail_basic_info(self, image_path):
|
||
"""
|
||
分析详情页首屏截图，提取场站名称和精确地址
|
||
"""
|
||
prompt = """
|
||
分析这张充电站详情页首屏截图，提取：
|
||
1. 场站名称 (通常在页面中部，大字体)
|
||
2. 详细地址 (通常在名称下方或页面下半部分，伴有地址图标)
|
||
|
||
输出格式为 JSON：
|
||
{
|
||
"name": "xxx充电站",
|
||
"address": "xxx省xxx市xxx区xxx路xxx号"
|
||
}
|
||
"""
|
||
try:
|
||
res_text = await self.vlm.analyze_image(image_path, prompt)
|
||
json_str = self.vlm.extract_json(res_text)
|
||
return json.loads(json_str)
|
||
except Exception as e:
|
||
logger.error(f"分析详情页基础信息失败: {e}")
|
||
return {}
|
||
|
||
async def analyze_station_list(self, image_path):
|
||
"""
|
||
分析场站列表页图片，提取场站位置和基本信息
|
||
"""
|
||
prompt = """
|
||
分析这张充电站列表截图，提取所有真实的充电站卡片。
|
||
|
||
要求：
|
||
1. 忽略页面上半部分（如顶部导航栏、搜索框、广告 Banner、筛选标签等）。
|
||
2. 仅识别下半部分一条条“充电站卡片”，每张卡片通常包含：场站名称、评分、最近充电时间、距离、价格、快/慢空闲数量等。
|
||
3. 不要把同一张卡片拆成多块；每条场站只对应一个矩形框。
|
||
|
||
对于每张卡片，请输出：
|
||
- name: 场站名称
|
||
- address: 场站地址（如果无法确定可置为 null）
|
||
- point: 卡片中心点击坐标 [x, y]，使用归一化坐标 [0-1000]（0 表示最左/最上，1000 表示最右/最下）
|
||
- bbox: 卡片外接矩形边界 [x1, y1, x2, y2]，同样使用归一化坐标 [0-1000]
|
||
|
||
以 JSON 数组形式输出，例如：
|
||
[
|
||
{
|
||
"name": "某某充电站",
|
||
"address": "某某路 100 号",
|
||
"point": [500, 750],
|
||
"bbox": [50, 600, 950, 820]
|
||
}
|
||
]
|
||
"""
|
||
|
||
try:
|
||
res_text = await self.vlm.analyze_image(image_path, prompt)
|
||
json_str = self.vlm.extract_json(res_text)
|
||
vlm_results = json.loads(json_str)
|
||
|
||
final_stations = []
|
||
vlm_list = vlm_results if isinstance(vlm_results, list) else []
|
||
img = read_image(image_path)
|
||
h, w = img.shape[:2] if img is not None else (2400, 1080)
|
||
y_threshold = h * SAFE_EXCLUDE_RATIO
|
||
bottom_threshold = int(h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO))
|
||
|
||
def to_pixel(v, max_len):
|
||
if v is None:
|
||
return None
|
||
try:
|
||
fv = float(v)
|
||
except:
|
||
return None
|
||
if fv <= 1000.0:
|
||
return int(fv * max_len / 1000.0)
|
||
return int(fv)
|
||
|
||
for res in vlm_list:
|
||
if not res:
|
||
continue
|
||
p = res.get("point")
|
||
b = res.get("bbox")
|
||
actual_p = None
|
||
actual_bbox = None
|
||
|
||
if p and len(p) == 2:
|
||
px = to_pixel(p[0], w)
|
||
py = to_pixel(p[1], h)
|
||
if px is not None and py is not None:
|
||
actual_p = [px, py]
|
||
|
||
if b and len(b) == 4:
|
||
x1 = to_pixel(b[0], w)
|
||
y1 = to_pixel(b[1], h)
|
||
x2 = to_pixel(b[2], w)
|
||
y2 = to_pixel(b[3], h)
|
||
if None not in (x1, y1, x2, y2):
|
||
actual_bbox = [max(0, int(x1)), max(0, int(y1)), min(w, int(x2)), min(h, int(y2))]
|
||
|
||
if actual_p is None and actual_bbox:
|
||
actual_p = [(actual_bbox[0] + actual_bbox[2]) // 2, (actual_bbox[1] + actual_bbox[3]) // 2]
|
||
|
||
if actual_p is None and actual_bbox is None:
|
||
continue
|
||
|
||
if actual_p and (actual_p[1] < y_threshold or actual_p[1] > bottom_threshold):
|
||
logger.warning(f"过滤掉可能的误点击点 (y={actual_p[1]}): {res.get('name')}")
|
||
continue
|
||
|
||
if actual_bbox is None and actual_p:
|
||
half_w = int(w * 0.4)
|
||
half_h = max(MIN_CARD_HEIGHT // 2, 90)
|
||
x1 = max(0, actual_p[0] - half_w)
|
||
x2 = min(w, actual_p[0] + half_w)
|
||
y1 = max(0, actual_p[1] - half_h)
|
||
y2 = min(h, actual_p[1] + half_h)
|
||
actual_bbox = [x1, y1, x2, y2]
|
||
|
||
if actual_bbox is not None:
|
||
bx1, by1, bx2, by2 = actual_bbox
|
||
current_h = by2 - by1
|
||
if current_h < MIN_CARD_HEIGHT * 0.8 or by2 > bottom_threshold:
|
||
continue
|
||
|
||
final_stations.append({
|
||
"name": res.get("name"),
|
||
"address": res.get("address"),
|
||
"point": actual_p,
|
||
"bbox": actual_bbox
|
||
})
|
||
|
||
if final_stations:
|
||
final_stations.sort(key=lambda s: (s.get("point")[1] if s.get("point") else (s.get("bbox")[1] if s.get("bbox") else 0)))
|
||
|
||
# 保存诊断图片 (_vl.jpg, _flag.jpg)
|
||
if final_stations:
|
||
debug_bboxes = []
|
||
debug_points = []
|
||
for s in final_stations:
|
||
if s.get("bbox"):
|
||
debug_bboxes.append(s["bbox"])
|
||
if s.get("point"):
|
||
debug_points.append(s["point"])
|
||
|
||
# 如果没有 bbox，根据 point 生成一个虚拟 bbox 供可视化
|
||
if not debug_bboxes and debug_points:
|
||
for p in debug_points:
|
||
debug_bboxes.append([p[0]-50, p[1]-30, p[0]+50, p[1]+30])
|
||
|
||
if debug_bboxes or debug_points:
|
||
draw_rectangles(image_path, bboxes=debug_bboxes, click_points=debug_points)
|
||
logger.info(f"已生成诊断图片: {image_path.replace('.jpg', '_vl.jpg')} 和 _flag.jpg")
|
||
|
||
return final_stations
|
||
except Exception as e:
|
||
logger.error(f"分析列表页失败: {e}")
|
||
return []
|