huanghai/aiData
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

'commit'

Browse Source
This commit is contained in:
HuangHai
2026-01-12 08:26:13 +08:00
parent 4a45b27638
commit 7b4548c12c
13 changed files with 471 additions and 449 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
Apps/XinDianTu/Config/__pycache__/Setting.cpython-310.pyc Normal file
View File

Binary file not shown.

119
Apps/XinDianTu/Crawler.py
View File

@@ -16,18 +16,21 @@ if project_root not in sys.path:
import uiautomator2 as u2 import uiautomator2 as u2
from Core.BaseCrawler import BaseCrawler from Core.BaseCrawler import BaseCrawler
from Util import Kit from Apps.XinDianTu import Kit
from Util.Kit import take_screenshot from Apps.XinDianTu.Kit import take_screenshot
from Util.ObsUtil import ObsUploader from Util.ObsUtil import ObsUploader
from Util.RedisKit import RedisKit from Util.RedisKit import RedisKit
from Util.PaddleOCRKit import get_ocr_kit from Util.PaddleOCRKit import get_ocr_kit
import cv2 import cv2
from Apps.XinDianTu.Service import XinDianTuService from Apps.XinDianTu.Service import XinDianTuService
from Config.Config import ( from Config.Config import (
OBS_TMP_PREFIX, CDN_DOMAIN, SCROLL_DISTANCE_RATIO, OBS_TMP_PREFIX, CDN_DOMAIN, TEMP_IMAGE_DIR
)
from Apps.XinDianTu.Config.Setting import (
SCROLL_DISTANCE_RATIO,
MAX_SCROLLS, REDIS_STATION_EXPIRE, MAX_SCROLLS, REDIS_STATION_EXPIRE,
WAIT_DETAIL_PAGE_LOAD, WAIT_BACK_TO_LIST, WAIT_AFTER_SCROLL, WAIT_DETAIL_PAGE_LOAD, WAIT_BACK_TO_LIST, WAIT_AFTER_SCROLL,
MAX_CRAWL_DISTANCE, TEMP_IMAGE_DIR MAX_CRAWL_DISTANCE
) )
# --- 用户配置区域 --- # --- 用户配置区域 ---
@@ -57,14 +60,15 @@ logger = logging.getLogger("StationList")
# 强制设置所有相关模块的日志级别为 INFO，防止被第三方库干扰 # 强制设置所有相关模块的日志级别为 INFO，防止被第三方库干扰
logging.getLogger("PaddleOCRKit").setLevel(logging.INFO) logging.getLogger("PaddleOCRKit").setLevel(logging.INFO)
logging.getLogger("Util.Kit").setLevel(logging.INFO)
logging.getLogger("OpenXinDianTu").setLevel(logging.INFO) logging.getLogger("OpenXinDianTu").setLevel(logging.INFO)
logging.getLogger("FullProcess").setLevel(logging.INFO) logging.getLogger("FullProcess").setLevel(logging.INFO)
class XinDianTuCrawler(BaseCrawler): class XinDianTuCrawler(BaseCrawler):
""" """
新电途小程序爬虫实现 新电途小程序爬虫实现
""" """
def __init__(self, service=None): def __init__(self, service=None):
super().__init__(service) super().__init__(service)
# 初始化配置参数 # 初始化配置参数
@@ -90,6 +94,7 @@ class XinDianTuCrawler(BaseCrawler):
async def crawl_detail(self, station_info): async def crawl_detail(self, station_info):
pass pass
async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
""" """
获取场站列表并处理翻页 获取场站列表并处理翻页
@@ -140,7 +145,8 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
logger.info(f"截图实际尺寸: {img_w}x{img_h}") logger.info(f"截图实际尺寸: {img_w}x{img_h}")
# 如果截图尺寸与窗口尺寸不符，更新 device_info 以适配坐标换算 # 如果截图尺寸与窗口尺寸不符，更新 device_info 以适配坐标换算
if img_w != w or img_h != h: if img_w != w or img_h != h:
logger.warning(f"检测到截图尺寸 ({img_w}x{img_h}) 与窗口尺寸 ({w}x{h}) 不一致，将以截图尺寸为准进行坐标换算。") logger.warning(
f"检测到截图尺寸 ({img_w}x{img_h}) 与窗口尺寸 ({w}x{h}) 不一致，将以截图尺寸为准进行坐标换算。")
device_info['width'] = img_w device_info['width'] = img_w
device_info['height'] = img_h device_info['height'] = img_h
except Exception as e: except Exception as e:
@@ -158,7 +164,8 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
vl_img_path = screenshot_path.replace(".jpg", "_vl.jpg") vl_img_path = screenshot_path.replace(".jpg", "_vl.jpg")
if not os.path.exists(json_path) or not os.path.exists(vl_img_path): if not os.path.exists(json_path) or not os.path.exists(vl_img_path):
logger.error(f"❌ Step [2/6] 失败: 未生成有效的 JSON 或 VL 图片 (JSON: {os.path.exists(json_path)}, VL: {os.path.exists(vl_img_path)})") logger.error(
f"❌ Step [2/6] 失败: 未生成有效的 JSON 或 VL 图片 (JSON: {os.path.exists(json_path)}, VL: {os.path.exists(vl_img_path)})")
logger.warning("可能未识别到任何卡片，跳过当前页") logger.warning("可能未识别到任何卡片，跳过当前页")
continue continue
@@ -201,53 +208,55 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
cropped_card = original_img[y1:y2, x1:x2] cropped_card = original_img[y1:y2, x1:x2]
# 识别 # 识别
logger.info(f"正在识别第 {idx+1}/{num_cards} 个卡片 (区域: {rect})...") logger.info(f"正在识别第 {idx + 1}/{num_cards} 个卡片 (区域: {rect})...")
t_card_start = time.time() t_card_start = time.time()
parsed_data = ocr_kit.recognize(cropped_card) parsed_data = ocr_kit.recognize(cropped_card)
t_card_end = time.time() t_card_end = time.time()
# Log detailed OCR result # Log detailed OCR result
logger.info(f"卡片 {idx+1} OCR 识别耗时: {t_card_end - t_card_start:.2f}s") logger.info(f"卡片 {idx + 1} OCR 识别耗时: {t_card_end - t_card_start:.2f}s")
if parsed_data and parsed_data.get("station_name"): if parsed_data and parsed_data.get("station_name"):
# 格式化数据以匹配原有结构 # 格式化数据以匹配原有结构
piles_list = parsed_data.get("piles", []) piles_list = parsed_data.get("piles", [])
piles_str_parts = [] piles_str_parts = []
for p in piles_list: for p in piles_list:
p_type = p.get("type", "") p_type = p.get("type", "")
p_free = p.get("free", 0) p_free = p.get("free", 0)
p_total = p.get("total", 0) p_total = p.get("total", 0)
piles_str_parts.append(f"{p_type}:{p_free}/{p_total}") piles_str_parts.append(f"{p_type}:{p_free}/{p_total}")
piles_str = " ".join(piles_str_parts) piles_str = " ".join(piles_str_parts)
station_info = { station_info = {
"station_name": parsed_data.get("station_name"), "station_name": parsed_data.get("station_name"),
"price": str(parsed_data.get("price")) if parsed_data.get("price") is not None else "", "price": str(parsed_data.get("price")) if parsed_data.get("price") is not None else "",
"pro_price": "", # 暂未解析 "pro_price": "", # 暂未解析
"piles": piles_str, "piles": piles_str,
"distance": parsed_data.get("distance", ""), "distance": parsed_data.get("distance", ""),
"uia_center_x": card["click_point"][0], "uia_center_x": card["click_point"][0],
"uia_center_y": card["click_point"][1], "uia_center_y": card["click_point"][1],
"tags": parsed_data.get("tags", []), "tags": parsed_data.get("tags", []),
"parking_info": parsed_data.get("parking", "") "parking_info": parsed_data.get("parking", "")
} }
logger.info(f"✅ 成功解析场站: {station_info['station_name']} | 距离: {station_info['distance']}") logger.info(
f"✅ 成功解析场站: {station_info['station_name']} | 距离: {station_info['distance']}")
# 立即记录场站基础状态信息 (电桩、价格等) # 立即记录场站基础状态信息 (电桩、价格等)
try: try:
await service.record_station_status(station_info) await service.record_station_status(station_info)
except Exception as e: except Exception as e:
logger.error(f"❌ 记录场站状态失败: {e}") logger.error(f"❌ 记录场站状态失败: {e}")
stations.append(station_info) stations.append(station_info)
else: else:
logger.warning(f"⚠️ 卡片 {idx+1} 未能识别出有效场站名称") logger.warning(f"⚠️ 卡片 {idx + 1} 未能识别出有效场站名称")
except Exception as e: except Exception as e:
logger.error(f"❌ Local OCR processing failed: {e}", exc_info=True) logger.error(f"❌ Local OCR processing failed: {e}", exc_info=True)
logger.info(f"Step [5/6] 本地 OCR 解析完成 (总耗时: {time.time() - t_ocr_total:.2f}s) | 成功识别: {len(stations)}/{len(json_metadata.get('cards', []))}") logger.info(
f"Step [5/6] 本地 OCR 解析完成 (总耗时: {time.time() - t_ocr_total:.2f}s) | 成功识别: {len(stations)}/{len(json_metadata.get('cards', []))}")
if not stations: if not stations:
logger.warning("当前页面未检测到任何有效的场站信息") logger.warning("当前页面未检测到任何有效的场站信息")
@@ -304,7 +313,8 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
dist_log_part = f" (解析: {dist_km:.2f}km / 限: {MAX_CRAWL_DISTANCE}km)" dist_log_part = f" (解析: {dist_km:.2f}km / 限: {MAX_CRAWL_DISTANCE}km)"
if dist_km >= MAX_CRAWL_DISTANCE: if dist_km >= MAX_CRAWL_DISTANCE:
logger.info(f"当前场站 {name} 距离 {dist_km}km (原始: {distance_str}) 已达到或超过限制 {MAX_CRAWL_DISTANCE}km。") logger.info(
f"当前场站 {name} 距离 {dist_km}km (原始: {distance_str}) 已达到或超过限制 {MAX_CRAWL_DISTANCE}km。")
max_distance_reached = True max_distance_reached = True
except Exception as e: except Exception as e:
logger.warning(f"解析距离出错: {distance_str}, {e}") logger.warning(f"解析距离出错: {distance_str}, {e}")
@@ -312,7 +322,8 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
pro_info = f" | PRO会员价: {pro_price}" if pro_price else "" pro_info = f" | PRO会员价: {pro_price}" if pro_price else ""
# 确保距离始终显示，如果为空则显示 "未知" # 确保距离始终显示，如果为空则显示 "未知"
display_distance = distance_str if distance_str else "未知" display_distance = distance_str if distance_str else "未知"
logger.info(f"正在处理第 {s_idx+1} 个场站: {name} | 价格: {price}{pro_info} | 枪: {piles_str} | 距离: {display_distance}{dist_log_part}") logger.info(
f"正在处理第 {s_idx + 1} 个场站: {name} | 价格: {price}{pro_info} | 枪: {piles_str} | 距离: {display_distance}{dist_log_part}")
if max_distance_reached: if max_distance_reached:
break break
@@ -328,7 +339,7 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
logger.info(f"👉 正在点击进入详情页: {name} (绝对坐标: {x}, {y})") logger.info(f"👉 正在点击进入详情页: {name} (绝对坐标: {x}, {y})")
d.click(x, y) d.click(x, y)
await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) # 等待详情页加载 await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) # 等待详情页加载
# 尝试获取并保存地址 # 尝试获取并保存地址
address_recorded = False address_recorded = False
@@ -348,11 +359,13 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
if up_success: if up_success:
detail_url = f"https://{CDN_DOMAIN}/{detail_object_key}" detail_url = f"https://{CDN_DOMAIN}/{detail_object_key}"
# 3. 解析并保存地址 (异步后台处理) # 3. 解析并保存地址 (异步后台处理)
# 定义异步任务函数 # 定义异步任务函数
async def _process_address_task(task_name, task_url, task_dev_info, task_redis_key): async def _process_address_task(task_name, task_url, task_dev_info, task_redis_key):
t_addr_task = time.time() t_addr_task = time.time()
try: try:
addr_res = await service.process_station_address(task_name, task_url, device_info=task_dev_info) addr_res = await service.process_station_address(task_name, task_url,
device_info=task_dev_info)
logger.info(f"Step [解析并保存地址-后台] 耗时: {time.time() - t_addr_task:.4f}s") logger.info(f"Step [解析并保存地址-后台] 耗时: {time.time() - t_addr_task:.4f}s")
if addr_res.get("address"): if addr_res.get("address"):
logger.info(f"成功记录地址: {addr_res.get('address')}") logger.info(f"成功记录地址: {addr_res.get('address')}")
@@ -373,7 +386,7 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
if not KEEP_SCREENSHOTS and os.path.exists(detail_screenshot_path): if not KEEP_SCREENSHOTS and os.path.exists(detail_screenshot_path):
os.remove(detail_screenshot_path) os.remove(detail_screenshot_path)
break # 直接跳出 retry 循环 break # 直接跳出 retry 循环
# --- 抓取价格时段信息 (New Feature) --- # --- 抓取价格时段信息 (New Feature) ---
# 仅在测试模式下或特定需求下启用 # 仅在测试模式下或特定需求下启用
@@ -387,7 +400,8 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
# 尝试识别，将调试图片保存到 Images 目录 # 尝试识别，将调试图片保存到 Images 目录
t_find = time.time() t_find = time.time()
pos = Kit.find_expand_button_position(screenshot_path, debug_dir=save_dir, debug_filename_prefix=name) pos = Kit.find_expand_button_position(screenshot_path, debug_dir=save_dir,
debug_filename_prefix=name)
logger.info(f"Step [识别展开按钮] 耗时: {time.time() - t_find:.4f}s") logger.info(f"Step [识别展开按钮] 耗时: {time.time() - t_find:.4f}s")
# 清理截图 # 清理截图
@@ -435,7 +449,8 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
async def _process_price_task(task_name, task_urls, task_dev_info): async def _process_price_task(task_name, task_urls, task_dev_info):
t_price_task = time.time() t_price_task = time.time()
try: try:
await service.process_price_schedule(task_name, task_urls, device_info=task_dev_info) await service.process_price_schedule(task_name, task_urls,
device_info=task_dev_info)
logger.info(f"Step [解析电费时段数据-后台] 耗时: {time.time() - t_price_task:.4f}s") logger.info(f"Step [解析电费时段数据-后台] 耗时: {time.time() - t_price_task:.4f}s")
logger.info(f"电费时段数据解析完成: {task_name}") logger.info(f"电费时段数据解析完成: {task_name}")
except Exception as e: except Exception as e:
@@ -457,7 +472,7 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
# 返回列表页 # 返回列表页
d.press("back") d.press("back")
await asyncio.sleep(WAIT_BACK_TO_LIST) # 等待列表页重新稳定 await asyncio.sleep(WAIT_BACK_TO_LIST) # 等待列表页重新稳定
if not new_stations_found and i > 0: if not new_stations_found and i > 0:
logger.info("未发现更多新场站，停止下拉。") logger.info("未发现更多新场站，停止下拉。")
@@ -472,16 +487,16 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
start_y = 0.5 + (SCROLL_DISTANCE_RATIO / 2) start_y = 0.5 + (SCROLL_DISTANCE_RATIO / 2)
end_y = 0.5 - (SCROLL_DISTANCE_RATIO / 2) end_y = 0.5 - (SCROLL_DISTANCE_RATIO / 2)
d.swipe(w * 0.5, h * start_y, w * 0.5, h * end_y, duration=0.5) d.swipe(w * 0.5, h * start_y, w * 0.5, h * end_y, duration=0.5)
await asyncio.sleep(WAIT_AFTER_SCROLL) # 等待页面加载和列表稳定 await asyncio.sleep(WAIT_AFTER_SCROLL) # 等待页面加载和列表稳定
# 清理列表页截图 # 清理列表页截图
if not KEEP_SCREENSHOTS: if not KEEP_SCREENSHOTS:
if os.path.exists(screenshot_path): if os.path.exists(screenshot_path):
os.remove(screenshot_path) os.remove(screenshot_path)
if os.path.exists(vl_img_path): if os.path.exists(vl_img_path):
os.remove(vl_img_path) os.remove(vl_img_path)
if os.path.exists(json_path): if os.path.exists(json_path):
os.remove(json_path) os.remove(json_path)
logger.info(f"爬取任务结束，正在等待 {len(background_tasks)} 个后台处理任务完成...") logger.info(f"爬取任务结束，正在等待 {len(background_tasks)} 个后台处理任务完成...")
if background_tasks: if background_tasks:
@@ -491,6 +506,7 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
logger.info(f"任务结束，共采集到 {len(all_stations)} 个场站。停止原因: {stop_reason}") logger.info(f"任务结束，共采集到 {len(all_stations)} 个场站。停止原因: {stop_reason}")
return all_stations return all_stations
def clean_images_dir(): def clean_images_dir():
"""清理 Images 目录下除保留文件外的所有文件""" """清理 Images 目录下除保留文件外的所有文件"""
base_dir = os.path.dirname(os.path.abspath(__file__)) base_dir = os.path.dirname(os.path.abspath(__file__))
@@ -517,6 +533,7 @@ def clean_images_dir():
logger.info(f"清理完成，共删除 {deleted_count} 个文件。") logger.info(f"清理完成，共删除 {deleted_count} 个文件。")
async def clean_redis_data(redis_kit): async def clean_redis_data(redis_kit):
"""清除 Redis 中的场站处理记录""" """清除 Redis 中的场站处理记录"""
if TEST_CLEAR_REDIS: if TEST_CLEAR_REDIS:
@@ -528,6 +545,7 @@ async def clean_redis_data(redis_kit):
else: else:
logger.info("未发现旧的场站处理记录") logger.info("未发现旧的场站处理记录")
async def main(service=None, do_cleanup=True): async def main(service=None, do_cleanup=True):
# 连接设备 # 连接设备
d = u2.connect() d = u2.connect()
@@ -572,6 +590,7 @@ async def main(service=None, do_cleanup=True):
if should_close_service: if should_close_service:
await service.close_db() await service.close_db()
if __name__ == "__main__": if __name__ == "__main__":
try: try:
asyncio.run(main()) asyncio.run(main())

