aiData/Apps/YeLiTe/Kit.py

# coding=utf-8
import os
import time
import logging

logger = logging.getLogger(__name__)

def clean_station_name(name):
    """
    清理场站名称，去除特殊字符和距离信息
    """
    if not name:
        return ""
    # 驿来特可能特有的清理逻辑
    import re
    # 移除常见的括号备注
    name = re.sub(r'\(.*?\)', '', name)
    name = re.sub(r'（.*?）', '', name)
    return name.strip()

def take_screenshot(d, filename, save_dir=None):
    """
    获取屏幕截图并保存
    """
    if not save_dir:
        from Config.Config import TEMP_IMAGE_DIR
        save_dir = TEMP_IMAGE_DIR
        
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)
        
    # 确保文件名有后缀
    if not filename.endswith(".jpg") and not filename.endswith(".png"):
        filename = f"{filename}.jpg"
        
    full_path = os.path.join(save_dir, filename)
    d.screenshot(full_path)
    return full_path

import hashlib
import numpy as np
import cv2

def get_file_md5(file_path):
    """计算文件的 MD5 值"""
    if not os.path.exists(file_path):
        return None
    hash_md5 = hashlib.md5()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            hash_md5.update(chunk)
    return hash_md5.hexdigest()

def get_name_md5(name):
    if not name:
        return "unknown"
    if not isinstance(name, str):
        name = str(name)
    return hashlib.md5(name.encode("utf-8")).hexdigest()

def get_image_content_md5(file_path, top_ratio=0.1, bottom_ratio=0.1):
    """
    计算图片核心内容的 MD5 值（排除状态栏和导航栏）
    """
    img = read_image(file_path)
    if img is None:
        return None

    h, w = img.shape[:2]
    top = int(h * top_ratio)
    bottom = int(h * (1 - bottom_ratio))

    # 裁剪中间部分
    content = img[top:bottom, :]

    # 直接使用原始像素数组计算 MD5，避免压缩损失
    return hashlib.md5(content.tobytes()).hexdigest()

def read_image(path):
    """读取图片，支持中文路径"""
    if not path or not os.path.exists(path):
        return None
    try:
        data = np.fromfile(path, dtype=np.uint8)
        if data.size == 0:
            return None
        img = cv2.imdecode(data, -1)
        return img
    except Exception as e:
        logger.error(f"Error reading image {path}: {e}")
        return None

def save_image(path, img):
    """保存图片，支持中文路径"""
    try:
        ext = os.path.splitext(path)[1]
        if not ext:
            ext = ".jpg"
        cv2.imencode(ext, img)[1].tofile(path)
        return True
    except Exception as e:
        logger.error(f"Error saving image {path}: {e}")
        return False

def find_template(image_path, template_name, threshold=0.8):
    """
    在图片中查找模板位置
    """
    img = read_image(image_path)
    if img is None:
        return None
        
    template_path = os.path.join(os.path.dirname(__file__), "Template", template_name)
    if not os.path.exists(template_path):
        logger.warning(f"Template not found: {template_path}")
        return None
        
    template = read_image(template_path)
    if template is None:
        return None
        
    res = cv2.matchTemplate(img, template, cv2.TM_CCOEFF_NORMED)
    min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(res)
    
    if max_val >= threshold:
        t_h, t_w = template.shape[:2]
        center_x = max_loc[0] + t_w // 2
        center_y = max_loc[1] + t_h // 2
        return (center_x, center_y)
    return None

