浏览代码

popi reimbursement

raylu 3 天之前
父节点
当前提交
209a10f1ac
共有 10 个文件被更改,包括 80 次插入16 次删除
  1. 4 4
      buy.py
  2. 2 0
      cache.py
  3. 1 0
      config.py
  4. 1 1
      junk.py
  5. 2 2
      market.py
  6. 61 0
      popi_reimbursement.py
  7. 1 1
      readme.md
  8. 2 2
      sell.py
  9. 2 2
      sell_liquidity.py
  10. 4 4
      supply.py

+ 4 - 4
buy.py

@@ -80,7 +80,7 @@ def get_raw_prices() -> typing.Mapping[str, market.RawPrice]:
 
 def get_total_buy(days: int) -> typing.Mapping[str, int]:
 	planets = [supply.Planet(fio_burn) for fio_burn in cache.get('https://rest.fnar.net/fioweb/burn/user/' + config.username,
-			headers={'Authorization': config.fio_api_key})]
+			headers={'Authorization': config.fio_rest_key})]
 
 	total_consumption: dict[str, float] = collections.defaultdict(float)
 	for planet in planets:
@@ -101,14 +101,14 @@ def get_planet_exports_and_ship_storage() -> typing.Mapping[str, int]:
 	avail = collections.defaultdict(int)
 
 	fio_burn: typing.Sequence[supply.FIOBurn] = cache.get('https://rest.fnar.net/fioweb/burn/user/' + config.username,
-			headers={'Authorization': config.fio_api_key})
+			headers={'Authorization': config.fio_rest_key})
 	for burn in fio_burn:
 		planet = supply.Planet(burn)
 		for mat in planet.exporting:
 			avail[mat] += planet.inventory.get(mat, 0)
 	
 	stores: typing.Sequence[market.Storage] = cache.get('https://rest.fnar.net/storage/' + config.username,
-			headers={'Authorization': config.fio_api_key})
+			headers={'Authorization': config.fio_rest_key})
 	for store in stores:
 		if store['Type'] != 'SHIP_STORE':
 			continue
@@ -119,7 +119,7 @@ def get_planet_exports_and_ship_storage() -> typing.Mapping[str, int]:
 
 def get_bids() -> tuple[typing.Mapping[str, int], list[market.ExchangeOrder]]:
 	orders: typing.Sequence[market.ExchangeOrder] = cache.get('https://rest.fnar.net/cxos/' + config.username,
-			headers={'Authorization': config.fio_api_key})
+			headers={'Authorization': config.fio_rest_key})
 	orders = [order for order in orders
 			if order['OrderType'] == 'BUYING' and order['Status'] != 'FILLED' and order['ExchangeCode'] == 'IC1']
 

+ 2 - 0
cache.py

@@ -28,6 +28,8 @@ def get(url: str, *, json=True, headers=None, expiry=datetime.timedelta(minutes=
 	parsed = urllib.parse.urlparse(url)
 	assert parsed.hostname is not None
 	cache_filename = urllib.parse.quote(parsed.path.removeprefix('/'), safe='')
+	if parsed.query:
+		cache_filename += '?' + urllib.parse.quote(parsed.query, safe='')
 	if json:
 		cache_filename += '.cbor'
 	cache_filename += '.xz'

+ 1 - 0
config.py

@@ -7,6 +7,7 @@ import typing
 @dataclasses.dataclass(eq=False, frozen=True, slots=True)
 class Config:
 	username: str
+	fio_rest_key: str
 	fio_api_key: str
 	punoted_api_key: str
 	market: MarketConfig

+ 1 - 1
junk.py

@@ -9,7 +9,7 @@ import supply
 
 def main() -> None:
 	fio_burns: typing.Sequence[supply.FIOBurn] = cache.get('https://rest.fnar.net/fioweb/burn/user/' + config.username,
-			headers={'Authorization': config.fio_api_key})
+			headers={'Authorization': config.fio_rest_key})
 	planets = [supply.Planet(fio_burn) for fio_burn in fio_burns]
 
 	total_consumption: dict[str, float] = collections.defaultdict(float)

+ 2 - 2
market.py

@@ -45,10 +45,10 @@ def main() -> None:
 
 def check_cxos() -> None:
 	warehouses: typing.Sequence[Warehouse] = cache.get('https://rest.fnar.net/sites/warehouses/' + config.username,
-			headers={'Authorization': config.fio_api_key})
+			headers={'Authorization': config.fio_rest_key})
 	for warehouse in warehouses:
 		storage: Storage = cache.get(f'https://rest.fnar.net/storage/{config.username}/{warehouse["StoreId"]}',
-			headers={'Authorization': config.fio_api_key})
+			headers={'Authorization': config.fio_rest_key})
 		for item in storage['StorageItems']:
 			threshold = config.market.mm_items.get(item['MaterialTicker'])
 			if threshold is not None and item['MaterialAmount'] > threshold:

