aiData/Apps/XinDianTu/ReadImageKit.py

import numpy as np
from PIL import Image, ImageDraw
import os
import asyncio
import hashlib
import json
import aiohttp
import logging
import base64
from openai import AsyncOpenAI, BadRequestError
from Apps.XinDianTu.Kit import (
    read_image, save_image, detect_rabbit_ad_close, detect_price_info_container_cv
)
from Config.Config import (
    ALY_LLM_API_KEY, VL_MODEL_NAME, VL_MODEL_NAME_AD, TEMP_IMAGE_DIR
)
from Apps.XinDianTu.Config.Setting import (
    SAFE_EXCLUDE_RATIO, FALLBACK_WIDTH, FALLBACK_HEIGHT,
    BOTTOM_SAFE_EXCLUDE_RATIO
)

logger = logging.getLogger(__name__)

# 场站过滤黑名单
STATION_EXCLUDED_TITLES = {
    "畅充会员", "组团充电", "微信社群", "申卡有礼", "夜间免停", 
    "洗手间", "不限车长", "不限车高", "组团", "综合排序", "距离优先",
    "价格最低", "空闲较多", "筛选", "地图", "列表"
}

STATION_BLACKLIST_KEYWORDS = [
    "广告", "优惠券", "点击进入", "立即参与", "咨询电话", "活动详情"
]