def detect_cards_cv(image_path, top_ratio=0.15, bottom_ratio=0.1):
    """
    使用计算机图形学 (OpenCV) 检测列表中的场站卡片
    方案：使用 Canny 边缘检测 + 轮廓查找，提取圆角矩形卡片
    """
    img = read_image(image_path)
    if img is None:
        return []
        
    h, w = img.shape[:2]
    
    # 1. 预处理：转灰度 -> 高斯模糊去噪
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    blurred = cv2.GaussianBlur(gray, (5, 5), 0)
    
    # 2. Canny 边缘检测
    # 阈值设置：降低阈值以适应低对比度边缘 (如浅灰背景上的白卡片)
    # 经过测试，(10, 50) 对于浅灰背景上的白卡片效果最佳
    edges = cv2.Canny(blurred, 10, 50)
    
    # 3. 查找轮廓
    contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
    
    cards = []
    min_card_width = int(w * 0.8) # 卡片宽度通常占屏幕 80% 以上
    min_card_height = 150         # 卡片高度至少 150px
    top_limit = int(h * top_ratio)
    bottom_limit = int(h * (1 - bottom_ratio))
    
    valid_contours = []
    
    for cnt in contours:
        x, y, cw, ch = cv2.boundingRect(cnt)
        
        # 筛选条件：
        # 1. 宽度足够大 (排除小图标、文字)
        # 2. 高度适中
        # 3. 位置在安全区域内 (排除顶部导航和底部 TabBar)
        if (cw > min_card_width and 
            ch > min_card_height and 
            y > top_limit and 
            y + ch < bottom_limit):
            
            # 简单去重：检查是否与已有的卡片高度重叠
            is_duplicate = False
            center_y = y + ch // 2
            for vx, vy, vcw, vch in valid_contours:
                v_center_y = vy + vch // 2
                if abs(center_y - v_center_y) < 50: # 中心点接近，视为同一个
                    is_duplicate = True
                    break
            
            if not is_duplicate:
                valid_contours.append((x, y, cw, ch))
                # 增加一点 padding，收缩一下边界，避免点到边缘
                padding = 5
                cards.append([
                    max(0, x + padding), 
                    max(0, y + padding), 
                    min(w, x + cw - padding), 
                    min(h, y + ch - padding)
                ])
    
    # 按 y 坐标排序
    cards.sort(key=lambda c: c[1])
    
    # 如果没找到，尝试稍微放宽条件兜底 (比如宽度 > 60%)
    if not cards:
        logger.warning("严格轮廓检测未找到卡片，尝试放宽条件...")
        for cnt in contours:
            x, y, cw, ch = cv2.boundingRect(cnt)
            if (cw > int(w * 0.6) and ch > 100 and y > top_limit and y + ch < bottom_limit):
                 # 简单去重
                center_y = y + ch // 2
                is_duplicate = False
                for existing in cards:
                     ec_y = (existing[1] + existing[3]) // 2
                     if abs(center_y - ec_y) < 50:
                         is_duplicate = True
                         break
                if not is_duplicate:
                    cards.append([x, y, x+cw, y+ch])
        cards.sort(key=lambda c: c[1])

    logger.info(f"最终检测到 {len(cards)} 个场站卡片区域")
    return cards

def clear_temp_dir(save_dir=None):
    """清空临时目录中的所有文件"""
    if save_dir is None:
        from Config.Config import TEMP_IMAGE_DIR
        save_dir = TEMP_IMAGE_DIR
        
    if not os.path.exists(save_dir):
        return
        
    for f in os.listdir(save_dir):
        file_path = os.path.join(save_dir, f)
        try:
            if os.path.isfile(file_path):
                os.remove(file_path)
        except Exception as e:
            logger.error(f"Error deleting file {file_path}: {e}")

def get_image_md5(image_path):
    """
    计算图片的 MD5 值
    """
    import hashlib
    if not os.path.exists(image_path):
        return ""
    with open(image_path, 'rb') as f:
        data = f.read()
        return hashlib.md5(data).hexdigest()

