#!/usr/bin/python3 import logging from argparse import ArgumentParser, ArgumentError from datetime import datetime from os import fstat from re import compile TIME_FORMAT = "%Y/%m/%d %H:%M:%S" def extract_delta(logline: str) -> int: """ Extracts a delta from a Client.txt log line. The 3rd column is a millisecond delta from some specific but unknowable time in the past. This is useful to group together /passives messages which, historically, have all had the same delta. """ return int(logline.split(" ")[2]) def extract_time(line: str): """ Extracts a time from a Client.txt log line. """ # ls = line.split(' ')[0:2] # strptime(f"{ls[0]} {ls[1]} {ls[2][-3:]}000", "%Y/%m/%d %H:%M:%S %f") return datetime.strptime(" ".join(line.split(" ")[0:2]), TIME_FORMAT) class PassivesList: """ An object encapsulating all of the data output by one /passives command """ MATCHLINE = compile(r": (\d+) Passive Skill Points from quests:$") QUESTLINE = compile(r": \((\d) from ([^\)]+)") def __init__(self, logline): m = self.MATCHLINE.search(logline) if not m: raise ValueError("Invalid init logline") self.start_time = extract_time(logline) self.delta = extract_delta(logline) self.uncompleted_quests = { "The Marooned Mariner": "A1: Ship Graveyard (Allflame) - Fairgraves)", "The Dweller of the Deep": "A1: Flood depths 🦀 - Tarkleigh", "The Way Forward": "A2: Western Forest (Thaumetic Emblem, Blackguards) - Bestel, A1", "Victario's Secrets": "A3: Sewer (busts) - Hargan", "Piety's Pets": "A3 - Grigor", "An Indomitable Spirit": " A4: Mines level 2 (Deshret spirit) - Tasuni", "In Service to Science": "A5: Control Blocks (Miasmeter) - Vilenta", "Kitava's Torments": "A5: Reliquary (3 torments) - Lani", "The Father of War": "A6: Karui Fortress (Tukohama) - Tarkleigh", "The Cloven One": "A6: Prisoner's Gate (Abberath) - Bestel", "The Puppet Mistress": "A6: The Wetlands via Riverways (Ryslatha) - Tarkleigh", "The Master of a Million Faces": "A7 - Eramir", "Queen of Despair": "A7: Dread Thicket via Northern Forest (Gruthkul) - Eramir", "Kishara's Star": "A7: Causeway (Kishara's Star) - Weylam", "Love is Dead": "A8: Quay (Ankh) - Clarissa", "Reflection of Terror": "A8: High Gardens via Bath House (YUGUUUUUUUUUL) - Sin", "The Gemling Legion": "A8: Grain Gate (Legionnaires) - Maramoa", "Queen of the Sands": "A9: Oasis (Shakari) - Irasha", "The Ruler of Highgate": "A9: Quarry (Garukhan) - Irasha or Tasuni", "Vilenta's Vengeance": "A10: Control Blocks via Ravaged Square (Vilenta) - Lani", "An End to Hunger": "Kitava: Oriath - Lani", } self.completed_quests = {} self.completed_count = int(m.group(1)) self.parsed_count = 0 self._parsed_line_count = 0 @property def name(self) -> str: return self.start_time.strftime(TIME_FORMAT) @property def full(self) -> bool: return self.parsed_count == self.completed_count def try_append(self, logline) -> bool: m = self.QUESTLINE.search(logline) if m: passive_count = int(m.group(1)) quest_name = m.group(2) logging.debug(f"Found quest line: {passive_count}, {quest_name}") if -250 < extract_delta(logline) - self.delta < 250: logging.debug(f"Delta matches >{self.delta}<") if "Bandit" in quest_name: self.parsed_count += passive_count logging.debug( f"{self.name} | {self.parsed_count}/{self.completed_count} | {self.full}" ) return True else: try: # Move description over because why not self.completed_quests[quest_name] = self.uncompleted_quests[ quest_name ] del self.uncompleted_quests[quest_name] self.parsed_count += passive_count logging.debug( f"{self.name} | {self.parsed_count}/{self.completed_count} | {self.full}" ) return True except KeyError as e: raise KeyError( f"Quest '{quest_name}' was not defined or already parsed." ) from e else: # TODO: Make this a parsing exception? # Delta mismatch pass return False class PassivesListContainer: FASTMATCH = "quests" def __init__(self): self.container = [] self.parsed_count = 0 @property def _is_open(self): return self.container and not self._last.full @property def _last(self): if self.container: return self.container[-1] def parse(self, logline): if self._is_open: # logging.debug(f"OPEN: {logline}") self._last.try_append(logline) if self.FASTMATCH in logline: # logging.debug(f"Passed FASTMATCH: {logline}") try: n = PassivesList(logline) self.container.append(n) except ValueError as e: # Line impersonated /passives starting flow pass if self.parsed_count <= 10: logging.info(f"Starting log: {logline}") self.parsed_count += 1 def print(self, index=-1): if not self.container: raise Exception("No data loaded") c = self.container[index] if not c: raise KeyError("Invalid passives index (check skip value?)") c = self.container[index] if self.validate(index): if not c.uncompleted_quests or not len(c.uncompleted_quests): print("PASSIVES GET!") else: uq = c.uncompleted_quests if "End to Hunger" in uq.keys(): missing = 1 + len(uq.keys()) else: missing = len(uq.keys()) print(f"Missing passives ({missing}):") for key, value in c.uncompleted_quests.items(): print("{: <34}{}".format(key, value)) else: logging.info(f"UnFound: {self.container[index].uncompleted_quests.items()}") logging.info(f"Found: {self.container[index].completed_quests.items()}") raise ValueError( f"Parser didn't find all of the quests for /passive call {self.container[index].name}!" ) def printall(self): for i in range(0, len(self.container)): self.print(i) print("-----") def validate(self, index): return self.container[index].full def main(): parser = ArgumentParser( description="reveal which passives are missing", ) parser.add_argument( "--file", "-f", default="/mnt/c/Program Files (x86)/Grinding Gear Games/Path of Exile/logs/Client.txt", dest="client_file", help="path to Client.txt", ) parser.add_argument( "--bytes", type=int, default=2000000, help="number of bytes from the end of Client.txt to check", ) parser.add_argument( "--skip", type=int, default=None, help="for debugging only: view data for a /passives call other than the most recent", ) parser.add_argument( "--debug", "-d", action="count", default=0, help="can be specified multiple times to increase log level", ) args = parser.parse_args() if args.bytes < 1: raise ArgumentError("--line count must be greater than 1") if args.debug: logging.getLogger().setLevel(logging.DEBUG) # Create datasets passives = PassivesListContainer() with open(args.client_file, "r") as f: # Client.txt stores the whole world, only parse the last ~1MB size = fstat(f.fileno()).st_size if size > args.bytes: f.seek(size - args.bytes) f.readline() # discard the partial line for line in f: passives.parse(line) # Print dataset if not args.skip: passives.print() elif args.skip == -1: passives.printall() else: passives.print(-1 - args.skip) if __name__ == "__main__": main()