mirror of
https://github.com/RedDeadDepresso/KKAFIO.git
synced 2025-12-22 09:20:02 +00:00
refactor: logger
This commit is contained in:
@@ -25,13 +25,10 @@ class Logger:
|
||||
handler1.setFormatter(formatter)
|
||||
self.logger.setLevel(logging.INFO)
|
||||
self.logger.addHandler(handler1)
|
||||
self.text = ['INFO', 'SUCCESS', 'ERROR', 'SKIPPED', 'REPLACED', 'RENAMED', 'REMOVED']
|
||||
# Status Text: INFO, SUCCESS, ERROR, SKIPPED, REPLACED, RENAMED, REMOVED
|
||||
self.status = [' INFO', ' SUCCESS', ' ERROR',
|
||||
' SKIPPED', ' REPLACED', ' RENAMED', ' REMOVED']
|
||||
self.status = ['INFO', 'SUCCESS', 'ERROR', 'WARNING', 'SKIPPED', 'REPLACED', 'RENAMED', 'REMOVED']
|
||||
# Status Color: Blue, Red, Green, Orange
|
||||
self.statusColor = ['#2d8cf0', '#00c12b', '#ed3f14', '#f90', '#f90', '#f90', '#f90']
|
||||
|
||||
self.statusColor = ['#2d8cf0', '#00c12b', '#ed3f14', '#f90', '#f90', '#f90', '#f90', '#f90']
|
||||
# Status HTML: <b style="color:$color">status</b>
|
||||
self.statusHtml = [
|
||||
f'<b style="color:{_color};">{status}</b>'
|
||||
@@ -58,7 +55,7 @@ class Logger:
|
||||
if self.logger_signal is not None:
|
||||
message = message.replace('\n', '<br>').replace(' ', ' ')
|
||||
adding = (f'''
|
||||
<div style="font-family: Consolas, monospace;color:{self.statusColor[level - 1]};">
|
||||
<div style="font-family: Inter;color:{self.statusColor[level - 1]};">
|
||||
{self.statusHtml[level - 1]} | {category} | {message}
|
||||
</div>
|
||||
''')
|
||||
@@ -69,10 +66,10 @@ class Logger:
|
||||
|
||||
def colorize(self, line):
|
||||
adding = line
|
||||
for i, s in enumerate(self.text):
|
||||
for i, s in enumerate(self.status):
|
||||
if s in line:
|
||||
adding = (f'''
|
||||
<div style="font-family: Consolas, monospace;color:{self.statusColor[i]};">
|
||||
<div style="font-family: Inter;color:{self.statusColor[i]};">
|
||||
{line}
|
||||
</div>
|
||||
|
||||
@@ -113,7 +110,7 @@ class Logger:
|
||||
"""
|
||||
self.__out__(category, message, 4)
|
||||
|
||||
def replaced(self, category: str, message: str) -> None:
|
||||
def warning(self, category: str, message: str) -> None:
|
||||
"""
|
||||
:param message: log message
|
||||
|
||||
@@ -121,7 +118,7 @@ class Logger:
|
||||
"""
|
||||
self.__out__(category, message, 5)
|
||||
|
||||
def renamed(self, category: str, message: str) -> None:
|
||||
def replaced(self, category: str, message: str) -> None:
|
||||
"""
|
||||
:param message: log message
|
||||
|
||||
@@ -129,7 +126,7 @@ class Logger:
|
||||
"""
|
||||
self.__out__(category, message, 6)
|
||||
|
||||
def removed(self, category: str, message: str) -> None:
|
||||
def renamed(self, category: str, message: str) -> None:
|
||||
"""
|
||||
:param message: log message
|
||||
|
||||
@@ -137,16 +134,27 @@ class Logger:
|
||||
"""
|
||||
self.__out__(category, message, 7)
|
||||
|
||||
def removed(self, category: str, message: str) -> None:
|
||||
"""
|
||||
:param message: log message
|
||||
|
||||
Output warn log
|
||||
"""
|
||||
self.__out__(category, message, 8)
|
||||
|
||||
def line(self) -> None:
|
||||
"""
|
||||
Output line
|
||||
"""
|
||||
# While the line print do not need wrapping, we
|
||||
# use raw_print=True to output log to logger box
|
||||
self.__out__(
|
||||
'<div style="font-family: Consolas, monospace;color:#2d8cf0;">--------------'
|
||||
'-------------------------------------------------------------'
|
||||
'-------------------</div>', raw_print=True)
|
||||
if self.logger_signal is not None:
|
||||
self.__out__("",
|
||||
'<div style="font-family: Consolas, monospace;color:#2d8cf0;">'
|
||||
'--------------------------------------------------------------------'
|
||||
'</div>', raw_print=True)
|
||||
else:
|
||||
print('--------------------------------------------------------------------')
|
||||
|
||||
|
||||
logger = Logger()
|
||||
@@ -10,6 +10,7 @@ class ThreadManager:
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._script = None
|
||||
self._line = '--------------------------------------------------------------------'
|
||||
|
||||
def start(self):
|
||||
if self._script is not None:
|
||||
@@ -28,6 +29,8 @@ class ThreadManager:
|
||||
line = self._script.stdout.readline().decode('utf-8')
|
||||
if not line:
|
||||
signalBus.stopSignal.emit()
|
||||
elif line == self._line:
|
||||
logger.line()
|
||||
else:
|
||||
logger.colorize(line)
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@ import os
|
||||
import re as regex
|
||||
import codecs
|
||||
import shutil
|
||||
from util.logger import Logger
|
||||
from app.common.logger import logger
|
||||
|
||||
class FilterConvertKKS:
|
||||
def __init__(self, config, file_manager):
|
||||
@@ -33,7 +33,7 @@ class FilterConvertKKS:
|
||||
card_type = 2
|
||||
elif data.find(b"KoiKatuCharaSun") != -1:
|
||||
card_type = 3
|
||||
Logger.log_info(f"[{card_type}]", f"{card_path}")
|
||||
logger.info(f"[{card_type}]", f"{card_path}")
|
||||
return card_type
|
||||
|
||||
def convert_kk(self, card_name, card_path, destination_path):
|
||||
@@ -65,12 +65,12 @@ class FilterConvertKKS:
|
||||
|
||||
count = len(png_list)
|
||||
if count > 0:
|
||||
Logger.log_info("SCRIPT", "0: unknown / 1: kk / 2: kksp / 3: kks")
|
||||
logger.info("SCRIPT", "0: unknown / 1: kk / 2: kksp / 3: kks")
|
||||
for png in png_list:
|
||||
if self.check_png(png) == 3:
|
||||
kks_card_list.append(png)
|
||||
else:
|
||||
Logger.log_success("SCRIPT", f"no PNG found")
|
||||
logger.success("SCRIPT", f"no PNG found")
|
||||
return
|
||||
|
||||
count = len(kks_card_list)
|
||||
@@ -83,7 +83,7 @@ class FilterConvertKKS:
|
||||
os.mkdir(target_folder)
|
||||
|
||||
if self.convert:
|
||||
Logger.log_info("SCRIPT", f"Conversion to KK is [{self.convert}]")
|
||||
logger.info("SCRIPT", f"Conversion to KK is [{self.convert}]")
|
||||
if not os.path.isdir(target_folder2):
|
||||
os.mkdir(target_folder2)
|
||||
|
||||
@@ -100,8 +100,8 @@ class FilterConvertKKS:
|
||||
shutil.move(source, target)
|
||||
|
||||
if self.convert:
|
||||
Logger.log_success("SCRIPT", f"[{count}] cards moved to [{kks_folder}] folder, converted and save to [{kks_folder2}] folder")
|
||||
logger.success("SCRIPT", f"[{count}] cards moved to [{kks_folder}] folder, converted and save to [{kks_folder2}] folder")
|
||||
else:
|
||||
Logger.log_success("SCRIPT", f"[{count}] cards moved to [{kks_folder}] folder")
|
||||
logger.success("SCRIPT", f"[{count}] cards moved to [{kks_folder}] folder")
|
||||
else:
|
||||
Logger.log_success("SCRIPT", f"no KKS card found")
|
||||
logger.success("SCRIPT", f"no KKS card found")
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import os
|
||||
import codecs
|
||||
from util.logger import Logger
|
||||
from app.common.logger import logger
|
||||
|
||||
class InstallChara:
|
||||
def __init__(self, config, file_manager):
|
||||
@@ -20,7 +20,7 @@ class InstallChara:
|
||||
if data.find(b"KoiKatuChara") != -1:
|
||||
if data.find(b"KoiKatuCharaSP") != -1 or data.find(b"KoiKatuCharaSun") != -1:
|
||||
basename = os.path.basename(image_path[0])
|
||||
Logger.log_error("CHARA", f"{basename} is a KKS card")
|
||||
logger.error("CHARA", f"{basename} is a KKS card")
|
||||
return
|
||||
self.file_manager.copy_and_paste("CHARA", image_path, self.game_path["chara"])
|
||||
elif data.find(b"KoiKatuClothes") != -1:
|
||||
@@ -32,7 +32,8 @@ class InstallChara:
|
||||
if folder_path is None:
|
||||
folder_path = self.input_path
|
||||
foldername = os.path.basename(folder_path)
|
||||
Logger.log_msg("FOLDER", foldername)
|
||||
logger.line()
|
||||
logger.info("FOLDER", foldername)
|
||||
file_list, compressed_file_list = self.file_manager.find_all_files(folder_path)
|
||||
|
||||
for file in file_list:
|
||||
@@ -44,8 +45,8 @@ class InstallChara:
|
||||
self.resolve_png(file)
|
||||
case _:
|
||||
basename = os.path.basename(file[0])
|
||||
Logger.log_error("UKNOWN", f"Cannot classify {basename}")
|
||||
print("[MSG]")
|
||||
logger.error("UKNOWN", f"Cannot classify {basename}")
|
||||
logger.line()
|
||||
|
||||
for compressed in compressed_file_list:
|
||||
extract_path = self.file_manager.extract_archive(compressed[0])
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import os
|
||||
import codecs
|
||||
from util.logger import Logger
|
||||
from app.common.logger import logger
|
||||
|
||||
|
||||
class RemoveChara:
|
||||
def __init__(self, config, file_manager):
|
||||
@@ -28,7 +29,7 @@ class RemoveChara:
|
||||
|
||||
def logic_wrapper(self):
|
||||
foldername = os.path.basename(self.input_path)
|
||||
Logger.log_msg("FOLDER", foldername)
|
||||
logger.info("FOLDER", foldername)
|
||||
file_list, archive_list = self.file_manager.find_all_files(self.input_path)
|
||||
|
||||
for file in file_list:
|
||||
@@ -40,7 +41,7 @@ class RemoveChara:
|
||||
self.resolve_png(file)
|
||||
case _:
|
||||
pass
|
||||
print("[MSG]")
|
||||
logger.line()
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ try:
|
||||
pass
|
||||
|
||||
from util.config import Config
|
||||
from util.logger import Logger
|
||||
from app.common.logger import logger
|
||||
from util.file_manager import FileManager
|
||||
from modules.install_chara import InstallChara
|
||||
from modules.remove_chara import RemoveChara
|
||||
@@ -21,6 +21,7 @@ try:
|
||||
Args:
|
||||
config (Config): BAAuto Config instance
|
||||
"""
|
||||
logger.logger_signal = None
|
||||
self.config = config
|
||||
self.file_manager = file_manager
|
||||
self.modules = {
|
||||
@@ -41,11 +42,11 @@ try:
|
||||
def run(self):
|
||||
for task in self.config.tasks:
|
||||
if self.modules[task]:
|
||||
Logger.log_info("SCRIPT", f'Start Task: {task}')
|
||||
logger.info("SCRIPT", f'Start Task: {task}')
|
||||
try:
|
||||
self.modules[task].logic_wrapper()
|
||||
except:
|
||||
Logger.log_error("SCRIPT", f'Task error: {task}. For more info, check the traceback.log file.')
|
||||
logger.error("SCRIPT", f'Task error: {task}. For more info, check the traceback.log file.')
|
||||
with open('traceback.log', 'a') as f:
|
||||
f.write(f'[{task}]\n')
|
||||
traceback.print_exc(None, f, True)
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
import sys
|
||||
import json
|
||||
import os
|
||||
from util.logger import Logger
|
||||
from app.common.logger import logger
|
||||
|
||||
class Config:
|
||||
def __init__(self, config_file):
|
||||
Logger.log_info("SCRIPT", "Initializing config module")
|
||||
logger.info("SCRIPT", "Initializing config module")
|
||||
self.config_file = config_file
|
||||
self.ok = False
|
||||
self.initialized = False
|
||||
@@ -20,31 +20,31 @@ class Config:
|
||||
with open(self.config_file, 'r') as json_file:
|
||||
self.config_data = json.load(json_file)
|
||||
except FileNotFoundError:
|
||||
Logger.log_error("SCRIPT", f"Config file '{self.config_file}' not found.")
|
||||
logger.error("SCRIPT", f"Config file '{self.config_file}' not found.")
|
||||
sys.exit(1)
|
||||
except json.JSONDecodeError:
|
||||
Logger.log_error("SCRIPT", f"Invalid JSON format in '{self.config_file}'.")
|
||||
logger.error("SCRIPT", f"Invalid JSON format in '{self.config_file}'.")
|
||||
sys.exit(1)
|
||||
|
||||
self.validate()
|
||||
|
||||
if self.ok and not self.initialized:
|
||||
Logger.log_info("SCRIPT", "Starting KKAFIO!")
|
||||
logger.info("SCRIPT", "Starting KKAFIO!")
|
||||
self.initialized = True
|
||||
self.changed = True
|
||||
elif not self.ok and not self.initialized:
|
||||
Logger.log_error("SCRIPT", "Invalid config. Please check your config file.")
|
||||
logger.error("SCRIPT", "Invalid config. Please check your config file.")
|
||||
sys.exit(1)
|
||||
elif not self.ok and self.initialized:
|
||||
Logger.log_warning("SCRIPT", "Config change detected, but with problems. Rolling back config.")
|
||||
logger.warning("SCRIPT", "Config change detected, but with problems. Rolling back config.")
|
||||
self._rollback_config(backup_config)
|
||||
elif self.ok and self.initialized:
|
||||
if backup_config != self.__dict__:
|
||||
Logger.log_warning("SCRIPT", "Config change detected. Hot-reloading.")
|
||||
logger.warning("SCRIPT", "Config change detected. Hot-reloading.")
|
||||
self.changed = True
|
||||
|
||||
def validate(self):
|
||||
Logger.log_info("SCRIPT", "Validating config")
|
||||
logger.info("SCRIPT", "Validating config")
|
||||
self.ok = True
|
||||
self.tasks = ["CreateBackup", "FilterConvertKKS", "InstallChara", "RemoveChara"]
|
||||
self.create_gamepath()
|
||||
@@ -56,7 +56,7 @@ class Config:
|
||||
elif "OutputPath" in self.config_data[task]:
|
||||
path = self.config_data[task]["OutputPath"]
|
||||
if not os.path.exists(path):
|
||||
Logger.log_error("SCRIPT", f"Path invalid for task {task}")
|
||||
logger.error("SCRIPT", f"Path invalid for task {task}")
|
||||
raise Exception()
|
||||
|
||||
self.install_chara = self.config_data.get("InstallChara", {})
|
||||
@@ -72,13 +72,13 @@ class Config:
|
||||
"BepInEx": os.path.join(base, "BepInEx"),
|
||||
"mods": os.path.join(base, "mods"),
|
||||
"chara": os.path.join(base, "UserData\\chara\\female"),
|
||||
"coordinate": os.path.join(base, "UserData\coordinate"),
|
||||
"Overlays": os.path.join(base, "UserData\Overlays")
|
||||
"coordinate": os.path.join(base, "UserData\\coordinate"),
|
||||
"Overlays": os.path.join(base, "UserData\\Overlays")
|
||||
}
|
||||
|
||||
for path in self.game_path.values():
|
||||
if not os.path.exists(path):
|
||||
Logger.log_error("SCRIPT", "Game path not valid")
|
||||
logger.error("SCRIPT", "Game path not valid")
|
||||
raise Exception()
|
||||
|
||||
def _deepcopy_dict(self, dictionary):
|
||||
|
||||
@@ -5,7 +5,7 @@ import patoolib
|
||||
import customtkinter
|
||||
import subprocess
|
||||
import time
|
||||
from util.logger import Logger
|
||||
from app.common.logger import logger
|
||||
|
||||
class FileManager:
|
||||
|
||||
@@ -41,11 +41,11 @@ class FileManager:
|
||||
already_exists = os.path.exists(destination_path)
|
||||
|
||||
if already_exists and conflicts == "Skip":
|
||||
Logger.log_skipped(type, base_name)
|
||||
logger.skipped(type, base_name)
|
||||
return
|
||||
|
||||
elif already_exists and conflicts == "Replace":
|
||||
Logger.log_replaced(type, base_name)
|
||||
logger.replaced(type, base_name)
|
||||
|
||||
elif already_exists and conflicts == "Rename":
|
||||
max_retries = 3
|
||||
@@ -56,7 +56,7 @@ class FileManager:
|
||||
new_source_path = f"{filename}_{new_name}{file_extension}"
|
||||
os.rename(source_path, new_source_path)
|
||||
source_path = new_source_path
|
||||
Logger.log_renamed(type, base_name)
|
||||
logger.renamed(type, base_name)
|
||||
|
||||
filename, file_extension = os.path.splitext(destination_path)
|
||||
destination_path = f"{filename}_{new_name}{file_extension}"
|
||||
@@ -65,20 +65,20 @@ class FileManager:
|
||||
if attempt < max_retries - 1:
|
||||
time.sleep(1) # Wait for 1 second before retrying
|
||||
else:
|
||||
Logger.log_error(type, f"Failed to rename {base_name} after {max_retries} attempts.")
|
||||
logger.error(type, f"Failed to rename {base_name} after {max_retries} attempts.")
|
||||
return
|
||||
|
||||
try:
|
||||
shutil.copy(source_path, destination_path)
|
||||
print(f"File copied successfully from {source_path} to {destination_path}")
|
||||
if not already_exists:
|
||||
Logger.log_success(type, base_name)
|
||||
logger.success(type, base_name)
|
||||
except FileNotFoundError:
|
||||
Logger.log_error(type, f"{base_name} does not exist.")
|
||||
logger.error(type, f"{base_name} does not exist.")
|
||||
except PermissionError:
|
||||
Logger.log_error(type, f"Permission denied for {base_name}")
|
||||
logger.error(type, f"Permission denied for {base_name}")
|
||||
except Exception as e:
|
||||
Logger.log_error(type, f"An error occurred: {e}")
|
||||
logger.error(type, f"An error occurred: {e}")
|
||||
|
||||
def find_and_remove(self, type, source_path, destination_folder):
|
||||
source_path = source_path[0]
|
||||
@@ -87,15 +87,15 @@ class FileManager:
|
||||
if os.path.exists(destination_path):
|
||||
try:
|
||||
os.remove(destination_path)
|
||||
Logger.log_removed(type, base_name)
|
||||
logger.removed(type, base_name)
|
||||
except OSError as e:
|
||||
Logger.log_error(type, base_name)
|
||||
logger.error(type, base_name)
|
||||
|
||||
def create_archive(self, folders, archive_path):
|
||||
# Specify the full path to the 7zip executable
|
||||
path_to_7zip = patoolib.util.find_program("7z")
|
||||
if not path_to_7zip:
|
||||
Logger.log_error("SCRIPT", "7zip not found. Unable to create backup")
|
||||
logger.error("SCRIPT", "7zip not found. Unable to create backup")
|
||||
raise Exception()
|
||||
|
||||
if os.path.exists(archive_path+".7z"):
|
||||
@@ -133,7 +133,7 @@ class FileManager:
|
||||
# Print the output
|
||||
for line in process.stdout.decode().split('\n'):
|
||||
if line.strip() != "":
|
||||
Logger.log_info("7-Zip", line)
|
||||
logger.info("7-Zip", line)
|
||||
# Check the return code
|
||||
if process.returncode not in [0, 1]:
|
||||
raise Exception()
|
||||
@@ -141,7 +141,7 @@ class FileManager:
|
||||
def extract_archive(self, archive_path):
|
||||
try:
|
||||
archive_name = os.path.basename(archive_path)
|
||||
Logger.log_info("ARCHIVE", f"Extracting {archive_name}")
|
||||
logger.info("ARCHIVE", f"Extracting {archive_name}")
|
||||
extract_path = os.path.join(f"{os.path.splitext(archive_path)[0]}_{datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')}")
|
||||
patoolib.extract_archive(archive_path, outdir=extract_path)
|
||||
return extract_path
|
||||
@@ -161,5 +161,5 @@ class FileManager:
|
||||
except:
|
||||
text = f"Wrong password or {archive_name} is corrupted. Please enter password again or click Cancel"
|
||||
|
||||
Logger.log_skipped("ARCHIVE", archive_name)
|
||||
logger.skipped("ARCHIVE", archive_name)
|
||||
|
||||
|
||||
@@ -1,34 +0,0 @@
|
||||
class Logger(object):
|
||||
|
||||
@classmethod
|
||||
def log_msg(cls, type, msg):
|
||||
print(f"[MSG][{type}] {msg}")
|
||||
|
||||
@classmethod
|
||||
def log_success(cls, type, msg):
|
||||
print(f"[SUCCESS][{type}] {msg}")
|
||||
|
||||
@classmethod
|
||||
def log_skipped(cls, type, msg):
|
||||
print(f"[SKIPPED][{type}] {msg}")
|
||||
|
||||
@classmethod
|
||||
def log_replaced(cls, type, msg):
|
||||
print(f"[REPLACED][{type}] {msg}")
|
||||
|
||||
@classmethod
|
||||
def log_renamed(cls, type, msg):
|
||||
print(f"[RENAMED][{type}] {msg}")
|
||||
|
||||
@classmethod
|
||||
def log_removed(cls, type, msg):
|
||||
print(f"[REMOVED][{type}] {msg}")
|
||||
|
||||
@classmethod
|
||||
def log_error(cls, type, msg):
|
||||
print(f"[ERROR][{type}] {msg}")
|
||||
|
||||
@classmethod
|
||||
def log_info(cls, type, msg):
|
||||
print(f"[INFO][{type}] {msg}")
|
||||
|
||||
Reference in New Issue
Block a user