pruncalc/newbies.py

from __future__ import annotations

import datetime
import itertools
import sys
import typing
import urllib.parse

import httpx

import cache

def main() -> None:
	since = sys.argv[1]
	c = httpx.Client()

	codes: dict[str, str] = {company['UserName'].casefold(): company['CompanyCode']
			for company in cache.get('https://rest.fnar.net/company/all', expiry=datetime.timedelta(hours=1))
			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] = c.get('https://api.fnar.net/chat/messages',
				params={'channel_names': planet + ' Global Site Owners'}).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']
	threshold = int(datetime.datetime.strptime(since, '%Y-%m-%d').replace(tzinfo=datetime.UTC).timestamp() * 1000)
	for username, join_ts in list(joined.items()):
		if join_ts < threshold:
			del joined[username]

	relevant_cos = {co for username in joined.keys() if (co := codes.get(username))}
	for username in joined.keys():
		co = codes.get(username)
		if co is None:
			response = c.get('https://rest.fnar.net/user/' + urllib.parse.quote(username)).raise_for_status()
			if response.status_code != 204:
				co = response.json()['CompanyCode']
		if co is not None:
			relevant_cos.add(co)

	rep: dict[str, int] = {}
	for co_batch in itertools.batched(relevant_cos, 10):
		companies = c.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(username, join_str, '', '', rep.get(username, ''), sep='\t')

class Message(typing.TypedDict):
	Type: typing.Literal['JOINED', 'LEFT', 'MESSAGE']
	SenderUserName: str
	MessageTimestamp: int

if __name__ == '__main__':
	main()