newbies.py 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960
  1. from __future__ import annotations
  2. import datetime
  3. import itertools
  4. import sys
  5. import typing
  6. import urllib.parse
  7. import httpx
  8. import cache
  9. def main() -> None:
  10. since = sys.argv[1] + 'T00:00:00Z'
  11. c = httpx.Client()
  12. codes: dict[str, str] = {company['UserName'].casefold(): company['CompanyCode']
  13. for company in cache.get('https://rest.fnar.net/company/all', expiry=datetime.timedelta(hours=1))
  14. if company['UserName'] is not None}
  15. planets = ['Avalon', 'Berthier', 'Boucher', 'Nova Honshu', 'Promitor']
  16. joined: dict[str, int] = {}
  17. for planet in planets:
  18. print('querying', planet, 'chat...', file=sys.stderr)
  19. messages: typing.Sequence[Message] = c.get('https://api.fnar.net/chat/messages',
  20. params={'channel_names': planet + ' Global Site Owners', 'updated_since': since}).raise_for_status().json()
  21. for message in messages:
  22. if message['Type'] == 'JOINED':
  23. old_join = joined.get(message['SenderUserName'])
  24. if old_join is None or old_join > message['MessageTimestamp']:
  25. joined[message['SenderUserName']] = message['MessageTimestamp']
  26. relevant_cos = {co for username in joined.keys() if (co := codes.get(username))}
  27. for username in joined.keys():
  28. co = codes.get(username)
  29. if co is None:
  30. response = c.get('https://rest.fnar.net/user/' + urllib.parse.quote(username)).raise_for_status()
  31. if response.status_code != 204:
  32. co = response.json()['CompanyCode']
  33. if co is not None:
  34. relevant_cos.add(co)
  35. rep: dict[str, int] = {}
  36. for co_batch in itertools.batched(relevant_cos, 10):
  37. companies = c.get('https://api.fnar.net/company/lookup?' + '&'.join(f'company={co}' for co in co_batch)).raise_for_status().json()
  38. for company in companies:
  39. if company is not None:
  40. rep[company['UserName'].casefold()] = company['Reputation']
  41. for username, join_ts in sorted(joined.items(), key=lambda i: i[1]):
  42. join_str = datetime.datetime.fromtimestamp(join_ts/1000, tz=datetime.UTC).strftime('%Y-%m-%d %H:%M:%S')
  43. print(username, join_str, '', '', rep.get(username, ''), sep='\t')
  44. class Message(typing.TypedDict):
  45. Type: typing.Literal['JOINED', 'LEFT', 'MESSAGE']
  46. SenderUserName: str
  47. MessageTimestamp: int
  48. if __name__ == '__main__':
  49. main()