class ReadImageKit:
    _client = AsyncOpenAI(
        api_key=ALY_LLM_API_KEY,
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    )

    # 通用回退设备信息，仅在无法动态获取设备信息时使用
    _FALLBACK_DEVICE_INFO = {
        "displayWidth": FALLBACK_WIDTH,
        "displayHeight": FALLBACK_HEIGHT,
        "productName": "generic"
    }

    _prompt = (
        "仅输出JSON数组（不含任何说明文字），按从左到右、从上到下的顺序识别图片中的充电站区域。识别规则如下：\n"
        "1. 必须是卡片形式的充电站信息区域。\n"
        "2. 每一个卡片必须同时具备以下所有要素，否则严禁识别：\n"
        "   - **场站名称 (station_name)**: 场站名称通常位于卡片的第一行（左上角），字体最大且加粗。严禁提取'距离优先'、'空闲较多'、'价格最低'等筛选标签。严禁将卡片中下部的描述信息（如'导航到...'、'停车说明...'、'距您...'等）识别为名称。\n"
        "   - 距离信息 (distance, 例如 '5.3km' 或 '90m')，位于卡片右上角；\n"
        "   - 金额/电费 (price，例如 '0.8490')；\n"
        "   - **充电枪信息 (piles)**: 包含'超'、'快'或'慢'的类型、总枪数和空闲枪数（例如 '快 闲4/4'）。如果图中只显示数字（如 '34/42'）而没有明确标明类型，请统一识别为 `type: \"慢\"`。\n"
        "3. 如果缺少上述任何一项要素（例如只有名称 and 距离，但没有电费或枪数信息），说明它不是真正的场站卡片（可能是广告或功能入口），请直接跳过。\n"
        "\n"
        "JSON对象字段要求：\n"
        "1. b_use: 状态标识（1或0）。如果场站名称为灰色或带有“暂停使用”等标签，则为0，否则为1。\n"
        "2. station_name: 场站名称；\n"
        "3. price: 一度电的价格（数字）；\n"
        "4. pro_price: Pro会员价格（数字），无则为null；\n"
        "5. piles: 充电枪列表 [{type: '快'|'慢'|'超', free: 4, total: 4}]；\n"
        "6. parking: 停车费用描述（通常在'P'图标后，例如 '收费停车：以场站实际收费规则为准' 或 '限时免费停车...'）。\n"
        "7. distance: 距离信息字符串（例如 '5.3km' 或 '90m'）。\n"
        "8. bounds: {x1,y1,x2,y2} 区域像素坐标（0-1000）；\n"
        "9. bounds_norm: {left,top,right,bottom} 归一化坐标(0-1)；\n"
        "10. station_name_bounds: 场站名称文字区域坐标 {x1,y1,x2,y2}（0-1000）；\n"
        "11. station_name_bounds_norm: 场站名称文字归一化坐标(0-1)。\n"
        "\n"
        "重要约束（违反者不予识别）：\n"
        "A. 严禁识别广告位和筛选标签。如“畅充会员”、“组团充电”、“微信社群”、“申卡有礼”、“夜间免停”、“洗手间”、“不限车长”、“不限车高”、“组团”、“综合排序”等均不是场站。\n"
        "B. 真正的场站卡片必须是一个横跨屏幕的大卡片，包含：场站名称（大号加粗）、金额（¥开头）、距离（km或m结尾）、充电枪状态（闲x/x）。\n"
        "C. 严禁将屏幕顶部的功能图标（如畅充会员、签到等）和中间的筛选标签误认为场站卡片。区域太小或缺少要素的一律跳过。\n"
        "\n"
        "严格返回纯JSON格式。"
    )

    @classmethod
    async def find_price_entrance_ocr(cls, image_path):
        """
        使用 OCR 在详情页寻找价格入口 (全部时段 >)
        """
        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
            
            return {
                "found": True,
                "reason": "OCR 精准定位到价格入口关键字 '全部时段'",
                "point": [center_x, center_y]
            }
        
        logger.warning(f"⚠️ [OCR识别失败] 在图片中未发现价格入口关键字 (全部时段): {image_path}")
        return {"found": False}

    @classmethod
    async def detect_ad_popup(cls, image_path: str, device_info=None):
        """
        使用 VL 模型 + 图形学算法检测是否存在弹窗广告，并返回关闭按钮坐标
        """
        if not os.path.exists(image_path):
            return None

        if not device_info:
            device_info = cls._FALLBACK_DEVICE_INFO

        primary_width = device_info.get("displayWidth") or device_info.get("width", FALLBACK_WIDTH)
        primary_height = device_info.get("displayHeight") or device_info.get("height", FALLBACK_HEIGHT)

        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小时了！）。\n"
            "2. **'新电兔AI'专属悬浮广告**（高频）：通常出现在列表区域的左下方，左侧为卡通兔子头像和“新电兔AI”文字，关闭按钮是一个带 X 的圆形按钮；当右侧同一高度附近还有一个独立的浮动广告时，关闭按钮会出现在右侧浮动广告下方的圆形按钮处（一般为灰色圆圈，内部有 X，可能部分覆盖在文字上）。\n"
            "3. **底部的横幅广告**：带有明显的关闭按钮。\n\n"
            "如果同时存在左侧“新电兔AI”浮动区域和右侧浮动广告，请优先返回右侧浮动广告下方圆形关闭按钮的坐标；只有在右侧没有浮动广告时，才返回左侧“新电兔AI”下方圆形关闭按钮的坐标。\n\n"
            "请返回最主要的关闭按钮的中心坐标。\n\n"
            "请以纯 JSON 格式输出：\n"
            "{\n"
            "  \"has_ad\": true/false,\n"
            "  \"ad_type\": \"rabbit\" | \"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") and len(result.get("close_point")) == 2:
                norm_point = result["close_point"]
                norm_x, norm_y = norm_point
                ad_type = result.get("ad_type", "unknown")

                if ad_type == "rabbit":
                    ratio_x = norm_x / 1000.0
                    if ratio_x > 0.7:
                        x = int(primary_width * 0.88)
                        y = int(primary_height * 0.83)
                        logger.info("采用右侧浮动广告固定坐标关闭按钮 (0.88W, 0.83H)")
                    else:
                        x = int(norm_x / 1000 * primary_width)
                        y = int(norm_y / 1000 * primary_height)
                        offset_y = int(primary_height * 0.05)
                        offset_x = int(primary_width * 0.014)
                        y = min(primary_height - 1, y + offset_y)
                        x = min(primary_width - 1, x + offset_x)
                else:
                    x = int(norm_x / 1000 * primary_width)
                    y = int(norm_y / 1000 * primary_height)

                try:
                    os.makedirs(TEMP_IMAGE_DIR, exist_ok=True)
                    img = Image.open(image_path).convert("RGB")
                    draw = ImageDraw.Draw(img)
                    r = max(6, int(primary_height * 0.012))
                    draw.ellipse((x - r, y - r, x + r, y + r), outline="red", width=4)
                    base = os.path.basename(image_path)
                    debug_path = os.path.join(TEMP_IMAGE_DIR, f"debug_click_{base}")
                    img.save(debug_path)
                    logger.info(f"保存广告关闭按钮点击预览图: {debug_path}")
                except Exception as e:
                    logger.warning(f"保存广告关闭按钮点击预览图失败: {e}")
                return {
                    "x": x,
                    "y": y,
                    "ad_type": ad_type
                }

        except Exception as e:
            logger.error(f"Failed to detect ad popup with VL model: {e}")

        cv_rabbit_point = detect_rabbit_ad_close(image_path, debug_dir=TEMP_IMAGE_DIR)
        if cv_rabbit_point:
            logger.info(f">>> 图形学算法捕捉到兔子广告关闭按钮: {cv_rabbit_point}")
            norm_point = cv_rabbit_point
            x = int(norm_point[0] * primary_width)
            y = int(norm_point[1] * primary_height)
            try:
                os.makedirs(TEMP_IMAGE_DIR, exist_ok=True)
                img = Image.open(image_path).convert("RGB")
                draw = ImageDraw.Draw(img)
                r = max(6, int(primary_height * 0.012))
                draw.ellipse((x - r, y - r, x + r, y + r), outline="red", width=4)
                base = os.path.basename(image_path)
                debug_path = os.path.join(TEMP_IMAGE_DIR, f"debug_click_{base}")
                img.save(debug_path)
                logger.info(f"保存广告关闭按钮点击预览图(CV): {debug_path}")
            except Exception as e:
                logger.warning(f"保存广告关闭按钮点击预览图(CV)失败: {e}")
            return {"x": x, "y": y, "ad_type": "rabbit"}

        return None

    @staticmethod
    def _to_minutes(t_str: str) -> int:
        """HH:MM -> 分钟数"""
        if not t_str or ":" not in t_str:
            return 0
        try:
            h, m = map(int, t_str.split(":"))
            return h * 60 + m
        except:
            return 0

    @staticmethod
    def _fmt(t: int) -> str:
        """分钟数 -> HH:MM"""
        h = t // 60
        m = t % 60
        return f"{h:02d}:{m:02d}"

    @staticmethod
    def _extract_json(text: str) -> str:
        if not text:
            return "[]"

        cleaned = text.strip()
        if "```" in cleaned:
            lines = []
            for line in cleaned.splitlines():
                if line.strip().startswith("```"):
                    continue
                lines.append(line)
            cleaned = "\n".join(lines).strip()

        decoder = json.JSONDecoder()

        pos = 0
        while pos < len(cleaned):
            idx_dict = cleaned.find("{", pos)
            idx_list = cleaned.find("[", pos)

            candidates = [i for i in (idx_dict, idx_list) if i != -1]
            if not candidates:
                break

            start = min(candidates)
            snippet = cleaned[start:]
            try:
                _, end = decoder.raw_decode(snippet)
                return snippet[:end]
            except json.JSONDecodeError:
                pos = start + 1
                continue

        return "[]"

    @staticmethod
    def _add_center(obj, device_info):
        return ReadImageKit._add_click_point(obj, device_info, anchor="center")

    @staticmethod
    def _add_click_point(obj, device_info, anchor: str = "center"):
        # 获取显示宽高，增加容错
        display_width = float(device_info.get("displayWidth") or device_info.get("width") or 1080)
        display_height = float(device_info.get("displayHeight") or device_info.get("height") or 2400)

        bounds = obj.get("bounds")
        bn = obj.get("bounds_norm")
        text_bounds = obj.get("station_name_bounds")
        text_bn = obj.get("station_name_bounds_norm")
        uia_x = None
        uia_y = None

        def get_pixel_coords(b_data, d_w, d_h):
            if not b_data:
                return None

            # Extract values based on dict or list
            if isinstance(b_data, list) and len(b_data) == 4:
                v1, v2, v3, v4 = b_data
            elif isinstance(b_data, dict):
                # 兼容多种可能的键名
                v1 = b_data.get("left") if b_data.get("left") is not None else b_data.get("x1")
                v2 = b_data.get("top") if b_data.get("top") is not None else b_data.get("y1")
                v3 = b_data.get("right") if b_data.get("right") is not None else b_data.get("x2")
                v4 = b_data.get("bottom") if b_data.get("bottom") is not None else b_data.get("y2")

                if any(v is None for v in (v1, v2, v3, v4)):
                    return None
            else:
                return None

            try:
                v1, v2, v3, v4 = float(v1), float(v2), float(v3), float(v4)
            except (ValueError, TypeError):
                return None

            max_v = max(v1, v2, v3, v4)

            # 1. 0-1 归一化坐标
            if max_v <= 1.05:
                x1, y1, x2, y2 = v1 * d_w, v2 * d_h, v3 * d_w, v4 * d_h
            # 2. 0-1000 归一化坐标 (Qwen-VL 常用)
            elif max_v <= 1005:
                x1, y1, x2, y2 = (v1 / 1000.0) * d_w, (v2 / 1000.0) * d_h, (v3 / 1000.0) * d_w, (v4 / 1000.0) * d_h
            # 3. 绝对像素坐标
            else:
                x1, y1, x2, y2 = v1, v2, v3, v4

            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)
        if coords:
            x1, y1, x2, y2 = coords
            uia_x = int(x1 + (x2 - x1) / 2)
            uia_y = int(y1 + (y2 - y1) / 2)
            logger.info(
                f"坐标计算: 使用文字区域 -> ({uia_x}, {uia_y}) | 区域: {coords} | 屏幕: {display_width}x{display_height}")

        # 备选：使用整个卡片区域
        if uia_x is None:
            coords = get_pixel_coords(bn, display_width, display_height) or get_pixel_coords(bounds, display_width,
                                                                                             display_height)
            if coords:
                x1, y1, x2, y2 = coords
                w, h = x2 - x1, y2 - y1
                if anchor == "top_left":
                    uia_x = int(x1 + max(5.0, w * 0.15))
                    uia_y = int(y1 + max(5.0, h * 0.20))
                else:
                    # 默认中心点，但稍微偏上一点，避开底部可能的按钮
                    uia_x = int(x1 + w / 2)
                    uia_y = int(y1 + h * 0.4)
                logger.info(
                    f"坐标计算: 使用卡片区域 -> ({uia_x}, {uia_y}) | 区域: {coords} | 屏幕: {display_width}x{display_height}")

        # --- 安全过滤：过滤掉屏幕顶部的点击坐标 (通常是广告或菜单) ---
        if uia_y is not None:
            # 如果点击点在屏幕顶部 SAFE_EXCLUDE_RATIO 范围内，极大概率是误触广告位，将其排除
            if uia_y < (display_height * SAFE_EXCLUDE_RATIO):
                logger.warning(
                    f"安全排除: 坐标 ({uia_x}, {uia_y}) 位于屏幕顶部 {int(SAFE_EXCLUDE_RATIO * 100)}% 区域，疑似广告或菜单，已忽略。")
                return None

            # 如果点击点在屏幕底部 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)}% 区域，疑似底部功能区，已忽略。")
                return None

        if uia_x is not None and uia_y is not None:
            obj["uia_center_x"] = uia_x
            obj["uia_center_y"] = uia_y

        # Clean up temporary fields
        for k in ["bounds", "bounds_norm", "station_name_bounds", "station_name_bounds_norm"]:
            if k in obj:
                del obj[k]
        return obj

    @staticmethod
    async def find_all_time_button_coordinate(image_url_or_path: str, device_info: dict = None) -> dict:
        if device_info is None:
            logger.warning("未提供动态设备信息，使用通用回退配置。")
            device_info = ReadImageKit._FALLBACK_DEVICE_INFO

        if os.path.exists(image_url_or_path):
            with open(image_url_or_path, "rb") as f:
                encoded = base64.b64encode(f.read()).decode("utf-8")
                img_input = {"url": f"data:image/jpeg;base64,{encoded}"}
        else:
            img_input = {"url": image_url_or_path}

        prompt = (
            "仅输出JSON对象（不含任何说明文字），请找到图片中带有“全部时段”字样的按钮区域（通常在价格表下方，是一个带有右箭头的文字按钮）。\n"
            "返回格式示例：\n"
            "{\n"
            '  "bounds": {"x1": 100, "y1": 200, "x2": 300, "y2": 400}, \n'
            '  "bounds_norm": {"left": 0.1, "top": 0.2, "right": 0.3, "bottom": 0.4}\n'
            "}\n"
            "注意：bounds应使用0-1000的归一化坐标空间。\n"
            "如果未找到，返回空JSON {}。"
        )

        resp = await ReadImageKit._client.chat.completions.create(
            model=VL_MODEL_NAME,
            timeout=30,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "image_url",
                            "image_url": img_input,
                        },
                        {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)},
                        {"type": "text", "text": prompt},
                    ],
                },
            ],
        )

        content = resp.choices[0].message.content or ""
        raw = ReadImageKit._extract_json(content)

        try:
            data = json.loads(raw)
            if isinstance(data, dict) and (data.get("bounds") or data.get("bounds_norm")):
                data = ReadImageKit._add_center(data, device_info)
                # 只返回中心坐标
                return {
                    "uia_center_x": data.get("uia_center_x"),
                    "uia_center_y": data.get("uia_center_y")
                }
            return {}
        except Exception as e:
            logger.error(f"Error parsing JSON: {e}")
            logger.error(f"Raw content: {raw}")
            return {}

    @staticmethod
    async def find_station_coordinate_first_page(image_url_or_path: str, station_name: str, device_info: dict = None) -> dict:
        if device_info is None:
            logger.warning("未提供动态设备信息，使用通用回退配置。")
            device_info = ReadImageKit._FALLBACK_DEVICE_INFO

        if os.path.exists(image_url_or_path):
            with open(image_url_or_path, "rb") as f:
                encoded = base64.b64encode(f.read()).decode("utf-8")
                img_input = {"url": f"data:image/jpeg;base64,{encoded}"}
        else:
            img_input = {"url": image_url_or_path}

        prompt = (
                "仅输出JSON对象（不含任何说明文字）。"
                "请在图片中找到场站名称完全匹配“" + str(station_name) + "”的那一行/卡片，"
                                                                        "返回该场站名称文字本身的区域坐标（优先返回归一化坐标）。"
                                                                        "返回格式示例："
                                                                        "{\"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}}"
                                                                        "注意：bounds使用0-1000归一化坐标空间；如果找不到，返回空JSON {}。"
        )

        resp = await ReadImageKit._client.chat.completions.create(
            model=VL_MODEL_NAME,
            timeout=30,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "image_url", "image_url": img_input},
                        {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)},
                        {"type": "text", "text": prompt},
                    ],
                },
            ],
        )

        content = resp.choices[0].message.content or ""
        raw = ReadImageKit._extract_json(content)

        try:
            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")):
                data = ReadImageKit._add_click_point(data, device_info, anchor="station_text")
                return {
                    "uia_center_x": data.get("uia_center_x"),
                    "uia_center_y": data.get("uia_center_y")
                }
            return {}
        except Exception:
            return {}

    @staticmethod
    async def parse_first_level_image_url(image_url_or_path: str, device_info: dict = None) -> list:
        if device_info is None:
            logger.warning("未提供动态设备信息，使用通用回退配置。")
            device_info = ReadImageKit._FALLBACK_DEVICE_INFO

        if os.path.exists(image_url_or_path):
            with open(image_url_or_path, "rb") as f:
                encoded = base64.b64encode(f.read()).decode("utf-8")
                img_input = {"url": f"data:image/jpeg;base64,{encoded}"}
        else:
            img_input = {"url": image_url_or_path}

        resp = await ReadImageKit._client.chat.completions.create(
            model=VL_MODEL_NAME,
            timeout=30,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "image_url",
                            "image_url": img_input,
                        },
                        {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)},
                        {"type": "text", "text": ReadImageKit._prompt},
                    ],
                },
            ],
        )

        content = resp.choices[0].message.content or ""
        raw = ReadImageKit._extract_json(content)

        try:
            data = json.loads(raw)
            if isinstance(data, list):
                # 过滤掉 None (即被 _add_click_point 排除掉的顶部项)
                data = [ReadImageKit._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]
            elif isinstance(data, dict):
                if data.get("b_use", 1) == 1:
                    data = ReadImageKit._add_click_point(data, device_info, anchor="station_text")
                    data = [data] if data is not None else []
                else:
                    data = []

            # Clean up temporary field b_use if it still exists
            if isinstance(data, list):
                for item in data:
                    if "b_use" in item:
                        del item["b_use"]
            elif isinstance(data, dict) and "b_use" in data:
                del data["b_use"]

            # If dict result, wrap in list for consistency if needed, but keeping as is based on original logic logic implies list return
            if isinstance(data, dict):
                return [data] if data else []

            if isinstance(data, list) and data:
                # 移除旧的关键词过滤逻辑，完全信任视觉模型的识别结果
                # 如果未来发现模型识别了太多广告，可以在 Prompt 中强化 "严禁识别广告" 的约束
                filtered = []
                for item in data:
                    if not isinstance(item, dict):
                        continue
                    name = item.get("station_name")
                    if not name:
                        continue
                    filtered.append(item)
                data = filtered

            if isinstance(data, list) and data:
                tasks = []
                idxs = []
                for i, item in enumerate(data):
                    if not isinstance(item, dict):
                        continue
                    if item.get("uia_center_x") is None or item.get("uia_center_y") is None:
                        name = item.get("station_name")
                        if name:
                            idxs.append(i)
                            tasks.append(asyncio.create_task(
                                ReadImageKit.find_station_coordinate_first_page(image_url_or_path, name, device_info)
                            ))
                if tasks:
                    results = await asyncio.gather(*tasks, return_exceptions=True)
                    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:
                            data[i]["uia_center_x"] = r.get("uia_center_x")
                            data[i]["uia_center_y"] = r.get("uia_center_y")

            return data
        except Exception as e:
            logger.error(f"Error parsing JSON: {e}")
            logger.error(f"Raw content: {raw}")
            return []

    @staticmethod
    async def parse_hybrid_image(image_path, uploader, cdn_domain):
        """
        统一入口：混合识别模式
        """
        return await ReadImageKit.get_stations_hybrid(image_path, uploader, cdn_domain)

    @staticmethod
    async def parse_vl_image(image_url_or_path: str, json_metadata, device_info=None):
        """
        基于 _vl.jpg (带绿框) 和 JSON 元数据进行识别
        """
        if not json_metadata or "cards" not in json_metadata:
            return []

        if os.path.exists(image_url_or_path):
            with open(image_url_or_path, "rb") as f:
                encoded = base64.b64encode(f.read()).decode("utf-8")
                img_input = {"url": f"data:image/jpeg;base64,{encoded}"}
        else:
            img_input = {"url": image_url_or_path}

        cards_meta = json_metadata["cards"]
        # 按 id 或 Y 坐标排序，确保顺序一致 (Kit 生成时已经是 top-down)
        cards_meta.sort(key=lambda x: x["rect"][1])

        prompt = (
            "图片中用绿色矩形框标记了若干个充电站卡片区域。\n"
            "请按从上到下的顺序，依次识别每个绿色框内的场站信息，并返回一个JSON数组。\n"
            "数组中元素的顺序必须与图片中绿色框从上到下的顺序严格一致。\n"
            "如果某个框内不是有效的场站卡片（例如是广告），请返回null或空对象，不要跳过顺序。\n"
            "\n"
            "每个JSON对象包含以下字段：\n"
            "1. station_name: 场站名称；\n"
            "2. price: 价格（数字）；\n"
            "3. pro_price: Pro会员价（数字，无则null）；\n"
            "4. piles: 充电枪列表 [{type: '快'|'慢'|'超', free: 4, total: 4}]。如果图中只显示数字（如 '34/42'）而没有明确标明类型，请统一识别为 `type: \"慢\"`；\n"
            "5. tags: 标签列表（如 ['限时免费']）；\n"
            "6. parking: 停车费用描述（通常在'P'图标后，例如 '收费停车：以场站实际收费规则为准' 或 '限时免费停车...'）。\n"
            "7. distance: 距离信息字符串（例如 '5.3km'）。\n"
            "\n"
            "严格返回纯JSON格式。"
        )

        resp = await ReadImageKit._client.chat.completions.create(
            model=VL_MODEL_NAME,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "image_url", "image_url": img_input},
                        {"type": "text", "text": prompt},
                    ],
                },
            ],
        )

        content = resp.choices[0].message.content or ""
        raw = ReadImageKit._extract_json(content)

        final_stations = []
        try:
            llm_data = json.loads(raw)
            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]

                    if not item or not isinstance(item, dict):
                        continue

                    name = item.get("station_name")
                    if not name:
                        continue

                    # 注入元数据中的点击坐标
                    click_pt = meta.get("click_point", [0, 0])
                    cx, cy = click_pt

                    # 安全检查
                    img_h = 2400
                    if device_info:
                        img_h = float(device_info.get("displayHeight") or device_info.get("height") or 2400)
                    else:
                        img_h = json_metadata.get("height", 2400)

                    # 顶部安全排除
                    if cy < (img_h * SAFE_EXCLUDE_RATIO):
                        logger.warning(
                            f"VL安全排除: 坐标 ({cx}, {cy}) 位于屏幕顶部 {int(SAFE_EXCLUDE_RATIO * 100)}% 区域，已忽略。")
                        continue

                    # 底部安全排除
                    if cy > (img_h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)):
                        logger.warning(
                            f"VL安全排除: 坐标 ({cx}, {cy}) 位于屏幕底部 {int(BOTTOM_SAFE_EXCLUDE_RATIO * 100)}% 区域，疑似底部功能区，已忽略。")
                        continue

                    item["uia_center_x"] = cx
                    item["uia_center_y"] = cy
                    item["rect"] = meta.get("rect")

                    final_stations.append(item)
            else:
                logger.warning(f"LLM return format error: expected list, got {type(llm_data)}")

        except Exception as e:
            logger.error(f"Error parsing VL response: {e}")

        return final_stations

    @staticmethod
    async def get_stations_hybrid(image_path, uploader, cdn_domain):
        """
        混合识别模式：图形学切片 + 大模型小图 VL 识别
        """
        # 1. 图形学切片
        segments = ReadImageKit.get_card_segments(image_path)
        if not segments:
            return []

        # 2. 对每个切片并行进行 VL 识别
        tasks = []
        for seg in segments:
            # 上传切片
            patch_path = seg["patch_path"]
            remote_path = f"tmp/patches/{os.path.basename(patch_path)}"
            uploader.upload_file(patch_path, remote_path)
            patch_url = f"{cdn_domain}/{remote_path}"

            # 记录上传后的 URL 供识别使用
            seg["patch_url"] = patch_url
            tasks.append(ReadImageKit.recognize_card_with_vl(patch_url))

        # 等待所有 VL 完成
        results = await asyncio.gather(*tasks)

        # 3. 组装结果
        final_stations = []
        for i, res in enumerate(results):
            name = res.get("station_name")
            if name and name != "未知":
                seg = segments[i]
                res["x"] = seg["center_x"]
                res["y"] = seg["center_y"]
                final_stations.append(res)

        # 4. 后置过滤 (复用原有过滤逻辑)
        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

    @staticmethod
    def get_card_segments(image_path, output_dir="./Debug/Patches"):
        """
        基于水平灰色分割带 (Divider Band) 切取场站卡片
        返回: list of dict {"patch_path": str, "center_x": int, "center_y": int, "y_range": tuple}
        """
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)

        try:
            img = Image.open(image_path).convert('RGB')
            width, height = img.size
            img_data = np.array(img)

            logger.info(f"页面分析 - 分辨率: {width}x{height}")

            # 1. 行特征分析：识别分割线 (Divider)
            # 分割线特征：横贯全屏，亮度均匀，通常比纯白(255)稍暗，比文字内容亮
            # 典型值：Mean=242-247, Std<10

            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)

            debug_rows = []

            for y in range(height):
                # 避开顶部和底部导航栏 (15% - 85%) - 扩大排除范围以避免误识别 Header/Footer
                if y < height * 0.15 or y > height * 0.85:
                    row_types.append(1)  # 视为无关区域 (标记为 Divider 以防止形成 Segment)
                    continue

                row = img_data[y]
                row_mean = np.mean(row)
                row_std = np.std(row)

                # 判定逻辑：
                # 1. 纯白行 (卡片底色) -> mean > 252 (允许微小噪点)
                # 2. 分割线 (Divider) -> 230 < mean < 252 且 std < 15 (颜色均匀)
                # 3. 内容行 (Content) -> mean <= 230 或 (mean > 230 且 std >= 15) (有文字/图片导致方差大)

                if row_mean > 252:
                    r_type = 2  # White/Empty
                elif 230 < row_mean <= 252 and row_std < 15:
                    r_type = 1  # Divider
                else:
                    r_type = 0  # Content

                row_types.append(r_type)

                # Debug log sampling
                if y % 50 == 0:
                    debug_rows.append(f"Row {y}: Mean={row_mean:.1f}, Std={row_std:.1f} -> Type={r_type}")

            if debug_rows:
                logger.info("行特征采样 (调试用):\n" + "\n".join(debug_rows))

            # 2. 聚合连续的 Content 区域 (Type 0 or Type 2 sandwiched by Type 0)
            # 实际上，卡片是由 Divider 分隔开的区域。
            # 我们寻找两个 Divider 之间的区域，且该区域必须包含 Content (Type 0)。

            segments = []

            # 简化状态机：
            # 寻找非 Divider 的连续段
            in_segment = False
            seg_start = -1
            has_content = False

            raw_blocks = []

            for y, r_type in enumerate(row_types):
                if r_type != 1:  # Not Divider
                    if not in_segment:
                        in_segment = True
                        seg_start = y
                        has_content = False

                    if r_type == 0:
                        has_content = True
                else:  # Is Divider
                    if in_segment:
                        # 结束一段
                        if has_content:  # 只有包含内容的段才算
                            raw_blocks.append((seg_start, y))
                        in_segment = False

            # 处理最后一段
            if in_segment and has_content:
                raw_blocks.append((seg_start, len(row_types)))

            # 3. 过滤和后处理
            # 合并距离很近的块？或者过滤太小的块
            base_name = os.path.splitext(os.path.basename(image_path))[0]

            valid_segments = []
            logger.info(f"Initial raw blocks count: {len(raw_blocks)}")

            for i, (y1, y2) in enumerate(raw_blocks):
                h = y2 - y1
                # 过滤太小的块 (可能是杂噪或单纯的文字行)
                if h < 50:
                    # logger.debug(f"Block {i} too small: {h}")
                    continue
                # 过滤太大的块 (可能是全屏错误)
                if h > 1000:
                    continue

                # 再次确认内部是否有足够的“暗像素”（内容）
                # 避免切出纯白的空隙
                region = img_data[y1:y2]
                region_mean = np.mean(region)
                if region_mean > 254.5:  # 整体太白 (放宽阈值，避免误杀大面积白色的卡片)
                    logger.info(f"  [-] 忽略区域 {i}: Y({y1}-{y2}), H={h}, 整体太白 (Mean={region_mean:.1f} > 254.5)")
                    continue

                # --- 优化：边缘背景检查与宽度裁剪 (Margin Check & Crop) ---
                # 不再直接拒绝白边，而是尝试计算内容的有效宽度
                # 假设：有效内容行的方差较高，或者亮度显著不同于背景

                # 简单策略：保留全宽，但增加高度限制
                if h < 60:  # 稍微降低阈值以测试
                    logger.info(f"  [-] 忽略区域 {i}: Y({y1}-{y2}), H={h}, 高度不足 (<60)")
                    continue

                # 计算左右边距的平均亮度，辅助判断（仅记录日志，不强行过滤）
                margin_w = max(5, int(width * 0.05))
                l_margin = img_data[y1:y2, 0:margin_w]
                r_margin = img_data[y1:y2, width - margin_w:width]
                l_mean = np.mean(l_margin)
                r_mean = np.mean(r_margin)

                # 如果是全宽卡片，左右边缘可能是白色的。
                # 之前的逻辑: if l_mean > 252 or r_mean > 252: continue (导致漏检)
                # 现在移除该逻辑。

                # --- 新增：自动裁剪水平宽度 (Auto Horizontal Crop) ---
                # 尝试找到内容的左右边界 (基于列的方差或亮度差异)
                x1, x2 = 0, width

                # 从左向右扫描
                for x in range(0, int(width * 0.4), 2):
                    col = img_data[y1:y2, x]
                    # 如果这一列不是纯色背景 (std > 5) 或者明显比背景暗 (mean < 245)
                    if np.std(col) > 5 or np.mean(col) < 245:
                        x1 = x
                        break

                # 从右向左扫描
                for x in range(width - 1, int(width * 0.6), -2):
                    col = img_data[y1:y2, x]
                    if np.std(col) > 5 or np.mean(col) < 245:
                        x2 = x + 1
                        break

                # 增加一点 padding
                x1 = max(0, x1 - 10)
                x2 = min(width, x2 + 10)

                # 如果裁剪后宽度太小，可能不是有效卡片
                if (x2 - x1) < width * 0.5:
                    logger.info(f"  [-] 忽略区域 {i}: 裁剪后宽度过小 ({x2 - x1})")
                    continue

                # 保存
                patch = img.crop((x1, y1, x2, y2))
                patch_name = f"{base_name}_p{i}_{y1}.jpg"
                patch_path = os.path.join(output_dir, patch_name)
                patch.save(patch_path)

                center_y = (y1 + y2) // 2
                valid_segments.append({
                    "patch_path": os.path.abspath(patch_path),
                    "center_x": (x1 + x2) // 2,
                    "center_y": center_y,
                    "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"分析完成：识别到 {len(valid_segments)} 个区域")
            return valid_segments

        except Exception as e:
            logger.error(f"图形学切片失败: {e}", exc_info=True)
            return []

    @staticmethod
    async def recognize_card_with_vl(patch_url):
        """
        对切片进行小图 VL 识别，提取场站详细信息
        """
        resp = await ReadImageKit._client.chat.completions.create(
            model=VL_MODEL_NAME,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": (
                                "你是一个专业的充电桩信息提取助手。请识别图片中充电站卡片的核心信息。\n"
                                "注意以下识别规则：\n"
                                "1. **场站名称 (station_name)**: 场站名称通常位于卡片的第一行，字体较大且加粗。严禁提取'距离优先'、'空闲较多'、'价格最低'等筛选标签。严禁将卡片中间的长段描述信息识别为名称。\n"
                                "2. **价格 (price)**: 提取数字部分，通常以'元/度'结尾。\n"
                                "3. **桩数 (piles)**: 识别所有类型的充电桩。请提取'快'、'慢'或'超'字样及其对应的'空闲/总数'（如 '快 4/4'）。\n"
                                "   - 如果图中只显示数字（如 '34/42'）而没有明确标明类型，请统一识别为 `type: \"慢\"`。\n"
                                "   - 必须以列表形式输出所有识别到的类型。\n"
                                "请以 JSON 格式输出：\n"
                                "{\"station_name\": \"场站名称\", \"price\": \"价格\", \"piles\": [{\"type\": \"快\", \"free\": 4, \"total\": 4}]}\n"
                                "只输出 JSON，不要有任何解释文字。"
                            )
                        },
                        {"type": "image_url", "image_url": {"url": patch_url}}
                    ]
                }
            ],
            max_tokens=200
        )
        content = resp.choices[0].message.content.strip()
        # 尝试解析 JSON
        try:
            import json
            import re
            json_match = re.search(r'\{.*\}', content, re.DOTALL)
            if json_match:
                return json.loads(json_match.group())
        except Exception:
            logger.warning(f"VL 结果解析 JSON 失败: {content}")

        return {"station_name": "未知"}

    @staticmethod
    async def detect_ad(image_url_or_path: str, device_info: dict = None) -> dict:
        """
        检测图片中是否存在广告弹窗，并定位关闭按钮坐标
        
        参数：
            image_url_or_path: 图片地址或本地路径
            device_info: 设备信息
        
        返回：
            {
                "has_ad": bool,
                "uia_center_x": int|None,
                "uia_center_y": int|None
            }
        """
        if device_info is None:
            logger.warning("未提供动态设备信息，使用通用回退配置。")
            device_info = ReadImageKit._FALLBACK_DEVICE_INFO

        if os.path.exists(image_url_or_path):
            with open(image_url_or_path, "rb") as f:
                encoded = base64.b64encode(f.read()).decode("utf-8")
                img_input = {"url": f"data:image/jpeg;base64,{encoded}"}
        else:
            img_input = {"url": image_url_or_path}

        prompt = (
            "分析图片中是否存在覆盖在主界面上的广告弹窗（Popup Ad）。"
            "如果存在，请找到关闭该弹窗的按钮（通常是一个带有 'X' 的图标，可能在弹窗的右上角、右下角 or 正下方）。"
            "仅输出JSON对象（不含任何说明文字），包含以下字段："
            "1. has_ad: 布尔值，是否存在广告弹窗；"
            "2. close_button_bounds: 关闭按钮的像素坐标或归一化坐标(0-1000) {x1,y1,x2,y2}。如果不存在广告则为 null。"
            "注意：只需识别最明显的那个关闭按钮。严格返回纯JSON。"
        )

        resp = await ReadImageKit._client.chat.completions.create(
            model=VL_MODEL_NAME_AD,  # 使用更强的视觉模型
            timeout=30,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "image_url",
                            "image_url": img_input,
                        },
                        {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)},
                        {"type": "text", "text": prompt},
                    ],
                },
            ],
        )

        content = resp.choices[0].message.content or ""
        raw = ReadImageKit._extract_json(content)

        try:
            data = json.loads(raw)
            has_ad = data.get("has_ad", False)
            if has_ad and data.get("close_button_bounds"):
                # 使用现有的辅助方法转换坐标
                temp_obj = {"bounds": data.get("close_button_bounds")}
                ReadImageKit._add_click_point(temp_obj, device_info)
                return {
                    "has_ad": True,
                    "uia_center_x": temp_obj.get("uia_center_x"),
                    "uia_center_y": temp_obj.get("uia_center_y")
                }
            return {"has_ad": False, "uia_center_x": None, "uia_center_y": None}
        except Exception as e:
            logger.error(f"Error parsing ad detection JSON: {e}")
            return {"has_ad": False, "uia_center_x": None, "uia_center_y": None}

    @staticmethod
    async def _download_as_base64(url: str) -> str:
        """Helper to download image and convert to base64 for VL model fallback"""
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as resp:
                    if resp.status != 200:
                        return None
                    content = await resp.read()
                    if not content:
                        return None
                    return base64.b64encode(content).decode('utf-8')
        except Exception as e:
            logger.error(f"Failed to download image for base64 conversion: {e}")
            return None

    @staticmethod
    async def get_image_md5(image_url_or_path: str) -> str:
        """
        计算图片的 MD5（用于内容去重），支持 URL 或 本地路径
        """
        if not image_url_or_path:
            return ""

        # 如果是本地文件
        if os.path.exists(image_url_or_path):
            from Apps.XinDianTu.Kit import get_file_md5
            return get_file_md5(image_url_or_path)

        # 如果是 URL
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(image_url_or_path, timeout=10) as resp:
                    if resp.status != 200:
                        return ""
                    content = await resp.read()
                    return hashlib.md5(content).hexdigest()
        except Exception as e:
            logger.error(f"MD5 获取失败 ({image_url_or_path}): {e}")
            return ""

    @staticmethod
    async def parse_price_schedule(station_name: str, image_url_or_path: str, device_info: dict = None) -> list:
        """
        解析价格时段表，支持 URL 或 本地路径
        """
        if device_info is None:
            logger.warning("未提供动态设备信息，使用通用回退配置。")
            device_info = ReadImageKit._FALLBACK_DEVICE_INFO

        # 处理图片输入 (URL or Local Path)
        if os.path.exists(image_url_or_path):
            with open(image_url_or_path, "rb") as f:
                encoded = base64.b64encode(f.read()).decode("utf-8")
                img_input = {"url": f"data:image/jpeg;base64,{encoded}"}
        else:
            img_input = {"url": image_url_or_path}

        # 视觉解析提示词
        prompt = (
            "仅输出JSON数组（不含任何说明文字）。识别图片中所有时段的价格信息，返回每一行："
            "1) start: 开始时间（HH:MM），2) end: 结束时间（HH:MM），"
            "3) price_kwh: 价格（元/度，站点价或总价），"
            "4) electric_fee_kwh: 电费（元/度），"
            "5) service_fee_kwh: 服务费（元/度）。"
            "所有数值以数字返回，例如 1.1800。若缺失某项则填 null。严格返回纯JSON数组。"
            "注意：如果某行价格信息为空或表示同上，请尝试复用上一行的价格信息。"
        )

        try:
            resp = await ReadImageKit._client.chat.completions.create(
                model=VL_MODEL_NAME,
                timeout=30,
                messages=[
                    {
                        "role": "user",
                        "content": [
                            {"type": "image_url", "image_url": img_input},
                            {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)},
                            {"type": "text", "text": prompt},
                        ],
                    },
                ],
            )
            content = resp.choices[0].message.content or ""
            raw = ReadImageKit._extract_json(content)
            return json.loads(raw)
        except Exception as e:
            logger.error(f"解析价格失败 ({station_name}): {e}")
            return []

    @staticmethod
    async def parse_address(station_name: str, image_url_or_path: str, device_info: dict = None) -> dict:
        """
        解析地址信息，支持 URL 或 本地路径
        """
        if device_info is None:
            logger.warning("未提供动态设备信息，使用通用回退配置。")
            device_info = ReadImageKit._FALLBACK_DEVICE_INFO

        # 处理图片输入 (URL or Local Path)
        if os.path.exists(image_url_or_path):
            with open(image_url_or_path, "rb") as f:
                encoded = base64.b64encode(f.read()).decode("utf-8")
                img_input = {"url": f"data:image/jpeg;base64,{encoded}"}
        else:
            img_input = {"url": image_url_or_path}

        # 寻找按钮坐标任务
        button_task = asyncio.create_task(ReadImageKit.find_all_time_button_coordinate(image_url_or_path, device_info))

        prompt = (
            "仅输出JSON对象（不含任何说明文字）。"
            "任务1：识别图片中充电站的完整名称（full_station_name）。"
            f"提示：列表中看到的名称可能是截断的（例如“{station_name}”），请在图片上方找到最匹配的完整名称。"
            "任务2：识别充电站的详细地址（address）。"
            "寻找规则：地址通常紧跟在场站名称下方，或者在‘距离’图标（定位小图表）附近，或者在带有‘导航’按钮的同一行。"
            "返回包含 full_station_name 和 address 字段的JSON对象，例如 {\"full_station_name\": \"完整名称\", \"address\": \"详细地址\"}。"
            "如果找不到，对应字段返回空字符串。"
            "严格返回纯JSON格式。"
        )

        result = {}
        try:
            resp = await ReadImageKit._client.chat.completions.create(
                model=VL_MODEL_NAME,
                timeout=30,
                messages=[
                    {
                        "role": "user",
                        "content": [
                            {"type": "image_url", "image_url": img_input},
                            {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)},
                            {"type": "text", "text": prompt},
                        ],
                    },
                ],
            )
            content = resp.choices[0].message.content or ""
            raw = ReadImageKit._extract_json(content)
            result = json.loads(raw)
        except Exception as e:
            logger.error(f"解析地址失败 ({station_name}): {e}")

        # 合并按钮坐标
        try:
            button_result = await button_task
            if button_result:
                result.update(button_result)
        except Exception as e:
            logger.error(f"按钮坐标获取失败: {e}")

        return result

    @staticmethod
    async def parse_price_schedule_multi(station_name: str, image_urls_or_paths: list, device_info: dict = None) -> list:
        """
        多图解析并去重合并
        """
        if not image_urls_or_paths:
            return []
        
        # MD5 去重
        md5_to_img = {}
        for img in image_urls_or_paths:
            m = await ReadImageKit.get_image_md5(img)
            if m and m not in md5_to_img:
                md5_to_img[m] = img
        
        unique_imgs = list(md5_to_img.values())
        combined = []
        for img in unique_imgs:
            rows = await ReadImageKit.parse_price_schedule(station_name, img, device_info=device_info)
            if rows:
                combined.extend(rows)
        return combined

    @staticmethod
    def hourly_full_day(rows: list) -> list:
        """
        将时段列表规整为全天24个整点小时段
        
        参数：
            rows: 原始时段列表（可来自多图合并）
        
        返回：
            固定24条记录（00:00-01:00 到 23:00-24:00），
            若某小时未被任何时段覆盖，则费用为 None
        说明：
            - 选择覆盖该小时段的时段（若多个，则选择重叠时间最长的一个）
            - 保证返回结构完整，便于后续消费端显示或补全
        """
        # 预处理：转换为分钟区间
        intervals = []
        for r in rows:
            s = ReadImageKit._to_minutes(r.get("start"))
            e = ReadImageKit._to_minutes(r.get("end"))
            if e <= s:
                continue
            s = max(0, s)
            e = min(1440, e)
            intervals.append({
                "s": s, "e": e,
                "price_kwh": r.get("price_kwh"),
                "electric_fee_kwh": r.get("electric_fee_kwh"),
                "service_fee_kwh": r.get("service_fee_kwh"),
            })
        intervals.sort(key=lambda x: (x["s"], x["e"]))

        result = []
        for h in range(24):
            hs = h * 60
            he = (h + 1) * 60
            best = None
            best_overlap = 0
            for it in intervals:
                overlap = max(0, min(he, it["e"]) - max(hs, it["s"]))
                if overlap > best_overlap:
                    best_overlap = overlap
                    best = it
            result.append({
                "start": ReadImageKit._fmt(hs),
                "end": ReadImageKit._fmt(he),
                "price_kwh": best["price_kwh"] if best else None,
                "electric_fee_kwh": best["electric_fee_kwh"] if best else None,
                "service_fee_kwh": best["service_fee_kwh"] if best else None,
            })
        return result


