Minor changes

This commit is contained in:
Nocturn9x 2022-03-10 11:07:23 +01:00
parent 92c36cb6e8
commit a0c17df7da
1 changed files with 73 additions and 64 deletions

View File

@ -207,8 +207,14 @@ async def main(arguments: argparse.Namespace) -> int:
if not arguments.tax_code: if not arguments.tax_code:
logger.info("You have not provided your tax/fiscal code, but I can get it for you. Please provide your GIA SSO" logger.info("You have not provided your tax/fiscal code, but I can get it for you. Please provide your GIA SSO"
" credentials below") " credentials below")
username = input("Type your GIA SSO username: ") try:
password = getpass("Type your GIA SSO password (hidden): ") username = input("Type your GIA SSO username: ")
password = getpass("Type your GIA SSO password (hidden): ")
except KeyboardInterrupt:
# Asyncio's signal handlers won't work
# when blocked in synchronous code like
# this
return
logger.info(f"Logging in as {username!r}") logger.info(f"Logging in as {username!r}")
if access_token := await login_with_gia(logger, username, password, arguments.verbose): if access_token := await login_with_gia(logger, username, password, arguments.verbose):
logger.debug(f"Access token is {access_token!r}") logger.debug(f"Access token is {access_token!r}")
@ -222,7 +228,7 @@ async def main(arguments: argparse.Namespace) -> int:
return -1 return -1
logger.info(f"Authenticating as '{arguments.tax_code}'") logger.info(f"Authenticating as '{arguments.tax_code}'")
async with httpx.AsyncClient( async with httpx.AsyncClient(
# We mimic the app's headers # We mimic the app's headers. Specifically, this is my Xiaomi Mi 11i, lol
headers={ headers={
"User-Agent": "Dalvik/2.1.0 (Linux; U; Android 11; M2012K11G Build/RKQ1.201112.002)", "User-Agent": "Dalvik/2.1.0 (Linux; U; Android 11; M2012K11G Build/RKQ1.201112.002)",
# I seriously have no idea what the app designers were thinking, but # I seriously have no idea what the app designers were thinking, but
@ -249,65 +255,68 @@ async def main(arguments: argparse.Namespace) -> int:
): ):
return 1 return 1
logger.debug(f"Request to {result.url} sent, status code is {result.status_code}") logger.debug(f"Request to {result.url} sent, status code is {result.status_code}")
if json.loads(result.text) != {}: try:
# The API returns an empty JSON object to unauthenticated if json.loads(result.text) != {}:
# requests # The API returns an empty JSON object to unauthenticated
logger.info(f"Tax code is valid! You can now leave this program running in the background") # requests
while True: logger.info(f"Tax code is valid! You can now leave this program running in the background")
if check_response( while True:
logger, if check_response(
result := await send_request(client, "get", GET_BOOKABLE_LESSONS_URL.format(arguments.tax_code)), logger,
arguments.verbose, result := await send_request(client, "get", GET_BOOKABLE_LESSONS_URL.format(arguments.tax_code)),
): arguments.verbose,
continue # Tries again ):
else: continue # Tries again
entries = [] else:
for chunk in json.loads(result.text): entries = []
# Lessons are divided according to chunks of the for chunk in json.loads(result.text):
# day, usually from 7:00 to 14:00 and from 14:00 to 22:00 # Lessons are divided according to chunks of the
for lesson in chunk["prenotazioni"]: # day, usually from 7:00 to 14:00 and from 14:00 to 22:00
if lesson["prenotabile"] and not lesson["prenotata"]: for lesson in chunk["prenotazioni"]:
if lesson["presenti"] < lesson["capacita"]: if lesson["prenotabile"] and not lesson["prenotata"]:
logger.info( if lesson["presenti"] < lesson["capacita"]:
f"Booking lesson {lesson['nome']!r} ({lesson['entry_id']}) scheduled at " logger.info(
f"{chunk['data']} from {lesson['ora_inizio']} to" f"Booking lesson {lesson['nome']!r} ({lesson['entry_id']}) scheduled at "
f" {lesson['ora_fine']} in {chunk['sede']!r} in classroom {lesson['aula']!r} " f"{chunk['data']} from {lesson['ora_inizio']} to"
f"({lesson['capacita'] - lesson['presenti']}/{lesson['capacita']}" f" {lesson['ora_fine']} in {chunk['sede']!r} in classroom {lesson['aula']!r} "
f" seats remaining)" f"({lesson['capacita'] - lesson['presenti']}/{lesson['capacita']}"
) f" seats remaining)"
entries.append(lesson["entry_id"]) )
else: entries.append(lesson["entry_id"])
logger.warning( else:
f"Lesson {lesson['nome']!r} ({lesson['entry_id']}) scheduled at" logger.warning(
f"{chunk['data']} from {lesson['ora_inizio']} to" f"Lesson {lesson['nome']!r} ({lesson['entry_id']}) scheduled at"
f" {lesson['ora_fine']} in {chunk['sede']} in classroom" f"{chunk['data']} from {lesson['ora_inizio']} to"
f" {lesson['aula']!r} has 0 remaining seats out of {lesson['capacita']}!" f" {lesson['ora_fine']} in {chunk['sede']} in classroom"
) f" {lesson['aula']!r} has 0 remaining seats out of {lesson['capacita']}!"
for entry in entries: )
# We _could_ send all entries at once, since the entry parameter is an for entry in entries:
# array, but this gives us finer error handling and makes it so that if # We _could_ send all entries at once, since the entry parameter is an
# one lesson is not bookable it doesn't affect the others. Maybe the API # array, but this gives us finer error handling and makes it so that if
# already does this, but I'm too lazy to find out # one lesson is not bookable it doesn't affect the others. Maybe the API
logger.debug(f"Sending request to {BOOK_LESSON_URL} for entry {entry}") # already does this, but I'm too lazy to find out
if check_response( logger.debug(f"Sending request to {BOOK_LESSON_URL} for entry {entry}")
logger, if check_response(
result := await send_request( logger,
client, result := await send_request(
"post", client,
BOOK_LESSON_URL, "post",
data=json.dumps({"CodiceFiscale": arguments.tax_code, "entry": [entry]}), BOOK_LESSON_URL,
), data=json.dumps({"CodiceFiscale": arguments.tax_code, "entry": [entry]}),
arguments.verbose, ),
): arguments.verbose,
entries.remove(entry) ):
logger.debug(f"Request to {result.url} sent, status code is {result.status_code}, payload is {result.content}") entries.remove(entry)
logger.info( logger.debug(f"Request to {result.url} sent, status code is {result.status_code}, payload is {result.content}")
f"Booked {len(entries)} lesson{'' if len(entries) == 1 else 's'}, sleeping for {arguments.delay} seconds" logger.info(
) f"Booked {len(entries)} lesson{'' if len(entries) == 1 else 's'}, sleeping for {arguments.delay} seconds"
await asyncio.sleep(arguments.delay) )
else: await asyncio.sleep(arguments.delay)
logger.error(f"The provided tax code does not appear to be valid, please check for any typos and try again") else:
return -1 logger.error(f"The provided tax code does not appear to be valid, please check for any typos and try again")
return -1
except json.decoder.JSONDecodeError as json_error:
logger.error(f"A fatal JSON decoding error occurred -> {type(json_error).__name__}: {json_error}")
if __name__ == "__main__": if __name__ == "__main__":
@ -331,8 +340,8 @@ if __name__ == "__main__":
"-v", "--verbose", help="Increase log message verbosity. For advanced users only!", action="store_true" "-v", "--verbose", help="Increase log message verbosity. For advanced users only!", action="store_true"
) )
parser.add_argument("-l", "--log-file", help="Tells the script to also write logs on the specified file (relative" parser.add_argument("-l", "--log-file", help="Tells the script to also write logs on the specified file (relative"
"or absolute paths are both accepted). Defaults to no file (i.e. no" " or absolute paths are both accepted). Defaults to no file (i.e. no"
"file logging)", default=None) " file logging)", default=None)
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
try: try:
main_task = asyncio.ensure_future(main(parser.parse_args())) main_task = asyncio.ensure_future(main(parser.parse_args()))