import traceback import argparse from argparse import Namespace import re from glob import glob from datetime import datetime, timedelta from typing import Any from functools import wraps from time import time as ti from ffxiv_aku import * from ffxiv_aku import get_any_Logdata, storeFilesInTmp, loadDataTheQuickestWay, getPlayerActionIDs, getPlayerStatusIDs, get_list_of_all_pet_names, splitLogLineByVersion, pretty_json from ffxiv_aku import print_color_green, print_color_yellow, print_color_red, print_pretty_json, convertLogdataFromClassToJson, getLogdataAsClassNew from dateutil import parser as dp def timing(f: Any) -> Any: @wraps(f) def wrap(*largs: Any, **kw: Any) -> Any: ts: float = ti() xresult: Any = f(*largs, **kw) te: float = ti() print(f'func:{f.__name__} took: {te-ts:.4f} sec') return xresult return wrap class CONFIG: """ This is the generall CONFIG used in this programm and will be used to store most data """ def __init__(self) -> None: self.exclude_stats_by_version = 22106 self.version: int = 0 self.is_bluemage: bool = False self.is_baldesion_active: bool = False self.is_castrum_active: bool = False self.is_dalriada_active: bool = False self.saved_location_when_entering_special_zone: str = "" self.found_correct_language: bool = False self.switchedLanguageCode: bool = False self.placename_locked: str = "" self.location: str = "" self.location_id: str = "" self.location_name: str = "" self.player: str = "" self.current_map: str = "" self.old_map: str = "" self.current_time_of_mapchange: datetime self.old_time_of_mapchange: datetime self.fates_to_compare: list[tuple[datetime, str, str]] = [] self.user_to_classes: dict[str, str] = {} ttype: dict[str, dict[str, str]] = loadDataTheQuickestWay("territorytype_all.json", translate=True) cfc: dict[str, dict[str, str]] = loadDataTheQuickestWay("contentfindercondition_all.json", translate=True) cj: dict[str, dict[str, str]] = loadDataTheQuickestWay("classjob_all.json", translate=True) action: dict[str, dict[str, str]] = loadDataTheQuickestWay("action_all.json", translate=True) t_bnpcname: dict[str, dict[str, str]] = loadDataTheQuickestWay("bnpcname_all.json", translate=True) bnpcname: dict[str, dict[str, str]] = loadDataTheQuickestWay("BNpcName") bnpc = loadDataTheQuickestWay("bnpcname_all.json", translate=True) status: dict[str, dict[str, str]] = loadDataTheQuickestWay("status_all.json", translate=True) placename: dict[str, dict[str, str]] = loadDataTheQuickestWay("placename_all.json", translate=True) result: dict[str, Any] = {} config: CONFIG = CONFIG() # language to safe results lang = "de" # language of the logfile itself e.g. from Azubi set it to en compare_lang = "de" example_lines: dict[Any, Any] = {} LANGUAGES:dict[str, str] = { "English": "en", "German": "de", "French": "fr", "Japanese": "ja", "Chinese": "cn", "Korean": "ko", "3": "de", "1": "en" } # main > run > work_on_each_line[00] > this def getPlacenameLocked(line: str, code: str) -> None: global config # look for close message in both german and english if code == "0839": m: re.Match[str] | None = re.search("Noch 15 Sekunden, bis sich (?:(?:der|die|das) )?(?:Zugang zu(?:[rm]| den)? )?(.*) schließt", line) n: re.Match[str] | None = re.search("(.*) will be sealed off(?: in (?:[0-9]+ seconds)?)?", line) if m: config.placename_locked = m.group(1) if n: config.placename_locked = getPlacenameName(n.group(1)) # main > run > work_on_each_line[00] > getPlacenameLocked > this def getPlacenameName(place_name: str) -> str: if compare_lang != lang: for _, npc in placename.items(): if npc[f"Name_{compare_lang}"].lower() == place_name.lower() or npc[f"Name_{compare_lang}"].lower().replace("the ", "") == place_name.lower(): return npc[f"Name_{lang}"].title() return place_name.title() return place_name.title() # main > run > work_on_each_line[00] > this def get_blu_content_name(line: str, code: str) -> None: global config global result lx: re.Match[str] | None = re.search("„(.+)“ hat begonnen", line) if lx and code in ['0839', '0039']: config.location_name = lx.group(1) if result.get(config.location_name, None) is None: result[config.location_name] = {} config.is_bluemage = False print(f"\nRenamed BLU instancce from '{config.location_name}' to '{config.location_name}'!") # main > run > work_on_each_line[00] > this def handleBaldesionArsenalStuff(line: str) -> None: global config # get location for Castrum Lacus Litore if "Eureka Hydatos" == config.location_name and "Du erleidest den Effekt von  Wiederbelebungsbegrenzung." in line: print("Found Eureka Baldesions Arsenal") config.is_baldesion_active = True config.saved_location_when_entering_special_zone = config.location_name config.location_name = "Eureka Baldesions Arsenal" elif config.is_baldesion_active and ("Du erholst dich von dem Effekt von  Wiederbelebungsbegrenzung" in line or "hat Proto-Yadis besiegt" in line or "hast Proto-Yadis besiegt" in line): print("Exit Eureka Baldesions Arsenal") config.is_baldesion_active = False config.location_name = config.saved_location_when_entering_special_zone # main > run > work_on_each_line[00] > this def handleBozjaStuff(line: str) -> None: global config # get location for Castrum Lacus Litore if "Bozja-Südfront" == config.location_name and "Du nimmst an „Sturm auf Castrum Lacus Litore“ teil." in line: print("Found Castrum Lacus Litore") config.is_castrum_active = True config.saved_location_when_entering_special_zone = config.location_name config.location_name = "Castrum Lacus Litore" elif config.is_castrum_active and ("hat Dawon besiegt" in line or "hast Dawon besiegt" in line): print("Exit Castrum Lacus Litore") config.is_castrum_active = False config.location_name = config.saved_location_when_entering_special_zone # main > run > work_on_each_line[00] > this def handleZladnorStuff(line: str) -> None: global config # get location for Castrum Lacus Litore if "Zadnor-Hochebene" == config.location_name and "Du nimmst an „Sturm auf die Dalriada“ teil." in line: print("Found Dalriada") config.is_dalriada_active = True config.saved_location_when_entering_special_zone = config.location_name config.location_name = "Die Dalriada" elif config.is_dalriada_active and ("hat das Diablo-Armament besiegt" in line or "hast das Diablo-Armament besiegt" in line): print("Exit Dalriada") config.is_dalriada_active = False config.location_name = config.saved_location_when_entering_special_zone # main > run > work_on_each_line[01] > this def check_for_location(jline: dict[str, str]) -> None: global result global config config.location = jline['zonename'] config.location_id = jline['zoneid'] config.is_bluemage = True if config.location_id == "31C" else False config.placename_locked = "" config.location_name = getLocationName() if result.get(config.location_name, None) is None: result[config.location_name] = { 'combatants': {}, "contentzoneid": [], "music": [], "zone": [] } print_color_yellow(f"\nFound new Location {config.location_name}' - '{config.location_id}' ({config.location})!", end="") # main > run > work_on_each_line[01] > check_for_location > this def getLocationName() -> str: global config _id: str = config.location_id location: str = config.location try: entry: dict[str, str] = ttype[str(int(_id, 16))] value: str = getInstanceFromTerritoryType(entry).replace("\u00ad", "").replace("(Episch)", "(episch)").replace("(Schwer)", "(schwer)").replace("(Extrem)", "(extrem)").replace("(Fatal)", "(fatal)") except KeyError: traceback.print_exc() value = location.replace("(Episch)", "(episch)").replace("(Schwer)", "(schwer)").replace("(Extrem)", "(extrem)").replace("(Fatal)", "(fatal)") return value # main > run > work_on_each_line[01] > check_for_location > getLocationName > this def getInstanceFromTerritoryType(c_id: dict[str, str]) -> str: for _id, content in cfc.items(): if content["TerritoryType"] == c_id["TerritoryType"]: return content[f"Name_{lang}"].replace("<", "").replace(">", "").replace("\u00ad", "") #print_color_red("Could not find Instance, use prewrote unreals in [getInstanceFromTerritoryType]") if c_id["TerritoryType"] == "r1fc_3": return "Traumprüfung Shiva" if c_id["TerritoryType"] == "s1fa_4": return "Traumprüfung Titan" if c_id["TerritoryType"] == "a2fd_3": return "Traumprüfung Sephiroth" if c_id["TerritoryType"] == "a2fe_3": return "Traumprüfung Sophia" #print_color_red("Error at getInstace: " + str(c_id)) return c_id[f"Name_{lang}"].replace("\u00ad", "") # main > run > work_on_each_line[249] > this def check_language_stuff(line: str) -> None: global config tmp_lang: str = getLanguageToImport(line) if tmp_lang == "": return None if f"Language ID: {tmp_lang}" in line: config.found_correct_language = True #print_color_green(f"Found Language Code {tmp_lang}") elif "Language ID: " in line: config.found_correct_language = False print_color_red("Incorrect Languagecode found, change settings") # main > run > work_on_each_line[249] > check_language_stuff > this def getLanguageToImport(line: str) -> str: m: Any = re.search(r'(?:Language ID: )([A-Za-z0-9]+)', line) if not m: return "" language_code_int: Any = m.group(1) if LANGUAGES.get(language_code_int, None): return language_code_int return "" # main > run > work_on_each_line[253] > this def getVersionFromLogfile(line: str) -> int: version: str = "" version_list: list[str] = [] version_nr: int = 0 if "|FFXIV PLUGIN VERSION:" in line: version = line.split("|FFXIV PLUGIN VERSION: ")[1] version = version.split("|")[0] version = version.split(", CLIENT MODE")[0] else: version = line.split("|FFXIV_ACT_Plugin Version: ")[1] version = version.split(" (")[0] version_list = version.split(".") version_list.insert(2, "0") version = "".join(version_list) version_nr = int(version) print_color_green(f"Found version: {version_nr}") return version_nr # main > run > work_on_each_line > this def split_line_to_json(line: str) -> Any: global config file_version: int = config.version lline_dict, llogname = splitLogLineByVersion(file_version, line) # type: ignore handle_example_lines(lline_dict, llogname, line) # type: ignore #try: # del lline_dict['timestamp'] # type: ignore #except Exception: # lline_dict.__deleteitems__('timestamp') # type: ignore return lline_dict # main > run > work_on_each_line > split_line_to_json > this def handle_example_lines(lline_dict: dict[str, str], llogname: str, line: str) -> None: global example_lines # seperate 00 into an own array if lline_dict["lineid"] == "00": if not example_lines.get("0" + lline_dict["lineid"] + " - " + llogname, None): example_lines["0" + lline_dict["lineid"] + " - " + llogname] = {} if not example_lines["0" + lline_dict["lineid"] + " - " + llogname].get(lline_dict["code"], None): example_lines["0" + lline_dict["lineid"] + " - " + llogname][lline_dict["code"]] = lline_dict elif len(lline_dict["lineid"]) > 1: key_string = lline_dict["lineid"] + " - " + llogname if len(lline_dict["lineid"]) < 3: key_string = "0" + lline_dict["lineid"] + " - " + llogname if not example_lines.get(key_string, None): example_lines[key_string] = lline_dict else: print_color_red(line) def work_on_01_lines(jline: dict[str, str]) -> None: global config config.current_map = config.location_name config.current_time_of_mapchange = dp.parse(jline['timestamp']) #print(config.fates_to_compare) for f in config.fates_to_compare: _time, _type, fatename = f diff: timedelta = config.current_time_of_mapchange - _time diff2: timedelta = _time - config.old_time_of_mapchange if diff.total_seconds() > 5 and diff2.total_seconds() < 5: if not result[config.old_map].get("fates", None): result[config.old_map]['fates'] = [] if fatename not in result[config.old_map]['fates']: result[config.old_map]['fates'].append(fatename) # reset values config.fates_to_compare = [] config.user_to_classes = {} # set current values to old values config.old_map = config.current_map config.old_time_of_mapchange = config.current_time_of_mapchange # main > run > this def work_on_each_line(_file: str) -> None: global result global config config = CONFIG() for line in open(_file, "r", encoding="utf8"): jline: dict[str, str] = split_line_to_json(line) match jline["lineid"]: case "00": if config.is_bluemage: get_blu_content_name(jline['line'], jline['code']) # get specials for eureka und bozja handleBaldesionArsenalStuff(jline['line']) handleBozjaStuff(jline['line']) handleZladnorStuff(jline['line']) # get placename from closing message getPlacenameLocked(jline['line'], jline['code']) # MapChange case "01": # this was originally done way before the rest of line content check_for_location(jline) work_on_01_lines(jline) # ChangePrimaryPlayer case "02": config.player = jline['playername'] # AddCombatant - check case for line entry 03 (line.split("|")[9] will be the key from bnpcid) case "03": ... case "04": ... case "11": ... # PlayerStats - e.g. jobswitch case "12": config.user_to_classes[config.player] = cj[jline['jobid']][f"Name_{compare_lang}"] case "20": ... case "21": ... case "22": ... case "23": ... case "24": ... case "25": ... case "26": ... case "27": ... case "28": ... case "29": ... case "30": ... case "31": ... case "32": ... case "33": ... case "34": ... case "35": ... case "36": ... case "37": ... case "38": ... case "39": ... case "40": ... case "41": ... case "42": ... case "249": # if language code was not correct check_language_stuff(jline["line"]) case "250": ... case "251": ... case "252": ... case "253": # Line when ACT was restarted config.version = getVersionFromLogfile(line) if jline["line"].startswith("FFXIV PLUGIN VERSION: "): config = CONFIG() config.user_to_classes = {} case "254": ... case "256": ... case "257": ... case "258": fate: str = str(int(jline['fateId'], 16)) current_time: datetime = dp.parse(jline['timestamp']) config.fates_to_compare.append((current_time, jline["category"], fate)) case "259": ... case "260": ... case "261": ... case "262": ... case "263": ... case "264": ... case "265": ... case "266": ... case "267": ... case "268": ... case "269": ... case "270": ... case "271": ... case "272": ... case "273": ... case "274": ... case _: print_color_red(jline) config.fates_to_compare = [] for k, v in vars(config).items(): print(f"{k}: {v}") # main > run > this def addAllMissingZones() -> None: global result compare_data: list[str] = [x.lower() for x in result] for _, content in cfc.items(): if content[f'Name_{lang}']: if content[f'Name_{lang}'].lower() not in compare_data: name: str = content[f'Name_{lang}'].strip().replace("(Episch)", "(episch)").replace("(Fatal)", "(fatal)") data[name] = {} print(f"Added Zone name for: {name}") # main > this def run(largs: Namespace) -> None: global compare_lang global result result = get_any_Logdata() addAllMissingZones() files: list[str] = glob(f"{largs.main_path}/{largs.sub_path}/{largs.file_regex}") number_of_files: int = len(files) counter_of_files = 0 for _file in files: counter_of_files += 1 if largs.skip_files >= counter_of_files: print_color_red(f"Skip file {counter_of_files}/{number_of_files}: {_file}") continue print_color_green(f"Work on file {counter_of_files}/{number_of_files}: {_file}") work_on_each_line(_file) # main > this def checkArgumentsAndsetDefaults(largs: Namespace) -> Namespace: largs.main_path = largs.main_path if largs.main_path else "." largs.sub_path = largs.sub_path if largs.sub_path else "Neuer Ordner" largs.file_regex = largs.file_regex if largs.file_regex else "Network_*.log" largs.skip_files = int(largs.skip_files) if largs.skip_files else 0 if largs.file_language in ["de", "en", "fr", "ja", "cn", "ko"]: global compare_lang compare_lang = largs.file_language return largs if __name__ == "__main__": example_usage = "" parser = argparse.ArgumentParser( description="Recreates the logdata json file", epilog=example_usage, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("-mp", "--main-path", help="main path to file") parser.add_argument("-sp", "--sub-path", help="sub path to file") parser.add_argument("-fr", "--file-regex", help="file regex") parser.add_argument("-fl", "--file-language", help="file language") parser.add_argument("-ib", "--include-buffs", action='store_true', help="file language") parser.add_argument("-sfw", "--skip-file-write", action='store_true', help="If set to true, logdata will not be saved") parser.add_argument("-sf", "--skip-files", help="Skip already processed files") parser.add_argument("-pll", "--print-log-lines", action='store_true', help="Print log line examples") args: Namespace = parser.parse_args() args= checkArgumentsAndsetDefaults(args) run(args)