# -*- coding: utf-8 -*- import json from typing import Dict from httpx import AsyncClient, URL from app.core.config import settings from app.services.service import Service from app.utils.date import get_time_str class WeatherService(Service): def __init__(self, client: AsyncClient, server_settings=settings): super(WeatherService, self).__init__(client) self._base_url = URL(server_settings.WEATHER_HOST) self._now_time = get_time_str(fmt='YYYY-MM-DD HH:mm:ss') async def get_realtime_weather(self, project_id: str) -> Dict: url = self._base_url.join('EMS_Weather/Spring/MVC/entrance/unifier/HourHistoryData') headers = { 'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8', 'Connection': 'close' } params = { 'projectId': project_id, 'startTime': get_time_str(60 * 60 * 2, flag='ago', fmt='YYYY-MM-DD HH:mm:ss'), 'endTime': self._now_time, } raw_info = await self._post(url, params={'jsonString': json.dumps(params)}, headers=headers) try: result = raw_info.get('content')[-1] except (KeyError, TypeError, IndexError): result = {} return result