82
Apps/XinDianTu/Kit.py
View File

@@ -3,11 +3,13 @@ import os
import cv2 import cv2
import numpy as np import numpy as np
import time import time
from Config.Config import BOTTOM_SAFE_EXCLUDE_RATIO, TEMP_IMAGE_DIR from Apps.XinDianTu.Config.Setting import BOTTOM_SAFE_EXCLUDE_RATIO
from Config.Config import TEMP_IMAGE_DIR
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def read_image(path): def read_image(path):
"""读取图片，支持中文路径""" """读取图片，支持中文路径"""
try: try:
@@ -16,6 +18,7 @@ def read_image(path):
logger.info(f"Error reading image {path}: {e}") logger.info(f"Error reading image {path}: {e}")
return None return None
def save_image(path, img): def save_image(path, img):
"""保存图片，支持中文路径""" """保存图片，支持中文路径"""
try: try:
@@ -28,6 +31,7 @@ def save_image(path, img):
logger.error(f"Error saving image {path}: {e}") logger.error(f"Error saving image {path}: {e}")
return False return False
# 截图 # 截图
def take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR): def take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR):
path = f"{save_dir}/{image_uuid}.jpg" path = f"{save_dir}/{image_uuid}.jpg"
@@ -35,6 +39,7 @@ def take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR):
d.screenshot(path) d.screenshot(path)
return path return path
def click_image_template(d, template_path, timeout=5.0, threshold=0.8): def click_image_template(d, template_path, timeout=5.0, threshold=0.8):
""" """
使用 OpenCV 模板匹配查找并点击图片 使用 OpenCV 模板匹配查找并点击图片
@@ -88,7 +93,7 @@ def click_image_template(d, template_path, timeout=5.0, threshold=0.8):
if max_val >= threshold: if max_val >= threshold:
found = (max_val, max_loc, r_w, r_h) found = (max_val, max_loc, r_w, r_h)
break # 找到满足阈值的即停止当前截图的搜索 break # 找到满足阈值的即停止当前截图的搜索
# 清理临时文件 # 清理临时文件
try: try:
@@ -112,6 +117,7 @@ def click_image_template(d, template_path, timeout=5.0, threshold=0.8):
logger.info(f"Image not found after {timeout}s (Best confidence: {best_val_overall:.2f})") logger.info(f"Image not found after {timeout}s (Best confidence: {best_val_overall:.2f})")
return False return False
def detect_black_agree_button(image_path, debug_dir=None): def detect_black_agree_button(image_path, debug_dir=None):
""" """
通过计算机图形学检测黑色的"同意"按钮 (Image 1 场景) 通过计算机图形学检测黑色的"同意"按钮 (Image 1 场景)
@@ -134,7 +140,7 @@ def detect_black_agree_button(image_path, debug_dir=None):
# 定义黑色的范围 # 定义黑色的范围
# 黑色：V (Brightness) 很低 # 黑色：V (Brightness) 很低
lower_black = np.array([0, 0, 0]) lower_black = np.array([0, 0, 0])
upper_black = np.array([180, 255, 40]) # V < 40 认为是黑色 upper_black = np.array([180, 255, 40]) # V < 40 认为是黑色
mask = cv2.inRange(hsv, lower_black, upper_black) mask = cv2.inRange(hsv, lower_black, upper_black)
@@ -186,7 +192,7 @@ def detect_black_agree_button(image_path, debug_dir=None):
if debug_dir: if debug_dir:
os.makedirs(debug_dir, exist_ok=True) os.makedirs(debug_dir, exist_ok=True)
debug_img = img.copy() debug_img = img.copy()
cv2.rectangle(debug_img, (x, y), (x+cw, y+ch), (0, 0, 255), 2) cv2.rectangle(debug_img, (x, y), (x + cw, y + ch), (0, 0, 255), 2)
cv2.circle(debug_img, (center_x, center_y), 5, (0, 255, 0), -1) cv2.circle(debug_img, (center_x, center_y), 5, (0, 255, 0), -1)
save_image(os.path.join(debug_dir, "debug_agree_btn.jpg"), debug_img) save_image(os.path.join(debug_dir, "debug_agree_btn.jpg"), debug_img)
@@ -194,6 +200,7 @@ def detect_black_agree_button(image_path, debug_dir=None):
return None return None
def detect_ad_close_x(image_path, template_path, debug_dir=None, threshold=0.7): def detect_ad_close_x(image_path, template_path, debug_dir=None, threshold=0.7):
""" """
通过模板匹配检测"关闭(X)"按钮 (Image 2 场景) 通过模板匹配检测"关闭(X)"按钮 (Image 2 场景)
@@ -264,7 +271,8 @@ def detect_ad_close_x(image_path, template_path, debug_dir=None, threshold=0.7):
rel_x = center_x / target_w rel_x = center_x / target_w
rel_y = center_y / target_h rel_y = center_y / target_h
logger.info(f"Ad Close Button Match: Confidence={max_val:.4f}, Scale={best_scale:.2f}, Pos=({center_x}, {center_y}), Rel=({rel_x:.2f}, {rel_y:.2f})") logger.info(
f"Ad Close Button Match: Confidence={max_val:.4f}, Scale={best_scale:.2f}, Pos=({center_x}, {center_y}), Rel=({rel_x:.2f}, {rel_y:.2f})")
is_valid_pos = True is_valid_pos = True
@@ -277,22 +285,25 @@ def detect_ad_close_x(image_path, template_path, debug_dir=None, threshold=0.7):
# 规则2: 如果置信度不是特别高 (>0.9), 强制要求在典型区域 (右上角 或 底部中间) # 规则2: 如果置信度不是特别高 (>0.9), 强制要求在典型区域 (右上角 或 底部中间)
# 右上角: X > 0.7, Y < 0.5 # 右上角: X > 0.7, Y < 0.5
# 底部中间: Y > 0.6 # 底部中间: Y > 0.6
if is_valid_pos and max_val < 0.98: # 如果置信度非常高(0.98+)，可能是极其标准的X，暂时放过(除非在禁区) if is_valid_pos and max_val < 0.98: # 如果置信度非常高(0.98+)，可能是极其标准的X，暂时放过(除非在禁区)
if not ((rel_x > 0.7 and rel_y < 0.5) or (rel_y > 0.6)): if not ((rel_x > 0.7 and rel_y < 0.5) or (rel_y > 0.6)):
logger.warning(f"Ignored match at ({center_x}, {center_y}) - not in typical Close Button regions (Top-Right or Bottom).") logger.warning(
is_valid_pos = False f"Ignored match at ({center_x}, {center_y}) - not in typical Close Button regions (Top-Right or Bottom).")
is_valid_pos = False
if max_val >= threshold: if max_val >= threshold:
if is_valid_pos: if is_valid_pos:
# [Safety Check] 底部安全区排除 # [Safety Check] 底部安全区排除
# 如果检测到的关闭按钮位于屏幕底部 BOTTOM_SAFE_EXCLUDE_RATIO 区域内，认为是误判（如误触底部功能按钮） # 如果检测到的关闭按钮位于屏幕底部 BOTTOM_SAFE_EXCLUDE_RATIO 区域内，认为是误判（如误触底部功能按钮）
if center_y > (target_h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)): if center_y > (target_h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)):
logger.warning(f"Ignored Ad Close Button at ({center_x}, {center_y}) - in Bottom Safety Zone ({int(BOTTOM_SAFE_EXCLUDE_RATIO*100)}%).") logger.warning(
f"Ignored Ad Close Button at ({center_x}, {center_y}) - in Bottom Safety Zone ({int(BOTTOM_SAFE_EXCLUDE_RATIO * 100)}%).")
# 也可以保存一下调试图 # 也可以保存一下调试图
if debug_dir: if debug_dir:
os.makedirs(debug_dir, exist_ok=True) os.makedirs(debug_dir, exist_ok=True)
debug_img = target.copy() debug_img = target.copy()
cv2.rectangle(debug_img, top_left, (top_left[0] + best_w, top_left[1] + best_h), (0, 0, 128), 2) # Dark Red for Safety Ignored cv2.rectangle(debug_img, top_left, (top_left[0] + best_w, top_left[1] + best_h), (0, 0, 128),
2) # Dark Red for Safety Ignored
cv2.putText(debug_img, f"SAFETY IGNORED", cv2.putText(debug_img, f"SAFETY IGNORED",
(top_left[0], top_left[1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 128), 1) (top_left[0], top_left[1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 128), 1)
save_image(os.path.join(debug_dir, "debug_ad_close_safety_ignored.jpg"), debug_img) save_image(os.path.join(debug_dir, "debug_ad_close_safety_ignored.jpg"), debug_img)
@@ -311,11 +322,12 @@ def detect_ad_close_x(image_path, template_path, debug_dir=None, threshold=0.7):
return (center_x, center_y) return (center_x, center_y)
else: else:
# 虽然置信度高，但是位置不对，保存为 False Positive 供调试 # 虽然置信度高，但是位置不对，保存为 False Positive 供调试
if debug_dir: if debug_dir:
os.makedirs(debug_dir, exist_ok=True) os.makedirs(debug_dir, exist_ok=True)
debug_img = target.copy() debug_img = target.copy()
cv2.rectangle(debug_img, top_left, (top_left[0] + best_w, top_left[1] + best_h), (0, 165, 255), 2) # Orange for ignored cv2.rectangle(debug_img, top_left, (top_left[0] + best_w, top_left[1] + best_h), (0, 165, 255),
2) # Orange for ignored
cv2.putText(debug_img, f"IGNORED Pos", cv2.putText(debug_img, f"IGNORED Pos",
(top_left[0], top_left[1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 165, 255), 1) (top_left[0], top_left[1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 165, 255), 1)
save_image(os.path.join(debug_dir, "debug_ad_close_ignored.jpg"), debug_img) save_image(os.path.join(debug_dir, "debug_ad_close_ignored.jpg"), debug_img)
@@ -332,6 +344,7 @@ def detect_ad_close_x(image_path, template_path, debug_dir=None, threshold=0.7):
return None return None
def detect_any_ad_close(image_path, template_dir, debug_dir=None): def detect_any_ad_close(image_path, template_dir, debug_dir=None):
""" """
遍历模板目录下的所有 ad_close*.jpg 进行匹配 遍历模板目录下的所有 ad_close*.jpg 进行匹配
@@ -348,6 +361,7 @@ def detect_any_ad_close(image_path, template_dir, debug_dir=None):
return pos return pos
return None return None
def detect_bottom_close_circle(image_path, debug_dir=None): def detect_bottom_close_circle(image_path, debug_dir=None):
""" """
通过几何特征检测底部的圆形关闭按钮 (常见于插屏广告) 通过几何特征检测底部的圆形关闭按钮 (常见于插屏广告)
@@ -379,8 +393,8 @@ def detect_bottom_close_circle(image_path, debug_dir=None):
# 2. 霍夫圆变换检测圆形 # 2. 霍夫圆变换检测圆形
# dp=1.2 (累加器分辨率), minDist=w/5 (圆心最小距离), param1=100 (Canny高阈值), param2=30 (圆心累加阈值), minR=w*0.04, maxR=w*0.1 # dp=1.2 (累加器分辨率), minDist=w/5 (圆心最小距离), param1=100 (Canny高阈值), param2=30 (圆心累加阈值), minR=w*0.04, maxR=w*0.1
# 缩小最大半径范围，防止识别到过大的按钮（如扫码充电按钮） # 缩小最大半径范围，防止识别到过大的按钮（如扫码充电按钮）
circles = cv2.HoughCircles(gray_blurred, cv2.HOUGH_GRADIENT, dp=1.2, minDist=w/5, circles = cv2.HoughCircles(gray_blurred, cv2.HOUGH_GRADIENT, dp=1.2, minDist=w / 5,
param1=100, param2=30, minRadius=int(w*0.04), maxRadius=int(w*0.09)) param1=100, param2=30, minRadius=int(w * 0.04), maxRadius=int(w * 0.09))
if circles is not None: if circles is not None:
circles = np.round(circles[0, :]).astype("int") circles = np.round(circles[0, :]).astype("int")
@@ -394,7 +408,7 @@ def detect_bottom_close_circle(image_path, debug_dir=None):
global_cx = cx global_cx = cx
# 过滤1: 必须在水平中心附近 (容差 15%，收紧限制) # 过滤1: 必须在水平中心附近 (容差 15%，收紧限制)
if abs(global_cx - w//2) > (w * 0.15): if abs(global_cx - w // 2) > (w * 0.15):
continue continue
# 过滤2: 垂直方向限制 (必须在屏幕 80% - 95% 之间) # 过滤2: 垂直方向限制 (必须在屏幕 80% - 95% 之间)
@@ -403,7 +417,7 @@ def detect_bottom_close_circle(image_path, debug_dir=None):
if rel_y < 0.80 or rel_y > 0.95: if rel_y < 0.80 or rel_y > 0.95:
continue continue
dist = abs(global_cx - w//2) dist = abs(global_cx - w // 2)
if dist < min_dist_to_center: if dist < min_dist_to_center:
min_dist_to_center = dist min_dist_to_center = dist
best_circle = (global_cx, global_cy, r) best_circle = (global_cx, global_cy, r)
@@ -416,7 +430,8 @@ def detect_bottom_close_circle(image_path, debug_dir=None):
# [Safety Check] 底部安全区排除 # [Safety Check] 底部安全区排除
# 如果检测到的圆形按钮位于屏幕底部 BOTTOM_SAFE_EXCLUDE_RATIO 区域内，认为是误判（如误触底部扫码充电等） # 如果检测到的圆形按钮位于屏幕底部 BOTTOM_SAFE_EXCLUDE_RATIO 区域内，认为是误判（如误触底部扫码充电等）
if cy > (h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)): if cy > (h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)):
logger.warning(f"Ignored Bottom Circle at ({cx}, {cy}) - in Bottom Safety Zone ({int(BOTTOM_SAFE_EXCLUDE_RATIO*100)}%).") logger.warning(
f"Ignored Bottom Circle at ({cx}, {cy}) - in Bottom Safety Zone ({int(BOTTOM_SAFE_EXCLUDE_RATIO * 100)}%).")
return None return None
logger.info(f"Found Bottom Circle Button via Hough: ({cx}, {cy}), r={r}") logger.info(f"Found Bottom Circle Button via Hough: ({cx}, {cy}), r={r}")
@@ -432,6 +447,7 @@ def detect_bottom_close_circle(image_path, debug_dir=None):
return None return None
def find_expand_button_position(image_path, debug_dir=None, debug_filename_prefix=None): def find_expand_button_position(image_path, debug_dir=None, debug_filename_prefix=None):
""" """
通过几何特征识别"全部时段"按钮的位置 通过几何特征识别"全部时段"按钮的位置
@@ -473,8 +489,8 @@ def find_expand_button_position(image_path, debug_dir=None, debug_filename_prefi
# 筛选条件：左右边缘点很少，中间边缘点较多 # 筛选条件：左右边缘点很少，中间边缘点较多
# 阈值可以根据实际情况调整 # 阈值可以根据实际情况调整
# 允许少量噪点，所以不是严格的0，而是小于某个较小值（例如宽度的1%） # 允许少量噪点，所以不是严格的0，而是小于某个较小值（例如宽度的1%）
noise_threshold = 2 # 允许2个像素的噪点 noise_threshold = 2 # 允许2个像素的噪点
content_threshold = 5 # 中间至少有5个像素的边缘 content_threshold = 5 # 中间至少有5个像素的边缘
candidates = (row_sum_left <= noise_threshold) & \ candidates = (row_sum_left <= noise_threshold) & \
(row_sum_right <= noise_threshold) & \ (row_sum_right <= noise_threshold) & \
@@ -492,7 +508,7 @@ def find_expand_button_position(image_path, debug_dir=None, debug_filename_prefi
start = y_indices[0] start = y_indices[0]
prev = y_indices[0] prev = y_indices[0]
for y in y_indices[1:]: for y in y_indices[1:]:
if y > prev + 5: # 允许5像素断裂 if y > prev + 5: # 允许5像素断裂
segments.append((start, prev)) segments.append((start, prev))
start = y start = y
prev = y prev = y
@@ -572,7 +588,7 @@ def get_row_stats(gray):
# 中央区域 (用于检测内容) # 中央区域 (用于检测内容)
center_x = w // 2 center_x = w // 2
strip_w = 100 strip_w = 100
center_strip = gray[:, center_x - 50 : center_x + 50] center_strip = gray[:, center_x - 50: center_x + 50]
# 边缘区域 (用于检测背景/边距) # 边缘区域 (用于检测背景/边距)
# 假设边距至少有 10px # 假设边距至少有 10px
@@ -584,6 +600,7 @@ def get_row_stats(gray):
return row_means, row_stds, edge_means return row_means, row_stds, edge_means
def crop_cards_from_image(img_path, output_dir=None, save_debug=True): def crop_cards_from_image(img_path, output_dir=None, save_debug=True):
""" """
从图片中裁剪场站卡片 从图片中裁剪场站卡片
@@ -688,7 +705,7 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True):
height = end_y - start_y height = end_y - start_y
# 过滤太矮的区域 # 过滤太矮的区域
if height > 100: # 稍微放宽一点，150 -> 100 if height > 100: # 稍微放宽一点，150 -> 100
segments.append((start_y, end_y)) segments.append((start_y, end_y))
# 处理最后一个 segment # 处理最后一个 segment
@@ -700,7 +717,7 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True):
# [新增] 触底过滤：过滤掉延伸到图片底部的 Segment # [新增] 触底过滤：过滤掉延伸到图片底部的 Segment
# 底部通常是导航栏或被截断的卡片 # 底部通常是导航栏或被截断的卡片
BOTTOM_MARGIN = 50 # 距离底部多少像素内算触底 BOTTOM_MARGIN = 50 # 距离底部多少像素内算触底
valid_segments_scan = [] valid_segments_scan = []
for y1, y2 in segments: for y1, y2 in segments:
if y2 >= h - BOTTOM_MARGIN: if y2 >= h - BOTTOM_MARGIN:
@@ -718,7 +735,7 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True):
if segments: if segments:
current_group = [segments[0]] current_group = [segments[0]]
for i in range(1, len(segments)): for i in range(1, len(segments)):
prev_end = segments[i-1][1] prev_end = segments[i - 1][1]
curr_start = segments[i][0] curr_start = segments[i][0]
gap = curr_start - prev_end gap = curr_start - prev_end
@@ -765,7 +782,7 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True):
# 取中间一段行来分析 # 取中间一段行来分析
mid_y = (y1 + y2) // 2 mid_y = (y1 + y2) // 2
sample_h = min(10, y2 - y1) sample_h = min(10, y2 - y1)
sample_rows = gray[mid_y - sample_h//2 : mid_y + sample_h//2, :] sample_rows = gray[mid_y - sample_h // 2: mid_y + sample_h // 2, :]
col_means = np.mean(sample_rows, axis=0) col_means = np.mean(sample_rows, axis=0)
# 从左向右找第一个非灰像素 # 从左向右找第一个非灰像素
@@ -783,10 +800,10 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True):
break break
if x1 >= x2 or (x2 - x1) < w * 0.5: if x1 >= x2 or (x2 - x1) < w * 0.5:
# 如果这个 segment 无法确定宽度，可能不是有效卡片，暂不参与宽度投票 # 如果这个 segment 无法确定宽度，可能不是有效卡片，暂不参与宽度投票
# 但为了不漏掉，暂时先记录，用默认值填充 # 但为了不漏掉，暂时先记录，用默认值填充
logger.warning(f" Warning: Segment {y1}-{y2} has weird width {x2-x1}.") logger.warning(f" Warning: Segment {y1}-{y2} has weird width {x2 - x1}.")
pass pass
else: else:
candidate_x1.append(x1) candidate_x1.append(x1)
candidate_x2.append(x2) candidate_x2.append(x2)
@@ -819,7 +836,7 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True):
else: else:
heights = [y2 - y1 for y1, y2 in temp_valid_segments] heights = [y2 - y1 for y1, y2 in temp_valid_segments]
if not heights: if not heights:
logger.info(" No heights to calculate.") logger.info(" No heights to calculate.")
else: else:
max_h = max(heights) max_h = max(heights)
median_h = np.median(heights) median_h = np.median(heights)
@@ -831,7 +848,8 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True):
for (y1, y2), h in zip(temp_valid_segments, heights): for (y1, y2), h in zip(temp_valid_segments, heights):
if h < threshold_h: if h < threshold_h:
logger.info(f" Filtering out segment Y={y1}-{y2} (H={h}) because it's too short (Threshold={threshold_h:.1f}).") logger.info(
f" Filtering out segment Y={y1}-{y2} (H={h}) because it's too short (Threshold={threshold_h:.1f}).")
else: else:
final_cards.append((y1, y2, final_x1, final_x2)) final_cards.append((y1, y2, final_x1, final_x2))
logger.info(f" Card: Y={y1}-{y2}, X={final_x1}-{final_x2}, H={h}") logger.info(f" Card: Y={y1}-{y2}, X={final_x1}-{final_x2}, H={h}")

