|
|
@@ -1,26 +1,43 @@
|
|
|
from __future__ import annotations
|
|
|
|
|
|
import datetime
|
|
|
+import itertools
|
|
|
import sys
|
|
|
import typing
|
|
|
|
|
|
+import httpx
|
|
|
+
|
|
|
import cache
|
|
|
|
|
|
def main() -> None:
|
|
|
+ since = sys.argv[1] + 'T00:00:00Z'
|
|
|
+
|
|
|
+ codes: dict[str, str] = {company['UserName'].casefold(): company['CompanyCode']
|
|
|
+ for company in cache.get('https://rest.fnar.net/company/all') if company['UserName'] is not None}
|
|
|
+
|
|
|
planets = ['Avalon', 'Berthier', 'Boucher', 'Nova Honshu', 'Promitor']
|
|
|
joined: dict[str, int] = {}
|
|
|
for planet in planets:
|
|
|
print('querying', planet, 'chat...', file=sys.stderr)
|
|
|
- messages: typing.Sequence[Message] = cache.get(f'https://api.fnar.net/chat/messages?channel_names={planet} Global Site Owners')
|
|
|
+ messages: typing.Sequence[Message] = httpx.get('https://api.fnar.net/chat/messages',
|
|
|
+ params={'channel_names': planet + ' Global Site Owners', 'updated_since': since}).raise_for_status().json()
|
|
|
for message in messages:
|
|
|
if message['Type'] == 'JOINED':
|
|
|
old_join = joined.get(message['SenderUserName'])
|
|
|
if old_join is None or old_join > message['MessageTimestamp']:
|
|
|
joined[message['SenderUserName']] = message['MessageTimestamp']
|
|
|
|
|
|
- for user_id, join_ts in sorted(joined.items(), key=lambda i: i[1]):
|
|
|
+ relevant_cos = {co for username in joined.keys() if (co := codes.get(username))}
|
|
|
+ rep: dict[str, int] = {}
|
|
|
+ for co_batch in itertools.batched(relevant_cos, 10):
|
|
|
+ companies = httpx.get('https://api.fnar.net/company/lookup?' + '&'.join(f'company={co}' for co in co_batch)).raise_for_status().json()
|
|
|
+ for company in companies:
|
|
|
+ if company is not None:
|
|
|
+ rep[company['UserName'].casefold()] = company['Reputation']
|
|
|
+
|
|
|
+ for username, join_ts in sorted(joined.items(), key=lambda i: i[1]):
|
|
|
join_str = datetime.datetime.fromtimestamp(join_ts/1000, tz=datetime.UTC).strftime('%Y-%m-%d %H:%M:%S')
|
|
|
- print(user_id, join_str, sep='\t')
|
|
|
+ print(username, join_str, '', '', rep.get(username, ''), sep='\t')
|
|
|
|
|
|
class Message(typing.TypedDict):
|
|
|
Type: typing.Literal['JOINED', 'LEFT', 'MESSAGE']
|