+ 61 - 0
popi_reimbursement.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import sys
+import typing
+
+import cache
+from config import config
+
+def main() -> None:
+	planet = sys.argv[1]
+
+	print('getting prices...')
+	prices = get_prices()
+	print('\r')
+
+	infras: typing.Sequence[Infrastructure] = cache.get(
+			'https://api.fnar.net/infrastructure?include_contributions=true&infrastructure=' + planet)
+	for infra in infras:
+		print(infra['Type'])
+		for contribution in infra['Contributions']:
+			if contribution['ContributerName'] != config.username:
+				continue
+			print('\t' + contribution['Time'])
+			for mat in contribution['Materials']:
+				price = prices[mat['Ticker']]
+				print(f'\t\t{mat["Ticker"]} {mat["Amount"]} @ {price} = {mat["Amount"] * price}')
+
+def get_prices() -> typing.Mapping[str, float]:
+	cxos_users: typing.Mapping = cache.get('https://api.fnar.net/cx/cxos/?username=' + config.username,
+			headers={'Authorization': 'FIOAPIKey ' + config.fio_api_key})
+	(cxos_user,) = cxos_users['UserNameToData'].values()
+	cxos: list[CXOSOrder] = cxos_user['Data']
+	cxos.reverse()
+
+	prices: dict[str, float] = {}
+	for order in cxos:
+		if order['OrderType'] == 'BUYING' and order['Status'] == 'FILLED':
+			prices[order['MaterialTicker']] = order['LimitAmount']
+	return prices
+
+class Infrastructure(typing.TypedDict):
+	Type: str
+	Contributions: typing.Sequence[Contribution]
+
+class Contribution(typing.TypedDict):
+	ContributerName: str
+	Time: str
+	Materials: typing.Sequence[Material]
+
+class Material(typing.TypedDict):
+	Ticker: str
+	Amount: int
+
+class CXOSOrder(typing.TypedDict):
+	OrderType: typing.Literal['BUYING', 'SELLING']
+	Status: typing.Literal['PLACED', 'PARTIALLY_FILLED', 'FILLED']
+	MaterialTicker: str
+	LimitAmount: float
+
+if __name__ == '__main__':
+	main()

+ 1 - 1
readme.md

