weather.py 1.2 KB

123456789101112131415161718192021222324252627282930313233343536373839
  1. # -*- coding: utf-8 -*-
  2. import json
  3. from typing import Dict
  4. from httpx import AsyncClient, URL
  5. from app.core.config import settings
  6. from app.services.service import Service
  7. from app.utils.date import get_time_str
  8. class WeatherService(Service):
  9. def __init__(self, client: AsyncClient, server_settings=settings):
  10. super(WeatherService, self).__init__(client)
  11. self._base_url = URL(server_settings.WEATHER_HOST)
  12. self._now_time = get_time_str()
  13. async def get_realtime_weather(self, project_id: str) -> Dict:
  14. url = self._base_url.join('EMS_Weather/Spring/MVC/entrance/unifier/CommonCityHourWeatherService')
  15. headers = {
  16. 'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8',
  17. 'Connection': 'close'
  18. }
  19. params = {
  20. 'cityCode': project_id[2:8],
  21. 'startDate': get_time_str(60 * 60, flag='ago'),
  22. 'endDate': self._now_time,
  23. 'minResponse': True
  24. }
  25. raw_info = await self._post(url, params={'jsonString': json.dumps(params)}, headers=headers)
  26. try:
  27. result = raw_info.get('content')[-1]
  28. except KeyError:
  29. result = {}
  30. return result