# coding=utf-8 import hashlib import logging import os 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 Model.StationProfile import StationProfile from Model.StationStatus import StationStatus from Model.StationPriceSchedule import StationPriceSchedule from Apps.TeLaiDian.Kit import setup_logger # 初始化日志 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, s: str) -> str: 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_data(self, station_name, address, prices): """ 保存场站及其价格数据到数据库 """ if not prices: return False station_hash = self.get_hash(station_name) now = datetime.now() # 将价格转换为 24 小时的 schedule 格式 (0-23) # 每个小时存储一个包含多重价格的字典,以记录优惠价、PLUS价和挂牌价 hourly_schedule = [None] * 24 for p in prices: try: start_parts = p['start'].split(':') end_parts = p['end'].split(':') start_hour = int(start_parts[0]) end_hour = int(end_parts[0]) # 处理 00:00 作为结束时间的情况 (表示 24:00) if end_hour == 0 and (int(end_parts[1]) == 0 if len(end_parts) > 1 else True): if start_hour != 0: end_hour = 24 # 提取各项价格 price_data = { "price": float(p.get('price')) if p.get('price') is not None else 0.0, "plus_price": float(p.get('plus_price')) if p.get('plus_price') is not None else None, "market_price": float(p.get('market_price')) if p.get('market_price') is not None else None, "elec_price": float(p.get('elec_price')) if p.get('elec_price') is not None else None, "service_price": float(p.get('service_price')) if p.get('service_price') is not None else None } # 填充对应的小时槽位 curr = start_hour # 如果是跨天的,比如 23:00 - 01:00 if end_hour < start_hour: end_hour += 24 while curr < end_hour: hourly_schedule[curr % 24] = price_data curr += 1 except Exception as e: logger.error(f"解析价格时段失败: {p}, error: {e}") 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=None, # 特来电暂时没抓取总桩数 free_piles=None, piles_detail_json=None, 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}") return True