@@ -5,7 +5,7 @@ install uv: https://docs.astral.sh/uv/getting-started/installation/
 get a FIO API key from https://fio.fnar.net/settings and create a `config.toml` like
 ```toml
 username = 'raylu'
-fio_api_key = 'uuid_here'
+fio_rest_key = 'uuid_here'
 punoted_api_key = ''
 
 [market.mm_items]

+ 2 - 2
sell.py

@@ -58,7 +58,7 @@ def main() -> None:
 def get_ship_storage(ship_name: str) -> typing.Sequence[market.StorageItem]:
 	ship_name = ship_name.casefold()
 	stores: typing.Sequence[market.Storage] = cache.get('https://rest.fnar.net/storage/' + config.username,
-			headers={'Authorization': config.fio_api_key})
+			headers={'Authorization': config.fio_rest_key})
 	for store in stores:
 		if store['Type'] == 'SHIP_STORE' and store['Name'].casefold() == ship_name:
 			return store['StorageItems']
@@ -67,7 +67,7 @@ def get_ship_storage(ship_name: str) -> typing.Sequence[market.StorageItem]:
 def get_consumption() -> typing.Mapping[str, float]:
 	'''get consumption on planets that consume each mat'''
 	planets = [supply.Planet(fio_burn) for fio_burn in cache.get('https://rest.fnar.net/fioweb/burn/user/' + config.username,
-			headers={'Authorization': config.fio_api_key})]
+			headers={'Authorization': config.fio_rest_key})]
 
 	consumption: dict[str, float] = collections.defaultdict(float)
 	for planet in planets:

+ 2 - 2
sell_liquidity.py

@@ -16,7 +16,7 @@ def main() -> None:
 			for p in cache.get('https://refined-prun.github.io/refined-prices/all.json') if p['ExchangeCode'] == 'IC1'}
 
 	planets: typing.Sequence[supply.FIOBurn] = cache.get('https://rest.fnar.net/fioweb/burn/user/' + config.username,
-			headers={'Authorization': config.fio_api_key})
+			headers={'Authorization': config.fio_rest_key})
 	producing = frozenset(mat['MaterialTicker'] for planet in planets for mat in planet['OrderProduction']
 			if mat['MaterialTicker'] not in (c['MaterialTicker'] for c in planet['OrderConsumption']))
 	materials: list[Material] = []
@@ -32,7 +32,7 @@ def main() -> None:
 	materials.sort()
 
 	all_orders: typing.Sequence[market.ExchangeOrder] = cache.get('https://rest.fnar.net/cxos/' + config.username,
-			headers={'Authorization': config.fio_api_key})
+			headers={'Authorization': config.fio_rest_key})
 	orders: dict[str, int] = collections.defaultdict(int)
 	for order in all_orders:
 		if order['OrderType'] != 'SELLING' or order['Status'] == 'FILLED' or order['ExchangeCode'] != 'IC1':

+ 4 - 4
supply.py

@@ -27,7 +27,7 @@ def main() -> None:
 	planets = [Planet(fio_burn) for fio_burn in get_fio_burn(args.planets)]
 	if args.include_ship:
 		stores: typing.Sequence[market.Storage] = cache.get('https://rest.fnar.net/storage/' + config.username,
-				headers={'Authorization': config.fio_api_key})
+				headers={'Authorization': config.fio_rest_key})
 		for ship in args.include_ship:
 			ship_name, planet_name = ship.casefold().split('=')
 			for store in stores:
@@ -88,7 +88,7 @@ def main() -> None:
 
 def get_fio_burn(planet_names: typing.Sequence[str]) -> typing.Iterator[FIOBurn]:
 	planets: list[FIOBurn] = cache.get('https://rest.fnar.net/fioweb/burn/user/' + config.username,
-			headers={'Authorization': config.fio_api_key})
+			headers={'Authorization': config.fio_rest_key})
 	for name in planet_names:
 		name = name.casefold()
 		for planet_data in planets:
@@ -195,11 +195,11 @@ def shipping_used(materials: typing.Mapping[str, Material], ignore: typing.Colle
 
 def warehouse_inventory() -> dict[str, int]:
 	warehouses: typing.Sequence[market.Warehouse] = cache.get('https://rest.fnar.net/sites/warehouses/' + config.username,
-			headers={'Authorization': config.fio_api_key})
+			headers={'Authorization': config.fio_rest_key})
 	for warehouse in warehouses:
 		if warehouse['LocationNaturalId'] == 'HRT':
 			storage: market.Storage = cache.get(f'https://rest.fnar.net/storage/{config.username}/{warehouse["StoreId"]}',
-				headers={'Authorization': config.fio_api_key})
+				headers={'Authorization': config.fio_rest_key})
 			assert storage['Type'] == 'WAREHOUSE_STORE'
 			return {item['MaterialTicker']: item['MaterialAmount'] for item in storage['StorageItems']}
 	raise Exception("couldn't find HRT warehouse")