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-15 10:14:44 +08:00
parent ccac3e889c
commit 23dfa433b0
6 changed files with 96 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
Apps/TeLaiDian/Crawler.py
View File

@@ -234,6 +234,7 @@ class TeLaiDianCrawler(BaseCrawler):
# 爬取详情
await self.crawl_detail_logic(d, station)
if os.path.exists(detail_check_path): os.remove(detail_check_path)
# 标记为已处理
await self.redis_kit.set_data(redis_key, "1", expire=REDIS_STATION_EXPIRE)
@@ -272,6 +273,7 @@ class TeLaiDianCrawler(BaseCrawler):
first_screen_path = take_screenshot(d, f"tld_detail_basic_{int(time.time())}.jpg")
station_name = station_info.get("name")
address = station_info.get("address")
distance = station_info.get("distance")
total_piles = None
free_piles = None
piles_detail = None
@@ -314,22 +316,22 @@ class TeLaiDianCrawler(BaseCrawler):
await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD)
# --- [优化] 小步快跑滚动查找逻辑 ---
# 目标：通过多次小幅度滚动 + OCR 实时探测，精准捕获“全部时段”入口，避免滑过头
# 目标：通过多次小幅度滚动 + OCR 实时探测，精准捕获价格入口（“全部时段”或“全天价格统一”），避免滑过头
max_scroll_attempts = 6 # 最大滚动尝试次数
scroll_step_ratio = 0.3 # 每次滚动的步长（屏幕高度的 30%）
found_entry = None
logger.info(f"[详情页] 开始“小步快跑”滚动查找‘全部时段’入口 (最多 {max_scroll_attempts} 次)...")
logger.info(f"[详情页] 开始“小步快跑”滚动查找价格入口 (全部时段/全天价格统一) (最多 {max_scroll_attempts} 次)...")
for i in range(max_scroll_attempts):
# 1. 截图识别当前屏
curr_screen = take_screenshot(d, f"tld_scroll_ocr_{i}_{int(time.time())}.jpg")
# 2. 尝试 OCR 识别“全部时段”
# 2. 尝试 OCR 识别价格入口
entry_data = await self.read_image_kit.find_price_entrance_ocr(curr_screen)
if entry_data.get("found"):
logger.info(f"[详情页] 第 {i+1} 次尝试：成功探测到‘全部时段’入口！")
logger.info(f"[详情页] 第 {i+1} 次尝试：成功探测到价格入口！")
found_entry = {
"screen": curr_screen,
"point": entry_data["point"]
@@ -353,7 +355,7 @@ class TeLaiDianCrawler(BaseCrawler):
price_tab_screen = found_entry["screen"]
p = found_entry["point"]
# 2. 点击“全部时段”入口
# 2. 点击价格入口
entry_x = int(p[0] * w / 1000)
entry_y = int(p[1] * h / 1000)
@@ -376,7 +378,7 @@ class TeLaiDianCrawler(BaseCrawler):
else:
# 按照用户要求：找不到文字时输出日志、截图并停止程序
fail_screen = take_screenshot(d, f"tld_ocr_fail_{int(time.time())}.jpg")
logger.error(f"❌ [OCR失败] 经过 {max_scroll_attempts} 次滚动仍未在页面中找到‘全部时段’文字！")
logger.error(f"❌ [OCR失败] 经过 {max_scroll_attempts} 次滚动仍未在页面中找到价格入口文字 (全部时段/全天价格统一)！")
logger.error(f"❌ [OCR失败] 最终截图已保存至: {fail_screen}")
logger.error("❌ [OCR失败] 程序将停止运行，请检查页面内容或识别逻辑。")
sys.exit(1) # 停止程序
@@ -489,7 +491,8 @@ class TeLaiDianCrawler(BaseCrawler):
total_piles=total_piles,
free_piles=free_piles,
piles_detail=piles_detail,
parking_info=parking_info
parking_info=parking_info,
distance=distance
)
else:
logger.warning(f"❌ 未能提取到任何价格信息，请检查页面识别逻辑")

21
Apps/TeLaiDian/Kit.py
View File

