aiData/Apps/TeLaiDian/Service.py

# 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())