# 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 # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class YiLaiTeService: 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() def _to_float(self, value) -> float: """ 从字符串中提取浮点数,过滤掉非数字字符 """ if value is None: return 0.0 if isinstance(value, (int, float)): return float(value) import re try: # 提取第一个匹配的数字部分(包含小数点) match = re.search(r"[-+]?\d*\.?\d+", str(value)) if match: return float(match.group()) return 0.0 except: return 0.0 async def init_db(self): await self.db.init_db() async def close_db(self): await self.db.close() async def log_task_start(self, task_id): """记录任务开始""" from datetime import datetime try: await self.db.save("t_crawl_task_log", { "task_id": task_id, "operator": self.operator, "start_time": datetime.now(), "status": "running" }, "task_id") except Exception as e: logger.error(f"记录任务开始日志失败: {e}") async def log_task_end(self, task_id, count, status, error_msg=None): """记录任务结束""" from datetime import datetime try: end_time = datetime.now() # 假设 start_time 已经在数据库中,我们可以通过 task_id 更新 # 这里的 duration 计算可以在 SQL 中做,或者先查出来再算 await self.db.update("t_crawl_task_log", { "task_id": task_id, "end_time": end_time, "station_count": count, "status": status, "error_msg": error_msg }, "task_id") except Exception as e: logger.error(f"更新任务结束日志失败: {e}") async def process_price_detail_data(self, station_name, hourly_schedule, address=None) -> bool: """ 直接保存已处理好的小时段价格数据 """ if not station_name or not hourly_schedule: return False station_hash = self.get_hash(station_name) 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. 保存价格 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 ) await session.commit() return True async def process_station_list_vl(self, image_path, json_metadata, device_info=None, max_count=None) -> list: """ 基于 VL 模式处理场站列表 (整页识别) """ # 驿来特目前逻辑待实现,先占位 return []