@@ -161,7 +161,7 @@ def get_ocr_reader():
def detect_price_info_container_cv(image_path):
"""
使用 OCR 精准定位详情页中的“全部时段”文本。
使用 OCR 精准定位详情页中的价格入口文本（“全部时段”或“全天价格统一”）。
返回: [x1, y1, x2, y2] 归一化坐标，如果未找到则返回 None
"""
img = read_image(image_path)
@@ -169,17 +169,20 @@ def detect_price_info_container_cv(image_path):
return None
h, w = img.shape[:2]
keywords = ['全部时段', '全天价格统一']
try:
reader = get_ocr_reader()
# 使用封装后的方法查找文本
found = reader.find_text_position(img, '全部时段')
# 获取所有识别结果，避免重复 OCR
results = reader.read_text(img)
if found:
text, quad, prob = found
# 使用封装后的方法计算归一化矩形
res = reader.get_normalized_rect(quad, w, h)
print(f"[OCR识别] 找到文本: '{text}', 置信度: {prob:.4f}, 归一化坐标: {res}")
return res
for (quad, text, prob) in results:
# 检查是否包含任一关键字
if any(kw in text for kw in keywords) and prob >= 0.5:
# 使用封装后的方法计算归一化矩形
res = reader.get_normalized_rect(quad, w, h)
print(f"[OCR识别] 找到文本: '{text}', 置信度: {prob:.4f}, 归一化坐标: {res}")
return res
except Exception as e:
print(f"OCR 识别发生异常: {e}")

11
Apps/TeLaiDian/ReadImageKit.py
View File

