weather.py 1.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243
  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(fmt="YYYY-MM-DD HH:mm:ss")
  13. async def get_realtime_weather(self, project_id: str) -> Dict:
  14. url = self._base_url.join(
  15. "EMS_Weather/Spring/MVC/entrance/unifier/HourHistoryData"
  16. )
  17. headers = {
  18. "Content-Type": "application/x-www-form-urlencoded;charset=UTF-8",
  19. "Connection": "close",
  20. }
  21. params = {
  22. "projectId": project_id,
  23. "startTime": get_time_str(
  24. 60 * 60 * 2, flag="ago", fmt="YYYY-MM-DD HH:mm:ss"
  25. ),
  26. "endTime": self._now_time,
  27. }
  28. raw_info = await self._post(
  29. url, params={"jsonString": json.dumps(params)}, headers=headers
  30. )
  31. try:
  32. result = raw_info.get("content")[-1]
  33. except (KeyError, TypeError, IndexError):
  34. result = {}
  35. return result