# coding=utf-8 import asyncio import hashlib import logging import os import sys import uuid from datetime import datetime 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 try: from Config.Config import DB_URL, PRICE_FLATTEN_TO_24H_GLOBAL except ModuleNotFoundError: import importlib.util _config_path_cfg = os.path.join(project_root, "Config", "Config.py") _spec_cfg = importlib.util.spec_from_file_location("project_config_fallback", _config_path_cfg) _cfg_mod = importlib.util.module_from_spec(_spec_cfg) assert _spec_cfg.loader is not None _spec_cfg.loader.exec_module(_cfg_mod) DB_URL = _cfg_mod.DB_URL PRICE_FLATTEN_TO_24H_GLOBAL = _cfg_mod.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") class TeLaiDianService: def __init__(self): self.db = Db(db_url=DB_URL) self.station_profile_model = StationProfile() self.station_status_model = StationStatus() self.station_price_schedule_model = StationPriceSchedule() self.operator = "特来电" def generate_id(self): return str(uuid.uuid4()) def get_hash(self, station_name: str, address: str = None) -> str: s = f"{station_name}{address or ''}" return hashlib.md5(s.encode('utf-8')).hexdigest() async def init_db(self): await self.db.init_db() async def close_db(self): await self.db.close() async def save_station_profile_only(self, station_name, address, distance=None): if not station_name: return False station_hash = self.get_hash(station_name, address) now = datetime.now() async with await self.db.get_session() as session: # 1. 保存 Profile profile_id = self.generate_id() await self.station_profile_model.save( session=session, id=profile_id, station_hash=station_hash, operator=self.operator, station_name=station_name, address=address, valid_start_time=now ) # 2. 如果有距离信息，也保存一个状态记录，确保距离被记录下来 if distance: status_id = self.generate_id() await self.station_status_model.save( session=session, id=status_id, station_hash=station_hash, total_piles=None, free_piles=None, piles_detail_json=None, current_price=None, distance=distance, valid_start_time=now ) await session.commit() logger.info(f"仅保存场站基础信息: {station_name} | 地址: {address} | 距离: {distance}") return True async def save_station_data(self, station_name, address, prices, total_piles=None, free_piles=None, piles_detail=None, parking_info=None, distance=None): """ 保存场站全量数据：名称、地址、24小时价格计划、当前状态 """ now = datetime.now() station_hash = self.get_hash(station_name, address) # 预处理价格：生成 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_parts = start_str.split(':') end_parts = end_str.split(':') start_hour = int(start_parts[0]) end_hour = int(end_parts[0]) # 处理 00:00 或 24:00 作为结束时间的情况 if end_hour == 0 or end_hour == 24: # 如果开始是 0，说明是跨越全天或到午夜 if start_hour == 0: end_hour = 24 else: # 比如 23:00 - 00:00 end_hour = 24 # 特殊处理：如果 start == end 且只有一条，视为全天 if start_hour == end_hour and len(prices) == 1: end_hour = start_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 # 如果还是没有价格，且有 plus_price 或 market_price，尝试回填 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') is not None and p.get('plus_price') != 0 else None, "market_price": float(p.get('market_price')) if p.get('market_price') is not None and p.get('market_price') != 0 else None, "elec_price": float(p.get('elec_price')) if p.get('elec_price') is not None and p.get('elec_price') != 0 else None, "service_price": float(p.get('service_price')) if p.get('service_price') is not None and p.get('service_price') != 0 else None } # 填充对应的小时槽位 curr = start_hour # 如果是跨天的，比如 23:00 - 01:00 if end_hour < start_hour: end_hour += 24 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_data } curr += 1 except Exception as e: logger.error(f"解析价格时段失败: {p}, error: {e}") # 最后兜底：如果有任何价格数据但 hourly_schedule 还有空位，用最近的一个填充 # 这通常处理 VLM 识别不全或者时段没对齐的情况 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 小时价格表") async with await self.db.get_session() as session: # 1. 保存 Profile profile_id = self.generate_id() await self.station_profile_model.save( session=session, id=profile_id, station_hash=station_hash, operator=self.operator, station_name=station_name, address=address, 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 ) # 3. 保存当前状态快照 (包含当前小时的价格) current_hour = now.hour current_price_info = hourly_schedule[current_hour] or {} status_id = self.generate_id() await self.station_status_model.save( session=session, id=status_id, station_hash=station_hash, total_piles=total_piles, free_piles=free_piles, piles_detail_json=piles_detail, parking_info=parking_info, distance=distance, current_price=current_price_info.get('price'), pro_price=current_price_info.get('plus_price'), market_price=current_price_info.get('market_price'), valid_start_time=now ) await session.commit() logger.info(f"成功保存场站数据: {station_name} (桩数: {total_piles}, 空闲: {free_piles}, 距离: {distance}, 价格计划条数: {len([x for x in hourly_schedule if x])})") # 记录 24 小时价格概览，方便调试 summary = [] for i in range(24): p = hourly_schedule[i] if p: summary.append(f"{i:02d}h:{p.get('price')}") logger.info(f"24小时价格概览: {' | '.join(summary)}") return True async def process_price_detail_data(self, station_name, hourly_schedule, total_piles=None, free_piles=None, piles_detail=None, parking_info=None, distance=None): if not station_name or not hourly_schedule: return False 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=schedule_to_save, valid_start_time=now, ) current_hour = now.hour current_price_info = None 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 = {} status_id = self.generate_id() await self.station_status_model.save( session=session, id=status_id, station_hash=station_hash, total_piles=total_piles, free_piles=free_piles, piles_detail_json=piles_detail, parking_info=parking_info, distance=distance, current_price=current_price_info.get("price"), pro_price=current_price_info.get("plus_price"), market_price=current_price_info.get("market_price"), valid_start_time=now, ) await session.commit() return True async def save_station_profile_and_status(self, station_name, address, total_piles=None, free_piles=None, piles_detail=None, parking_info=None, distance=None): if not station_name: return False now = datetime.now() station_hash = self.get_hash(station_name, address) async with await self.db.get_session() as session: profile_id = self.generate_id() await self.station_profile_model.save( session=session, id=profile_id, station_hash=station_hash, operator=self.operator, station_name=station_name, address=address, valid_start_time=now ) status_id = self.generate_id() await self.station_status_model.save( session=session, id=status_id, station_hash=station_hash, total_piles=total_piles, free_piles=free_piles, piles_detail_json=piles_detail, parking_info=parking_info, distance=distance, current_price=None, valid_start_time=now ) await session.commit() return True async def _test_save_profile(): service = TeLaiDianService() await service.init_db() try: ok = await service.save_station_profile_only( "测试场站-AI调试", "测试地址-杭州滨江", "1.2km" ) print("save_station_profile_only_return:", ok) finally: await service.close_db() if __name__ == "__main__": asyncio.run(_test_save_profile())