@@ -23,7 +23,7 @@ class ReadImageKit:
async def find_price_entrance_ocr(self, image_path):
"""
使用 OCR 在详情页寻找价格入口 (全部时段 >)
使用 OCR 在详情页寻找价格入口 (全部时段 > 或 全天价格统一 >)
不再使用 VLM 和 掩码 CV
"""
container_norm = detect_price_info_container_cv(image_path)
@@ -45,12 +45,12 @@ class ReadImageKit:
return {
"found": True,
"reason": "OCR 精准定位到‘全部时段’文字",
"reason": "OCR 精准定位到价格入口关键字",
"point": [center_x, center_y]
}
# 记录警告日志
logger.warning(f"⚠️ [OCR识别失败] 在图片中未发现‘全部时段’关键字: {image_path}")
logger.warning(f"⚠️ [OCR识别失败] 在图片中未发现价格入口关键字 (全部时段/全天价格统一): {image_path}")
return {"found": False}
async def find_close_button_vlm(self, image_path):
@@ -278,11 +278,12 @@ class ReadImageKit:
要求：
1. 忽略页面上半部分（如顶部导航栏、搜索框、广告 Banner、筛选标签等）。
2. 仅识别下半部分一条条“充电站卡片”，每张卡片通常包含：场站名称、评分、最近充电时间、距离、价格、快/慢空闲数量等。
2. 仅识别下半部分一条条“充电站卡片”，每张卡片通常包含：场站名称、评分、最近充电时间、距离（如 '5.3km' 或 '90m'）、价格、快/慢空闲数量等。
3. 不要把同一张卡片拆成多块；每条场站只对应一个矩形框。
对于每张卡片，请输出：
- name: 场站名称
- distance: 距离信息字符串（例如 '5.3km' 或 '90m'）
- address: 场站地址（如果无法确定可置为 null）
- point: 卡片中心点击坐标 [x, y]，使用归一化坐标 [0-1000]（0 表示最左/最上，1000 表示最右/最下）
- bbox: 卡片外接矩形边界 [x1, y1, x2, y2]，同样使用归一化坐标 [0-1000]
@@ -291,6 +292,7 @@ class ReadImageKit:
[
{
"name": "某某充电站",
"distance": "1.2km",
"address": "某某路 100 号",
"point": [500, 750],
"bbox": [50, 600, 950, 820]
@@ -370,6 +372,7 @@ class ReadImageKit:
final_stations.append({
"name": res.get("name"),
"distance": res.get("distance"),
"address": res.get("address"),
"point": actual_p,
"bbox": actual_bbox

3
Apps/TeLaiDian/Service.py
View File

@@ -64,7 +64,7 @@ class TeLaiDianService:
logger.info(f"仅保存场站基础信息: {station_name}")
return True
async def save_station_data(self, station_name, address, prices, total_piles=None, free_piles=None, piles_detail=None, parking_info=None):
async def save_station_data(self, station_name, address, prices, total_piles=None, free_piles=None, piles_detail=None, parking_info=None, distance=None):
"""
保存场站全量数据：名称、地址、24小时价格计划、当前状态
"""
@@ -143,6 +143,7 @@ class TeLaiDianService:
free_piles=free_piles,
piles_detail_json=piles_detail,
parking_info=parking_info,
distance=distance,
current_price=current_price_info.get('price'),
pro_price=current_price_info.get('plus_price'),
market_price=current_price_info.get('market_price'),

BIN
Apps/TelaiDian/__pycache__/Kit.cpython-310.pyc
View File

Binary file not shown.

102
Tools/T7_Export.py
View File

@@ -324,46 +324,74 @@ def _build_station_sheet(ws, row: Dict[str, Any]):
row_idx = header_row + 1
# Track last valid values for forward filling
last_valid_price = None
last_valid_electric = None
last_valid_service = None
# 预处理 schedule_items：如果只有 24 个且没有 start/end，补充时间
processed_items = []
if len(schedule_items) == 24:
has_time = any(item.get("start") or item.get("begin") or item.get("start_time") for item in schedule_items if isinstance(item, dict))
if not has_time:
for i, item in enumerate(schedule_items):
if not isinstance(item, dict): continue
new_item = item.copy()
new_item["start"] = f"{i:02d}:00"
new_item["end"] = f"{i+1:02d}:00"
processed_items.append(new_item)
else:
processed_items = schedule_items
else:
processed_items = schedule_items
for item in schedule_items:
if not isinstance(item, dict):
continue
# 进一步优化：合并连续相同价格的时段
merged_items = []
if processed_items:
current_group = None
for item in processed_items:
if not isinstance(item, dict): continue
# 提取关键价格信息用于比较
price = _to_float(item.get("price_kwh") if "price_kwh" in item else item.get("price"))
elec = _to_float(item.get("electric_fee_kwh") if "electric_fee_kwh" in item else item.get("electric_fee") if "electric_fee" in item else item.get("elec_price"))
serv = _to_float(item.get("service_fee_kwh") if "service_fee_kwh" in item else item.get("service_fee") if "service_fee" in item else item.get("service_price"))
remark = item.get("type") or item.get("tag") or item.get("period") or ""
start = item.get("start") or item.get("begin") or item.get("start_time") or ""
end = item.get("end") or item.get("finish") or item.get("end_time") or ""
start = item.get("start") or item.get("begin") or item.get("start_time") or ""
end = item.get("end") or item.get("finish") or item.get("end_time") or ""
if current_group is None:
current_group = {
"start": start,
"end": end,
"price": price,
"elec": elec,
"serv": serv,
"remark": remark
}
else:
# 如果价格、电费、服务费、备注都相同，则合并
if (current_group["price"] == price and
current_group["elec"] == elec and
current_group["serv"] == serv and
current_group["remark"] == remark):
current_group["end"] = end
else:
merged_items.append(current_group)
current_group = {
"start": start,
"end": end,
"price": price,
"elec": elec,
"serv": serv,
"remark": remark
}
if current_group:
merged_items.append(current_group)
price = _to_float(item.get("price_kwh") if "price_kwh" in item else item.get("price"))
electric_fee = _to_float(item.get("electric_fee_kwh") if "electric_fee_kwh" in item else item.get("electric_fee"))
service_fee = _to_float(item.get("service_fee_kwh") if "service_fee_kwh" in item else item.get("service_fee"))
# Forward fill if None
if price is None and last_valid_price is not None:
price = last_valid_price
if electric_fee is None and last_valid_electric is not None:
electric_fee = last_valid_electric
if service_fee is None and last_valid_service is not None:
service_fee = last_valid_service
# Update last valid values if current is not None
if price is not None:
last_valid_price = price
if electric_fee is not None:
last_valid_electric = electric_fee
if service_fee is not None:
last_valid_service = service_fee
remark = item.get("type") or item.get("tag") or item.get("period") or ""
ws.cell(row=row_idx, column=1, value=start)
ws.cell(row=row_idx, column=2, value=end)
ws.cell(row=row_idx, column=3, value=price)
ws.cell(row=row_idx, column=4, value=electric_fee)
ws.cell(row=row_idx, column=5, value=service_fee)
ws.cell(row=row_idx, column=6, value=remark)
for item in merged_items:
ws.cell(row=row_idx, column=1, value=item["start"])
ws.cell(row=row_idx, column=2, value=item["end"])
ws.cell(row=row_idx, column=3, value=item["price"])
ws.cell(row=row_idx, column=4, value=item["elec"])
ws.cell(row=row_idx, column=5, value=item["serv"])
ws.cell(row=row_idx, column=6, value=item["remark"])
_apply_table_cell_style(ws, row_idx, 1, len(headers), center_cols=[1, 2, 3, 4, 5])
row_idx += 1