209 lines
7.2 KiB
Python
209 lines
7.2 KiB
Python
# 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, prices, address=None, distance=None) -> bool:
|
||
"""
|
||
保存场站数据：名称、地址、24小时价格计划、距离
|
||
"""
|
||
if not station_name or not prices:
|
||
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}")
|
||
|
||
# 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 小时价格表")
|
||
|
||
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'),
|
||
distance=distance,
|
||
valid_start_time=now
|
||
)
|
||
|
||
await session.commit()
|
||
|
||
logger.info(f"成功保存驿来特场站数据: {station_name} | 距离: {distance} | 价格条数: {len([x for x in hourly_schedule if x])}")
|
||
return True
|
||
|
||
async def process_station_list_vl(self, image_path, json_metadata, device_info=None, max_count=None) -> list:
|
||
"""
|
||
基于 VL 模式处理场站列表 (整页识别)
|
||
"""
|
||
# 驿来特目前逻辑待实现，先占位
|
||
return []
|