323 lines
11 KiB
Python
323 lines
11 KiB
Python
# coding=utf-8
|
||
import os
|
||
import logging
|
||
|
||
import hashlib
|
||
import numpy as np
|
||
import cv2
|
||
|
||
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
|
||
|
||
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_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 detect_price_click_point_cv(image_path):
|
||
"""
|
||
使用 HSV 颜色过滤定位详情页的橘红色价格区域，返回最左侧区域的中心点击点
|
||
"""
|
||
img = read_image(image_path)
|
||
if img is None:
|
||
return None
|
||
|
||
h, w = img.shape[:2]
|
||
# 1. 转换为 HSV 空间
|
||
hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
|
||
|
||
# 2. 橘红色的 HSV 范围 (适配特来电价格颜色)
|
||
lower_orange = np.array([0, 150, 150])
|
||
upper_orange = np.array([20, 255, 255])
|
||
mask = cv2.inRange(hsv, lower_orange, upper_orange)
|
||
|
||
# 3. 对掩码进行膨胀，连接数字
|
||
kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (50, 20))
|
||
dilated = cv2.dilate(mask, kernel)
|
||
|
||
# 4. 寻找轮廓
|
||
contours, _ = cv2.findContours(dilated, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
|
||
|
||
detected_areas = []
|
||
for cnt in contours:
|
||
x, y, cw, ch = cv2.boundingRect(cnt)
|
||
# 1. 过滤掉宽度过大（可能是横幅广告）或过小（可能是杂点）的区域
|
||
# 2. 价格区域 P0 通常在屏幕的中部，且宽度约为屏幕的一半
|
||
if 200 < y < h * 0.8 and 100 < cw < w * 0.6 and ch > 30:
|
||
detected_areas.append([x, y, x + cw, y + ch])
|
||
|
||
if not detected_areas:
|
||
# 备选：如果 HSV 失败，尝试通过轮廓大小寻找
|
||
# 1.1556 这种大数字通常会有很明显的轮廓
|
||
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
|
||
thresh = cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY_INV, 11, 2)
|
||
kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (30, 10))
|
||
closed = cv2.morphologyEx(thresh, cv2.MORPH_CLOSE, kernel)
|
||
contours, _ = cv2.findContours(closed, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
|
||
for cnt in contours:
|
||
x, y, cw, ch = cv2.boundingRect(cnt)
|
||
if 200 < y < h * 0.6 and 150 < cw < 300 and 50 < ch < 150:
|
||
detected_areas.append([x, y, x + cw, y + ch])
|
||
|
||
if not detected_areas:
|
||
return None
|
||
|
||
# 5. 按 X 轴排序，取最左边的区域 (即用户确认的 P0)
|
||
# 但要排除掉可能在最左侧的导航栏返回按钮等小元素，所以前面加了宽度限制
|
||
detected_areas.sort(key=lambda b: b[0])
|
||
target = detected_areas[0]
|
||
|
||
center_x = (target[0] + target[2]) // 2
|
||
center_y = (target[1] + target[3]) // 2
|
||
|
||
return [center_x, center_y]
|
||
|
||
def detect_cards_cv(image_path, top_ratio=0.40, bottom_ratio=0.12):
|
||
"""
|
||
使用计算机图形学 (OpenCV) 检测列表中的场站卡片。
|
||
"""
|
||
from Apps.TeLaiDian.Config.Setting import MIN_CARD_HEIGHT
|
||
|
||
img = read_image(image_path)
|
||
if img is None:
|
||
return []
|
||
|
||
h, w = img.shape[:2]
|
||
|
||
# 转换为灰度图
|
||
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
|
||
|
||
# 限制检测范围
|
||
top_limit = int(h * top_ratio)
|
||
bottom_limit = int(h * (1 - bottom_ratio))
|
||
|
||
# 使用自适应阈值
|
||
thresh = cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY_INV, 11, 2)
|
||
|
||
# 闭运算：连接断开的边缘
|
||
kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (w // 2, 1))
|
||
closed = cv2.morphologyEx(thresh, cv2.MORPH_CLOSE, kernel)
|
||
|
||
# 寻找轮廓
|
||
contours, _ = cv2.findContours(closed, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
|
||
|
||
cards = []
|
||
min_card_width = int(w * 0.7)
|
||
|
||
temp_boxes = []
|
||
for cnt in contours:
|
||
x, y, cw, ch = cv2.boundingRect(cnt)
|
||
|
||
# 增加最小高度验证 MIN_CARD_HEIGHT
|
||
if (cw > min_card_width and
|
||
ch > MIN_CARD_HEIGHT and
|
||
y > top_limit and
|
||
y + ch < bottom_limit):
|
||
|
||
temp_boxes.append((x, y, cw, ch))
|
||
|
||
# 按 Y 轴排序
|
||
temp_boxes.sort(key=lambda b: b[1])
|
||
|
||
# 再次过滤和去重
|
||
for i, box in enumerate(temp_boxes):
|
||
x, y, cw, ch = box
|
||
|
||
# 检查是否与已有的框重叠
|
||
is_duplicate = False
|
||
for v in cards:
|
||
if abs(y - v[1]) < 100: # 增加去重间距
|
||
is_duplicate = True
|
||
break
|
||
if not is_duplicate:
|
||
padding = 2
|
||
cards.append([
|
||
max(0, x + padding),
|
||
max(top_limit, y + padding),
|
||
min(w, x + cw - padding),
|
||
min(bottom_limit, y + ch - padding)
|
||
])
|
||
|
||
return cards
|
||
|
||
def draw_rectangles(image_path, points, output_path=None):
|
||
"""
|
||
使用 OpenCV 在图片上绘制矩形框
|
||
"""
|
||
try:
|
||
from Apps.TeLaiDian.Config.Setting import DRAW_DEBUG_BOXES
|
||
# 这里暂时写死颜色和粗细，或者从 Setting 读取
|
||
DEBUG_BOX_COLOR = (0, 255, 0)
|
||
DEBUG_BOX_THICKNESS = 3
|
||
|
||
# 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:
|
||
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:
|
||
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
|
||
|
||
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 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. 获取供应商代号 (如 TelaiDian)
|
||
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 (如 TelaiDian.T4_TelaiDian)
|
||
if name == supplier_code:
|
||
return parent_logger
|
||
return logging.getLogger(f"{supplier_code}.{name}")
|