This commit is contained in:
HuangHai
2026-01-12 21:33:21 +08:00
parent 2051cb61c2
commit 4612ee2d81
18 changed files with 457 additions and 191 deletions

View File

@@ -21,6 +21,16 @@ from Apps.AiTeJiYiChong import Kit
logger = logging.getLogger(__name__)
# 场站过滤黑名单
STATION_EXCLUDED_TITLES = {
"推荐站点", "距离优先", "价格最低", "空闲较多", "筛选", "地图", "列表",
"我的订单", "收藏站点", "优惠券", "站内搜索"
}
STATION_BLACKLIST_KEYWORDS = [
"广告", "点击进入", "立即参与", "咨询电话", "活动详情", "导航到", "停车说明"
]
class ReadImageKit:
_client = AsyncOpenAI(
@@ -42,14 +52,17 @@ class ReadImageKit:
" - 场站名称 (station_name);\n"
" - 距离信息 (distance, 例如 '2.82km' 或 '90m'),通常位于卡片右侧蓝色胶囊区域内;\n"
" - 金额/电费 (price,例如 '1.2500'),通常以红色字体显示;\n"
" - 充电枪信息 (piles,包含'快'或'慢'的类型、总枪数和空闲枪数,例如 '快 4/4')。\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"
"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"
@@ -303,15 +316,27 @@ class ReadImageKit:
img_input = {"url": f"data:image/jpeg;base64,{encoded}"}
prompt = (
"请识别图片中的充电站卡片信息,并以 JSON 格式输出:\n"
"你是一个专业的充电桩信息提取助手。请识别图片中充电站卡片的核心信息。\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\": \"空闲/总数(如 4/4)\",\n"
" \"distance\": \"距离(如 2.5km)\",\n"
" \"parking\": \"停车描述\"\n"
" \"price\": \"价格\",\n"
" \"piles\": [\n"
" {\"type\": \"快\", \"free\": 4, \"total\": 4}\n"
" ],\n"
" \"distance\": \"距离\",\n"
" \"parking\": \"停车信息\"\n"
"}\n"
"只输出 JSON,不要有其他文字。"
"只输出 JSON,不要有任何解释文字。"
)
try:
@@ -410,6 +435,20 @@ class ReadImageKit:
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
@@ -460,3 +499,55 @@ class ReadImageKit:
except Exception as e:
logger.error(f"Error calling VL model: {e}")
return []
@classmethod
async def find_time_price_button_coordinate(cls, image_path: str, device_info=None):
"""
使用 VL 模型定位详情页中的 '分时价格' 按钮
"""
if not os.path.exists(image_path):
return {}
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"
"请直接以 JSON 格式输出该按钮的中心点坐标:\n"
"{\"uia_center_x\": 500, \"uia_center_y\": 500}\n"
"注意:坐标系为 0-1000 归一化坐标,其中 (0,0) 为左上角,(1000,1000) 为右下角。\n"
"只输出 JSON,不要有任何解释文字。"
)
try:
resp = await cls._client.chat.completions.create(
model=VL_MODEL_NAME,
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": img_input}
]
}
],
max_tokens=100,
temperature=0.01
)
content = resp.choices[0].message.content.strip()
raw = cls._extract_json(content)
res = json.loads(raw)
# 转换 0-1000 坐标到实际像素坐标
if res.get("uia_center_x") is not None and device_info:
w = device_info.get("width", FALLBACK_WIDTH)
h = device_info.get("height", FALLBACK_HEIGHT)
res["uia_center_x"] = int(res["uia_center_x"] * w / 1000)
res["uia_center_y"] = int(res["uia_center_y"] * h / 1000)
return res
except Exception as e:
logger.error(f"VL find_time_price_button_coordinate error: {e}")
return {}