entry.py 1.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041
  1. from __future__ import annotations
  2. import datetime
  3. import typing
  4. import urllib.parse
  5. from workers import Response, WorkerEntrypoint, fetch
  6. if typing.TYPE_CHECKING:
  7. from workers import Request
  8. class Default(WorkerEntrypoint):
  9. async def fetch(self, request: Request) -> Response:
  10. url = urllib.parse.urlparse(request.url)
  11. match url.path:
  12. case '/gy-694c':
  13. return await gy_694c(self.env.punoted_api_key)
  14. case _:
  15. return Response('404', status=404)
  16. async def gy_694c(punoted_api_key: str) -> Response:
  17. headers = {'X-Data-Token': punoted_api_key}
  18. response = await fetch('https://api.punoted.net/v1/storages/user', headers=headers)
  19. storages = await response.json() # pyright: ignore[reportGeneralTypeIssues]
  20. (storage,) = (s for s in storages if s['Location'] == 'GY-694c')
  21. tio = 0
  22. for item in storage['StorageItems']:
  23. if item['MaterialTicker'] == 'TIO':
  24. tio = item['MaterialAmount']
  25. last_update = datetime.datetime.fromtimestamp(storage['LastUpdatedEpochMs'] / 1000)
  26. delta = datetime.datetime.now() - last_update
  27. response = await fetch('https://api.punoted.net/v1/production/user/burn?location=gy-694c', headers=headers)
  28. production = await response.json() # pyright: ignore[reportGeneralTypeIssues]
  29. (tio_production,) = (p for p in production['GY-694c'] if p['MaterialTicker'] == 'TIO')
  30. daily_production = tio_production['Production']
  31. estimated = int(tio + daily_production * (delta.total_seconds() / 60 / 60 / 24))
  32. return Response(f'raylu had {tio} TIO in storage as of {last_update} UTC ({delta.total_seconds() / 60 / 60:.1f} hours ago)\n'
  33. f'raylu is producing {daily_production}/day\n'
  34. f'estimated TIO in inventory: {estimated}\n')