123456789101112131415161718192021222324252627282930313233343536373839 |
- # -*- 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()
- async def get_realtime_weather(self, project_id: str) -> Dict:
- url = self._base_url.join('EMS_Weather/Spring/MVC/entrance/unifier/CommonCityHourWeatherService')
- headers = {
- 'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8',
- 'Connection': 'close'
- }
- params = {
- 'cityCode': project_id[2:8],
- 'startDate': get_time_str(60 * 60, flag='ago'),
- 'endDate': self._now_time,
- 'minResponse': True
- }
- raw_info = await self._post(url, params={'jsonString': json.dumps(params)}, headers=headers)
- try:
- result = raw_info.get('content')[-1]
- except KeyError:
- result = {}
- return result
|