diff --git a/Apps/AiTeJiYiChong/Config/Setting.py b/Apps/AiTeJiYiChong/Config/Setting.py index 36045b7..38d05c8 100644 --- a/Apps/AiTeJiYiChong/Config/Setting.py +++ b/Apps/AiTeJiYiChong/Config/Setting.py @@ -1,34 +1,24 @@ # 采集配置 -# 滑动距离比例 (0.1 ~ 0.9)，数值越大滑动幅度越大 SCROLL_DISTANCE_RATIO = 0.5 -# 最大采集场站数量，达到此次数后停止采集 MAX_STATIONS_COUNT = 100 FIRST_RUN_ONLY_ONE_STATION = False -FIRST_RUN_ONLY_ONE_STATION = False -# 场站去重过期时间（秒），在此时间内重复出现的场站不会再次点击进入详情页 -REDIS_STATION_EXPIRE = 120 -# 数据库数据保留时长（天），超过此时长的历史数据（is_current=0）将被删除 -DATA_RETENTION_DAYS = 365 - -# 测试配置 -# [Testing] 是否在启动时清除 Redis 中的已爬取记录 (用于快速重复测试) -TEST_CLEAR_REDIS = False - -# 等待时间配置 (秒) -# 点击进入详情页后等待加载的时间 -WAIT_DETAIL_PAGE_LOAD = 2.0 -# 从详情页返回列表页后等待页面刷新的时间 -WAIT_BACK_TO_LIST = 0.8 -# 执行滑动操作后等待页面内容加载和稳定的时间 -WAIT_AFTER_SCROLL = 2.0 - -# 坐标计算与安全防护 -# 屏幕顶部安全排除比例 (0.0~1.0)，此比例区域内不进行点击（避开状态栏、顶部菜单、横幅广告等） -SAFE_EXCLUDE_RATIO = 0.20 -# 屏幕底部安全排除比例 (0.0~1.0)，此比例区域内不进行点击（避开底部导航栏、功能按钮等） -BOTTOM_SAFE_EXCLUDE_RATIO = 0.12 -# 默认回退屏幕宽度，当无法自动获取设备信息时使用 -FALLBACK_WIDTH = 1080 -# 默认回退屏幕高度，当无法自动获取设备信息时使用 -FALLBACK_HEIGHT = 2400 +REDIS_STATION_EXPIRE = 120 +DATA_RETENTION_DAYS = 365 + +# 测试配置 +TEST_CLEAR_REDIS = False + +# 等待时间配置 (秒) +WAIT_DETAIL_PAGE_LOAD = 2.0 +WAIT_BACK_TO_LIST = 0.8 +WAIT_AFTER_SCROLL = 2.0 + +# 坐标计算与安全防护 +SAFE_EXCLUDE_RATIO = 0.20 +BOTTOM_SAFE_EXCLUDE_RATIO = 0.12 +FALLBACK_WIDTH = 1080 +FALLBACK_HEIGHT = 2400 + +# 分时价格配置 +PRICE_FLATTEN_TO_24H = False diff --git a/Apps/AiTeJiYiChong/Crawler.py b/Apps/AiTeJiYiChong/Crawler.py index 50f5b3e..6f40045 100644 --- a/Apps/AiTeJiYiChong/Crawler.py +++ b/Apps/AiTeJiYiChong/Crawler.py @@ -377,8 +377,7 @@ async def analyze_prices_background(service, station_name, image_paths): if all_prices: all_prices = [p for p in all_prices if isinstance(p, dict)] all_prices.sort(key=lambda x: x.get("start", "00:00")) - hourly_schedule = ReadImageKit.expand_schedule_to_24h(all_prices) - await service.process_price_detail_data(station_name, hourly_schedule) + await service.process_price_detail_data(station_name, all_prices) logger.info(f"后台处理完成: {station_name} 的分时价格已保存。") # 处理完成后清理图片，释放磁盘空间 diff --git a/Apps/AiTeJiYiChong/Service.py b/Apps/AiTeJiYiChong/Service.py index 05d4aa0..03e074a 100644 --- a/Apps/AiTeJiYiChong/Service.py +++ b/Apps/AiTeJiYiChong/Service.py @@ -6,20 +6,19 @@ import sys import uuid from datetime import datetime -# 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__)))) if project_root not in sys.path: sys.path.append(project_root) from Apps.AiTeJiYiChong.ReadImageKit import ReadImageKit from DbKit.Db import Db -from Config.Config import DB_URL +from Config.Config import DB_URL, PRICE_FLATTEN_TO_24H_GLOBAL from Model.StationProfile import StationProfile from Model.StationStatus import StationStatus from Model.StationPriceSchedule import StationPriceSchedule +from Apps.AiTeJiYiChong.Config.Setting import PRICE_FLATTEN_TO_24H -# 配置日志 -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__) class AiTeJiYiChongService: @@ -62,23 +61,29 @@ class AiTeJiYiChongService: await self.db.close() async def process_price_detail_data(self, station_name, hourly_schedule) -> bool: - """ - 直接保存已处理好的小时段价格数据 - """ if not station_name or not hourly_schedule: return False station_hash = self.get_hash(station_name) now = datetime.now() + schedule_to_save = hourly_schedule + + use_flatten = (PRICE_FLATTEN_TO_24H_GLOBAL or PRICE_FLATTEN_TO_24H) and isinstance(hourly_schedule, list) + + if use_flatten: + rows = ReadImageKit.expand_schedule_to_24h(hourly_schedule) + if rows: + schedule_to_save = rows + async with await self.db.get_session() as session: schedule_id = self.generate_id() await self.station_price_schedule_model.save( session=session, id=schedule_id, station_hash=station_hash, - schedule_json=hourly_schedule, - valid_start_time=now + schedule_json=schedule_to_save, + valid_start_time=now, ) await session.commit() return True @@ -94,8 +99,12 @@ class AiTeJiYiChongService: station_hash = self.get_hash(station_name) now = datetime.now() - # 将识别到的原始分时价格扩展为 24 小时整点数据 - hourly_schedule = ReadImageKit.expand_schedule_to_24h(prices) + schedule_to_save = prices + use_flatten = PRICE_FLATTEN_TO_24H_GLOBAL or PRICE_FLATTEN_TO_24H + if use_flatten: + rows = ReadImageKit.expand_schedule_to_24h(prices) + if rows: + schedule_to_save = rows async with await self.db.get_session() as session: schedule_id = self.generate_id() @@ -103,13 +112,13 @@ class AiTeJiYiChongService: session=session, id=schedule_id, station_hash=station_hash, - schedule_json=hourly_schedule, - valid_start_time=now + schedule_json=schedule_to_save, + valid_start_time=now, ) await session.commit() - logger.info(f"三级页面价格详情处理完成: {station_name}, 共 {len(hourly_schedule)} 个小时段") - return hourly_schedule + logger.info(f"三级页面价格详情处理完成: {station_name}, 共 {len(schedule_to_save)} 条记录") + return schedule_to_save async def process_station_detail(self, image_path, station_name=None) -> dict: """ diff --git a/Apps/TeLaiDian/Config/Setting.py b/Apps/TeLaiDian/Config/Setting.py index e365a77..bf4f226 100644 --- a/Apps/TeLaiDian/Config/Setting.py +++ b/Apps/TeLaiDian/Config/Setting.py @@ -2,9 +2,8 @@ # 采集配置 SCROLL_DISTANCE_RATIO = 0.3 -MAX_STATIONS_COUNT = 30 +MAX_STATIONS_COUNT = 1 FIRST_RUN_ONLY_ONE_STATION = False -# 场站去重过期时间（秒），在此时间内重复出现的场站不会再次点击进入详情页 REDIS_STATION_EXPIRE = 120 # 调试绘图配置 @@ -22,3 +21,6 @@ SAFE_EXCLUDE_RATIO = 0.4 BOTTOM_SAFE_EXCLUDE_RATIO = 0.12 MIN_CARD_HEIGHT = 250 DETAIL_SCROLL_DISTANCE_RATIO = 0.9 + +# 分时价格配置 +PRICE_FLATTEN_TO_24H = False diff --git a/Apps/TeLaiDian/Crawler.py b/Apps/TeLaiDian/Crawler.py index 16889c4..c2b4683 100644 --- a/Apps/TeLaiDian/Crawler.py +++ b/Apps/TeLaiDian/Crawler.py @@ -34,6 +34,8 @@ class TeLaiDianCrawler(BaseCrawler): super().__init__(service or TeLaiDianService()) self.read_image_kit = ReadImageKit() self.redis_kit = RedisKit() + self.pending_price_tasks = [] + self.vlm_concurrency = 3 async def start(self): """ @@ -239,6 +241,13 @@ class TeLaiDianCrawler(BaseCrawler): logger.info("已完成首个场站的全流程采集，根据配置退出爬取任务。") if os.path.exists(screenshot_path): os.remove(screenshot_path) + if self.pending_price_tasks: + logger.info(f"[收尾] 等待后台价格任务完成，共 {len(self.pending_price_tasks)} 个...") + try: + await asyncio.gather(*self.pending_price_tasks, return_exceptions=True) + finally: + self.pending_price_tasks.clear() + logger.info("[收尾] 后台价格任务已全部完成") return # 滑动到下一页 @@ -246,6 +255,14 @@ class TeLaiDianCrawler(BaseCrawler): await asyncio.sleep(WAIT_AFTER_SCROLL) if os.path.exists(screenshot_path): os.remove(screenshot_path) + + if self.pending_price_tasks: + logger.info(f"[收尾] 等待后台价格任务完成，共 {len(self.pending_price_tasks)} 个...") + try: + await asyncio.gather(*self.pending_price_tasks, return_exceptions=True) + finally: + self.pending_price_tasks.clear() + logger.info("[收尾] 后台价格任务已全部完成") async def check_wrong_page(self, d, image_path): """ @@ -381,89 +398,64 @@ class TeLaiDianCrawler(BaseCrawler): logger.info(f"[电价页] 入口点击后的电价页截图已保存: {entered_price_path}") await asyncio.sleep(1.0) - - # 1. 向下滚动到底 (根据用户反馈：只有不断向下滚动，才能看到00点的) - logger.info("正在向下滚动价格列表到底部...") - max_scroll_down = 6 - for i in range(max_scroll_down): - before_scroll_path = take_screenshot(d, f"scroll_dn_{i}.jpg") + + # 1. 向上滚动到顶部（不断下拉直到看到最上面的 00:00） + logger.info("正在向上滚动价格列表到顶部...") + max_scroll_up_to_top = 8 + for i in range(max_scroll_up_to_top): + before_scroll_path = take_screenshot(d, f"scroll_up_{i}.jpg") before_scroll_md5 = get_image_content_md5(before_scroll_path) - d.swipe_ext("up", scale=0.8) # 向下滚动 = 手势向上 - await asyncio.sleep(1.2) + d.swipe_ext("down", scale=0.85) + await asyncio.sleep(1.5) - after_scroll_path = take_screenshot(d, f"scroll_dn_after_{i}.jpg") + after_scroll_path = take_screenshot(d, f"scroll_up_after_{i}.jpg") after_scroll_md5 = get_image_content_md5(after_scroll_path) - # 清理临时截图 if os.path.exists(before_scroll_path): os.remove(before_scroll_path) if os.path.exists(after_scroll_path): os.remove(after_scroll_path) if before_scroll_md5 == after_scroll_md5: - logger.info(f"价格列表已到达底部 (滚动次数: {i})") + logger.info(f"价格列表已到达顶部 (滚动次数: {i})") break - - # 2. 向上滚动并逐页截图 (从底向上抓取) - logger.info("正在向上滚动价格列表并逐页截图...") - all_prices = [] + + # 2. 从顶部开始向下逐页截图 + logger.info("正在从顶部开始向下逐页截图...") price_screenshots = [] - max_scroll_up = 8 - for p_idx in range(1, max_scroll_up + 1): + max_scroll_down_pages = 8 + for p_idx in range(1, max_scroll_down_pages + 1): # 截图当前页 p_shot = take_screenshot(d, f"tld_detail_price_{int(time.time())}_{p_idx}.jpg") - # 检查是否还能向上滚动 - before_up_md5 = get_image_content_md5(p_shot) - d.swipe_ext("down", scale=0.85) # 向上滚动 = 手势向下 - await asyncio.sleep(1.5) + # 检查是否还能向下滚动 + before_dn_md5 = get_image_content_md5(p_shot) + d.swipe_ext("up", scale=0.8) + await asyncio.sleep(1.2) # 检查是否还有新内容 - check_up_path = take_screenshot(d, f"check_up_{p_idx}.jpg") - after_up_md5 = get_image_content_md5(check_up_path) - if os.path.exists(check_up_path): os.remove(check_up_path) + check_dn_path = take_screenshot(d, f"check_dn_{p_idx}.jpg") + after_dn_md5 = get_image_content_md5(check_dn_path) + if os.path.exists(check_dn_path): os.remove(check_dn_path) price_screenshots.append(p_shot) - if before_up_md5 == after_up_md5: - logger.info(f"价格列表已到达顶部 (共抓取页数: {p_idx})") + if before_dn_md5 == after_dn_md5: + logger.info(f"价格列表已到达底部 (共抓取页数: {p_idx})") break - try: - # 后台处理价格图片 (TeLaiDian 目前是顺序处理，后续可优化为异步) - for p_shot in price_screenshots: - prices = await self.read_image_kit.analyze_detail_price_info(p_shot) - if prices: - all_prices.extend(prices) - if os.path.exists(p_shot): os.remove(p_shot) - - if all_prices: - # 去重并排序 - unique_prices = [] - seen_periods = set() - for p in all_prices: - key = f"{p.get('start')}-{p.get('end')}" - if key not in seen_periods: - unique_prices.append(p) - seen_periods.add(key) - - unique_prices.sort(key=lambda x: x.get("start", "00:00")) - - # 补充详情 - await self.service.process_price_detail_data( - station_name, - unique_prices, + if price_screenshots: + task = asyncio.create_task( + self._analyze_and_save_prices_async( + station_name=station_name, + price_screenshots=price_screenshots, total_piles=total_piles, free_piles=free_piles, piles_detail=piles_detail, parking_info=parking_info ) - logger.info(f"[详情页] {station_name} 价格信息处理完成，共 {len(unique_prices)} 条时段") - except Exception as e: - logger.error(f"[详情页] 处理价格截图失败: {e}") - finally: - # 清理 - for p_shot in price_screenshots: - if os.path.exists(p_shot): os.remove(p_shot) + ) + self.pending_price_tasks.append(task) + logger.info(f"[详情页] 已后台提交 {len(price_screenshots)} 张电价截图进行识别与保存，继续后续流程不阻塞。") async def crawl_list(self): """ @@ -479,6 +471,61 @@ class TeLaiDianCrawler(BaseCrawler): # 逻辑已在 crawl_list_logic 中通过 crawl_detail_logic 调用 pass + async def _analyze_and_save_prices_async(self, station_name, price_screenshots, total_piles=None, free_piles=None, piles_detail=None, parking_info=None): + all_prices = [] + sem = asyncio.Semaphore(self.vlm_concurrency) + + async def analyze_one(path): + try: + async with sem: + prices = await self.read_image_kit.analyze_detail_price_info(path) + return path, prices + except Exception as e: + logger.error(f"[详情页] 异步识别价格失败 ({os.path.basename(path)}): {e}") + return path, [] + + try: + tasks = [analyze_one(p) for p in price_screenshots] + results = await asyncio.gather(*tasks, return_exceptions=False) + + for path, prices in results: + if prices: + all_prices.extend(prices) + if os.path.exists(path): + try: os.remove(path) + except: pass + + if not all_prices: + logger.warning(f"[详情页] {station_name} 后台识别未提取到任何价格信息") + return + + unique_prices = [] + seen_periods = set() + for p in all_prices: + key = f"{p.get('start')}-{p.get('end')}" + if key not in seen_periods: + unique_prices.append(p) + seen_periods.add(key) + + unique_prices.sort(key=lambda x: x.get("start", "00:00")) + + await self.service.process_price_detail_data( + station_name, + unique_prices, + total_piles=total_piles, + free_piles=free_piles, + piles_detail=piles_detail, + parking_info=parking_info + ) + logger.info(f"[详情页] {station_name} 后台价格信息处理完成，共 {len(unique_prices)} 条时段") + except Exception as e: + logger.error(f"[详情页] 后台处理价格截图失败: {e}") + finally: + for p_shot in price_screenshots: + if os.path.exists(p_shot): + try: os.remove(p_shot) + except: pass + async def main(service=None): crawler = TeLaiDianCrawler(service=service) await crawler.start() diff --git a/Apps/TeLaiDian/Service.py b/Apps/TeLaiDian/Service.py index d40efd2..ce8e6e1 100644 --- a/Apps/TeLaiDian/Service.py +++ b/Apps/TeLaiDian/Service.py @@ -6,17 +6,17 @@ import sys import uuid from datetime import datetime -# 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__)))) if project_root not in sys.path: sys.path.append(project_root) from DbKit.Db import Db -from Config.Config import DB_URL +from Config.Config import DB_URL, PRICE_FLATTEN_TO_24H_GLOBAL from Model.StationProfile import StationProfile from Model.StationStatus import StationStatus from Model.StationPriceSchedule import StationPriceSchedule from Apps.TeLaiDian.Kit import setup_logger +from Apps.TeLaiDian.Config.Setting import PRICE_FLATTEN_TO_24H # 初始化日志 logger = setup_logger("TeLaiDianService") @@ -243,22 +243,89 @@ class TeLaiDianService: now = datetime.now() station_hash = self.get_hash(station_name, None) + schedule_to_save = hourly_schedule + + use_flatten = (PRICE_FLATTEN_TO_24H_GLOBAL or PRICE_FLATTEN_TO_24H) and isinstance(hourly_schedule, list) + + if use_flatten: + tmp = [None] * 24 + for p in hourly_schedule: + try: + start_str = p.get("start") or "00:00" + end_str = p.get("end") or "00:00" + start_hour = int(start_str.split(":")[0]) + end_hour = int(end_str.split(":")[0]) + if end_hour == 0 and start_hour != 0: + end_hour = 24 + if end_hour < start_hour: + end_hour += 24 + + price_val = p.get("price") + if price_val is None or price_val == 0: + try: + e = float(p.get("elec_price", 0) or 0) + s = float(p.get("service_price", 0) or 0) + if e > 0 or s > 0: + price_val = e + s + except: + pass + if price_val is None or price_val == 0: + price_val = p.get("plus_price") or p.get("market_price") or 0.0 + + price_data = { + "price": float(price_val) if price_val is not None else 0.0, + "plus_price": float(p.get("plus_price")) if p.get("plus_price") not in (None, 0) else None, + "market_price": float(p.get("market_price")) if p.get("market_price") not in (None, 0) else None, + "elec_price": float(p.get("elec_price")) if p.get("elec_price") not in (None, 0) else None, + "service_price": float(p.get("service_price")) if p.get("service_price") not in (None, 0) else None, + } + + curr = start_hour + while curr < end_hour: + hour_idx = curr % 24 + tmp[hour_idx] = { + "start": f"{hour_idx:02d}:00", + "end": f"{(hour_idx + 1):02d}:00", + **price_data, + } + curr += 1 + except Exception as e: + logger.error(f"解析价格时段失败: {p}, error: {e}") + + last_valid = None + for item in tmp: + if item: + last_valid = item + break + if last_valid: + for i in range(24): + if tmp[i] is None: + new_item = last_valid.copy() + new_item["start"] = f"{i:02d}:00" + new_item["end"] = f"{(i+1):02d}:00" + tmp[i] = new_item + else: + last_valid = tmp[i] + + if any(tmp): + schedule_to_save = tmp + async with await self.db.get_session() as session: schedule_id = self.generate_id() await self.station_price_schedule_model.save( session=session, id=schedule_id, station_hash=station_hash, - schedule_json=hourly_schedule, - valid_start_time=now + schedule_json=schedule_to_save, + valid_start_time=now, ) current_hour = now.hour current_price_info = None - if isinstance(hourly_schedule, list) and 0 <= current_hour < len(hourly_schedule): - current_price_info = hourly_schedule[current_hour] or {} - elif isinstance(hourly_schedule, dict): - current_price_info = hourly_schedule + if isinstance(schedule_to_save, list) and 0 <= current_hour < len(schedule_to_save): + current_price_info = schedule_to_save[current_hour] or {} + elif isinstance(schedule_to_save, dict): + current_price_info = schedule_to_save else: current_price_info = {} @@ -272,12 +339,11 @@ class TeLaiDianService: 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'), - valid_start_time=now + current_price=current_price_info.get("price"), + pro_price=current_price_info.get("plus_price"), + market_price=current_price_info.get("market_price"), + valid_start_time=now, ) - await session.commit() return True diff --git a/Apps/TelaiDian/Config/__pycache__/Setting.cpython-310.pyc b/Apps/TelaiDian/Config/__pycache__/Setting.cpython-310.pyc index 4e58265..6c38f9b 100644 Binary files a/Apps/TelaiDian/Config/__pycache__/Setting.cpython-310.pyc and b/Apps/TelaiDian/Config/__pycache__/Setting.cpython-310.pyc differ diff --git a/Apps/TelaiDian/__pycache__/Crawler.cpython-310.pyc b/Apps/TelaiDian/__pycache__/Crawler.cpython-310.pyc index 57564ff..6f06e6f 100644 Binary files a/Apps/TelaiDian/__pycache__/Crawler.cpython-310.pyc and b/Apps/TelaiDian/__pycache__/Crawler.cpython-310.pyc differ diff --git a/Apps/XinDianTu/Config/Setting.py b/Apps/XinDianTu/Config/Setting.py index 5069641..dd2bb59 100644 --- a/Apps/XinDianTu/Config/Setting.py +++ b/Apps/XinDianTu/Config/Setting.py @@ -1,31 +1,21 @@ # 采集配置 -# 滑动距离比例 (0.1 ~ 0.9)，数值越大滑动幅度越大 -# 调大步长 (从 0.25 调至 0.5)，提高采集效率 SCROLL_DISTANCE_RATIO = 0.5 -# 最大采集场站数量，达到此次数后停止采集 MAX_STATIONS_COUNT = 100 FIRST_RUN_ONLY_ONE_STATION = False -# 场站去重过期时间（秒），在此时间内重复出现的场站不会再次点击进入详情页 -REDIS_STATION_EXPIRE = 120 - -# 数据库数据保留时长（天），超过此时长的历史数据（is_current=0）将被删除 -DATA_RETENTION_DAYS = 365 - -# 等待时间配置 (秒) -# 点击进入详情页后等待加载的时间 -WAIT_DETAIL_PAGE_LOAD = 1.5 -# 从详情页返回列表页后等待页面刷新的时间 -WAIT_BACK_TO_LIST = 0.5 -# 执行滑动操作后等待页面内容加载和稳定的时间 -WAIT_AFTER_SCROLL = 2.0 - -# 坐标计算与安全防护 -# 屏幕顶部安全排除比例 (0.0~1.0)，此比例区域内不进行点击（避开状态栏、顶部菜单、横幅广告等） -SAFE_EXCLUDE_RATIO = 0.40 -# 屏幕底部安全排除比例 (0.0~1.0)，此比例区域内不进行点击（避开底部导航栏、功能按钮等） -BOTTOM_SAFE_EXCLUDE_RATIO = 0.12 -# 默认回退屏幕宽度，当无法自动获取设备信息时使用 -FALLBACK_WIDTH = 1080 -# 默认回退屏幕高度，当无法自动获取设备信息时使用 +REDIS_STATION_EXPIRE = 120 +DATA_RETENTION_DAYS = 365 + +# 等待时间配置 (秒) +WAIT_DETAIL_PAGE_LOAD = 1.5 +WAIT_BACK_TO_LIST = 0.5 +WAIT_AFTER_SCROLL = 2.0 + +# 坐标计算与安全防护 +SAFE_EXCLUDE_RATIO = 0.40 +BOTTOM_SAFE_EXCLUDE_RATIO = 0.12 +FALLBACK_WIDTH = 1080 FALLBACK_HEIGHT = 2400 + +# 分时价格配置 +PRICE_FLATTEN_TO_24H = False diff --git a/Apps/XinDianTu/Crawler.py b/Apps/XinDianTu/Crawler.py index ce6b385..9ecd176 100644 --- a/Apps/XinDianTu/Crawler.py +++ b/Apps/XinDianTu/Crawler.py @@ -118,9 +118,7 @@ async def analyze_prices_background(service, station_name, image_paths, device_i all_rows = await ReadImageKit.parse_price_schedule_multi(station_name, image_paths, device_info=device_info) if all_rows: - # 24小时铺平 - hourly_schedule = ReadImageKit.hourly_full_day(all_rows) - await service.process_price_detail_data(station_name, hourly_schedule) + await service.process_price_detail_data(station_name, all_rows) logger.info(f"后台处理完成: {station_name} 的分时价格已保存。") else: logger.warning(f"后台处理失败: {station_name} 未能解析出价格时段。") diff --git a/Apps/XinDianTu/Service.py b/Apps/XinDianTu/Service.py index b35d7f1..213f6f6 100644 --- a/Apps/XinDianTu/Service.py +++ b/Apps/XinDianTu/Service.py @@ -6,20 +6,19 @@ import sys import uuid from datetime import datetime -# 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__)))) if project_root not in sys.path: sys.path.append(project_root) from Apps.XinDianTu.ReadImageKit import ReadImageKit from DbKit.Db import Db -from Config.Config import DB_URL +from Config.Config import DB_URL, PRICE_FLATTEN_TO_24H_GLOBAL from Model.StationProfile import StationProfile from Model.StationStatus import StationStatus from Model.StationPriceSchedule import StationPriceSchedule +from Apps.XinDianTu.Config.Setting import PRICE_FLATTEN_TO_24H -# 配置日志 -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__) class XinDianTuService: @@ -529,23 +528,29 @@ class XinDianTuService: logger.info(f" - Station status recorded for: {name} (Piles: {free}/{total}, Price: {current_price})") async def process_price_detail_data(self, station_name, hourly_schedule) -> bool: - """ - 直接保存已处理好的小时段价格数据 - """ if not station_name or not hourly_schedule: return False station_hash = self.get_hash(station_name) now = datetime.now() + schedule_to_save = hourly_schedule + + use_flatten = (PRICE_FLATTEN_TO_24H_GLOBAL or PRICE_FLATTEN_TO_24H) and isinstance(hourly_schedule, list) + + if use_flatten: + rows = ReadImageKit.hourly_full_day(hourly_schedule) + if rows: + schedule_to_save = rows + async with await self.db.get_session() as session: schedule_id = self.generate_id() await self.station_price_schedule_model.save( session=session, id=schedule_id, station_hash=station_hash, - schedule_json=hourly_schedule, - valid_start_time=now + schedule_json=schedule_to_save, + valid_start_time=now, ) await session.commit() return True diff --git a/Apps/YeLiTe/Config/Setting.py b/Apps/YeLiTe/Config/Setting.py index 7c6f7e5..de23d9f 100644 --- a/Apps/YeLiTe/Config/Setting.py +++ b/Apps/YeLiTe/Config/Setting.py @@ -1,40 +1,29 @@ -# 采集配置 -# 滑动距离比例 (0.1 ~ 0.9)，数值越大滑动幅度越大 -SCROLL_DISTANCE_RATIO = 0.5 -# 最大采集场站数量，达到此次数后停止采集 -MAX_STATIONS_COUNT = 30 -FIRST_RUN_ONLY_ONE_STATION = False -# 场站去重过期时间（秒），在此时间内重复出现的场站不会再次点击进入详情页 -REDIS_STATION_EXPIRE = 120 -# 数据库数据保留时长（天），超过此时长的历史数据（is_current=0）将被删除 -DATA_RETENTION_DAYS = 365 - -# 测试配置 -# [Testing] 是否在启动时清除 Redis 中的已爬取记录 (用于快速重复测试) -TEST_CLEAR_REDIS = False - -# 调试绘图配置 -# [Debug] 是否在图片上绘制识别出的绿色矩形框 (计算机图形学可视化) -DRAW_DEBUG_BOXES = True -DEBUG_BOX_COLOR = (0, 255, 0) # BGR 格式的绿色 -DEBUG_BOX_THICKNESS = 3 # 线条粗细 - -# 等待时间配置 (秒) -# 点击进入详情页后等待加载的时间 -WAIT_DETAIL_PAGE_LOAD = 2.0 -# 从详情页返回列表页后等待页面刷新的时间 -WAIT_BACK_TO_LIST = 0.8 -# 执行滑动操作后等待页面内容加载和稳定的时间 -WAIT_AFTER_SCROLL = 2.0 - -# 坐标计算与安全防护 -# 屏幕顶部安全排除比例 (0.0~1.0)，此比例区域内不进行点击（避开状态栏、顶部菜单、横幅广告等） -SAFE_EXCLUDE_RATIO = 0.15 -# 屏幕底部安全排除比例 (0.0~1.0)，此比例区域内不进行点击（避开底部导航栏、功能按钮等） -BOTTOM_SAFE_EXCLUDE_RATIO = 0.10 -# 默认回退屏幕宽度，当无法自动获取设备信息时使用 -FALLBACK_WIDTH = 1080 -# 默认回退屏幕高度，当无法自动获取设备信息时使用 -FALLBACK_HEIGHT = 2400 -FIRST_RUN_ONLY_ONE_STATION = False +# 采集配置 +SCROLL_DISTANCE_RATIO = 0.5 +MAX_STATIONS_COUNT = 30 +FIRST_RUN_ONLY_ONE_STATION = False +REDIS_STATION_EXPIRE = 120 +DATA_RETENTION_DAYS = 365 + +# 测试配置 +TEST_CLEAR_REDIS = False + +# 调试绘图配置 +DRAW_DEBUG_BOXES = True +DEBUG_BOX_COLOR = (0, 255, 0) +DEBUG_BOX_THICKNESS = 3 + +# 等待时间配置 (秒) +WAIT_DETAIL_PAGE_LOAD = 2.0 +WAIT_BACK_TO_LIST = 0.8 +WAIT_AFTER_SCROLL = 2.0 + +# 坐标计算与安全防护 +SAFE_EXCLUDE_RATIO = 0.15 +BOTTOM_SAFE_EXCLUDE_RATIO = 0.10 +FALLBACK_WIDTH = 1080 +FALLBACK_HEIGHT = 2400 + +# 分时价格配置 +PRICE_FLATTEN_TO_24H = False diff --git a/Apps/YeLiTe/Service.py b/Apps/YeLiTe/Service.py index ffa4d3a..220ec61 100644 --- a/Apps/YeLiTe/Service.py +++ b/Apps/YeLiTe/Service.py @@ -6,19 +6,18 @@ import sys import uuid from datetime import datetime -# 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__)))) if project_root not in sys.path: sys.path.append(project_root) from DbKit.Db import Db -from Config.Config import DB_URL +from Config.Config import DB_URL, PRICE_FLATTEN_TO_24H_GLOBAL from Model.StationProfile import StationProfile from Model.StationStatus import StationStatus from Model.StationPriceSchedule import StationPriceSchedule +from Apps.YeLiTe.Config.Setting import PRICE_FLATTEN_TO_24H -# 配置日志 -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__) class YiLaiTeService: @@ -90,73 +89,63 @@ class YiLaiTeService: except Exception as e: logger.error(f"更新任务结束日志失败: {e}") - async def process_price_detail_data(self, station_name, prices, address=None, distance=None) -> bool: - """ - 保存场站数据：名称、地址、24小时价格计划、距离 - """ - if not station_name or not prices: + async def process_price_detail_data(self, station_name, hourly_schedule, address=None, distance=None) -> bool: + if not station_name or not hourly_schedule: return False station_hash = self.get_hash(station_name) now = datetime.now() - # 1. 预处理价格：生成 24 小时的价格映射 - hourly_schedule = [None] * 24 - - for p in prices: - try: - start_str = p.get('start') or '00:00' - end_str = p.get('end') or '00:00' - - # 解析小时 - start_hour = int(start_str.split(':')[0]) - end_hour = int(end_str.split(':')[0]) - - # 处理 00:00 作为结束时间 - if end_hour == 0 and start_hour != 0: - end_hour = 24 - elif end_hour == 0 and start_hour == 0 and len(prices) == 1: - end_hour = 24 - - # 如果跨天 - if end_hour < start_hour: - end_hour += 24 - - price_val = self._to_float(p.get('price')) - - curr = start_hour - while curr < end_hour: - hour_idx = curr % 24 - hourly_schedule[hour_idx] = { - "start": f"{hour_idx:02d}:00", - "end": f"{(hour_idx + 1):02d}:00", - "price": price_val - } - curr += 1 - except Exception as e: - logger.error(f"解析价格时段失败: {p}, error: {e}") + schedule_to_save = hourly_schedule - # 2. 兜底填充 (补全 24 小时) - last_valid = None - for item in hourly_schedule: - if item: - last_valid = item - break - - if last_valid: - for i in range(24): - if hourly_schedule[i] is None: - new_item = last_valid.copy() - new_item["start"] = f"{i:02d}:00" - new_item["end"] = f"{(i+1):02d}:00" - hourly_schedule[i] = new_item - else: - last_valid = hourly_schedule[i] - else: - logger.warning(f"场站 {station_name} 未能生成有效的 24 小时价格表") + use_flatten = (PRICE_FLATTEN_TO_24H_GLOBAL or PRICE_FLATTEN_TO_24H) and isinstance(hourly_schedule, list) + + if use_flatten: + tmp = [None] * 24 + for p in hourly_schedule: + try: + start_str = p.get("start") or "00:00" + end_str = p.get("end") or "00:00" + start_hour = int(start_str.split(":")[0]) + end_hour = int(end_str.split(":")[0]) + if end_hour == 0 and start_hour != 0: + end_hour = 24 + if end_hour < start_hour: + end_hour += 24 + + price_val = self._to_float(p.get("price")) + + curr = start_hour + while curr < end_hour: + hour_idx = curr % 24 + tmp[hour_idx] = { + "start": f"{hour_idx:02d}:00", + "end": f"{(hour_idx + 1):02d}:00", + "price": price_val, + } + curr += 1 + except Exception as e: + logger.error(f"解析价格时段失败: {p}, error: {e}") + + last_valid = None + for item in tmp: + if item: + last_valid = item + break + if last_valid: + for i in range(24): + if tmp[i] is None: + new_item = last_valid.copy() + new_item["start"] = f"{i:02d}:00" + new_item["end"] = f"{(i+1):02d}:00" + tmp[i] = new_item + else: + last_valid = tmp[i] + + if any(tmp): + schedule_to_save = tmp async with await self.db.get_session() as session: - # 1. 保存 Profile (如果不存在) profile_id = self.generate_id() await self.station_profile_model.save( session=session, @@ -165,23 +154,28 @@ class YiLaiTeService: operator=self.operator, station_name=station_name, address=address, - valid_start_time=now + valid_start_time=now, ) - # 2. 保存价格 schedule_id = self.generate_id() await self.station_price_schedule_model.save( session=session, id=schedule_id, station_hash=station_hash, - schedule_json=hourly_schedule, - valid_start_time=now + schedule_json=schedule_to_save, + valid_start_time=now, ) - # 3. 保存状态 (记录距离和当前价格) current_hour = now.hour - current_price_info = hourly_schedule[current_hour] or {} - + current_price_info = {} + if isinstance(schedule_to_save, list) and 0 <= current_hour < len(schedule_to_save): + try: + current_price_info = schedule_to_save[current_hour] or {} + except: + current_price_info = {} + elif isinstance(schedule_to_save, dict): + current_price_info = schedule_to_save + status_id = self.generate_id() await self.station_status_model.save( session=session, @@ -190,14 +184,15 @@ class YiLaiTeService: total_piles=None, free_piles=None, piles_detail_json=None, - current_price=current_price_info.get('price'), + current_price=current_price_info.get("price"), distance=distance, - valid_start_time=now + valid_start_time=now, ) - + await session.commit() - - logger.info(f"成功保存驿来特场站数据: {station_name} | 距离: {distance} | 价格条数: {len([x for x in hourly_schedule if x])}") + + cnt = len(schedule_to_save) if isinstance(schedule_to_save, list) else 1 + logger.info(f"成功保存驿来特场站数据: {station_name} | 距离: {distance} | 价格条数: {cnt}") return True async def process_station_list_vl(self, image_path, json_metadata, device_info=None, max_count=None) -> list: diff --git a/Config/Config.py b/Config/Config.py index b39c392..4791f46 100644 --- a/Config/Config.py +++ b/Config/Config.py @@ -50,6 +50,10 @@ REDIS_DECODE_RESPONSES = True REDIS_PORT = 18890 REDIS_MAX_CONNECTIONS = 200 -# 临时图片存储路径 -TEMP_IMAGE_DIR = r"d:\dsWork\aiData\Output" - +# 临时图片存储路径 +TEMP_IMAGE_DIR = r"d:\dsWork\aiData\Output" + +# 分时价格全局配置 +# False: 默认按各个供应商自己的 PRICE_FLATTEN_TO_24H 决定 +# True: 强制所有供应商都铺平成 24 小时整点数组 +PRICE_FLATTEN_TO_24H_GLOBAL = False