| 123456789101112131415161718192021222324252627282930313233343536373839404142434445 |
- from __future__ import annotations
- import concurrent.futures
- import typing
- import tap
- import cache
- class Args(tap.Tap):
- planets: list[str]
- sector: tuple[str, ...] = ()
- def configure(self) -> None:
- self.add_argument('planets', nargs='+', metavar='planet') # take planets as positional args instead of flag
- def main() -> None:
- args = Args().parse_args()
- if args.sector:
- all_planets: typing.Sequence[Planet] = cache.get('https://rest.fnar.net/planet/allplanets')
- for planet in all_planets:
- if planet['PlanetNaturalId'].startswith(args.sector):
- args.planets.append(planet['PlanetName'])
- with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
- for planet, bases in zip(args.planets, executor.map(get_bases, args.planets)):
- print(planet)
- for base in bases:
- if base['OwnerCode'] is None:
- continue
- print(f'\t{base["OwnerCode"]}\t{base["OwnerName"]}')
- def get_bases(planet: str) -> typing.Sequence[Site]:
- return cache.get('https://rest.fnar.net/planet/sites/' + planet)
- class Planet(typing.TypedDict):
- PlanetNaturalId: str
- PlanetName: str
- class Site(typing.TypedDict):
- OwnerName: int
- OwnerCode: str | None
- if __name__ == '__main__':
- main()
|