poe_passives/passives.py

#!/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()