async def test1():
    # 测试新电途第一层结构的读取
    url = "https://dsideal.obs.cn-north-1.myhuaweicloud.com/HuangHai/Temp/8fd79c68-fec6-4ca7-8d8e-fbff3c6862c8.jpg"
    print(f"Testing First Level with URL: {url}")
    result = await ReadImageKit.parse_first_level_image_url(url)
    print("First Level Result:")
    print(json.dumps(result, ensure_ascii=False, indent=2))


async def test2(station_name: str):
    # 测试新电途第二层结构的读取
    url = "https://dsideal.obs.myhuaweicloud.com/HuangHai/Temp/SecondPage.jpg"
    print(f"Testing address extraction from: {url}")
    result = await ReadImageKit.parse_address(station_name, url)
    result["station_name"] = station_name
    print("Address result:")
    print(json.dumps(result, ensure_ascii=False, indent=2))


async def test3(station_name: str):
    # 测试新电途价格表图片的解析
    samples = [
        "https://dsideal.obs.cn-north-1.myhuaweicloud.com/HuangHai/Temp/XinDianTu/1.jpg",
        "https://dsideal.obs.cn-north-1.myhuaweicloud.com/HuangHai/Temp/XinDianTu/2.jpg",
    ]
    rows = await ReadImageKit.parse_price_schedule_multi(station_name, samples)
    hourly = ReadImageKit.hourly_full_day(rows)
    print(json.dumps(hourly, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    # asyncio.run(test1())
    station_name = '吉林省看守所充电站'
    # asyncio.run(test2(station_name))
    asyncio.run(test3(station_name))