11
Apps/XinDianTu/Opener.py
View File

@@ -1,14 +1,15 @@
# coding=utf-8 # coding=utf-8
import asyncio import asyncio
import logging import logging
import os
import time import time
import uuid import uuid
import os
import uiautomator2 as u2 import uiautomator2 as u2
from Util.Kit import take_screenshot, detect_black_agree_button, click_image_template, detect_ad_close_x, detect_any_ad_close, detect_bottom_close_circle
from Util.ObsUtil import ObsUploader from Apps.XinDianTu.Kit import take_screenshot, detect_black_agree_button, detect_any_ad_close, \
from Util.XinDianTuReadImageKit import XinDianTuReadImageKit detect_bottom_close_circle
from Config.Config import OBS_TMP_PREFIX, CDN_DOMAIN, TEMP_IMAGE_DIR from Config.Config import TEMP_IMAGE_DIR
# pip install adbutils # pip install adbutils
# 配置日志输出，方便调试和监控 # 配置日志输出，方便调试和监控

31
Apps/XinDianTu/Run.py
View File

@@ -12,8 +12,6 @@ if project_root not in sys.path:
from Apps.XinDianTu.Service import XinDianTuService from Apps.XinDianTu.Service import XinDianTuService
from Apps.XinDianTu import Opener from Apps.XinDianTu import Opener
from Apps.XinDianTu import Crawler from Apps.XinDianTu import Crawler
from Apps.XinDianTu import Coord as Y4_Coord
from Apps.XinDianTu import GetCookie as T2_GetAmapCookie
# 配置日志 # 配置日志
logging.basicConfig( logging.basicConfig(
@@ -60,35 +58,6 @@ async def main():
logger.error(f"❌ 步骤 2 执行异常: {e}") logger.error(f"❌ 步骤 2 执行异常: {e}")
return return
# 步骤 3: 补全坐标 (调用 Y4_Coord)
logger.info(">>> 步骤 3: 启动高德地图坐标回填任务...")
try:
# 首次尝试，开启 Cookie 检查
await Y4_Coord.main(check_cookie_first=True)
except Y4_Coord.CookieInvalidError:
logger.warning("⚠️ 检测到高德地图 Cookie 失效或缺失。")
logger.info(">>> 正在启动 Cookie 获取工具 (T2)...")
logger.info("请在弹出的浏览器窗口中完成登录，然后在控制台按回车确认。")
# 调用同步的 T2 工具
# 由于 T2 包含 input() 阻塞，且这是客户端脚本，直接调用即可
try:
# 传入 force=True 跳过“文件已存在”的询问，因为我们确定它无效
# run_in_executor 用于避免完全阻塞事件循环（虽然在这里影响不大）
loop = asyncio.get_event_loop()
success = await loop.run_in_executor(None, T2_GetAmapCookie.run_login_tool, True)
if success:
logger.info("✅ Cookie 更新成功，正在重新启动坐标回填任务...")
# 再次尝试执行，这次可以跳过检查（刚获取完）或者继续检查
await Y4_Coord.main(check_cookie_first=False)
else:
logger.error("❌ Cookie 获取失败或用户取消，跳过坐标回填步骤。")
except Exception as e:
logger.error(f"❌ 调用 T2 工具失败: {e}")
except Exception as e:
logger.error(f"❌ 步骤 3 执行异常: {e}")
finally: finally:
# 确保数据库连接关闭 # 确保数据库连接关闭
if service: if service:

10
Apps/XinDianTu/Service.py
View File

@@ -1,19 +1,17 @@
import hashlib import hashlib
import logging
import os import os
import re
import sys import sys
import uuid import uuid
import logging from datetime import datetime
import json
import re
from datetime import datetime, timedelta
# Ensure sys path includes root for imports if not already # 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__)))) project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
if project_root not in sys.path: if project_root not in sys.path:
sys.path.append(project_root) sys.path.append(project_root)
from sqlalchemy import text from Apps.XinDianTu.XinDianTuReadImageKit import XinDianTuReadImageKit
from Util.XinDianTuReadImageKit import XinDianTuReadImageKit
from DbKit.Db import Db from DbKit.Db import Db
from Config.Config import DB_URL from Config.Config import DB_URL
from Model.StationProfile import StationProfile from Model.StationProfile import StationProfile

93
Apps/XinDianTu/XinDianTuReadImageKit.py
View File

@@ -9,13 +9,16 @@ import logging
import base64 import base64
from openai import OpenAI, BadRequestError from openai import OpenAI, BadRequestError
from Config.Config import ( from Config.Config import (
ALY_LLM_API_KEY, VL_MODEL_NAME, VL_MODEL_NAME_AD, ALY_LLM_API_KEY, VL_MODEL_NAME, VL_MODEL_NAME_AD
)
from Apps.XinDianTu.Config.Setting import (
SAFE_EXCLUDE_RATIO, FALLBACK_WIDTH, FALLBACK_HEIGHT, SAFE_EXCLUDE_RATIO, FALLBACK_WIDTH, FALLBACK_HEIGHT,
BOTTOM_SAFE_EXCLUDE_RATIO BOTTOM_SAFE_EXCLUDE_RATIO
) )
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class XinDianTuReadImageKit: class XinDianTuReadImageKit:
_client = OpenAI( _client = OpenAI(
api_key=ALY_LLM_API_KEY, api_key=ALY_LLM_API_KEY,
@@ -152,16 +155,20 @@ class XinDianTuReadImageKit:
return min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2) return min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2)
# 优先使用场站名称区域进行点击 (最安全) # 优先使用场站名称区域进行点击 (最安全)
coords = get_pixel_coords(text_bn, display_width, display_height) or get_pixel_coords(text_bounds, display_width, display_height) coords = get_pixel_coords(text_bn, display_width, display_height) or get_pixel_coords(text_bounds,
display_width,
display_height)
if coords: if coords:
x1, y1, x2, y2 = coords x1, y1, x2, y2 = coords
uia_x = int(x1 + (x2 - x1) / 2) uia_x = int(x1 + (x2 - x1) / 2)
uia_y = int(y1 + (y2 - y1) / 2) uia_y = int(y1 + (y2 - y1) / 2)
logger.info(f"坐标计算: 使用文字区域 -> ({uia_x}, {uia_y}) | 区域: {coords} | 屏幕: {display_width}x{display_height}") logger.info(
f"坐标计算: 使用文字区域 -> ({uia_x}, {uia_y}) | 区域: {coords} | 屏幕: {display_width}x{display_height}")
# 备选：使用整个卡片区域 # 备选：使用整个卡片区域
if uia_x is None: if uia_x is None:
coords = get_pixel_coords(bn, display_width, display_height) or get_pixel_coords(bounds, display_width, display_height) coords = get_pixel_coords(bn, display_width, display_height) or get_pixel_coords(bounds, display_width,
display_height)
if coords: if coords:
x1, y1, x2, y2 = coords x1, y1, x2, y2 = coords
w, h = x2 - x1, y2 - y1 w, h = x2 - x1, y2 - y1
@@ -172,18 +179,21 @@ class XinDianTuReadImageKit:
# 默认中心点，但稍微偏上一点，避开底部可能的按钮 # 默认中心点，但稍微偏上一点，避开底部可能的按钮
uia_x = int(x1 + w / 2) uia_x = int(x1 + w / 2)
uia_y = int(y1 + h * 0.4) uia_y = int(y1 + h * 0.4)
logger.info(f"坐标计算: 使用卡片区域 -> ({uia_x}, {uia_y}) | 区域: {coords} | 屏幕: {display_width}x{display_height}") logger.info(
f"坐标计算: 使用卡片区域 -> ({uia_x}, {uia_y}) | 区域: {coords} | 屏幕: {display_width}x{display_height}")
# --- 安全过滤：过滤掉屏幕顶部的点击坐标 (通常是广告或菜单) --- # --- 安全过滤：过滤掉屏幕顶部的点击坐标 (通常是广告或菜单) ---
if uia_y is not None: if uia_y is not None:
# 如果点击点在屏幕顶部 SAFE_EXCLUDE_RATIO 范围内，极大概率是误触广告位，将其排除 # 如果点击点在屏幕顶部 SAFE_EXCLUDE_RATIO 范围内，极大概率是误触广告位，将其排除
if uia_y < (display_height * SAFE_EXCLUDE_RATIO): if uia_y < (display_height * SAFE_EXCLUDE_RATIO):
logger.warning(f"安全排除: 坐标 ({uia_x}, {uia_y}) 位于屏幕顶部 {int(SAFE_EXCLUDE_RATIO*100)}% 区域，疑似广告或菜单，已忽略。") logger.warning(
f"安全排除: 坐标 ({uia_x}, {uia_y}) 位于屏幕顶部 {int(SAFE_EXCLUDE_RATIO * 100)}% 区域，疑似广告或菜单，已忽略。")
return None return None
# 如果点击点在屏幕底部 BOTTOM_SAFE_EXCLUDE_RATIO 范围内，极大概率是误触底部功能区（如扫码充电），将其排除 # 如果点击点在屏幕底部 BOTTOM_SAFE_EXCLUDE_RATIO 范围内，极大概率是误触底部功能区（如扫码充电），将其排除
if uia_y > (display_height * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)): if uia_y > (display_height * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)):
logger.warning(f"安全排除: 坐标 ({uia_x}, {uia_y}) 位于屏幕底部 {int(BOTTOM_SAFE_EXCLUDE_RATIO*100)}% 区域，疑似底部功能区，已忽略。") logger.warning(
f"安全排除: 坐标 ({uia_x}, {uia_y}) 位于屏幕底部 {int(BOTTOM_SAFE_EXCLUDE_RATIO * 100)}% 区域，疑似底部功能区，已忽略。")
return None return None
if uia_x is not None and uia_y is not None: if uia_x is not None and uia_y is not None:
@@ -261,13 +271,13 @@ class XinDianTuReadImageKit:
device_info = XinDianTuReadImageKit._FALLBACK_DEVICE_INFO device_info = XinDianTuReadImageKit._FALLBACK_DEVICE_INFO
prompt = ( prompt = (
"仅输出JSON对象（不含任何说明文字）。" "仅输出JSON对象（不含任何说明文字）。"
"请在图片中找到场站名称完全匹配“" + str(station_name) + "”的那一行/卡片，" "请在图片中找到场站名称完全匹配“" + str(station_name) + "”的那一行/卡片，"
"返回该场站名称文字本身的区域坐标（优先返回归一化坐标）。" "返回该场站名称文字本身的区域坐标（优先返回归一化坐标）。"
"返回格式示例：" "返回格式示例："
"{\"station_name_bounds\": {\"x1\": 100, \"y1\": 200, \"x2\": 300, \"y2\": 240}, " "{\"station_name_bounds\": {\"x1\": 100, \"y1\": 200, \"x2\": 300, \"y2\": 240}, "
"\"station_name_bounds_norm\": {\"left\": 0.1, \"top\": 0.2, \"right\": 0.3, \"bottom\": 0.24}}" "\"station_name_bounds_norm\": {\"left\": 0.1, \"top\": 0.2, \"right\": 0.3, \"bottom\": 0.24}}"
"注意：bounds使用0-1000归一化坐标空间；如果找不到，返回空JSON {}。" "注意：bounds使用0-1000归一化坐标空间；如果找不到，返回空JSON {}。"
) )
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
@@ -293,7 +303,9 @@ class XinDianTuReadImageKit:
try: try:
data = json.loads(raw) data = json.loads(raw)
if isinstance(data, dict) and (data.get("station_name_bounds") or data.get("station_name_bounds_norm") or data.get("bounds") or data.get("bounds_norm")): if isinstance(data, dict) and (
data.get("station_name_bounds") or data.get("station_name_bounds_norm") or data.get(
"bounds") or data.get("bounds_norm")):
data = XinDianTuReadImageKit._add_click_point(data, device_info, anchor="station_text") data = XinDianTuReadImageKit._add_click_point(data, device_info, anchor="station_text")
return { return {
"uia_center_x": data.get("uia_center_x"), "uia_center_x": data.get("uia_center_x"),
@@ -339,7 +351,8 @@ class XinDianTuReadImageKit:
data = json.loads(raw) data = json.loads(raw)
if isinstance(data, list): if isinstance(data, list):
# 过滤掉 None (即被 _add_click_point 排除掉的顶部项) # 过滤掉 None (即被 _add_click_point 排除掉的顶部项)
data = [XinDianTuReadImageKit._add_click_point(x, device_info, anchor="station_text") for x in data if x.get("b_use", 1) == 1] data = [XinDianTuReadImageKit._add_click_point(x, device_info, anchor="station_text") for x in data if
x.get("b_use", 1) == 1]
data = [x for x in data if x is not None] data = [x for x in data if x is not None]
elif isinstance(data, dict): elif isinstance(data, dict):
if data.get("b_use", 1) == 1: if data.get("b_use", 1) == 1:
@@ -389,7 +402,8 @@ class XinDianTuReadImageKit:
if tasks: if tasks:
results = await asyncio.gather(*tasks, return_exceptions=True) results = await asyncio.gather(*tasks, return_exceptions=True)
for i, r in zip(idxs, results): for i, r in zip(idxs, results):
if isinstance(r, dict) and r.get("uia_center_x") is not None and r.get("uia_center_y") is not None: if isinstance(r, dict) and r.get("uia_center_x") is not None and r.get(
"uia_center_y") is not None:
data[i]["uia_center_x"] = r.get("uia_center_x") data[i]["uia_center_x"] = r.get("uia_center_x")
data[i]["uia_center_y"] = r.get("uia_center_y") data[i]["uia_center_y"] = r.get("uia_center_y")
@@ -491,12 +505,14 @@ class XinDianTuReadImageKit:
# 顶部安全排除 # 顶部安全排除
if cy < (img_h * SAFE_EXCLUDE_RATIO): if cy < (img_h * SAFE_EXCLUDE_RATIO):
logger.warning(f"VL安全排除: 坐标 ({cx}, {cy}) 位于屏幕顶部 {int(SAFE_EXCLUDE_RATIO*100)}% 区域，已忽略。") logger.warning(
f"VL安全排除: 坐标 ({cx}, {cy}) 位于屏幕顶部 {int(SAFE_EXCLUDE_RATIO * 100)}% 区域，已忽略。")
continue continue
# 底部安全排除 # 底部安全排除
if cy > (img_h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)): if cy > (img_h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)):
logger.warning(f"VL安全排除: 坐标 ({cx}, {cy}) 位于屏幕底部 {int(BOTTOM_SAFE_EXCLUDE_RATIO*100)}% 区域，疑似底部功能区，已忽略。") logger.warning(
f"VL安全排除: 坐标 ({cx}, {cy}) 位于屏幕底部 {int(BOTTOM_SAFE_EXCLUDE_RATIO * 100)}% 区域，疑似底部功能区，已忽略。")
continue continue
item["uia_center_x"] = cx item["uia_center_x"] = cx
@@ -585,7 +601,7 @@ class XinDianTuReadImageKit:
# 分割线特征：横贯全屏，亮度均匀，通常比纯白(255)稍暗，比文字内容亮 # 分割线特征：横贯全屏，亮度均匀，通常比纯白(255)稍暗，比文字内容亮
# 典型值：Mean=242-247, Std<10 # 典型值：Mean=242-247, Std<10
row_types = [] # 0: Unknown/Content, 1: Divider, 2: White/Empty row_types = [] # 0: Unknown/Content, 1: Divider, 2: White/Empty
# 采样点：左中右 # 采样点：左中右
l_x, m_x, r_x = int(width * 0.05), int(width * 0.5), int(width * 0.95) l_x, m_x, r_x = int(width * 0.05), int(width * 0.5), int(width * 0.95)
@@ -595,7 +611,7 @@ class XinDianTuReadImageKit:
for y in range(height): for y in range(height):
# 避开顶部和底部导航栏 (15% - 85%) - 扩大排除范围以避免误识别 Header/Footer # 避开顶部和底部导航栏 (15% - 85%) - 扩大排除范围以避免误识别 Header/Footer
if y < height * 0.15 or y > height * 0.85: if y < height * 0.15 or y > height * 0.85:
row_types.append(1) # 视为无关区域 (标记为 Divider 以防止形成 Segment) row_types.append(1) # 视为无关区域 (标记为 Divider 以防止形成 Segment)
continue continue
row = img_data[y] row = img_data[y]
@@ -608,11 +624,11 @@ class XinDianTuReadImageKit:
# 3. 内容行 (Content) -> mean <= 230 或 (mean > 230 且 std >= 15) (有文字/图片导致方差大) # 3. 内容行 (Content) -> mean <= 230 或 (mean > 230 且 std >= 15) (有文字/图片导致方差大)
if row_mean > 252: if row_mean > 252:
r_type = 2 # White/Empty r_type = 2 # White/Empty
elif 230 < row_mean <= 252 and row_std < 15: elif 230 < row_mean <= 252 and row_std < 15:
r_type = 1 # Divider r_type = 1 # Divider
else: else:
r_type = 0 # Content r_type = 0 # Content
row_types.append(r_type) row_types.append(r_type)
@@ -638,7 +654,7 @@ class XinDianTuReadImageKit:
raw_blocks = [] raw_blocks = []
for y, r_type in enumerate(row_types): for y, r_type in enumerate(row_types):
if r_type != 1: # Not Divider if r_type != 1: # Not Divider
if not in_segment: if not in_segment:
in_segment = True in_segment = True
seg_start = y seg_start = y
@@ -646,10 +662,10 @@ class XinDianTuReadImageKit:
if r_type == 0: if r_type == 0:
has_content = True has_content = True
else: # Is Divider else: # Is Divider
if in_segment: if in_segment:
# 结束一段 # 结束一段
if has_content: # 只有包含内容的段才算 if has_content: # 只有包含内容的段才算
raw_blocks.append((seg_start, y)) raw_blocks.append((seg_start, y))
in_segment = False in_segment = False
@@ -678,7 +694,7 @@ class XinDianTuReadImageKit:
# 避免切出纯白的空隙 # 避免切出纯白的空隙
region = img_data[y1:y2] region = img_data[y1:y2]
region_mean = np.mean(region) region_mean = np.mean(region)
if region_mean > 254.5: # 整体太白 (放宽阈值，避免误杀大面积白色的卡片) if region_mean > 254.5: # 整体太白 (放宽阈值，避免误杀大面积白色的卡片)
logger.info(f" [-] 忽略区域 {i}: Y({y1}-{y2}), H={h}, 整体太白 (Mean={region_mean:.1f} > 254.5)") logger.info(f" [-] 忽略区域 {i}: Y({y1}-{y2}), H={h}, 整体太白 (Mean={region_mean:.1f} > 254.5)")
continue continue
@@ -687,14 +703,14 @@ class XinDianTuReadImageKit:
# 假设：有效内容行的方差较高，或者亮度显著不同于背景 # 假设：有效内容行的方差较高，或者亮度显著不同于背景
# 简单策略：保留全宽，但增加高度限制 # 简单策略：保留全宽，但增加高度限制
if h < 60: # 稍微降低阈值以测试 if h < 60: # 稍微降低阈值以测试
logger.info(f" [-] 忽略区域 {i}: Y({y1}-{y2}), H={h}, 高度不足 (<60)") logger.info(f" [-] 忽略区域 {i}: Y({y1}-{y2}), H={h}, 高度不足 (<60)")
continue continue
# 计算左右边距的平均亮度，辅助判断（仅记录日志，不强行过滤） # 计算左右边距的平均亮度，辅助判断（仅记录日志，不强行过滤）
margin_w = max(5, int(width * 0.05)) margin_w = max(5, int(width * 0.05))
l_margin = img_data[y1:y2, 0:margin_w] l_margin = img_data[y1:y2, 0:margin_w]
r_margin = img_data[y1:y2, width-margin_w:width] r_margin = img_data[y1:y2, width - margin_w:width]
l_mean = np.mean(l_margin) l_mean = np.mean(l_margin)
r_mean = np.mean(r_margin) r_mean = np.mean(r_margin)
@@ -727,8 +743,8 @@ class XinDianTuReadImageKit:
# 如果裁剪后宽度太小，可能不是有效卡片 # 如果裁剪后宽度太小，可能不是有效卡片
if (x2 - x1) < width * 0.5: if (x2 - x1) < width * 0.5:
logger.info(f" [-] 忽略区域 {i}: 裁剪后宽度过小 ({x2-x1})") logger.info(f" [-] 忽略区域 {i}: 裁剪后宽度过小 ({x2 - x1})")
continue continue
# 保存 # 保存
patch = img.crop((x1, y1, x2, y2)) patch = img.crop((x1, y1, x2, y2))
@@ -743,7 +759,8 @@ class XinDianTuReadImageKit:
"center_y": center_y, "center_y": center_y,
"y_range": (y1, y2) "y_range": (y1, y2)
}) })
logger.info(f" [+] 发现卡片 {i}: Y({y1}-{y2}), H={h}, Crop X({x1}-{x2}), 边缘(L={l_mean:.1f}, R={r_mean:.1f}), 已保存") logger.info(
f" [+] 发现卡片 {i}: Y({y1}-{y2}), H={h}, Crop X({x1}-{x2}), 边缘(L={l_mean:.1f}, R={r_mean:.1f}), 已保存")
logger.info(f"分析完成：识别到 {len(valid_segments)} 个区域") logger.info(f"分析完成：识别到 {len(valid_segments)} 个区域")
return valid_segments return valid_segments
@@ -940,7 +957,7 @@ class XinDianTuReadImageKit:
# 尝试捕获 DataInspection 错误并进行 Base64 回退 # 尝试捕获 DataInspection 错误并进行 Base64 回退
err_code = getattr(e, 'code', '') or '' err_code = getattr(e, 'code', '') or ''
if not err_code and hasattr(e, 'body') and isinstance(e.body, dict): if not err_code and hasattr(e, 'body') and isinstance(e.body, dict):
err_code = e.body.get('code', '') err_code = e.body.get('code', '')
# 如果是媒体格式或数据检查错误 # 如果是媒体格式或数据检查错误
if 'InvalidParameter.DataInspection' in str(err_code) or 'media format' in str(e).lower(): if 'InvalidParameter.DataInspection' in str(err_code) or 'media format' in str(e).lower():
@@ -966,8 +983,8 @@ class XinDianTuReadImageKit:
logger.error(f"API BadRequestError: {e}") logger.error(f"API BadRequestError: {e}")
return [] return []
except Exception as e: except Exception as e:
logger.error(f"API Unexpected Error: {e}") logger.error(f"API Unexpected Error: {e}")
return [] return []
content = resp.choices[0].message.content or "" content = resp.choices[0].message.content or ""
raw = XinDianTuReadImageKit._extract_json(content) raw = XinDianTuReadImageKit._extract_json(content)
@@ -1284,7 +1301,7 @@ async def test3(station_name: str):
if __name__ == "__main__": if __name__ == "__main__":
#asyncio.run(test1()) # asyncio.run(test1())
station_name = '吉林省看守所充电站' station_name = '吉林省看守所充电站'
#asyncio.run(test2(station_name)) # asyncio.run(test2(station_name))
asyncio.run(test3(station_name)) asyncio.run(test3(station_name))

BIN
Apps/XinDianTu/__pycache__/Crawler.cpython-310.pyc
View File

Binary file not shown.

BIN
Apps/XinDianTu/__pycache__/Kit.cpython-310.pyc Normal file
View File

Binary file not shown.

BIN
Apps/XinDianTu/__pycache__/Opener.cpython-310.pyc
View File

Binary file not shown.

BIN
Apps/XinDianTu/__pycache__/Run.cpython-310.pyc
View File

Binary file not shown.

BIN
Apps/XinDianTu/__pycache__/Service.cpython-310.pyc
View File

Binary file not shown.

BIN
Apps/XinDianTu/__pycache__/XinDianTuReadImageKit.cpython-310.pyc Normal file
View File

Binary file not shown.