import sys import os import traceback import datetime from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed from colorama import Fore from tqdm import tqdm from dotenv import load_dotenv # This needs to be before the module imports as some of them currently try to read and use some of these values # upon import, in which case if they are unset the script will crash before we can output these messages. envMissing = False load_dotenv() for env in [ "api", "key", "organization", "model", "language", "timeout", "fileThreads", "threads", "width", "listWidth", ]: if os.getenv(env) is None or str(os.getenv(env))[:1] == "<": tqdm.write(Fore.RED + f"Environment variable {env} is not set!") envMissing = True if envMissing: tqdm.write( Fore.RED + "Some of the required environment values may not be set correctly. You can set \ these values using an .env file, for an example see .env.example" ) from modules.rpgmakermvmz import handleMVMZ, setSpeakerParseMode, finalizeSpeakerParse from modules.rpgmakerace import handleACE from modules.csv import handleCSV from modules.tyrano import handleTyrano from modules.kirikiri import handleKirikiri from modules.json import handleJSON from modules.lune import handleLune from modules.nscript import handleOnscripter from modules.wolf import handleWOLF from modules.wolf2 import handleWOLF2 from modules.regex import handleRegex from modules.text import handleText from modules.renpy import handleRenpy from modules.unity import handleUnity from modules.images import handleImages from modules.rpgmakerplugin import handlePlugin from modules.srpg import handleSRPG # For GPT4 rate limit will be hit if you have more than 1 thread. # 1 Thread for each file. Controls how many files are worked on at once. THREADS = int(os.getenv("fileThreads")) # [Display name, file extension, handle function] MODULES = [ ["RPGMaker MV/MZ", ["json"], handleMVMZ], ["RPGMaker Plugins", ["js", "rb"], handlePlugin], ["RPGMaker ACE", ["yaml"], handleACE], ["CSV (From Translator++)", ["csv"], handleCSV], ["Tyrano", ["ks"], handleTyrano], ["Kirikiri", ["ks", "tjs", "ssd", "asd"], handleKirikiri], ["JSON", ["json"], handleJSON], ["Lune", ["json"], handleLune], ["NScript", ["txt"], handleOnscripter], ["Wolf", ["json"], handleWOLF], ["Wolf", ["txt"], handleWOLF2], ["Regex", ["txt", "json", "script", "csv"], handleRegex], ["Text", ["txt", "srt"], handleText], ["Renpy", ["rpy"], handleRenpy], ["Unity", ["txt"], handleUnity], ["SRPG Studio", ["json"], handleSRPG], ["Images", [""], handleImages], ] # Info Message tqdm.write( Fore.CYAN + "-Dazed MTL Tool -" + Fore.RESET, end="\n\n", ) def main(): estimate = "" speaker_parse = False # Deferred until after engine select while estimate == "": estimate = input("Select Mode:\n\n 1. Translate\n 2. Estimate\n") match estimate: case "1": estimate = False case "2": estimate = True case _: estimate = "" version = "" while True: tqdm.write("Select game engine:\n") for position, module in enumerate(MODULES): tqdm.write(f"{str(position + 1).rjust(2)}. {module[0]} (.{module[1]})") version = input() try: version = int(version) - 1 except: continue if version in range(len(MODULES)): break totalCost = ( Fore.RED + "Translation module didn't return the total cost. Make sure the \ files to translate are in the /files folder and that you picked the right game engine." ) # If translating RPGMaker MV/MZ (index 0) prompt for speaker parse mode if version == 0 and not estimate: sub = "" while sub == "": sub = input("RPGMaker MV/MZ options:\n\n 1. Standard Translate\n 2. Parse Speakers (collect speaker names only)\n") match sub: case "1": speaker_parse = False case "2": speaker_parse = True case _: sub = "" if speaker_parse: setSpeakerParseMode(True) # Open File (Threads) - recursively walk 'files' and preserve directory structure # Prepare per-run log file so CLI runs also write to a run-specific history file try: hist_dir = Path("log") / "history" hist_dir.mkdir(parents=True, exist_ok=True) fname = datetime.datetime.now().strftime("translationHistory_%Y%m%d_%H%M%S.txt") run_log_path = hist_dir / fname run_log_path.touch(exist_ok=True) # Export env var so other modules/util functions pick it up try: os.environ['TRANSLATION_RUN_LOG'] = str(run_log_path) except Exception: pass # Try to create a hard link from legacy path to this run file for compatibility legacy = Path("log") / "translationHistory.txt" try: if legacy.exists(): try: legacy.unlink() except Exception: pass os.link(str(run_log_path), str(legacy)) except Exception: # Fallback: ensure legacy file exists so modules writing to it won't fail try: legacy.parent.mkdir(parents=True, exist_ok=True) legacy.touch(exist_ok=True) except Exception: pass except Exception: pass with ThreadPoolExecutor(max_workers=THREADS) as executor: futures = [] files_root = "files" # Special-case: Images engine expects a folder, not a file; schedule per directory containing assets if MODULES[version][0] == "Images": for root, dirs, filenames in os.walk(files_root): # Skip hidden/system directories dirs[:] = [d for d in dirs if d not in {".git", "__pycache__"}] # Skip the root 'files' itself to avoid processing everything twice # We'll still allow scheduling for root if it contains assets # Only schedule directories that contain potential assets has_assets = any(fn.lower().endswith((".png", ".txt")) for fn in filenames) if not has_assets: continue # Compute relative directory path and ensure translated mirror exists rel_dir = os.path.relpath(root, files_root).replace(os.sep, "/") if rel_dir == ".": # Represent root as empty string so handler creates files under translated/ directly rel_dir = "" try: target_dir = os.path.join("translated", rel_dir.replace("/", os.sep)) if rel_dir else "translated" os.makedirs(target_dir, exist_ok=True) except Exception: pass futures.append( executor.submit(MODULES[version][2], rel_dir, estimate) ) else: # Gather all candidate files recursively for root, dirs, filenames in os.walk(files_root): # Skip hidden/system directories if any dirs[:] = [d for d in dirs if d not in {".git", "__pycache__"}] for fname in filenames: if fname == ".gitkeep": continue abs_path = os.path.join(root, fname) # Build relative path from 'files' root using POSIX-style separators so handlers can do 'files/' + rel rel_path = os.path.relpath(abs_path, files_root) rel_path_posix = rel_path.replace(os.sep, "/") # Check extension match for the selected module version for m in MODULES[version][1]: if rel_path_posix.endswith(m): # Ensure the corresponding directory exists under 'translated' rel_dir = os.path.dirname(rel_path_posix) if rel_dir: try: os.makedirs(os.path.join("translated", rel_dir.replace("/", os.sep)), exist_ok=True) except Exception: # Best-effort; handler may attempt write and fail if permissions are insufficient pass futures.append( executor.submit(MODULES[version][2], rel_path_posix, estimate) ) break # Avoid double-adding if multiple ext entries match for future in as_completed(futures): try: totalCost = future.result() except Exception as e: tracebackLineNo = str(traceback.extract_tb(sys.exc_info()[2])[-1].lineno) tqdm.write(Fore.RED + str(e) + "|" + tracebackLineNo + Fore.RESET) # Finalize speaker parse mode by writing collected speakers to vocab if speaker_parse: finalizeSpeakerParse() # Delete Tmp Files if os.path.isfile("csv.tmp"): os.remove("csv.tmp") # Sweep any leftover temp files in translated/ try: translated_dir = os.path.join("translated") if os.path.isdir(translated_dir): for fname in os.listdir(translated_dir): if fname.endswith(".tmp"): fpath = os.path.join(translated_dir, fname) try: os.remove(fpath) except Exception: # Best-effort cleanup; ignore files locked by other processes pass except Exception: pass # Finish if totalCost != "Fail": # if estimate is False: # This is to encourage people to grab what's in /translated instead # deleteFolderFiles("files") tqdm.write(str(totalCost)) def deleteFolderFiles(folderPath): for filename in os.listdir(folderPath): file_path = os.path.join(folderPath, filename) if file_path.endswith((".json", ".yaml", ".ks")): os.remove(file_path)