def draw_rectangles(image_path, points, output_path=None):
    """
    使用 OpenCV 在图片上绘制矩形框 (计算机图形学绘制)
    :param image_path: 原始图片路径
    :param points: 坐标点列表，格式如 [[x1, y1, x2, y2], ...] 或 [(x, y), ...]
    :param output_path: 输出图片路径，如果不指定则覆盖原图
    """
    try:
        import cv2
        import numpy as np
        from Apps.YeLiTe.Config.Setting import DRAW_DEBUG_BOXES, DEBUG_BOX_COLOR, DEBUG_BOX_THICKNESS
        
        if not DRAW_DEBUG_BOXES:
            return image_path
            
        img = read_image(image_path)
        if img is None:
            return image_path
            
        for p in points:
            if len(p) == 4: # 矩形 [x1, y1, x2, y2]
                cv2.rectangle(img, (int(p[0]), int(p[1])), (int(p[2]), int(p[3])), DEBUG_BOX_COLOR, DEBUG_BOX_THICKNESS)
            elif len(p) == 2: # 点 (x, y)，画一个小圆圈或正方形表示点击位
                center = (int(p[0]), int(p[1]))
                cv2.circle(img, center, 10, DEBUG_BOX_COLOR, -1)
                
        save_path = output_path if output_path else image_path
        save_image(save_path, img)
        return save_path
    except Exception as e:
        logger.error(f"绘制矩形框失败: {e}")
        return image_path

from Util.EasyOcrKit import get_easyocr_reader

# 预加载 EasyOCR Reader (单例模式)
def get_ocr_reader():
    return get_easyocr_reader(gpu=True)

def detect_price_info_container_cv(image_path):
    """
    使用 OCR 精准定位详情页中的价格入口文本（“阶段性电价”）。
    返回: [x1, y1, x2, y2] 绝对像素坐标，如果未找到则返回 None
    """
    img = read_image(image_path)
    if img is None:
        return None
    h, w = img.shape[:2]
    
    keywords = ['阶段性电价', '当前电费']
    
    try:
        reader = get_ocr_reader()
        results = reader.read_text(img)
        
        for (quad, text, prob) in results:
            if any(kw in text for kw in keywords) and prob >= 0.5:
                pts = np.array(quad).astype(int)
                x_min = np.min(pts[:, 0])
                y_min = np.min(pts[:, 1])
                x_max = np.max(pts[:, 0])
                y_max = np.max(pts[:, 1])
                
                center_x = int((x_min + x_max) // 2)
                center_y = int((y_min + y_max) // 2)
                logger.info(f"[OCR识别] 找到文本: '{text}', 置信度: {prob:.4f}, 中心坐标: ({center_x}, {center_y})")
                return (center_x, center_y)
    except Exception as e:
        logger.error(f"OCR 识别发生异常: {e}")
        
    return None

def setup_logger(name, log_file=None, clear_old_log=False):
    """
    配置日志，支持同时输出到控制台和文件。
    使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler，
    并通过 propagate=False 避免与根 Logger 重复。
    :param name: Logger 名称
    :param log_file: 指定日志文件路径，如果不指定则使用默认路径
    :param clear_old_log: 是否在启动时清空旧日志文件
    """
    # 1. 获取供应商代号 (如 YeLiTe)
    supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
    
    # 2. 获取父级 Logger 并配置
    parent_logger = logging.getLogger(supplier_code)
    parent_logger.setLevel(logging.INFO)
    parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复

    if log_file is None:
        # 获取项目根目录 (aiData)
        root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
        log_dir = os.path.join(root_dir, "Logs")
        if not os.path.exists(log_dir):
            os.makedirs(log_dir)
        log_file = os.path.join(log_dir, f"{supplier_code}.log")

    # 如果需要清空旧日志且文件存在
    if clear_old_log and os.path.exists(log_file):
        try:
            # 关闭现有的 handler 以便删除文件
            for handler in parent_logger.handlers[:]:
                handler.close()
                parent_logger.removeHandler(handler)
            os.remove(log_file)
        except Exception as e:
            print(f"无法清空旧日志文件 {log_file}: {e}")

    if not parent_logger.handlers:
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

        # 控制台 Handler
        ch = logging.StreamHandler()
        ch.setFormatter(formatter)
        parent_logger.addHandler(ch)

        # 文件 Handler
        fh = logging.FileHandler(log_file, encoding='utf-8')
        fh.setFormatter(formatter)
        parent_logger.addHandler(fh)

    # 3. 返回子 Logger
    if name == supplier_code:
        return parent_logger
    return logging.getLogger(f"{supplier_code}.{name}")