4 Commits

Author SHA1 Message Date
1c1620d91a First Time Setup Implementation 2025-05-16 00:25:46 +02:00
88f440dd5e Compilation Assessment Plan 2025-05-15 21:15:28 +02:00
8ae9eaed35 Merge branch 'GUI-and-Configs' into Dev 2025-05-15 20:58:53 +02:00
b43b2522d7 Implemented Item type priority handling ( DISP16 ) 2025-05-15 20:52:58 +02:00
20 changed files with 1582 additions and 806 deletions

View File

@@ -38,7 +38,8 @@
"delete_system_pattern_by_id", "delete_system_pattern_by_id",
"get_conport_schema", "get_conport_schema",
"get_recent_activity_summary", "get_recent_activity_summary",
"semantic_search_conport" "semantic_search_conport",
"search_system_patterns_fts"
] ]
} }
} }

View File

View File

@@ -1,3 +1,15 @@
{ {
"customModes": [] "customModes": [
{
"slug": "Task-Initiator",
"name": "Task Initiator",
"roleDefinition": "You are Task Initiator. Your exclusive function is comprehensive initial context gathering, focusing solely on ConPort data. Do NOT perform other tasks or use direct file system tools for context gathering.",
"customInstructions": "1. First, execute standard initial context setup procedures (as per global ConPort strategy).\n2. Next, if a specific user request is pending, YOU, as Task Initiator, should analyze it and proactively gather relevant information, strictly by querying ConPort. Your process for this is:\n a. Identify the key subject(s) of the request.\n b. Loosely search relevant ConPort data for information or summaries related to these identified subject(s).\n3. After completing both standard setup AND any ConPort-based task-specific gathering, briefly report the overall context status. This report must cover ConPort initialization and summarize any specific information found (or explicitly not found) within ConPort relevant to the user's request.\n4. Then, output `[TASK_INITIATOR_COMPLETE]`.\n5. Finally, to address the user's main request with the context you've gathered (or confirmed is missing from ConPort), use the `switch_mode` tool to transition to the determined most appropriate mode by analysing the initial request. you should ALWAYS finish context-gathering before switching modes.",
"groups": [
"mcp",
"read"
],
"source": "project"
}
]
} }

View File

@@ -34,7 +34,7 @@ The script accepts several command-line arguments to configure the test run. If
* A string to search for within the application logs generated during the test run. If found, matching log lines (with context) will be highlighted. * A string to search for within the application logs generated during the test run. If found, matching log lines (with context) will be highlighted.
* Default: None * Default: None
* `--additional-lines NUM_LINES` (optional): * `--additional-lines NUM_LINES` (optional):
* When using `--search`, this specifies how many lines of context before and after each matching log line should be displayed. * When using `--search`, this specifies how many lines of context before and after each matching log line should be displayed. A good non-zero value is 1-2.
* Default: `0` * Default: `0`
**Example Usage:** **Example Usage:**
@@ -81,3 +81,5 @@ When executed, `autotest.py` performs the following steps:
* **Output Directory:** Inspect the contents of the specified output directory to manually verify the processed assets if needed. * **Output Directory:** Inspect the contents of the specified output directory to manually verify the processed assets if needed.
This automated test helps ensure the stability of the core processing logic when driven by GUI-equivalent actions. This automated test helps ensure the stability of the core processing logic when driven by GUI-equivalent actions.
Note: Under some conditions, the autotest will exit with errorcode "3221226505". This has no consequence and can therefor be ignore.

View File

@@ -1,5 +1,5 @@
{ {
"preset_name": "Dinesen Custom", "preset_name": "Dinesen",
"supplier_name": "Dinesen", "supplier_name": "Dinesen",
"notes": "Preset for standard Poliigon downloads. Prioritizes _xxx16 files. Moves previews etc. to Extra/. Assumes Metal/Rough workflow.", "notes": "Preset for standard Poliigon downloads. Prioritizes _xxx16 files. Moves previews etc. to Extra/. Assumes Metal/Rough workflow.",
"source_naming": { "source_naming": {
@@ -10,11 +10,7 @@
}, },
"glossiness_keywords": [ "glossiness_keywords": [
"GLOSS" "GLOSS"
], ]
"bit_depth_variants": {
"NRM": "*_NRM16*",
"DISP": "*_DISP16*"
}
}, },
"move_to_extra_patterns": [ "move_to_extra_patterns": [
"*_Preview*", "*_Preview*",
@@ -25,7 +21,8 @@
"*.pdf", "*.pdf",
"*.url", "*.url",
"*.htm*", "*.htm*",
"*_Fabric.*" "*_Fabric.*",
"*_DISP_*METALNESS*"
], ],
"map_type_mapping": [ "map_type_mapping": [
{ {
@@ -46,6 +43,11 @@
"NORM*", "NORM*",
"NRM*", "NRM*",
"N" "N"
],
"priority_keywords": [
"*_NRM16*",
"*_NM16*",
"*Normal16*"
] ]
}, },
{ {
@@ -75,6 +77,14 @@
"DISP", "DISP",
"HEIGHT", "HEIGHT",
"BUMP" "BUMP"
],
"priority_keywords": [
"*_DISP16*",
"*_DSP16*",
"*DSP16*",
"*DISP16*",
"*Displacement16*",
"*Height16*"
] ]
}, },
{ {

View File

@@ -10,11 +10,7 @@
}, },
"glossiness_keywords": [ "glossiness_keywords": [
"GLOSS" "GLOSS"
], ]
"bit_depth_variants": {
"NRM": "*_NRM16*",
"DISP": "*_DISP16*"
}
}, },
"move_to_extra_patterns": [ "move_to_extra_patterns": [
"*_Preview*", "*_Preview*",
@@ -28,7 +24,114 @@
"*_Fabric.*", "*_Fabric.*",
"*_Albedo*" "*_Albedo*"
], ],
"map_type_mapping": [], "map_type_mapping": [
{
"target_type": "MAP_COL",
"keywords": [
"COLOR*",
"COL",
"COL-*",
"DIFFUSE",
"DIF",
"ALBEDO"
]
},
{
"target_type": "MAP_NRM",
"keywords": [
"NORMAL*",
"NORM*",
"NRM*",
"N"
],
"priority_keywords": [
"*_NRM16*",
"*_NM16*",
"*Normal16*"
]
},
{
"target_type": "MAP_ROUGH",
"keywords": [
"ROUGHNESS",
"ROUGH"
]
},
{
"target_type": "MAP_GLOSS",
"keywords": [
"GLOSS"
]
},
{
"target_type": "MAP_AO",
"keywords": [
"AMBIENTOCCLUSION",
"AO"
]
},
{
"target_type": "MAP_DISP",
"keywords": [
"DISPLACEMENT",
"DISP",
"HEIGHT",
"BUMP"
],
"priority_keywords": [
"*_DISP16*",
"*_DSP16*",
"*DSP16*",
"*DISP16*",
"*Displacement16*",
"*Height16*"
]
},
{
"target_type": "MAP_REFL",
"keywords": [
"REFLECTION",
"REFL",
"SPECULAR",
"SPEC"
]
},
{
"target_type": "MAP_SSS",
"keywords": [
"SSS",
"SUBSURFACE*"
]
},
{
"target_type": "MAP_FUZZ",
"keywords": [
"FUZZ"
]
},
{
"target_type": "MAP_IDMAP",
"keywords": [
"IDMAP"
]
},
{
"target_type": "MAP_MASK",
"keywords": [
"OPAC*",
"TRANSP*",
"MASK*",
"ALPHA*"
]
},
{
"target_type": "MAP_METAL",
"keywords": [
"METAL*",
"METALLIC"
]
}
],
"asset_category_rules": { "asset_category_rules": {
"model_patterns": [ "model_patterns": [
"*.fbx", "*.fbx",

View File

@@ -1,9 +1,9 @@
{ {
"source_rules": [ "source_rules": [
{ {
"input_path": "BoucleChunky001.zip", "input_path": "BoucleChunky001.zip",
"supplier_identifier": "Dinesen", "supplier_identifier": "Dinesen",
"preset_name": null, "preset_name": "Dinesen",
"assets": [ "assets": [
{ {
"asset_name": "BoucleChunky001", "asset_name": "BoucleChunky001",

View File

@@ -298,15 +298,32 @@ class AutoTester(QObject):
def run_test(self) -> None: def run_test(self) -> None:
"""Orchestrates the test steps.""" """Orchestrates the test steps."""
logger.info("Starting test run...") # Load expected rules first to potentially get the preset name
self._load_expected_rules() # Moved here
if not self.expected_rules_data: # Ensure rules were loaded if not self.expected_rules_data: # Ensure rules were loaded
logger.error("Expected rules not loaded. Aborting test.") logger.error("Expected rules not loaded. Aborting test.")
self.cleanup_and_exit(success=False) self.cleanup_and_exit(success=False)
return return
# Determine preset to use: from expected rules if available, else from CLI args
preset_to_use = self.cli_args.preset # Default
if self.expected_rules_data.get("source_rules") and \
isinstance(self.expected_rules_data["source_rules"], list) and \
len(self.expected_rules_data["source_rules"]) > 0 and \
isinstance(self.expected_rules_data["source_rules"][0], dict) and \
self.expected_rules_data["source_rules"][0].get("preset_name"):
preset_to_use = self.expected_rules_data["source_rules"][0]["preset_name"]
logger.info(f"Overriding preset with value from expected_rules.json: '{preset_to_use}'")
else:
logger.info(f"Using preset from CLI arguments: '{preset_to_use}' (this was self.cli_args.preset)")
# If preset_to_use is still self.cli_args.preset, ensure it's logged correctly
# The variable preset_to_use will hold the correct value to be used throughout.
logger.info("Starting test run...") # Moved after preset_to_use definition
# Add a specific summary log for essential context # Add a specific summary log for essential context
logger.info(f"Autotest Context: Input='{self.cli_args.zipfile.name}', Preset='{self.cli_args.preset}', Output='{self.cli_args.outputdir}'") # This now correctly uses preset_to_use
logger.info(f"Autotest Context: Input='{self.cli_args.zipfile.name}', Preset='{preset_to_use}', Output='{self.cli_args.outputdir}'")
# Step 1: Load ZIP # Step 1: Load ZIP
self.test_step = "LOADING_ZIP" self.test_step = "LOADING_ZIP"
@@ -326,20 +343,25 @@ class AutoTester(QObject):
# Step 2: Select Preset # Step 2: Select Preset
self.test_step = "SELECTING_PRESET" self.test_step = "SELECTING_PRESET"
logger.info(f"Step 2: Selecting preset: {self.cli_args.preset}") # KEEP INFO - Passes filter # Use preset_to_use (which is now correctly defined earlier)
logger.info(f"Step 2: Selecting preset: {preset_to_use}") # KEEP INFO - Passes filter
# The print statement below already uses preset_to_use, which is good.
print(f"DEBUG: Attempting to select preset: '{preset_to_use}' (derived from expected: {preset_to_use == self.expected_rules_data.get('source_rules',[{}])[0].get('preset_name') if self.expected_rules_data.get('source_rules') else 'N/A'}, cli_arg: {self.cli_args.preset})")
preset_found = False preset_found = False
preset_list_widget = self.main_window.preset_editor_widget.editor_preset_list preset_list_widget = self.main_window.preset_editor_widget.editor_preset_list
for i in range(preset_list_widget.count()): for i in range(preset_list_widget.count()):
item = preset_list_widget.item(i) item = preset_list_widget.item(i)
if item and item.text() == self.cli_args.preset: if item and item.text() == preset_to_use: # Use preset_to_use
preset_list_widget.setCurrentItem(item) preset_list_widget.setCurrentItem(item)
logger.debug(f"Preset '{self.cli_args.preset}' selected.") logger.debug(f"Preset '{preset_to_use}' selected.")
print(f"DEBUG: Successfully selected preset '{item.text()}' in GUI.")
preset_found = True preset_found = True
break break
if not preset_found: if not preset_found:
logger.error(f"Preset '{self.cli_args.preset}' not found in the list.") logger.error(f"Preset '{preset_to_use}' not found in the list.")
available_presets = [preset_list_widget.item(i).text() for i in range(preset_list_widget.count())] available_presets = [preset_list_widget.item(i).text() for i in range(preset_list_widget.count())]
logger.debug(f"Available presets: {available_presets}") logger.debug(f"Available presets: {available_presets}")
print(f"DEBUG: Failed to find preset '{preset_to_use}'. Available: {available_presets}")
self.cleanup_and_exit(success=False) self.cleanup_and_exit(success=False)
return return
@@ -449,8 +471,6 @@ class AutoTester(QObject):
else: else:
logger.warning("Log console or output widget not found. Cannot retrieve logs.") logger.warning("Log console or output widget not found. Cannot retrieve logs.")
self._process_and_display_logs(all_logs_text)
logger.info("Log analysis completed.")
# Final Step # Final Step
logger.info("Test run completed successfully.") # KEEP INFO - Passes filter logger.info("Test run completed successfully.") # KEEP INFO - Passes filter
@@ -523,7 +543,7 @@ class AutoTester(QObject):
comparable_sources_list.append({ comparable_sources_list.append({
"input_path": Path(source_rule_obj.input_path).name, # Use only the filename "input_path": Path(source_rule_obj.input_path).name, # Use only the filename
"supplier_identifier": source_rule_obj.supplier_identifier, "supplier_identifier": source_rule_obj.supplier_identifier,
"preset_name": source_rule_obj.preset_name, "preset_name": source_rule_obj.preset_name, # This is the actual preset name from the SourceRule object
"assets": comparable_asset_list "assets": comparable_asset_list
}) })
logger.debug("Conversion to comparable dictionary finished.") logger.debug("Conversion to comparable dictionary finished.")
@@ -569,6 +589,8 @@ class AutoTester(QObject):
if not self._compare_list_of_rules(actual_value, expected_value, "FileRule", current_context, "file_path"): if not self._compare_list_of_rules(actual_value, expected_value, "FileRule", current_context, "file_path"):
item_match = False item_match = False
else: # Regular field comparison else: # Regular field comparison
if key == "preset_name":
print(f"DEBUG: Comparing preset_name: Actual='{actual_value}', Expected='{expected_value}' for {item_type_name} ({current_context})")
if actual_value != expected_value: if actual_value != expected_value:
# Handle None vs "None" string for preset_name specifically if it's a common issue # Handle None vs "None" string for preset_name specifically if it's a common issue
if key == "preset_name" and actual_value is None and expected_value == "None": if key == "preset_name" and actual_value is None and expected_value == "None":
@@ -602,6 +624,11 @@ class AutoTester(QObject):
list_match = False # Count mismatch is an error list_match = False # Count mismatch is an error
# If counts differ, we still try to match what we can to provide more detailed feedback, # If counts differ, we still try to match what we can to provide more detailed feedback,
# but the overall list_match will remain False. # but the overall list_match will remain False.
if item_type_name == "FileRule":
print(f"DEBUG: FileRule count mismatch for {parent_context}. Actual: {len(actual_list)}, Expected: {len(expected_list)}")
print(f"DEBUG: Actual FileRule paths: {[item.get(item_key_field) for item in actual_list]}")
print(f"DEBUG: Expected FileRule paths: {[item.get(item_key_field) for item in expected_list]}")
actual_items_map = {item.get(item_key_field): item for item in actual_list if item.get(item_key_field) is not None} actual_items_map = {item.get(item_key_field): item for item in actual_list if item.get(item_key_field) is not None}
@@ -765,6 +792,10 @@ class AutoTester(QObject):
def cleanup_and_exit(self, success: bool = True) -> None: def cleanup_and_exit(self, success: bool = True) -> None:
"""Cleans up and exits the application.""" """Cleans up and exits the application."""
# Retrieve logs before clearing the handler
all_logs_text = "" # This variable is not used by _process_and_display_logs anymore, but kept for signature compatibility if needed elsewhere.
self._process_and_display_logs(all_logs_text) # Process and display logs BEFORE clearing the buffer
global autotest_memory_handler global autotest_memory_handler
if autotest_memory_handler: if autotest_memory_handler:
logger.debug("Clearing memory log handler buffer and removing handler.") logger.debug("Clearing memory log handler buffer and removing handler.")

View File

@@ -0,0 +1,167 @@
# Plan: Assessing Compilation of Asset Processor with PyInstaller and Cython
## Objective
To assess the feasibility and create a plan for compiling the Asset Processor project into standalone executables using PyInstaller, incorporating Cython for general speedup and source code obfuscation. A key requirement is to maintain user access to, and the ability to modify, configuration files (like `user_settings.json`, `asset_type_definitions.json`, etc.) and `Preset` files post-compilation.
---
## Phase 1: Initial Analysis & Information Gathering
* **Project Dependencies (from [`requirements.txt`](requirements.txt:1)):**
* `opencv-python`
* `numpy`
* `openexr`
* `PySide6`
* `py7zr`
* `rarfile`
* `requests`
* *Note: `PySide6`, `opencv-python`, and `openexr` may require special handling with PyInstaller (e.g., hidden imports, hooks).*
* **Configuration Loading (based on [`configuration.py`](configuration.py:1)):**
* Configuration files (`app_settings.json`, `llm_settings.json`, `asset_type_definitions.json`, `file_type_definitions.json`, `user_settings.json`, `suppliers.json`) are loaded from a `config/` subdirectory relative to [`configuration.py`](configuration.py:1).
* Preset files are loaded from a `Presets/` subdirectory relative to [`configuration.py`](configuration.py:1).
* `BASE_DIR` is `Path(__file__).parent`, which will refer to the bundled location in a PyInstaller build.
* [`user_settings.json`](configuration.py:16) is designed for overrides and is a candidate for external management.
* Saving functions write back to these relative paths, which needs adaptation.
* **Potential Cython Candidates:**
* Modules within the `processing/` directory.
* Specifically: `processing/utils/image_processing_utils.py` and individual stage files in `processing/pipeline/stages/` (e.g., `alpha_extraction_to_mask.py`, `gloss_to_rough_conversion.py`, etc.).
* Other modules (e.g., `processing/pipeline/orchestrator.py`) could be Cythonized primarily for obfuscation.
* **User-Accessible Files (Defaults):**
* The `config/` directory (containing `app_settings.json`, `asset_type_definitions.json`, `file_type_definitions.json`, `llm_settings.json`, `suppliers.json`).
* The `Presets/` directory and its contents.
---
## Phase 2: Strategy Development
1. **Cython Strategy:**
* **Build Integration:** Utilize a `setup.py` script with `setuptools` and `Cython.Build.cythonize` to compile `.py` files into C extensions (`.pyd` on Windows, `.so` on Linux/macOS).
* **Candidate Prioritization:** Focus on `processing/` modules for performance gains and obfuscation.
* **Compatibility & Challenges:**
* GUI modules (PySide6) are generally left as Python.
* Ensure compatibility with OpenCV, NumPy, and OpenEXR.
* Address potential issues with highly dynamic Python code.
* Consider iterative conversion to `.pyx` files with C-style type annotations for maximum performance in identified hot spots.
* **Obfuscation:** The primary goal for many modules might be obfuscation rather than pure speedup.
2. **PyInstaller Strategy:**
* **Bundle Type:** One-directory bundle (`--onedir`) is recommended for easier debugging and data file management.
* **Data Files (`.spec` file `datas` section):**
* Bundle default `config/` directory (containing `app_settings.json`, `asset_type_definitions.json`, `file_type_definitions.json`, `llm_settings.json`, `suppliers.json`).
* Bundle default `Presets/` directory.
* Include any other necessary GUI assets (icons, etc.).
* Consider bundling the `blender_addon/` if it's to be deployed with the app.
* **Hidden Imports & Hooks (`.spec` file):**
* Add explicit `hiddenimports` for `PySide6`, `opencv-python`, `openexr`, and any other problematic libraries.
* Utilize or create PyInstaller hooks if necessary.
* **Console Window:** Disable for GUI application (`console=False`).
3. **User-Accessible Files & First-Time Setup Strategy:**
* **First-Run Detection:** Application checks for a marker file or stored configuration path.
* **First-Time Setup UI (PySide6 Dialog):**
* **Configuration Location Choice:**
* Option A (Recommended): Store in user profile (e.g., `Documents/AssetProcessor` or `AppData/Roaming/AssetProcessor`).
* Option B (Advanced): User chooses a custom folder.
* The application copies default `config/` (excluding `app_settings.json` but including other definition files) and `Presets/` to the chosen location.
* The chosen path is saved.
* **Key Application Settings Configuration (saved to `user_settings.json` in user's chosen location):**
* Default Library Output Path (`OUTPUT_BASE_DIR`).
* Asset Structure (`OUTPUT_DIRECTORY_PATTERN`).
* Image Output Formats (`OUTPUT_FORMAT_16BIT_PRIMARY`, `OUTPUT_FORMAT_16BIT_FALLBACK`, `OUTPUT_FORMAT_8BIT`).
* JPG Threshold (`RESOLUTION_THRESHOLD_FOR_JPG`).
* Blender Paths (`DEFAULT_NODEGROUP_BLEND_PATH`, `DEFAULT_MATERIALS_BLEND_PATH`, `BLENDER_EXECUTABLE_PATH`).
* **Configuration Loading Logic Modification ([`configuration.py`](configuration.py:1)):**
* `BASE_DIR` for user-modifiable files will point to the user-chosen location.
* `app_settings.json` (master defaults) always loaded from the bundle.
* `user_settings.json` loaded from the user-chosen location, containing overrides.
* Other definition files and `Presets` loaded from the user-chosen location, with a fallback/re-copy mechanism from bundled defaults if missing.
* **Saving Logic Modification ([`configuration.py`](configuration.py:1)):**
* All configuration saving functions will write to the user-chosen configuration location. Bundled defaults remain read-only post-installation.
---
## Phase 3: Outline of Combined Build Process
1. **Environment Setup (Developer):** Install Python, Cython, PyInstaller, and project dependencies.
2. **Cythonization (`setup.py`):**
* Create `setup.py` using `setuptools` and `Cython.Build.cythonize`.
* List `.py` files/modules for compilation (e.g., `processing.utils.image_processing_utils`, `processing.pipeline.stages.*`).
* Include `numpy.get_include()` if Cython files use NumPy C-API.
* Run `python setup.py build_ext --inplace` to generate `.pyd`/`.so` files.
3. **PyInstaller Packaging (`.spec` file):**
* Generate initial `AssetProcessor.spec` with `pyinstaller --name AssetProcessor main.py`.
* Modify `.spec` file:
* `datas`: Add default `config/` and `Presets/` directories, and other assets.
* `hiddenimports`: List modules for `PySide6`, `opencv-python`, etc.
* `excludes`: Optionally exclude original `.py` files for Cythonized modules.
* Set `onedir = True`, `onefile = False`, `console = False`.
* Run `pyinstaller AssetProcessor.spec` to create `dist/AssetProcessor`.
4. **Post-Build Steps (Optional):**
* Clean up original `.py` files from `dist/` if obfuscation is paramount.
* Archive `dist/AssetProcessor` for distribution (ZIP, installer).
---
## Phase 4: Distribution Structure
**Inside `dist/AssetProcessor/` (Distribution Package):**
* `AssetProcessor.exe` (or platform equivalent)
* Core Python and library dependencies (DLLs/SOs)
* Cythonized modules (`.pyd`/`.so` files, e.g., `processing/utils/image_processing_utils.pyd`)
* Non-Cythonized Python modules (`.pyc` files)
* Bundled default `config/` directory (with `app_settings.json`, `asset_type_definitions.json`, etc.)
* Bundled default `Presets/` directory (with `_template.json`, `Dinesen.json`, etc.)
* Other GUI assets (icons, etc.)
* Potentially `blender_addon/` files if bundled.
**User's Configuration Directory (e.g., `Documents/AssetProcessor/`, created on first run):**
* `user_settings.json` (user's choices for paths, formats, etc.)
* Copied `config/` directory (for user modification of `asset_type_definitions.json`, etc.)
* Copied `Presets/` directory (for user modification/additions)
* Marker file for first-time setup choice.
---
## Phase 5: Plan for Testing & Validation
1. **Core Functionality:** Test GUI operations, Directory Monitor, CLI (if applicable).
2. **Configuration System:**
* Verify first-time setup UI, config location choice, copying of defaults.
* Confirm loading from and saving to the user's chosen config location.
* Test modification of user configs and application's reflection of changes.
3. **Dependency Checks:** Ensure bundled libraries (PySide6, OpenCV) function correctly.
4. **Performance (Cython):** Basic comparison of critical operations (Python vs. Cythonized).
5. **Obfuscation (Cython):** Verify absence of original `.py` files for Cythonized modules in distribution (if desired) and that `.pyd`/`.so` files are used.
6. **Cross-Platform Testing:** Repeat build and test process on all target OS.
---
## Phase 6: Documentation Outline
1. **Developer/Build Documentation:**
* Build environment setup.
* `setup.py` (Cython) and `pyinstaller` command usage.
* Structure of `setup.py` and `.spec` file, key configurations.
* Troubleshooting common build issues.
2. **User Documentation:**
* First-time setup guide (config location, initial settings).
* Managing user-specific configurations and presets (location, backup).
* How to reset to default configurations.
---
## Phase 7: Risk Assessment & Mitigation (Brief)
* **Risk:** Cython compilation issues.
* **Mitigation:** Incremental compilation, selective Cythonization.
* **Risk:** PyInstaller packaging complexities.
* **Mitigation:** Thorough testing, community hooks, iterative `.spec` refinement.
* **Risk:** Logic errors in new configuration loading/saving.
* **Mitigation:** Careful coding, detailed testing of config pathways.
* **Risk:** Cython performance not meeting expectations.
* **Mitigation:** Profile Python code first; focus Cython on CPU-bound loops.
* **Risk:** Increased build complexity.
* **Mitigation:** Automate build steps with scripts.

View File

@@ -195,14 +195,16 @@
"FILE_IGNORE": { "FILE_IGNORE": {
"bit_depth_rule": "", "bit_depth_rule": "",
"color": "#673d35", "color": "#673d35",
"description": "File to be ignored", "description": "File identified to be ignored due to prioritization rules (e.g., a lower bit-depth version when a higher one is present).",
"category": "Ignored",
"examples": [ "examples": [
"Thumbs.db", "Thumbs.db",
".DS_Store" ".DS_Store"
], ],
"is_grayscale": false, "is_grayscale": false,
"keybind": "X", "keybind": "X",
"standard_type": "" "standard_type": "",
"details": {}
} }
} }
} }

View File

@@ -1,26 +1,42 @@
import json import json
import os import os
import sys
import shutil
from pathlib import Path from pathlib import Path
import logging import logging
import re import re
import collections.abc import collections.abc
from typing import Optional from typing import Optional, Union
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
BASE_DIR = Path(__file__).parent # This BASE_DIR is primarily for fallback when not bundled or for locating bundled resources relative to the script.
APP_SETTINGS_PATH = BASE_DIR / "config" / "app_settings.json" _SCRIPT_DIR = Path(__file__).resolve().parent
LLM_SETTINGS_PATH = BASE_DIR / "config" / "llm_settings.json"
ASSET_TYPE_DEFINITIONS_PATH = BASE_DIR / "config" / "asset_type_definitions.json"
FILE_TYPE_DEFINITIONS_PATH = BASE_DIR / "config" / "file_type_definitions.json"
USER_SETTINGS_PATH = BASE_DIR / "config" / "user_settings.json"
SUPPLIERS_CONFIG_PATH = BASE_DIR / "config" / "suppliers.json"
PRESETS_DIR = BASE_DIR / "Presets"
class ConfigurationError(Exception): class ConfigurationError(Exception):
"""Custom exception for configuration loading errors.""" """Custom exception for configuration loading errors."""
pass pass
def _get_user_config_path_placeholder() -> Optional[Path]:
"""
Placeholder function. In a real scenario, this would retrieve the
saved user configuration path (e.g., from a settings file).
Returns None if not set, triggering first-time setup behavior.
"""
# For this subtask, we assume this path is determined externally and passed to Configuration.
# If we were to implement the settings.ini check here, it would look like:
# try:
# app_data_dir = Path(os.getenv('APPDATA')) / "AssetProcessor"
# settings_ini = app_data_dir / "settings.ini"
# if settings_ini.exists():
# with open(settings_ini, 'r') as f:
# path_str = f.read().strip()
# return Path(path_str)
# except Exception:
# return None
return None
def _get_base_map_type(target_map_string: str) -> str: def _get_base_map_type(target_map_string: str) -> str:
"""Extracts the base map type (e.g., 'COL') from a potentially numbered string ('COL-1').""" """Extracts the base map type (e.g., 'COL') from a potentially numbered string ('COL-1')."""
# Use regex to find the leading alphabetical part # Use regex to find the leading alphabetical part
@@ -92,49 +108,252 @@ def _deep_merge_dicts(base_dict: dict, override_dict: dict) -> dict:
class Configuration: class Configuration:
""" """
Loads and provides access to core settings combined with a specific preset. Loads and provides access to core settings combined with a specific preset,
managing bundled and user-specific configuration paths.
""" """
def __init__(self, preset_name: str): BASE_DIR_APP_BUNDLED_CONFIG_SUBDIR_NAME = "config"
PRESETS_DIR_APP_BUNDLED_NAME = "Presets"
USER_SETTINGS_FILENAME = "user_settings.json"
APP_SETTINGS_FILENAME = "app_settings.json"
ASSET_TYPE_DEFINITIONS_FILENAME = "asset_type_definitions.json"
FILE_TYPE_DEFINITIONS_FILENAME = "file_type_definitions.json"
LLM_SETTINGS_FILENAME = "llm_settings.json"
SUPPLIERS_CONFIG_FILENAME = "suppliers.json"
USER_CONFIG_SUBDIR_NAME = "config" # Subdirectory within user's chosen config root for most jsons
USER_PRESETS_SUBDIR_NAME = "Presets" # Subdirectory within user's chosen config root for presets
def __init__(self, preset_name: str, base_dir_user_config: Optional[Path] = None, is_first_run_setup: bool = False):
""" """
Loads core config, user overrides, and the specified preset file. Loads core config, user overrides, and the specified preset file.
Args: Args:
preset_name: The name of the preset (without .json extension). preset_name: The name of the preset (without .json extension).
base_dir_user_config: The root path for user-specific configurations.
If None, loading of user-specific files will be skipped or may fail.
is_first_run_setup: Flag indicating if this is part of the initial setup
process where user config dir might be empty and fallbacks
should not aggressively try to copy from bundle until UI confirms.
Raises: Raises:
ConfigurationError: If core config or preset cannot be loaded/validated. ConfigurationError: If critical configurations cannot be loaded/validated.
""" """
log.debug(f"Initializing Configuration with preset: '{preset_name}'") log.debug(f"Initializing Configuration with preset: '{preset_name}', user_config_dir: '{base_dir_user_config}', first_run_flag: {is_first_run_setup}")
self.preset_name = preset_name self._preset_filename_stem = preset_name
self.base_dir_user_config: Optional[Path] = base_dir_user_config
self.is_first_run_setup = is_first_run_setup
self.base_dir_app_bundled: Path = self._determine_base_dir_app_bundled()
# 1. Load core settings log.info(f"Determined BASE_DIR_APP_BUNDLED: {self.base_dir_app_bundled}")
self._core_settings: dict = self._load_core_config() log.info(f"Using BASE_DIR_USER_CONFIG: {self.base_dir_user_config}")
# 2. Load asset type definitions # 1. Load core application settings (always from bundled)
self._asset_type_definitions: dict = self._load_asset_type_definitions() app_settings_path = self.base_dir_app_bundled / self.BASE_DIR_APP_BUNDLED_CONFIG_SUBDIR_NAME / self.APP_SETTINGS_FILENAME
self._core_settings: dict = self._load_json_file(
app_settings_path,
is_critical=True,
description="Core application settings"
)
# 3. Load file type definitions # 2. Load user settings (from user config dir, if provided)
self._file_type_definitions: dict = self._load_file_type_definitions() user_settings_overrides: dict = {}
if self.base_dir_user_config:
user_settings_file_path = self.base_dir_user_config / self.USER_SETTINGS_FILENAME
user_settings_overrides = self._load_json_file(
user_settings_file_path,
is_critical=False, # Not critical if missing, especially on first run
description=f"User settings from {user_settings_file_path}"
) or {} # Ensure it's a dict
else:
log.info(f"{self.USER_SETTINGS_FILENAME} not loaded: User config directory not set.")
# 4. Load user settings # 3. Deep merge user settings onto core settings
user_settings_overrides: dict = self._load_user_settings()
# 5. Deep merge user settings onto core settings
if user_settings_overrides: if user_settings_overrides:
log.info("Applying user setting overrides to core settings.") log.info(f"Applying user setting overrides to core settings.")
# _deep_merge_dicts modifies self._core_settings in place
_deep_merge_dicts(self._core_settings, user_settings_overrides) _deep_merge_dicts(self._core_settings, user_settings_overrides)
# 6. Load LLM settings # 4. Load other definition files (from user config dir, with fallback from bundled)
self._llm_settings: dict = self._load_llm_config() self._asset_type_definitions: dict = self._load_definition_file_with_fallback(
self.ASSET_TYPE_DEFINITIONS_FILENAME, "ASSET_TYPE_DEFINITIONS"
)
self._file_type_definitions: dict = self._load_definition_file_with_fallback(
self.FILE_TYPE_DEFINITIONS_FILENAME, "FILE_TYPE_DEFINITIONS"
)
self._llm_settings: dict = self._load_definition_file_with_fallback(
self.LLM_SETTINGS_FILENAME, None # LLM settings might be flat (no root key)
)
self._suppliers_config: dict = self._load_definition_file_with_fallback(
self.SUPPLIERS_CONFIG_FILENAME, None # Suppliers config is flat
)
# 7. Load preset settings (conceptually overrides combined base + user for shared keys) # 5. Load preset settings (from user config dir, with fallback from bundled)
self._preset_settings: dict = self._load_preset(preset_name) self._preset_settings: dict = self._load_preset_with_fallback(self._preset_filename_stem)
# 8. Validate and compile (after all base/user/preset settings are established) self.actual_internal_preset_name = self._preset_settings.get("preset_name", self._preset_filename_stem)
log.info(f"Configuration instance: Loaded preset file '{self._preset_filename_stem}.json', internal preset_name is '{self.actual_internal_preset_name}'")
# 6. Validate and compile (after all base/user/preset settings are established)
self._validate_configs() self._validate_configs()
self._compile_regex_patterns() self._compile_regex_patterns()
log.info(f"Configuration loaded successfully using preset: '{self.preset_name}'") log.info(f"Configuration loaded successfully using preset: '{self.actual_internal_preset_name}'")
def _determine_base_dir_app_bundled(self) -> Path:
"""Determines the base directory for bundled application resources."""
if getattr(sys, 'frozen', False) and hasattr(sys, '_MEIPASS'):
# Running in a PyInstaller bundle
log.debug(f"Running as bundled app, _MEIPASS: {sys._MEIPASS}")
return Path(sys._MEIPASS)
else:
# Running as a script
log.debug(f"Running as script, using _SCRIPT_DIR: {_SCRIPT_DIR}")
return _SCRIPT_DIR
def _ensure_dir_exists(self, dir_path: Path):
"""Ensures a directory exists, creating it if necessary."""
try:
if not dir_path.exists():
log.info(f"Directory not found, creating: {dir_path}")
dir_path.mkdir(parents=True, exist_ok=True)
elif not dir_path.is_dir():
raise ConfigurationError(f"Expected directory but found file: {dir_path}")
except OSError as e:
raise ConfigurationError(f"Failed to create or access directory {dir_path}: {e}")
def _copy_default_if_missing(self, user_target_path: Path, bundled_source_subdir: str, filename: str) -> bool:
"""
Copies a default file from the bundled location to the user config directory
if it's missing in the user directory. This is for post-first-time-setup fallback.
"""
if not self.base_dir_user_config:
log.error(f"Cannot copy default for '{filename}': base_dir_user_config is not set.")
return False
if user_target_path.exists():
log.debug(f"User file '{user_target_path}' already exists. No copy needed from bundle.")
return False
# This fallback copy should NOT happen during the initial UI-driven setup phase
# where the UI is responsible for the first population of the user directory.
# It's for subsequent runs where a user might have deleted a file.
if self.is_first_run_setup:
log.debug(f"'{filename}' missing in user dir during first_run_setup phase. UI should handle initial copy. Skipping fallback copy.")
return False # File is missing, but UI should handle it.
bundled_file_path = self.base_dir_app_bundled / bundled_source_subdir / filename
if not bundled_file_path.is_file():
log.warning(f"Default bundled file '{bundled_file_path}' not found. Cannot copy to user location '{user_target_path}'.")
return False
log.warning(f"User file '{user_target_path}' is missing. Attempting to restore from bundled default: '{bundled_file_path}'.")
try:
self._ensure_dir_exists(user_target_path.parent)
shutil.copy2(bundled_file_path, user_target_path)
log.info(f"Successfully copied '{bundled_file_path}' to '{user_target_path}'.")
return True # File was copied
except Exception as e:
log.error(f"Failed to copy '{bundled_file_path}' to '{user_target_path}': {e}")
return False # Copy failed
def _load_json_file(self, file_path: Optional[Path], is_critical: bool = False, description: str = "configuration") -> dict:
"""Loads a JSON file, handling errors. Returns empty dict if not found and not critical."""
if not file_path:
if is_critical:
raise ConfigurationError(f"Critical {description} file path is not defined.")
log.debug(f"{description} file path is not defined. Returning empty dict.")
return {}
log.debug(f"Attempting to load {description} from: {file_path}")
if not file_path.is_file():
if is_critical:
raise ConfigurationError(f"Critical {description} file not found: {file_path}")
log.info(f"{description} file not found: {file_path}. Returning empty dict.")
return {}
try:
with open(file_path, 'r', encoding='utf-8') as f:
settings = json.load(f)
log.debug(f"{description} loaded successfully from {file_path}.")
return settings
except json.JSONDecodeError as e:
msg = f"Failed to parse {description} file {file_path}: Invalid JSON - {e}"
if is_critical: raise ConfigurationError(msg)
log.warning(msg + ". Returning empty dict.")
return {}
except Exception as e:
msg = f"Failed to read {description} file {file_path}: {e}"
if is_critical: raise ConfigurationError(msg)
log.warning(msg + ". Returning empty dict.")
return {}
def _load_definition_file_with_fallback(self, filename: str, root_key: Optional[str] = None) -> dict:
"""
Loads a definition JSON file from the user config subdir.
If not found and not first_run_setup, attempts to copy from bundled config subdir and then loads it.
If base_dir_user_config is not set, loads directly from bundled (read-only).
"""
data = {}
user_file_path = None
if self.base_dir_user_config:
user_file_path = self.base_dir_user_config / self.USER_CONFIG_SUBDIR_NAME / filename
data = self._load_json_file(user_file_path, is_critical=False, description=f"User {filename}")
if not data: # If not found or failed to load from user path
# Attempt fallback copy only if not in the initial setup phase by UI
# and if the file was genuinely missing (not a parse error for an existing file)
if not user_file_path.exists() and not self.is_first_run_setup:
if self._copy_default_if_missing(user_file_path, self.BASE_DIR_APP_BUNDLED_CONFIG_SUBDIR_NAME, filename):
data = self._load_json_file(user_file_path, is_critical=False, description=f"User {filename} after copy")
else:
# No user_config_dir, load directly from bundled (read-only)
log.warning(f"User config directory not set. Loading '{filename}' from bundled defaults (read-only).")
bundled_path = self.base_dir_app_bundled / self.BASE_DIR_APP_BUNDLED_CONFIG_SUBDIR_NAME / filename
data = self._load_json_file(bundled_path, is_critical=False, description=f"Bundled {filename}")
if not data:
# If still no data, it's an issue, especially for critical definitions
is_critical_def = filename in [self.ASSET_TYPE_DEFINITIONS_FILENAME, self.FILE_TYPE_DEFINITIONS_FILENAME]
err_msg = f"Failed to load '{filename}' from user dir '{user_file_path if user_file_path else 'N/A'}' or bundled defaults. Critical functionality may be affected."
if is_critical_def: raise ConfigurationError(err_msg)
log.error(err_msg)
return {}
if root_key:
if root_key not in data:
raise ConfigurationError(f"Key '{root_key}' not found in loaded {filename} data: {data.keys()}")
content = data[root_key]
# Ensure content is a dictionary if a root_key is expected to yield one
if not isinstance(content, dict):
raise ConfigurationError(f"Content under root key '{root_key}' in {filename} must be a dictionary, got {type(content)}.")
return content
return data # For flat files
def _load_preset_with_fallback(self, preset_name_stem: str) -> dict:
"""
Loads a preset JSON file from the user's Presets subdir.
If not found and not first_run_setup, attempts to copy from bundled Presets and then loads it.
If base_dir_user_config is not set, loads directly from bundled (read-only).
"""
preset_filename = f"{preset_name_stem}.json"
preset_data = {}
user_preset_file_path = None
if self.base_dir_user_config:
user_presets_dir = self.base_dir_user_config / self.USER_PRESETS_SUBDIR_NAME
user_preset_file_path = user_presets_dir / preset_filename
preset_data = self._load_json_file(user_preset_file_path, is_critical=False, description=f"User preset '{preset_filename}'")
if not preset_data: # If not found or failed to load
if not user_preset_file_path.exists() and not self.is_first_run_setup:
if self._copy_default_if_missing(user_preset_file_path, self.PRESETS_DIR_APP_BUNDLED_NAME, preset_filename):
preset_data = self._load_json_file(user_preset_file_path, is_critical=False, description=f"User preset '{preset_filename}' after copy")
else:
log.warning(f"User config directory not set. Loading preset '{preset_filename}' from bundled defaults (read-only).")
bundled_presets_dir = self.base_dir_app_bundled / self.PRESETS_DIR_APP_BUNDLED_NAME
bundled_preset_file_path = bundled_presets_dir / preset_filename
# Presets are generally critical for operation if one is specified
preset_data = self._load_json_file(bundled_preset_file_path, is_critical=True, description=f"Bundled preset '{preset_filename}'")
if not preset_data:
raise ConfigurationError(f"Preset file '{preset_filename}' could not be loaded from user dir '{user_preset_file_path if user_preset_file_path else 'N/A'}' or bundled defaults.")
return preset_data
def _compile_regex_patterns(self): def _compile_regex_patterns(self):
@@ -143,8 +362,8 @@ class Configuration:
self.compiled_extra_regex: list[re.Pattern] = [] self.compiled_extra_regex: list[re.Pattern] = []
self.compiled_model_regex: list[re.Pattern] = [] self.compiled_model_regex: list[re.Pattern] = []
self.compiled_bit_depth_regex_map: dict[str, re.Pattern] = {} self.compiled_bit_depth_regex_map: dict[str, re.Pattern] = {}
# Map: base_map_type -> list of tuples: (compiled_regex, original_keyword, rule_index) # Map: base_map_type -> list of tuples: (compiled_regex, original_keyword, rule_index, is_priority)
self.compiled_map_keyword_regex: dict[str, list[tuple[re.Pattern, str, int]]] = {} self.compiled_map_keyword_regex: dict[str, list[tuple[re.Pattern, str, int, bool]]] = {}
for pattern in self.move_to_extra_patterns: for pattern in self.move_to_extra_patterns:
try: try:
@@ -179,28 +398,53 @@ class Configuration:
for rule_index, mapping_rule in enumerate(self.map_type_mapping): for rule_index, mapping_rule in enumerate(self.map_type_mapping):
if not isinstance(mapping_rule, dict) or \ if not isinstance(mapping_rule, dict) or \
'target_type' not in mapping_rule or \ 'target_type' not in mapping_rule: # Removed 'keywords' check here as it's handled below
'keywords' not in mapping_rule or \ log.warning(f"Skipping invalid map_type_mapping rule at index {rule_index}: {mapping_rule}. Expected dict with 'target_type'.")
not isinstance(mapping_rule['keywords'], list):
log.warning(f"Skipping invalid map_type_mapping rule at index {rule_index}: {mapping_rule}. Expected dict with 'target_type' and 'keywords' list.")
continue continue
target_type = mapping_rule['target_type'].upper() target_type = mapping_rule['target_type'].upper()
source_keywords = mapping_rule['keywords']
# Ensure 'keywords' exists and is a list, default to empty list if not found or not a list
regular_keywords = mapping_rule.get('keywords', [])
if not isinstance(regular_keywords, list):
log.warning(f"Rule {rule_index} for target '{target_type}' has 'keywords' but it's not a list. Treating as empty.")
regular_keywords = []
for keyword in source_keywords: priority_keywords = mapping_rule.get('priority_keywords', []) # Optional, defaults to empty list
if not isinstance(priority_keywords, list):
log.warning(f"Rule {rule_index} for target '{target_type}' has 'priority_keywords' but it's not a list. Treating as empty.")
priority_keywords = []
# Process regular keywords
for keyword in regular_keywords:
if not isinstance(keyword, str): if not isinstance(keyword, str):
log.warning(f"Skipping non-string keyword '{keyword}' in rule {rule_index} for target '{target_type}'.") log.warning(f"Skipping non-string regular keyword '{keyword}' in rule {rule_index} for target '{target_type}'.")
continue
try:
kw_regex_part = _fnmatch_to_regex(keyword)
# Ensure the keyword is treated as a whole word or is at the start/end of a segment
regex_str = rf"(?:^|{separator})({kw_regex_part})(?:$|{separator})"
compiled_regex = re.compile(regex_str, re.IGNORECASE)
# Add False for is_priority
temp_compiled_map_regex[target_type].append((compiled_regex, keyword, rule_index, False))
log.debug(f" Compiled regular keyword '{keyword}' (rule {rule_index}) for target '{target_type}' as regex: {regex_str}")
except re.error as e:
log.warning(f"Failed to compile regular map keyword regex '{keyword}' for target type '{target_type}': {e}. Skipping keyword.")
# Process priority keywords
for keyword in priority_keywords:
if not isinstance(keyword, str):
log.warning(f"Skipping non-string priority keyword '{keyword}' in rule {rule_index} for target '{target_type}'.")
continue continue
try: try:
kw_regex_part = _fnmatch_to_regex(keyword) kw_regex_part = _fnmatch_to_regex(keyword)
regex_str = rf"(?:^|{separator})({kw_regex_part})(?:$|{separator})" regex_str = rf"(?:^|{separator})({kw_regex_part})(?:$|{separator})"
compiled_regex = re.compile(regex_str, re.IGNORECASE) compiled_regex = re.compile(regex_str, re.IGNORECASE)
temp_compiled_map_regex[target_type].append((compiled_regex, keyword, rule_index)) # Add True for is_priority
log.debug(f" Compiled keyword '{keyword}' (rule {rule_index}) for target '{target_type}' as regex: {regex_str}") temp_compiled_map_regex[target_type].append((compiled_regex, keyword, rule_index, True))
log.debug(f" Compiled priority keyword '{keyword}' (rule {rule_index}) for target '{target_type}' as regex: {regex_str}")
except re.error as e: except re.error as e:
log.warning(f"Failed to compile map keyword regex '{keyword}' for target type '{target_type}': {e}. Skipping keyword.") log.warning(f"Failed to compile priority map keyword regex '{keyword}' for target type '{target_type}': {e}. Skipping keyword.")
self.compiled_map_keyword_regex = dict(temp_compiled_map_regex) self.compiled_map_keyword_regex = dict(temp_compiled_map_regex)
log.debug(f"Compiled map keyword regex keys: {list(self.compiled_map_keyword_regex.keys())}") log.debug(f"Compiled map keyword regex keys: {list(self.compiled_map_keyword_regex.keys())}")
@@ -208,118 +452,6 @@ class Configuration:
log.debug("Finished compiling regex patterns.") log.debug("Finished compiling regex patterns.")
def _load_core_config(self) -> dict:
"""Loads settings from the core app_settings.json file."""
log.debug(f"Loading core config from: {APP_SETTINGS_PATH}")
if not APP_SETTINGS_PATH.is_file():
raise ConfigurationError(f"Core configuration file not found: {APP_SETTINGS_PATH}")
try:
with open(APP_SETTINGS_PATH, 'r', encoding='utf-8') as f:
settings = json.load(f)
log.debug(f"Core config loaded successfully.")
return settings
except json.JSONDecodeError as e:
raise ConfigurationError(f"Failed to parse core configuration file {APP_SETTINGS_PATH}: Invalid JSON - {e}")
except Exception as e:
raise ConfigurationError(f"Failed to read core configuration file {APP_SETTINGS_PATH}: {e}")
def _load_llm_config(self) -> dict:
"""Loads settings from the llm_settings.json file."""
log.debug(f"Loading LLM config from: {LLM_SETTINGS_PATH}")
if not LLM_SETTINGS_PATH.is_file():
# Log a warning but don't raise an error, allow fallback if possible
log.warning(f"LLM configuration file not found: {LLM_SETTINGS_PATH}. LLM features might be disabled or use defaults.")
return {}
try:
with open(LLM_SETTINGS_PATH, 'r', encoding='utf-8') as f:
settings = json.load(f)
log.debug(f"LLM config loaded successfully.")
return settings
except json.JSONDecodeError as e:
log.error(f"Failed to parse LLM configuration file {LLM_SETTINGS_PATH}: Invalid JSON - {e}")
return {}
except Exception as e:
log.error(f"Failed to read LLM configuration file {LLM_SETTINGS_PATH}: {e}")
return {}
def _load_preset(self, preset_name: str) -> dict:
"""Loads the specified preset JSON file."""
log.debug(f"Loading preset: '{preset_name}' from {PRESETS_DIR}")
if not PRESETS_DIR.is_dir():
raise ConfigurationError(f"Presets directory not found: {PRESETS_DIR}")
preset_file = PRESETS_DIR / f"{preset_name}.json"
if not preset_file.is_file():
raise ConfigurationError(f"Preset file not found: {preset_file}")
try:
with open(preset_file, 'r', encoding='utf-8') as f:
preset_data = json.load(f)
log.debug(f"Preset '{preset_name}' loaded successfully.")
return preset_data
except json.JSONDecodeError as e:
raise ConfigurationError(f"Failed to parse preset file {preset_file}: Invalid JSON - {e}")
except Exception as e:
raise ConfigurationError(f"Failed to read preset file {preset_file}: {e}")
def _load_asset_type_definitions(self) -> dict:
"""Loads asset type definitions from the asset_type_definitions.json file."""
log.debug(f"Loading asset type definitions from: {ASSET_TYPE_DEFINITIONS_PATH}")
if not ASSET_TYPE_DEFINITIONS_PATH.is_file():
raise ConfigurationError(f"Asset type definitions file not found: {ASSET_TYPE_DEFINITIONS_PATH}")
try:
with open(ASSET_TYPE_DEFINITIONS_PATH, 'r', encoding='utf-8') as f:
data = json.load(f)
if "ASSET_TYPE_DEFINITIONS" not in data:
raise ConfigurationError(f"Key 'ASSET_TYPE_DEFINITIONS' not found in {ASSET_TYPE_DEFINITIONS_PATH}")
settings = data["ASSET_TYPE_DEFINITIONS"]
if not isinstance(settings, dict):
raise ConfigurationError(f"'ASSET_TYPE_DEFINITIONS' in {ASSET_TYPE_DEFINITIONS_PATH} must be a dictionary.")
log.debug(f"Asset type definitions loaded successfully.")
return settings
except json.JSONDecodeError as e:
raise ConfigurationError(f"Failed to parse asset type definitions file {ASSET_TYPE_DEFINITIONS_PATH}: Invalid JSON - {e}")
except Exception as e:
raise ConfigurationError(f"Failed to read asset type definitions file {ASSET_TYPE_DEFINITIONS_PATH}: {e}")
def _load_file_type_definitions(self) -> dict:
"""Loads file type definitions from the file_type_definitions.json file."""
log.debug(f"Loading file type definitions from: {FILE_TYPE_DEFINITIONS_PATH}")
if not FILE_TYPE_DEFINITIONS_PATH.is_file():
raise ConfigurationError(f"File type definitions file not found: {FILE_TYPE_DEFINITIONS_PATH}")
try:
with open(FILE_TYPE_DEFINITIONS_PATH, 'r', encoding='utf-8') as f:
data = json.load(f)
if "FILE_TYPE_DEFINITIONS" not in data:
raise ConfigurationError(f"Key 'FILE_TYPE_DEFINITIONS' not found in {FILE_TYPE_DEFINITIONS_PATH}")
settings = data["FILE_TYPE_DEFINITIONS"]
if not isinstance(settings, dict):
raise ConfigurationError(f"'FILE_TYPE_DEFINITIONS' in {FILE_TYPE_DEFINITIONS_PATH} must be a dictionary.")
log.debug(f"File type definitions loaded successfully.")
return settings
except json.JSONDecodeError as e:
raise ConfigurationError(f"Failed to parse file type definitions file {FILE_TYPE_DEFINITIONS_PATH}: Invalid JSON - {e}")
except Exception as e:
raise ConfigurationError(f"Failed to read file type definitions file {FILE_TYPE_DEFINITIONS_PATH}: {e}")
def _load_user_settings(self) -> dict:
"""Loads user override settings from config/user_settings.json."""
log.debug(f"Attempting to load user settings from: {USER_SETTINGS_PATH}")
if not USER_SETTINGS_PATH.is_file():
log.info(f"User settings file not found: {USER_SETTINGS_PATH}. Proceeding without user overrides.")
return {}
try:
with open(USER_SETTINGS_PATH, 'r', encoding='utf-8') as f:
settings = json.load(f)
log.info(f"User settings loaded successfully from {USER_SETTINGS_PATH}.")
return settings
except json.JSONDecodeError as e:
log.warning(f"Failed to parse user settings file {USER_SETTINGS_PATH}: Invalid JSON - {e}. Using empty user settings.")
return {}
except Exception as e:
log.warning(f"Failed to read user settings file {USER_SETTINGS_PATH}: {e}. Using empty user settings.")
return {}
def _validate_configs(self): def _validate_configs(self):
"""Performs basic validation checks on loaded settings.""" """Performs basic validation checks on loaded settings."""
@@ -343,31 +475,43 @@ class Configuration:
] ]
for key in required_preset_keys: for key in required_preset_keys:
if key not in self._preset_settings: if key not in self._preset_settings:
raise ConfigurationError(f"Preset '{self.preset_name}' is missing required key: '{key}'.") raise ConfigurationError(f"Preset file '{self._preset_filename_stem}.json' (internal name: '{self.actual_internal_preset_name}') is missing required key: '{key}'.")
# Validate map_type_mapping structure (new format) # Validate map_type_mapping structure (new format)
if not isinstance(self._preset_settings['map_type_mapping'], list): if not isinstance(self._preset_settings['map_type_mapping'], list):
raise ConfigurationError(f"Preset '{self.preset_name}': 'map_type_mapping' must be a list.") raise ConfigurationError(f"Preset file '{self._preset_filename_stem}.json': 'map_type_mapping' must be a list.")
for index, rule in enumerate(self._preset_settings['map_type_mapping']): for index, rule in enumerate(self._preset_settings['map_type_mapping']):
if not isinstance(rule, dict): if not isinstance(rule, dict):
raise ConfigurationError(f"Preset '{self.preset_name}': Rule at index {index} in 'map_type_mapping' must be a dictionary.") raise ConfigurationError(f"Preset file '{self._preset_filename_stem}.json': Rule at index {index} in 'map_type_mapping' must be a dictionary.")
if 'target_type' not in rule or not isinstance(rule['target_type'], str): if 'target_type' not in rule or not isinstance(rule['target_type'], str):
raise ConfigurationError(f"Preset '{self.preset_name}': Rule at index {index} in 'map_type_mapping' is missing 'target_type' string.") raise ConfigurationError(f"Preset file '{self._preset_filename_stem}.json': Rule at index {index} in 'map_type_mapping' is missing 'target_type' string.")
valid_file_type_keys = self._file_type_definitions.keys() valid_file_type_keys = self._file_type_definitions.keys()
if rule['target_type'] not in valid_file_type_keys: if rule['target_type'] not in valid_file_type_keys:
raise ConfigurationError( raise ConfigurationError(
f"Preset '{self.preset_name}': Rule at index {index} in 'map_type_mapping' " f"Preset file '{self._preset_filename_stem}.json': Rule at index {index} in 'map_type_mapping' "
f"has an invalid 'target_type': '{rule['target_type']}'. " f"has an invalid 'target_type': '{rule['target_type']}'. "
f"Must be one of {list(valid_file_type_keys)}." f"Must be one of {list(valid_file_type_keys)}."
) )
if 'keywords' not in rule or not isinstance(rule['keywords'], list): # 'keywords' is optional if 'priority_keywords' is present and not empty,
raise ConfigurationError(f"Preset '{self.preset_name}': Rule at index {index} in 'map_type_mapping' is missing 'keywords' list.") # but if 'keywords' IS present, it must be a list of strings.
if 'keywords' in rule:
if not isinstance(rule['keywords'], list):
raise ConfigurationError(f"Preset file '{self._preset_filename_stem}.json': Rule at index {index} in 'map_type_mapping' has 'keywords' but it's not a list.")
for kw_index, keyword in enumerate(rule['keywords']): for kw_index, keyword in enumerate(rule['keywords']):
if not isinstance(keyword, str): if not isinstance(keyword, str):
raise ConfigurationError(f"Preset '{self.preset_name}': Keyword at index {kw_index} in rule {index} ('{rule['target_type']}') must be a string.") raise ConfigurationError(f"Preset file '{self._preset_filename_stem}.json': Keyword at index {kw_index} in rule {index} ('{rule['target_type']}') must be a string.")
elif not ('priority_keywords' in rule and rule['priority_keywords']): # if 'keywords' is not present, 'priority_keywords' must be
raise ConfigurationError(f"Preset file '{self._preset_filename_stem}.json': Rule at index {index} in 'map_type_mapping' must have 'keywords' or non-empty 'priority_keywords'.")
# Validate priority_keywords if present
if 'priority_keywords' in rule:
if not isinstance(rule['priority_keywords'], list):
raise ConfigurationError(f"Preset file '{self._preset_filename_stem}.json': Rule at index {index} in 'map_type_mapping' has 'priority_keywords' but it's not a list.")
for prio_kw_index, prio_keyword in enumerate(rule['priority_keywords']):
if not isinstance(prio_keyword, str):
raise ConfigurationError(f"Preset file '{self._preset_filename_stem}.json': Priority keyword at index {prio_kw_index} in rule {index} ('{rule['target_type']}') must be a string.")
if not isinstance(self._core_settings.get('TARGET_FILENAME_PATTERN'), str): if not isinstance(self._core_settings.get('TARGET_FILENAME_PATTERN'), str):
raise ConfigurationError("Core config 'TARGET_FILENAME_PATTERN' must be a string.") raise ConfigurationError("Core config 'TARGET_FILENAME_PATTERN' must be a string.")
@@ -404,9 +548,20 @@ class Configuration:
@property @property
def supplier_name(self) -> str: def supplier_name(self) -> str: # From preset
return self._preset_settings.get('supplier_name', 'DefaultSupplier') return self._preset_settings.get('supplier_name', 'DefaultSupplier')
@property
def suppliers_config(self) -> dict: # From suppliers.json
"""Returns the loaded suppliers configuration."""
return self._suppliers_config
@property
def internal_display_preset_name(self) -> str:
"""Returns the 'preset_name' field from within the loaded preset JSON,
or falls back to the filename stem if not present."""
return self.actual_internal_preset_name
@property @property
def default_asset_category(self) -> str: def default_asset_category(self) -> str:
"""Gets the default asset category from core settings.""" """Gets the default asset category from core settings."""
@@ -669,9 +824,64 @@ class Configuration:
return self._core_settings.get('LOW_RESOLUTION_THRESHOLD', 512) return self._core_settings.get('LOW_RESOLUTION_THRESHOLD', 512)
@property @property
def FILE_TYPE_DEFINITIONS(self) -> dict: def FILE_TYPE_DEFINITIONS(self) -> dict: # Kept for compatibility if used directly
return self._file_type_definitions return self._file_type_definitions
# --- Save Methods ---
def _save_json_to_user_config(self, data_to_save: dict, filename: str, subdir: Optional[str] = None, is_root_key_data: Optional[str] = None):
"""Helper to save a dictionary to a JSON file in the user config directory."""
if not self.base_dir_user_config:
raise ConfigurationError(f"Cannot save {filename}: User config directory (base_dir_user_config) is not set.")
target_dir = self.base_dir_user_config
if subdir:
target_dir = target_dir / subdir
self._ensure_dir_exists(target_dir)
path = target_dir / filename
data_for_json = {is_root_key_data: data_to_save} if is_root_key_data else data_to_save
log.debug(f"Saving data to: {path}")
try:
with open(path, 'w', encoding='utf-8') as f:
json.dump(data_for_json, f, indent=4)
log.info(f"Data saved successfully to {path}")
except Exception as e:
log.error(f"Failed to save file {path}: {e}")
raise ConfigurationError(f"Failed to save {filename}: {e}")
def save_user_settings(self, settings_dict: dict):
"""Saves the provided settings dictionary to user_settings.json in the user config directory."""
self._save_json_to_user_config(settings_dict, self.USER_SETTINGS_FILENAME)
def save_llm_settings(self, settings_dict: dict):
"""Saves LLM settings to the user config directory's 'config' subdir."""
self._save_json_to_user_config(settings_dict, self.LLM_SETTINGS_FILENAME, subdir=self.USER_CONFIG_SUBDIR_NAME)
def save_asset_type_definitions(self, data: dict):
"""Saves asset type definitions to the user config directory's 'config' subdir."""
self._save_json_to_user_config(data, self.ASSET_TYPE_DEFINITIONS_FILENAME, subdir=self.USER_CONFIG_SUBDIR_NAME, is_root_key_data="ASSET_TYPE_DEFINITIONS")
def save_file_type_definitions(self, data: dict):
"""Saves file type definitions to the user config directory's 'config' subdir."""
self._save_json_to_user_config(data, self.FILE_TYPE_DEFINITIONS_FILENAME, subdir=self.USER_CONFIG_SUBDIR_NAME, is_root_key_data="FILE_TYPE_DEFINITIONS")
def save_supplier_settings(self, data: dict):
"""Saves supplier settings to the user config directory's 'config' subdir."""
self._save_json_to_user_config(data, self.SUPPLIERS_CONFIG_FILENAME, subdir=self.USER_CONFIG_SUBDIR_NAME)
def save_preset(self, preset_data: dict, preset_name_stem: str):
"""Saves a preset to the user config directory's 'Presets' subdir."""
if not preset_name_stem:
raise ConfigurationError("Preset name stem cannot be empty for saving.")
preset_filename = f"{preset_name_stem}.json"
# Ensure the preset_data itself contains the correct 'preset_name' field
# or update it before saving if necessary.
# For example: preset_data['preset_name'] = preset_name_stem
self._save_json_to_user_config(preset_data, preset_filename, subdir=self.USER_PRESETS_SUBDIR_NAME)
@property @property
def keybind_config(self) -> dict[str, list[str]]: def keybind_config(self) -> dict[str, list[str]]:
""" """
@@ -695,275 +905,60 @@ class Configuration:
# For now, we rely on the order they appear in the config. # For now, we rely on the order they appear in the config.
return keybinds return keybinds
def load_base_config() -> dict: # The global load_base_config() is effectively replaced by Configuration.__init__
""" # Global save/load functions for individual files are refactored to be methods
Loads base configuration by merging app_settings.json, user_settings.json (if exists), # of the Configuration class or called by them, using instance paths.
asset_type_definitions.json, and file_type_definitions.json.
Does not load presets or perform full validation beyond basic file loading.
Returns a dictionary containing the merged settings. If app_settings.json
fails to load, an empty dictionary is returned. If other files
fail, errors are logged, and the function proceeds with what has been loaded.
"""
base_settings = {}
# 1. Load app_settings.json (critical) # For example, to get a list of preset names, one might need a static method
if not APP_SETTINGS_PATH.is_file(): # or a function that knows about both bundled and user preset directories.
log.error(f"Critical: Base application settings file not found: {APP_SETTINGS_PATH}. Returning empty configuration.") def get_available_preset_names(base_dir_user_config: Optional[Path], base_dir_app_bundled: Path) -> list[str]:
return {} """
try: Gets a list of available preset names (stems) by looking in user presets
with open(APP_SETTINGS_PATH, 'r', encoding='utf-8') as f: and then bundled presets. User presets take precedence.
base_settings = json.load(f) """
log.info(f"Successfully loaded base application settings from: {APP_SETTINGS_PATH}") preset_names = set()
except json.JSONDecodeError as e:
log.error(f"Critical: Failed to parse base application settings file {APP_SETTINGS_PATH}: Invalid JSON - {e}. Returning empty configuration.")
return {}
except Exception as e:
log.error(f"Critical: Failed to read base application settings file {APP_SETTINGS_PATH}: {e}. Returning empty configuration.")
return {}
# 2. Attempt to load user_settings.json # Check user presets first
user_settings_overrides = {} if base_dir_user_config:
if USER_SETTINGS_PATH.is_file(): user_presets_dir = base_dir_user_config / Configuration.USER_PRESETS_SUBDIR_NAME
try: if user_presets_dir.is_dir():
with open(USER_SETTINGS_PATH, 'r', encoding='utf-8') as f: for f in user_presets_dir.glob("*.json"):
user_settings_overrides = json.load(f) preset_names.add(f.stem)
log.info(f"User settings loaded successfully for base_config from {USER_SETTINGS_PATH}.")
except json.JSONDecodeError as e:
log.warning(f"Failed to parse user settings file {USER_SETTINGS_PATH} for base_config: Invalid JSON - {e}. Proceeding without these user overrides.")
except Exception as e:
log.warning(f"Failed to read user settings file {USER_SETTINGS_PATH} for base_config: {e}. Proceeding without these user overrides.")
# 3. Deep merge user settings onto base_settings # Check bundled presets
if user_settings_overrides: bundled_presets_dir = base_dir_app_bundled / Configuration.PRESETS_DIR_APP_BUNDLED_NAME
log.info("Applying user setting overrides to base_settings in load_base_config.") if bundled_presets_dir.is_dir():
# _deep_merge_dicts modifies base_settings in place for f in bundled_presets_dir.glob("*.json"):
_deep_merge_dicts(base_settings, user_settings_overrides) preset_names.add(f.stem) # Adds if not already present from user dir
# 4. Load asset_type_definitions.json (non-critical, merge if successful) if not preset_names:
if not ASSET_TYPE_DEFINITIONS_PATH.is_file(): log.warning("No preset files found in user or bundled preset directories.")
log.error(f"Asset type definitions file not found: {ASSET_TYPE_DEFINITIONS_PATH}. Proceeding without it.") # Consider adding a default/template preset if none are found, or ensure one always exists in bundle.
else: # For now, return empty list.
try:
with open(ASSET_TYPE_DEFINITIONS_PATH, 'r', encoding='utf-8') as f:
asset_defs_data = json.load(f)
if "ASSET_TYPE_DEFINITIONS" in asset_defs_data:
if isinstance(asset_defs_data["ASSET_TYPE_DEFINITIONS"], dict):
# Merge into base_settings, which might already contain user overrides
base_settings['ASSET_TYPE_DEFINITIONS'] = asset_defs_data["ASSET_TYPE_DEFINITIONS"]
log.info(f"Successfully loaded and merged ASSET_TYPE_DEFINITIONS from: {ASSET_TYPE_DEFINITIONS_PATH}")
else:
log.error(f"Value under 'ASSET_TYPE_DEFINITIONS' in {ASSET_TYPE_DEFINITIONS_PATH} is not a dictionary. Skipping merge.")
else:
log.error(f"Key 'ASSET_TYPE_DEFINITIONS' not found in {ASSET_TYPE_DEFINITIONS_PATH}. Skipping merge.")
except json.JSONDecodeError as e:
log.error(f"Failed to parse asset type definitions file {ASSET_TYPE_DEFINITIONS_PATH}: Invalid JSON - {e}. Skipping merge.")
except Exception as e:
log.error(f"Failed to read asset type definitions file {ASSET_TYPE_DEFINITIONS_PATH}: {e}. Skipping merge.")
# 5. Load file_type_definitions.json (non-critical, merge if successful) return sorted(list(preset_names))
if not FILE_TYPE_DEFINITIONS_PATH.is_file():
log.error(f"File type definitions file not found: {FILE_TYPE_DEFINITIONS_PATH}. Proceeding without it.")
else:
try:
with open(FILE_TYPE_DEFINITIONS_PATH, 'r', encoding='utf-8') as f:
file_defs_data = json.load(f)
if "FILE_TYPE_DEFINITIONS" in file_defs_data:
if isinstance(file_defs_data["FILE_TYPE_DEFINITIONS"], dict):
# Merge into base_settings
base_settings['FILE_TYPE_DEFINITIONS'] = file_defs_data["FILE_TYPE_DEFINITIONS"]
log.info(f"Successfully loaded and merged FILE_TYPE_DEFINITIONS from: {FILE_TYPE_DEFINITIONS_PATH}")
else:
log.error(f"Value under 'FILE_TYPE_DEFINITIONS' in {FILE_TYPE_DEFINITIONS_PATH} is not a dictionary. Skipping merge.")
else:
log.error(f"Key 'FILE_TYPE_DEFINITIONS' not found in {FILE_TYPE_DEFINITIONS_PATH}. Skipping merge.")
except json.JSONDecodeError as e:
log.error(f"Failed to parse file type definitions file {FILE_TYPE_DEFINITIONS_PATH}: Invalid JSON - {e}. Skipping merge.")
except Exception as e:
log.error(f"Failed to read file type definitions file {FILE_TYPE_DEFINITIONS_PATH}: {e}. Skipping merge.")
return base_settings # Global functions like load_asset_definitions, save_asset_definitions etc.
# are now instance methods of the Configuration class (e.g., self.save_asset_type_definitions).
# If any external code was calling these global functions, it will need to be updated
# to instantiate a Configuration object and call its methods, or these global
# functions need to be carefully adapted to instantiate Configuration internally
# or accept a Configuration instance.
def save_llm_config(settings_dict: dict): # For now, let's assume the primary interaction is via Configuration instance.
""" # The old global functions below this point are effectively deprecated by the class methods.
Saves the provided LLM settings dictionary to llm_settings.json. # I will remove them to avoid confusion and ensure all save/load operations
""" # are managed through the Configuration instance with correct path context.
log.debug(f"Saving LLM config to: {LLM_SETTINGS_PATH}")
try:
with open(LLM_SETTINGS_PATH, 'w', encoding='utf-8') as f:
json.dump(settings_dict, f, indent=4)
# Use info level for successful save
log.info(f"LLM config saved successfully to {LLM_SETTINGS_PATH}")
except Exception as e:
log.error(f"Failed to save LLM configuration file {LLM_SETTINGS_PATH}: {e}")
# Re-raise as ConfigurationError to signal failure upstream
raise ConfigurationError(f"Failed to save LLM configuration: {e}")
def save_user_config(settings_dict: dict):
"""Saves the provided settings dictionary to user_settings.json."""
log.debug(f"Saving user config to: {USER_SETTINGS_PATH}")
try:
# Ensure parent directory exists (though 'config/' should always exist)
USER_SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True)
with open(USER_SETTINGS_PATH, 'w', encoding='utf-8') as f:
json.dump(settings_dict, f, indent=4)
log.info(f"User config saved successfully to {USER_SETTINGS_PATH}")
except Exception as e:
log.error(f"Failed to save user configuration file {USER_SETTINGS_PATH}: {e}")
raise ConfigurationError(f"Failed to save user configuration: {e}")
def save_base_config(settings_dict: dict):
"""
Saves the provided settings dictionary to app_settings.json.
"""
log.debug(f"Saving base config to: {APP_SETTINGS_PATH}")
try:
with open(APP_SETTINGS_PATH, 'w', encoding='utf-8') as f:
json.dump(settings_dict, f, indent=4)
log.debug(f"Base config saved successfully.")
except Exception as e:
log.error(f"Failed to save base configuration file {APP_SETTINGS_PATH}: {e}")
raise ConfigurationError(f"Failed to save configuration: {e}")
def load_asset_definitions() -> dict: # Removing old global load/save functions as their logic is now
""" # part of the Configuration class or replaced by its new loading/saving mechanisms.
Reads config/asset_type_definitions.json. # load_base_config() - Replaced by Configuration.__init__()
Returns the dictionary under the "ASSET_TYPE_DEFINITIONS" key. # save_llm_config(settings_dict: dict) - Replaced by Configuration.save_llm_settings()
Handles file not found or JSON errors gracefully (e.g., return empty dict, log error). # save_user_config(settings_dict: dict) - Replaced by Configuration.save_user_settings()
""" # save_base_config(settings_dict: dict) - Bundled app_settings.json should be read-only.
log.debug(f"Loading asset type definitions from: {ASSET_TYPE_DEFINITIONS_PATH}") # load_asset_definitions() -> dict - Replaced by Configuration._load_definition_file_with_fallback() logic
if not ASSET_TYPE_DEFINITIONS_PATH.is_file(): # save_asset_definitions(data: dict) - Replaced by Configuration.save_asset_type_definitions()
log.error(f"Asset type definitions file not found: {ASSET_TYPE_DEFINITIONS_PATH}") # load_file_type_definitions() -> dict - Replaced by Configuration._load_definition_file_with_fallback() logic
return {} # save_file_type_definitions(data: dict) - Replaced by Configuration.save_file_type_definitions()
try: # load_supplier_settings() -> dict - Replaced by Configuration._load_definition_file_with_fallback() logic
with open(ASSET_TYPE_DEFINITIONS_PATH, 'r', encoding='utf-8') as f: # save_supplier_settings(data: dict) - Replaced by Configuration.save_supplier_settings()
data = json.load(f)
if "ASSET_TYPE_DEFINITIONS" not in data:
log.error(f"Key 'ASSET_TYPE_DEFINITIONS' not found in {ASSET_TYPE_DEFINITIONS_PATH}")
return {}
settings = data["ASSET_TYPE_DEFINITIONS"]
if not isinstance(settings, dict):
log.error(f"'ASSET_TYPE_DEFINITIONS' in {ASSET_TYPE_DEFINITIONS_PATH} must be a dictionary.")
return {}
log.debug(f"Asset type definitions loaded successfully.")
return settings
except json.JSONDecodeError as e:
log.error(f"Failed to parse asset type definitions file {ASSET_TYPE_DEFINITIONS_PATH}: Invalid JSON - {e}")
return {}
except Exception as e:
log.error(f"Failed to read asset type definitions file {ASSET_TYPE_DEFINITIONS_PATH}: {e}")
return {}
def save_asset_definitions(data: dict):
"""
Takes a dictionary (representing the content for the "ASSET_TYPE_DEFINITIONS" key).
Writes it to config/asset_type_definitions.json under the root key "ASSET_TYPE_DEFINITIONS".
Handles potential I/O errors.
"""
log.debug(f"Saving asset type definitions to: {ASSET_TYPE_DEFINITIONS_PATH}")
try:
with open(ASSET_TYPE_DEFINITIONS_PATH, 'w', encoding='utf-8') as f:
json.dump({"ASSET_TYPE_DEFINITIONS": data}, f, indent=4)
log.info(f"Asset type definitions saved successfully to {ASSET_TYPE_DEFINITIONS_PATH}")
except Exception as e:
log.error(f"Failed to save asset type definitions file {ASSET_TYPE_DEFINITIONS_PATH}: {e}")
raise ConfigurationError(f"Failed to save asset type definitions: {e}")
def load_file_type_definitions() -> dict:
"""
Reads config/file_type_definitions.json.
Returns the dictionary under the "FILE_TYPE_DEFINITIONS" key.
Handles errors gracefully.
"""
log.debug(f"Loading file type definitions from: {FILE_TYPE_DEFINITIONS_PATH}")
if not FILE_TYPE_DEFINITIONS_PATH.is_file():
log.error(f"File type definitions file not found: {FILE_TYPE_DEFINITIONS_PATH}")
return {}
try:
with open(FILE_TYPE_DEFINITIONS_PATH, 'r', encoding='utf-8') as f:
data = json.load(f)
if "FILE_TYPE_DEFINITIONS" not in data:
log.error(f"Key 'FILE_TYPE_DEFINITIONS' not found in {FILE_TYPE_DEFINITIONS_PATH}")
return {}
settings = data["FILE_TYPE_DEFINITIONS"]
if not isinstance(settings, dict):
log.error(f"'FILE_TYPE_DEFINITIONS' in {FILE_TYPE_DEFINITIONS_PATH} must be a dictionary.")
return {}
log.debug(f"File type definitions loaded successfully.")
return settings
except json.JSONDecodeError as e:
log.error(f"Failed to parse file type definitions file {FILE_TYPE_DEFINITIONS_PATH}: Invalid JSON - {e}")
return {}
except Exception as e:
log.error(f"Failed to read file type definitions file {FILE_TYPE_DEFINITIONS_PATH}: {e}")
return {}
def save_file_type_definitions(data: dict):
"""
Takes a dictionary (representing content for "FILE_TYPE_DEFINITIONS" key).
Writes it to config/file_type_definitions.json under the root key "FILE_TYPE_DEFINITIONS".
Handles errors.
"""
log.debug(f"Saving file type definitions to: {FILE_TYPE_DEFINITIONS_PATH}")
try:
with open(FILE_TYPE_DEFINITIONS_PATH, 'w', encoding='utf-8') as f:
json.dump({"FILE_TYPE_DEFINITIONS": data}, f, indent=4)
log.info(f"File type definitions saved successfully to {FILE_TYPE_DEFINITIONS_PATH}")
except Exception as e:
log.error(f"Failed to save file type definitions file {FILE_TYPE_DEFINITIONS_PATH}: {e}")
raise ConfigurationError(f"Failed to save file type definitions: {e}")
def load_supplier_settings() -> dict:
"""
Reads config/suppliers.json.
Returns the entire dictionary.
Handles file not found (return empty dict) or JSON errors.
If the loaded data is a list (old format), convert it in memory to the new
dictionary format, defaulting normal_map_type to "OpenGL" for each supplier.
"""
log.debug(f"Loading supplier settings from: {SUPPLIERS_CONFIG_PATH}")
if not SUPPLIERS_CONFIG_PATH.is_file():
log.warning(f"Supplier settings file not found: {SUPPLIERS_CONFIG_PATH}. Returning empty dict.")
return {}
try:
with open(SUPPLIERS_CONFIG_PATH, 'r', encoding='utf-8') as f:
data = json.load(f)
if isinstance(data, list):
log.warning(f"Supplier settings in {SUPPLIERS_CONFIG_PATH} is in the old list format. Converting to new dictionary format.")
new_data = {}
for supplier_name in data:
if isinstance(supplier_name, str):
new_data[supplier_name] = {"normal_map_type": "OpenGL"}
else:
log.warning(f"Skipping non-string item '{supplier_name}' during old format conversion of supplier settings.")
log.info(f"Supplier settings converted to new format: {new_data}")
return new_data
if not isinstance(data, dict):
log.error(f"Supplier settings in {SUPPLIERS_CONFIG_PATH} must be a dictionary. Found {type(data)}. Returning empty dict.")
return {}
log.debug(f"Supplier settings loaded successfully.")
return data
except json.JSONDecodeError as e:
log.error(f"Failed to parse supplier settings file {SUPPLIERS_CONFIG_PATH}: Invalid JSON - {e}. Returning empty dict.")
return {}
except Exception as e:
log.error(f"Failed to read supplier settings file {SUPPLIERS_CONFIG_PATH}: {e}. Returning empty dict.")
return {}
def save_supplier_settings(data: dict):
"""
Takes a dictionary (in the new format).
Writes it directly to config/suppliers.json.
Handles errors.
"""
log.debug(f"Saving supplier settings to: {SUPPLIERS_CONFIG_PATH}")
if not isinstance(data, dict):
log.error(f"Data for save_supplier_settings must be a dictionary. Got {type(data)}.")
raise ConfigurationError(f"Invalid data type for saving supplier settings: {type(data)}")
try:
with open(SUPPLIERS_CONFIG_PATH, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2) # Using indent=2 as per the example for suppliers.json
log.info(f"Supplier settings saved successfully to {SUPPLIERS_CONFIG_PATH}")
except Exception as e:
log.error(f"Failed to save supplier settings file {SUPPLIERS_CONFIG_PATH}: {e}")
raise ConfigurationError(f"Failed to save supplier settings: {e}")

Binary file not shown.

View File

@@ -0,0 +1,388 @@
import sys
import os
import shutil
import json
from pathlib import Path
from typing import Optional, Tuple
from PySide6.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QPushButton,
QFileDialog, QMessageBox, QGroupBox, QFormLayout, QSpinBox, QDialogButtonBox
)
from PySide6.QtCore import Qt, Slot
# Constants for bundled resource locations relative to app base
BUNDLED_CONFIG_SUBDIR_NAME = "config"
BUNDLED_PRESETS_SUBDIR_NAME = "Presets"
DEFAULT_USER_DATA_SUBDIR_NAME = "user_data" # For portable path attempt
# Files to copy from bundled config to user config
DEFAULT_CONFIG_FILES = [
"asset_type_definitions.json",
"file_type_definitions.json",
"llm_settings.json",
"suppliers.json"
]
# app_settings.json is NOT copied. user_settings.json is handled separately.
USER_SETTINGS_FILENAME = "user_settings.json"
PERSISTENT_PATH_MARKER_FILENAME = ".first_run_complete"
PERSISTENT_CONFIG_ROOT_STORAGE_FILENAME = "asset_processor_user_root.txt" # Stores USER_CHOSEN_PATH
APP_NAME = "AssetProcessor" # Used for AppData paths
def get_app_base_dir() -> Path:
"""Determines the base directory for the application (executable or script)."""
if getattr(sys, 'frozen', False) and hasattr(sys, '_MEIPASS'):
# Running in a PyInstaller bundle
return Path(sys._MEIPASS)
else:
# Running as a script
return Path(__file__).resolve().parent.parent # Assuming this file is in gui/ subdir
def get_os_specific_app_data_dir() -> Path:
"""Gets the OS-specific application data directory."""
if sys.platform == "win32":
path_str = os.getenv('APPDATA')
if path_str:
return Path(path_str) / APP_NAME
# Fallback if APPDATA is not set, though unlikely
return Path.home() / "AppData" / "Roaming" / APP_NAME
elif sys.platform == "darwin": # macOS
return Path.home() / "Library" / "Application Support" / APP_NAME
else: # Linux and other Unix-like
return Path.home() / ".config" / APP_NAME
class FirstTimeSetupDialog(QDialog):
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowTitle("Asset Processor - First-Time Setup")
self.setModal(True)
self.setMinimumWidth(600)
self.app_base_dir = get_app_base_dir()
self.user_chosen_path: Optional[Path] = None
self._init_ui()
self._propose_default_config_path()
def _init_ui(self):
main_layout = QVBoxLayout(self)
# Configuration Path Group
config_path_group = QGroupBox("Configuration Location")
config_path_layout = QVBoxLayout()
self.proposed_path_label = QLabel("Proposed default configuration path:")
config_path_layout.addWidget(self.proposed_path_label)
path_selection_layout = QHBoxLayout()
self.config_path_edit = QLineEdit()
self.config_path_edit.setReadOnly(False) # Allow editing, then validate
path_selection_layout.addWidget(self.config_path_edit)
browse_button = QPushButton("Browse...")
browse_button.clicked.connect(self._browse_config_path)
path_selection_layout.addWidget(browse_button)
config_path_layout.addLayout(path_selection_layout)
config_path_group.setLayout(config_path_layout)
main_layout.addWidget(config_path_group)
# User Settings Group
user_settings_group = QGroupBox("Initial User Settings")
user_settings_form_layout = QFormLayout()
self.output_base_dir_edit = QLineEdit()
output_base_dir_browse_button = QPushButton("Browse...")
output_base_dir_browse_button.clicked.connect(self._browse_output_base_dir)
output_base_dir_layout = QHBoxLayout()
output_base_dir_layout.addWidget(self.output_base_dir_edit)
output_base_dir_layout.addWidget(output_base_dir_browse_button)
user_settings_form_layout.addRow("Default Library Output Path:", output_base_dir_layout)
self.output_dir_pattern_edit = QLineEdit("[supplier]/[asset_category]/[asset_name]")
user_settings_form_layout.addRow("Asset Structure Pattern:", self.output_dir_pattern_edit)
self.output_format_16bit_primary_edit = QLineEdit("png")
user_settings_form_layout.addRow("Default 16-bit Output Format (Primary):", self.output_format_16bit_primary_edit)
self.output_format_8bit_edit = QLineEdit("png")
user_settings_form_layout.addRow("Default 8-bit Output Format:", self.output_format_8bit_edit)
self.resolution_threshold_jpg_spinbox = QSpinBox()
self.resolution_threshold_jpg_spinbox.setRange(256, 16384)
self.resolution_threshold_jpg_spinbox.setValue(4096)
self.resolution_threshold_jpg_spinbox.setSuffix(" px")
user_settings_form_layout.addRow("JPG Resolution Threshold (for 8-bit):", self.resolution_threshold_jpg_spinbox)
user_settings_group.setLayout(user_settings_form_layout)
main_layout.addWidget(user_settings_group)
# Dialog Buttons
self.button_box = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel)
self.button_box.button(QDialogButtonBox.StandardButton.Ok).setText("Finish Setup")
self.button_box.accepted.connect(self._on_finish_setup)
self.button_box.rejected.connect(self.reject)
main_layout.addWidget(self.button_box)
def _propose_default_config_path(self):
proposed_path = None
# 1. Try portable path: user_data/ next to the application base dir
# If running from script, app_base_dir is .../Asset_processor_tool/gui, so parent is .../Asset_processor_tool
# If bundled, app_base_dir is the directory of the executable.
# Let's refine app_base_dir for portable path logic
# If script: Path(__file__).parent.parent = Asset_processor_tool
# If frozen: sys._MEIPASS (which is the temp extraction dir, not ideal for persistent user_data)
# A better approach for portable if frozen: Path(sys.executable).parent
current_app_dir = Path(sys.executable).parent if getattr(sys, 'frozen', False) else self.app_base_dir
portable_path_candidate = current_app_dir / DEFAULT_USER_DATA_SUBDIR_NAME
try:
portable_path_candidate.mkdir(parents=True, exist_ok=True)
if os.access(str(portable_path_candidate), os.W_OK):
proposed_path = portable_path_candidate
self.proposed_path_label.setText(f"Proposed portable path (writable):")
else:
self.proposed_path_label.setText(f"Portable path '{portable_path_candidate}' not writable.")
except Exception as e:
self.proposed_path_label.setText(f"Could not use portable path '{portable_path_candidate}': {e}")
print(f"Error checking/creating portable path: {e}") # For debugging
# 2. Fallback to OS-specific app data directory
if not proposed_path:
os_specific_path = get_os_specific_app_data_dir()
try:
os_specific_path.mkdir(parents=True, exist_ok=True)
if os.access(str(os_specific_path), os.W_OK):
proposed_path = os_specific_path
self.proposed_path_label.setText(f"Proposed standard path (writable):")
else:
self.proposed_path_label.setText(f"Standard path '{os_specific_path}' not writable. Please choose a location.")
except Exception as e:
self.proposed_path_label.setText(f"Could not use standard path '{os_specific_path}': {e}. Please choose a location.")
print(f"Error checking/creating standard path: {e}") # For debugging
if proposed_path:
self.config_path_edit.setText(str(proposed_path.resolve()))
else:
# Should not happen if OS specific path creation works, but as a last resort:
self.config_path_edit.setText(str(Path.home())) # Default to home if all else fails
QMessageBox.warning(self, "Path Issue", "Could not determine a default writable configuration path. Please select one manually.")
@Slot()
def _browse_config_path(self):
directory = QFileDialog.getExistingDirectory(
self,
"Select Configuration Directory",
self.config_path_edit.text() or str(Path.home())
)
if directory:
self.config_path_edit.setText(directory)
@Slot()
def _browse_output_base_dir(self):
directory = QFileDialog.getExistingDirectory(
self,
"Select Default Library Output Directory",
self.output_base_dir_edit.text() or str(Path.home())
)
if directory:
self.output_base_dir_edit.setText(directory)
def _validate_inputs(self) -> bool:
# Validate chosen config path
path_str = self.config_path_edit.text().strip()
if not path_str:
QMessageBox.warning(self, "Input Error", "Configuration path cannot be empty.")
return False
self.user_chosen_path = Path(path_str)
try:
self.user_chosen_path.mkdir(parents=True, exist_ok=True)
if not os.access(str(self.user_chosen_path), os.W_OK):
QMessageBox.warning(self, "Path Error", f"The chosen configuration path '{self.user_chosen_path}' is not writable.")
return False
except Exception as e:
QMessageBox.warning(self, "Path Error", f"Error with chosen configuration path '{self.user_chosen_path}': {e}")
return False
# Validate output base dir
output_base_dir_str = self.output_base_dir_edit.text().strip()
if not output_base_dir_str:
QMessageBox.warning(self, "Input Error", "Default Library Output Path cannot be empty.")
return False
try:
Path(output_base_dir_str).mkdir(parents=True, exist_ok=True) # Check if creatable
if not os.access(output_base_dir_str, os.W_OK):
QMessageBox.warning(self, "Path Error", f"The chosen output base path '{output_base_dir_str}' is not writable.")
return False
except Exception as e:
QMessageBox.warning(self, "Path Error", f"Error with output base path '{output_base_dir_str}': {e}")
return False
if not self.output_dir_pattern_edit.text().strip():
QMessageBox.warning(self, "Input Error", "Asset Structure Pattern cannot be empty.")
return False
if not self.output_format_16bit_primary_edit.text().strip():
QMessageBox.warning(self, "Input Error", "Default 16-bit Output Format cannot be empty.")
return False
if not self.output_format_8bit_edit.text().strip():
QMessageBox.warning(self, "Input Error", "Default 8-bit Output Format cannot be empty.")
return False
return True
def _copy_default_files(self):
if not self.user_chosen_path:
return
bundled_config_dir = self.app_base_dir / BUNDLED_CONFIG_SUBDIR_NAME
user_target_config_dir = self.user_chosen_path / BUNDLED_CONFIG_SUBDIR_NAME # User files also go into a 'config' subdir
try:
user_target_config_dir.mkdir(parents=True, exist_ok=True)
except Exception as e:
QMessageBox.critical(self, "Error", f"Could not create user config subdirectory '{user_target_config_dir}': {e}")
return
for filename in DEFAULT_CONFIG_FILES:
source_file = bundled_config_dir / filename
target_file = user_target_config_dir / filename
if not target_file.exists():
if source_file.is_file():
try:
shutil.copy2(str(source_file), str(target_file))
print(f"Copied '{source_file}' to '{target_file}'")
except Exception as e:
QMessageBox.warning(self, "File Copy Error", f"Could not copy '{filename}' to '{target_file}': {e}")
else:
print(f"Default config file '{source_file}' not found in bundle.")
else:
print(f"User config file '{target_file}' already exists. Skipping copy.")
# Copy Presets
bundled_presets_dir = self.app_base_dir / BUNDLED_PRESETS_SUBDIR_NAME
user_target_presets_dir = self.user_chosen_path / BUNDLED_PRESETS_SUBDIR_NAME
if bundled_presets_dir.is_dir():
try:
user_target_presets_dir.mkdir(parents=True, exist_ok=True)
for item in bundled_presets_dir.iterdir():
target_item = user_target_presets_dir / item.name
if not target_item.exists():
if item.is_file():
shutil.copy2(str(item), str(target_item))
print(f"Copied preset '{item.name}' to '{target_item}'")
# Add elif item.is_dir() for recursive copy if presets can have subdirs
except Exception as e:
QMessageBox.warning(self, "Preset Copy Error", f"Could not copy presets to '{user_target_presets_dir}': {e}")
else:
print(f"Bundled presets directory '{bundled_presets_dir}' not found.")
def _save_initial_user_settings(self):
if not self.user_chosen_path:
return
user_settings_path = self.user_chosen_path / USER_SETTINGS_FILENAME
settings_data = {}
# Load existing if it exists (though unlikely for first-time setup, but good practice)
if user_settings_path.exists():
try:
with open(user_settings_path, 'r', encoding='utf-8') as f:
settings_data = json.load(f)
except Exception as e:
QMessageBox.warning(self, "Error Loading Settings", f"Could not load existing user settings from '{user_settings_path}': {e}. Will create a new one.")
settings_data = {}
# Update with new values from dialog
settings_data['OUTPUT_BASE_DIR'] = self.output_base_dir_edit.text().strip()
settings_data['OUTPUT_DIRECTORY_PATTERN'] = self.output_dir_pattern_edit.text().strip()
settings_data['OUTPUT_FORMAT_16BIT_PRIMARY'] = self.output_format_16bit_primary_edit.text().strip().lower()
settings_data['OUTPUT_FORMAT_8BIT'] = self.output_format_8bit_edit.text().strip().lower()
settings_data['RESOLUTION_THRESHOLD_FOR_JPG'] = self.resolution_threshold_jpg_spinbox.value()
# Ensure general_settings exists for app_version if needed, or other core settings
if 'general_settings' not in settings_data:
settings_data['general_settings'] = {}
# Example: settings_data['general_settings']['some_new_user_setting'] = True
try:
with open(user_settings_path, 'w', encoding='utf-8') as f:
json.dump(settings_data, f, indent=4)
print(f"Saved user settings to '{user_settings_path}'")
except Exception as e:
QMessageBox.critical(self, "Error Saving Settings", f"Could not save user settings to '{user_settings_path}': {e}")
def _save_persistent_info(self):
if not self.user_chosen_path:
return
# 1. Save USER_CHOSEN_PATH to a persistent location (e.g., AppData)
persistent_storage_dir = get_os_specific_app_data_dir()
try:
persistent_storage_dir.mkdir(parents=True, exist_ok=True)
persistent_path_file = persistent_storage_dir / PERSISTENT_CONFIG_ROOT_STORAGE_FILENAME
with open(persistent_path_file, 'w', encoding='utf-8') as f:
f.write(str(self.user_chosen_path.resolve()))
print(f"Saved chosen config path to '{persistent_path_file}'")
except Exception as e:
QMessageBox.warning(self, "Error Saving Path", f"Could not persistently save the chosen configuration path: {e}")
# This is not critical enough to stop the setup, but user might need to re-select on next launch.
# 2. Create marker file in USER_CHOSEN_PATH
marker_file = self.user_chosen_path / PERSISTENT_PATH_MARKER_FILENAME
try:
with open(marker_file, 'w', encoding='utf-8') as f:
f.write("Asset Processor first-time setup complete.")
print(f"Created marker file at '{marker_file}'")
except Exception as e:
QMessageBox.warning(self, "Error Creating Marker", f"Could not create first-run marker file at '{marker_file}': {e}")
@Slot()
def _on_finish_setup(self):
if not self._validate_inputs():
return
# Confirmation before proceeding
reply = QMessageBox.question(self, "Confirm Setup",
f"The following path will be used for configuration and user data:\n"
f"{self.user_chosen_path}\n\n"
f"Default configuration files and presets will be copied if they don't exist.\n"
f"Initial user settings will be saved.\n\nProceed with setup?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No)
if reply == QMessageBox.StandardButton.No:
return
try:
self._copy_default_files()
self._save_initial_user_settings()
self._save_persistent_info()
QMessageBox.information(self, "Setup Complete", "First-time setup completed successfully!")
self.accept()
except Exception as e:
QMessageBox.critical(self, "Setup Error", f"An unexpected error occurred during setup: {e}")
# Optionally, attempt cleanup or guide user
def get_chosen_config_path(self) -> Optional[Path]:
"""Returns the path chosen by the user after successful completion."""
if self.result() == QDialog.DialogCode.Accepted:
return self.user_chosen_path
return None
if __name__ == '__main__':
from PySide6.QtWidgets import QApplication
app = QApplication(sys.argv)
dialog = FirstTimeSetupDialog()
if dialog.exec():
chosen_path = dialog.get_chosen_config_path()
print(f"Dialog accepted. Chosen config path: {chosen_path}")
else:
print("Dialog cancelled.")
sys.exit()

View File

@@ -311,7 +311,7 @@ class MainWindow(QMainWindow):
log.info(f"Added {added_count} new asset paths: {newly_added_paths}") log.info(f"Added {added_count} new asset paths: {newly_added_paths}")
self.statusBar().showMessage(f"Added {added_count} asset(s). Updating preview...", 3000) self.statusBar().showMessage(f"Added {added_count} asset(s). Updating preview...", 3000)
mode, selected_preset_text = self.preset_editor_widget.get_selected_preset_mode() mode, selected_display_name, preset_file_path = self.preset_editor_widget.get_selected_preset_mode()
if mode == "llm": if mode == "llm":
log.info(f"LLM Interpretation selected. Preparing LLM prediction for {len(newly_added_paths)} new paths.") log.info(f"LLM Interpretation selected. Preparing LLM prediction for {len(newly_added_paths)} new paths.")
@@ -330,8 +330,9 @@ class MainWindow(QMainWindow):
log.info(f"Delegating {len(llm_requests_to_queue)} LLM requests to the handler.") log.info(f"Delegating {len(llm_requests_to_queue)} LLM requests to the handler.")
self.llm_interaction_handler.queue_llm_requests_batch(llm_requests_to_queue) self.llm_interaction_handler.queue_llm_requests_batch(llm_requests_to_queue)
# The handler manages starting its own processing internally. # The handler manages starting its own processing internally.
elif mode == "preset" and selected_preset_text: elif mode == "preset" and selected_display_name and preset_file_path:
log.info(f"Preset '{selected_preset_text}' selected. Triggering prediction for {len(newly_added_paths)} new paths.") preset_name_for_loading = preset_file_path.stem
log.info(f"Preset '{selected_display_name}' (file: {preset_name_for_loading}.json) selected. Triggering prediction for {len(newly_added_paths)} new paths.")
if self.prediction_thread and not self.prediction_thread.isRunning(): if self.prediction_thread and not self.prediction_thread.isRunning():
log.debug("Starting prediction thread from add_input_paths.") log.debug("Starting prediction thread from add_input_paths.")
self.prediction_thread.start() self.prediction_thread.start()
@@ -343,7 +344,8 @@ class MainWindow(QMainWindow):
self._source_file_lists[input_path_str] = file_list self._source_file_lists[input_path_str] = file_list
self._pending_predictions.add(input_path_str) self._pending_predictions.add(input_path_str)
log.debug(f"Added '{input_path_str}' to pending predictions. Current pending: {self._pending_predictions}") log.debug(f"Added '{input_path_str}' to pending predictions. Current pending: {self._pending_predictions}")
self.start_prediction_signal.emit(input_path_str, file_list, selected_preset_text) # Pass the filename stem for loading, not the display name
self.start_prediction_signal.emit(input_path_str, file_list, preset_name_for_loading)
else: else:
log.warning(f"Skipping prediction for {input_path_str} due to extraction error.") log.warning(f"Skipping prediction for {input_path_str} due to extraction error.")
elif mode == "placeholder": elif mode == "placeholder":
@@ -446,7 +448,12 @@ class MainWindow(QMainWindow):
self.statusBar().showMessage("No assets added to process.", 3000) self.statusBar().showMessage("No assets added to process.", 3000)
return return
mode, selected_preset_name = self.preset_editor_widget.get_selected_preset_mode() # mode, selected_preset_name, preset_file_path are relevant here if processing depends on the *loaded* preset's config
# For now, _on_process_requested uses the rules already in unified_model, which should have been generated
# using the correct preset context. The preset name itself isn't directly used by the processing engine,
# as the SourceRule object already contains the necessary preset-derived information or the preset name string.
# We'll rely on the SourceRule objects in unified_model.get_all_source_rules() to be correct.
# mode, selected_display_name, preset_file_path = self.preset_editor_widget.get_selected_preset_mode()
output_dir_str = settings.get("output_dir") output_dir_str = settings.get("output_dir")
@@ -694,7 +701,7 @@ class MainWindow(QMainWindow):
log.error("RuleBasedPredictionHandler not loaded. Cannot update preview.") log.error("RuleBasedPredictionHandler not loaded. Cannot update preview.")
self.statusBar().showMessage("Error: Prediction components not loaded.", 5000) self.statusBar().showMessage("Error: Prediction components not loaded.", 5000)
return return
mode, selected_preset_name = self.preset_editor_widget.get_selected_preset_mode() mode, selected_display_name, preset_file_path = self.preset_editor_widget.get_selected_preset_mode()
if mode == "placeholder": if mode == "placeholder":
log.debug("Update preview called with placeholder preset selected. Showing existing raw inputs (detailed view).") log.debug("Update preview called with placeholder preset selected. Showing existing raw inputs (detailed view).")
@@ -749,9 +756,10 @@ class MainWindow(QMainWindow):
# Do not return here; let the function exit normally after handling LLM case. # Do not return here; let the function exit normally after handling LLM case.
# The standard prediction path below will be skipped because mode is 'llm'. # The standard prediction path below will be skipped because mode is 'llm'.
elif mode == "preset" and selected_preset_name: elif mode == "preset" and selected_display_name and preset_file_path:
log.info(f"[{time.time():.4f}] Requesting background preview update for {len(input_paths)} items using Preset='{selected_preset_name}'") preset_name_for_loading = preset_file_path.stem
self.statusBar().showMessage(f"Updating preview for '{selected_preset_name}'...", 0) log.info(f"[{time.time():.4f}] Requesting background preview update for {len(input_paths)} items using Preset Display='{selected_display_name}' (File Stem='{preset_name_for_loading}')")
self.statusBar().showMessage(f"Updating preview for '{selected_display_name}'...", 0)
log.debug("Clearing accumulated rules for new standard preview batch.") log.debug("Clearing accumulated rules for new standard preview batch.")
self._accumulated_rules.clear() self._accumulated_rules.clear()
@@ -764,8 +772,8 @@ class MainWindow(QMainWindow):
for input_path_str in input_paths: for input_path_str in input_paths:
file_list = self._extract_file_list(input_path_str) file_list = self._extract_file_list(input_path_str)
if file_list is not None: if file_list is not None:
log.debug(f"[{time.time():.4f}] Emitting start_prediction_signal for: {input_path_str} with {len(file_list)} files.") log.debug(f"[{time.time():.4f}] Emitting start_prediction_signal for: {input_path_str} with {len(file_list)} files, using preset file stem: {preset_name_for_loading}.")
self.start_prediction_signal.emit(input_path_str, file_list, selected_preset_name) self.start_prediction_signal.emit(input_path_str, file_list, preset_name_for_loading) # Pass stem for loading
else: else:
log.warning(f"[{time.time():.4f}] Skipping standard prediction signal for {input_path_str} due to extraction error.") log.warning(f"[{time.time():.4f}] Skipping standard prediction signal for {input_path_str} due to extraction error.")
else: else:
@@ -1066,13 +1074,13 @@ class MainWindow(QMainWindow):
log.debug(f"<-- Exiting _handle_prediction_completion for '{input_path}'") log.debug(f"<-- Exiting _handle_prediction_completion for '{input_path}'")
@Slot(str, str) @Slot(str, str, Path) # mode, display_name, file_path (Path can be None)
def _on_preset_selection_changed(self, mode: str, preset_name: str | None): def _on_preset_selection_changed(self, mode: str, display_name: str | None, file_path: Path | None ):
""" """
Handles changes in the preset editor selection (preset, LLM, placeholder). Handles changes in the preset editor selection (preset, LLM, placeholder).
Switches between PresetEditorWidget and LLMEditorWidget. Switches between PresetEditorWidget and LLMEditorWidget.
""" """
log.info(f"Preset selection changed: mode='{mode}', preset_name='{preset_name}'") log.info(f"Preset selection changed: mode='{mode}', display_name='{display_name}', file_path='{file_path}'")
if mode == "llm": if mode == "llm":
log.debug("Switching editor stack to LLM Editor Widget.") log.debug("Switching editor stack to LLM Editor Widget.")
@@ -1094,11 +1102,11 @@ class MainWindow(QMainWindow):
self.editor_stack.setCurrentWidget(self.preset_editor_widget.json_editor_container) self.editor_stack.setCurrentWidget(self.preset_editor_widget.json_editor_container)
# The PresetEditorWidget's internal logic handles disabling/clearing the editor fields. # The PresetEditorWidget's internal logic handles disabling/clearing the editor fields.
if mode == "preset" and preset_name: if mode == "preset" and display_name: # Use display_name for window title
# This might be redundant if the editor handles its own title updates on save/load # This might be redundant if the editor handles its own title updates on save/load
# but good for consistency. # but good for consistency.
unsaved = self.preset_editor_widget.editor_unsaved_changes unsaved = self.preset_editor_widget.editor_unsaved_changes
self.setWindowTitle(f"Asset Processor Tool - {preset_name}{'*' if unsaved else ''}") self.setWindowTitle(f"Asset Processor Tool - {display_name}{'*' if unsaved else ''}")
elif mode == "llm": elif mode == "llm":
self.setWindowTitle("Asset Processor Tool - LLM Interpretation") self.setWindowTitle("Asset Processor Tool - LLM Interpretation")
else: else:

View File

@@ -39,10 +39,9 @@ if not log.hasHandlers():
def classify_files(file_list: List[str], config: Configuration) -> Dict[str, List[Dict[str, Any]]]: def classify_files(file_list: List[str], config: Configuration) -> Dict[str, List[Dict[str, Any]]]:
""" """
Analyzes a list of files based on configuration rules using a two-pass approach Analyzes a list of files based on configuration rules to group them by asset
to group them by asset and determine initial file properties. and determine initial file properties, applying prioritization based on
Pass 1: Identifies and classifies prioritized bit depth variants. 'priority_keywords' in map_type_mapping.
Pass 2: Classifies extras, general maps (downgrading if primary exists), and ignores.
Args: Args:
file_list: List of absolute file paths. file_list: List of absolute file paths.
@@ -53,19 +52,21 @@ def classify_files(file_list: List[str], config: Configuration) -> Dict[str, Lis
Example: Example:
{ {
'AssetName1': [ 'AssetName1': [
{'file_path': '/path/to/AssetName1_DISP16.png', 'item_type': 'DISP', 'asset_name': 'AssetName1'}, {'file_path': '/path/to/AssetName1_DISP16.png', 'item_type': 'MAP_DISP', 'asset_name': 'AssetName1'},
{'file_path': '/path/to/AssetName1_DISP.png', 'item_type': 'EXTRA', 'asset_name': 'AssetName1'}, {'file_path': '/path/to/AssetName1_Color.png', 'item_type': 'MAP_COL', 'asset_name': 'AssetName1'}
{'file_path': '/path/to/AssetName1_Color.png', 'item_type': 'COL', 'asset_name': 'AssetName1'}
], ],
# ... other assets # ... other assets
} }
Files marked as "FILE_IGNORE" will also be included in the output.
Returns an empty dict if classification fails or no files are provided. Returns an empty dict if classification fails or no files are provided.
""" """
temp_grouped_files = defaultdict(list) classified_files_info: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
extra_files_to_associate = [] file_matches: Dict[str, List[Tuple[str, int, bool]]] = defaultdict(list) # {file_path: [(target_type, rule_index, is_priority), ...]}
primary_asset_names = set() files_to_ignore: Set[str] = set()
primary_assignments = set()
processed_in_pass1 = set() # --- DEBUG: Log the input file_list ---
log.info(f"DEBUG_ROO_CLASSIFY_INPUT: classify_files received file_list (len={len(file_list)}): {file_list}")
# --- END DEBUG ---
# --- Validation --- # --- Validation ---
if not file_list or not config: if not file_list or not config:
@@ -73,20 +74,20 @@ def classify_files(file_list: List[str], config: Configuration) -> Dict[str, Lis
return {} return {}
if not hasattr(config, 'compiled_map_keyword_regex') or not config.compiled_map_keyword_regex: if not hasattr(config, 'compiled_map_keyword_regex') or not config.compiled_map_keyword_regex:
log.warning("Classification skipped: Missing compiled map keyword regex in config.") log.warning("Classification skipped: Missing compiled map keyword regex in config.")
# Proceeding might still classify EXTRA/FILE_IGNORE if those rules exist
if not hasattr(config, 'compiled_extra_regex'): if not hasattr(config, 'compiled_extra_regex'):
log.warning("Configuration object missing 'compiled_extra_regex'. Cannot classify extra files.") log.warning("Configuration object missing 'compiled_extra_regex'. Cannot classify extra files.")
if not hasattr(config, 'compiled_bit_depth_regex_map'): compiled_extra_regex = [] # Provide default to avoid errors
log.warning("Configuration object missing 'compiled_bit_depth_regex_map'. Cannot prioritize bit depth variants.") else:
compiled_extra_regex = getattr(config, 'compiled_extra_regex', [])
compiled_map_regex = getattr(config, 'compiled_map_keyword_regex', {}) compiled_map_regex = getattr(config, 'compiled_map_keyword_regex', {})
compiled_extra_regex = getattr(config, 'compiled_extra_regex', []) # Note: compiled_bit_depth_regex_map is no longer used for primary classification logic here
compiled_bit_depth_regex_map = getattr(config, 'compiled_bit_depth_regex_map', {})
num_map_rules = sum(len(patterns) for patterns in compiled_map_regex.values()) num_map_rules = sum(len(patterns) for patterns in compiled_map_regex.values())
num_extra_rules = len(compiled_extra_regex) num_extra_rules = len(compiled_extra_regex)
num_bit_depth_rules = len(compiled_bit_depth_regex_map)
log.debug(f"Starting classification for {len(file_list)} files using {num_map_rules} map keyword patterns, {num_bit_depth_rules} bit depth patterns, and {num_extra_rules} extra patterns.") log.debug(f"Starting classification for {len(file_list)} files using {num_map_rules} map keyword patterns and {num_extra_rules} extra patterns.")
# --- Asset Name Extraction Helper --- # --- Asset Name Extraction Helper ---
def get_asset_name(f_path: Path, cfg: Configuration) -> str: def get_asset_name(f_path: Path, cfg: Configuration) -> str:
@@ -120,155 +121,179 @@ def classify_files(file_list: List[str], config: Configuration) -> Dict[str, Lis
log.warning(f"Asset name extraction resulted in empty string for '{filename}'. Using stem: '{asset_name}'.") log.warning(f"Asset name extraction resulted in empty string for '{filename}'. Using stem: '{asset_name}'.")
return asset_name return asset_name
# --- Pass 1: Prioritized Bit Depth Variants --- # --- Pass 1: Collect all potential matches for each file ---
log.debug("--- Starting Classification Pass 1: Prioritized Variants ---") # For each file, find all map_type_mapping rules it matches (both regular and priority keywords).
# Store the target_type, original rule_index, and whether it was a priority match.
log.debug("--- Starting Classification Pass 1: Collect Potential Matches ---")
file_matches: Dict[str, List[Tuple[str, int, bool]]] = defaultdict(list) # {file_path: [(target_type, rule_index, is_priority), ...]}
files_classified_as_extra: Set[str] = set() # Files already classified as EXTRA
compiled_map_regex = getattr(config, 'compiled_map_keyword_regex', {})
compiled_extra_regex = getattr(config, 'compiled_extra_regex', [])
for file_path_str in file_list: for file_path_str in file_list:
file_path = Path(file_path_str) file_path = Path(file_path_str)
filename = file_path.name filename = file_path.name
asset_name = get_asset_name(file_path, config) asset_name = get_asset_name(file_path, config)
processed = False
for target_type, variant_regex in compiled_bit_depth_regex_map.items(): if "BoucleChunky001" in file_path_str:
match = variant_regex.search(filename) log.info(f"DEBUG_ROO: Processing file: {file_path_str}")
if match:
log.debug(f"PASS 1: File '{filename}' matched PRIORITIZED bit depth variant for type '{target_type}'.")
matched_item_type = target_type
if (asset_name, matched_item_type) in primary_assignments: # Check for EXTRA files first
log.warning(f"PASS 1: Primary assignment ({asset_name}, {matched_item_type}) already exists. File '{filename}' will be handled in Pass 2.")
else:
primary_assignments.add((asset_name, matched_item_type))
log.debug(f" PASS 1: Added primary assignment: ({asset_name}, {matched_item_type})")
primary_asset_names.add(asset_name)
temp_grouped_files[asset_name].append({
'file_path': file_path_str,
'item_type': matched_item_type,
'asset_name': asset_name
})
processed_in_pass1.add(file_path_str)
processed = True
break # Stop checking other variant patterns for this file
log.debug(f"--- Finished Pass 1. Primary assignments made: {primary_assignments} ---")
# --- Pass 2: Extras, General Maps, Ignores ---
log.debug("--- Starting Classification Pass 2: Extras, General Maps, Ignores ---")
for file_path_str in file_list:
if file_path_str in processed_in_pass1:
log.debug(f"PASS 2: Skipping '{Path(file_path_str).name}' (processed in Pass 1).")
continue
file_path = Path(file_path_str)
filename = file_path.name
asset_name = get_asset_name(file_path, config)
is_extra = False is_extra = False
is_map = False
# 1. Check for Extra Files FIRST in Pass 2
for extra_pattern in compiled_extra_regex: for extra_pattern in compiled_extra_regex:
if extra_pattern.search(filename): if extra_pattern.search(filename):
log.debug(f"PASS 2: File '{filename}' matched EXTRA pattern: {extra_pattern.pattern}") if "BoucleChunky001_DISP_1K_METALNESS.png" in filename and extra_pattern.search(filename):
extra_files_to_associate.append((file_path_str, filename)) log.info(f"DEBUG_ROO: EXTRA MATCH: File '{filename}' matched EXTRA pattern: {extra_pattern.pattern}")
log.debug(f"PASS 1: File '{filename}' matched EXTRA pattern: {extra_pattern.pattern}")
# For EXTRA, we assign it directly and don't check map rules for this file
classified_files_info[asset_name].append({
'file_path': file_path_str,
'item_type': "EXTRA",
'asset_name': asset_name
})
files_classified_as_extra.add(file_path_str)
is_extra = True is_extra = True
break break
if is_extra: if "BoucleChunky001_DISP_1K_METALNESS.png" in filename and not is_extra: # after the extra loop
continue log.info(f"DEBUG_ROO: EXTRA CHECK FAILED for {filename}. is_extra: {is_extra}")
# 2. Check for General Map Files in Pass 2 if "BoucleChunky001_DISP_1K_METALNESS.png" in filename and not is_extra:
log.info(f"DEBUG_ROO: EXTRA CHECK FAILED for {filename}. is_extra: {is_extra}")
if is_extra:
continue # Move to the next file
# If not EXTRA, check for MAP matches (collect all potential matches)
for target_type, patterns_list in compiled_map_regex.items(): for target_type, patterns_list in compiled_map_regex.items():
for compiled_regex, original_keyword, rule_index in patterns_list: for compiled_regex, original_keyword, rule_index, is_priority in patterns_list:
match = compiled_regex.search(filename) match = compiled_regex.search(filename)
if match: if match:
try: if "BoucleChunky001" in file_path_str:
# map_type_mapping_list = config.map_type_mapping # Old gloss logic source log.info(f"DEBUG_ROO: PASS 1 MAP MATCH: File '{filename}' matched keyword '{original_keyword}' (priority: {is_priority}) for target type '{target_type}' (Rule Index: {rule_index}).")
# matched_rule_details = map_type_mapping_list[rule_index] # Old gloss logic source log.debug(f" PASS 1: File '{filename}' matched keyword '{original_keyword}' (priority: {is_priority}) for target type '{target_type}' (Rule Index: {rule_index}).")
# is_gloss_flag = matched_rule_details.get('is_gloss_source', False) # Old gloss logic file_matches[file_path_str].append((target_type, rule_index, is_priority))
log.debug(f" PASS 2: Match found! Rule Index: {rule_index}, Keyword: '{original_keyword}', Target: '{target_type}'") # Removed Gloss from log
except Exception as e:
log.exception(f" PASS 2: Error accessing rule details for index {rule_index}: {e}")
# *** Crucial Check: Has a prioritized variant claimed this type? *** log.debug(f"--- Finished Pass 1. Collected matches for {len(file_matches)} files. ---")
if (asset_name, target_type) in primary_assignments:
log.debug(f"PASS 2: File '{filename}' matched '{original_keyword}' for type '{target_type}', but primary already assigned via Pass 1. Classifying as EXTRA.") # --- Pass 2: Determine Trumped Regular Matches ---
matched_item_type = "EXTRA" # Identify which regular matches are trumped by a priority match for the same rule_index within the asset.
# is_gloss_flag = False # Old gloss logic log.debug("--- Starting Classification Pass 2: Determine Trumped Regular Matches ---")
trumped_regular_matches: Set[Tuple[str, int]] = set() # Set of (file_path_str, rule_index) pairs that are trumped
# First, determine which rule_indices have *any* priority match across the entire asset
rule_index_has_priority_match_in_asset: Set[int] = set()
for file_path_str, matches in file_matches.items():
for match_target, match_rule_index, match_is_priority in matches:
if match_is_priority:
rule_index_has_priority_match_in_asset.add(match_rule_index)
log.debug(f" Rule indices with priority matches in asset: {sorted(list(rule_index_has_priority_match_in_asset))}")
# Then, for each file, check its matches against the rules that had priority matches
for file_path_str in file_list:
if file_path_str in files_classified_as_extra:
continue
matches_for_this_file = file_matches.get(file_path_str, [])
# Determine if this file has any priority match for a given rule_index
file_has_priority_match_for_rule: Dict[int, bool] = defaultdict(bool)
for match_target, match_rule_index, match_is_priority in matches_for_this_file:
if match_is_priority:
file_has_priority_match_for_rule[match_rule_index] = True
# Determine if this file has any regular match for a given rule_index
file_has_regular_match_for_rule: Dict[int, bool] = defaultdict(bool)
for match_target, match_rule_index, match_is_priority in matches_for_this_file:
if not match_is_priority:
file_has_regular_match_for_rule[match_rule_index] = True
# Identify trumped regular matches for this file
for match_target, match_rule_index, match_is_priority in matches_for_this_file:
if not match_is_priority: # Only consider regular matches
if match_rule_index in rule_index_has_priority_match_in_asset:
# This regular match is for a rule_index that had a priority match somewhere in the asset
if not file_has_priority_match_for_rule[match_rule_index]:
# And this specific file did NOT have a priority match for this rule_index
trumped_regular_matches.add((file_path_str, match_rule_index))
log.debug(f" File '{Path(file_path_str).name}': Regular match for Rule Index {match_rule_index} is trumped.")
if "BoucleChunky001" in file_path_str:
log.info(f"DEBUG_ROO: TRUMPED: File '{Path(file_path_str).name}': Regular match for Rule Index {match_rule_index} (target {match_target}) is trumped.")
if "BoucleChunky001" in file_path_str: # Check if it was actually added by checking the set, or just log if the condition was met
if (file_path_str, match_rule_index) in trumped_regular_matches:
log.info(f"DEBUG_ROO: TRUMPED: File '{Path(file_path_str).name}': Regular match for Rule Index {match_rule_index} (target {match_target}) is trumped.")
log.debug(f"--- Finished Pass 2. Identified {len(trumped_regular_matches)} trumped regular matches. ---")
# --- Pass 3: Final Assignment & Inter-Entry Resolution ---
# Iterate through files, apply ignore rules, and then apply earliest rule wins for remaining valid matches.
log.debug("--- Starting Classification Pass 3: Final Assignment ---")
final_file_assignments: Dict[str, str] = {} # {file_path: final_item_type}
for file_path_str in file_list:
# Check if the file was already classified as EXTRA in Pass 1 and added to classified_files_info
if file_path_str in files_classified_as_extra:
log.debug(f" Final Assignment: Skipping '{Path(file_path_str).name}' as it was already classified as EXTRA in Pass 1.")
continue # Skip this file in Pass 3 as it's already handled
asset_name = get_asset_name(Path(file_path_str), config) # Need asset name for the final output structure
# Get valid matches for this file after considering intra-entry priority trumps regular
valid_matches = []
for match_target, match_rule_index, match_is_priority in file_matches.get(file_path_str, []):
if (file_path_str, match_rule_index) not in trumped_regular_matches:
valid_matches.append((match_target, match_rule_index, match_is_priority))
log.debug(f" File '{Path(file_path_str).name}': Valid match - Target: '{match_target}', Rule Index: {match_rule_index}, Priority: {match_is_priority}")
else: else:
log.debug(f"PASS 2: File '{filename}' matched '{original_keyword}' for item_type '{target_type}'.") log.debug(f" File '{Path(file_path_str).name}': Invalid match (trumped by priority) - Target: '{match_target}', Rule Index: {match_rule_index}, Priority: {match_is_priority}")
matched_item_type = target_type
temp_grouped_files[asset_name].append({ if "BoucleChunky001" in file_path_str:
log.info(f"DEBUG_ROO: PASS 3 PRE-ASSIGN: File '{Path(file_path_str).name}'. Valid matches: {valid_matches}")
if "BoucleChunky001" in file_path_str:
log.info(f"DEBUG_ROO: PASS 3 PRE-ASSIGN: File '{Path(file_path_str).name}'. Valid matches: {valid_matches}")
final_item_type = "FILE_IGNORE" # Default to ignore if no valid matches
if valid_matches:
# Apply earliest rule wins among valid matches
best_match = min(valid_matches, key=lambda x: x[1]) # Find match with lowest rule_index
final_item_type = best_match[0] # Assign the target_type of the best match
log.debug(f" File '{Path(file_path_str).name}': Best valid match -> Target: '{best_match[0]}', Rule Index: {best_match[1]}. Final type: '{final_item_type}'.")
else:
log.debug(f" File '{Path(file_path_str).name}'': No valid matches after filtering. Final type: '{final_item_type}'.")
if "BoucleChunky001" in file_path_str:
log.info(f"DEBUG_ROO: PASS 3 FINAL ASSIGN: File '{Path(file_path_str).name}' -> Final Type: '{final_item_type}'")
final_file_assignments[file_path_str] = final_item_type
if "BoucleChunky001" in file_path_str:
log.info(f"DEBUG_ROO: PASS 3 FINAL ASSIGN: File '{Path(file_path_str).name}' -> Final Type: '{final_item_type}'")
# Add the file info to the classified_files_info structure
log.info(f"DEBUG_ROO: PASS 3 APPEND: Appending file '{Path(file_path_str).name}' with type '{final_item_type}' to classified_files_info['{asset_name}']")
classified_files_info[asset_name].append({
'file_path': file_path_str, 'file_path': file_path_str,
'item_type': matched_item_type, 'item_type': final_item_type,
'asset_name': asset_name 'asset_name': asset_name
}) })
is_map = True log.debug(f" Final Grouping: '{Path(file_path_str).name}' -> '{final_item_type}' (Asset: '{asset_name}')")
break
if is_map:
break
# 3. Handle Unmatched Files in Pass 2 (Not Extra, Not Map)
if not is_extra and not is_map:
log.debug(f"PASS 2: File '{filename}' did not match any map/extra pattern. Grouping under asset '{asset_name}' as FILE_IGNORE.")
temp_grouped_files[asset_name].append({
'file_path': file_path_str,
'item_type': "FILE_IGNORE",
'asset_name': asset_name
})
log.debug("--- Finished Pass 2 ---")
# --- Determine Primary Asset Name for Extra Association (using Pass 1 results) ---
final_primary_asset_name = None
if primary_asset_names:
primary_map_asset_names_pass1 = [
f_info['asset_name']
for asset_files in temp_grouped_files.values()
for f_info in asset_files
if f_info['asset_name'] in primary_asset_names and (f_info['asset_name'], f_info['item_type']) in primary_assignments
]
if primary_map_asset_names_pass1:
name_counts = Counter(primary_map_asset_names_pass1)
most_common_names = name_counts.most_common()
final_primary_asset_name = most_common_names[0][0]
if len(most_common_names) > 1 and most_common_names[0][1] == most_common_names[1][1]:
tied_names = sorted([name for name, count in most_common_names if count == most_common_names[0][1]])
final_primary_asset_name = tied_names[0]
log.warning(f"Multiple primary asset names tied for most common based on Pass 1: {tied_names}. Using '{final_primary_asset_name}' for associating extra files.")
log.debug(f"Determined primary asset name for extras based on Pass 1 primary maps: '{final_primary_asset_name}'")
else:
log.warning("Primary asset names set (from Pass 1) was populated, but no corresponding groups found. Falling back.")
if not final_primary_asset_name:
if temp_grouped_files and extra_files_to_associate:
fallback_name = sorted(temp_grouped_files.keys())[0]
final_primary_asset_name = fallback_name
log.warning(f"No primary map files found in Pass 1. Associating extras with first group found alphabetically: '{final_primary_asset_name}'.")
elif extra_files_to_associate:
log.warning(f"Could not determine any asset name to associate {len(extra_files_to_associate)} extra file(s) with. They will be ignored.")
else:
log.debug("No primary asset name determined (no maps or extras found).")
# --- Associate Extra Files (collected in Pass 2) --- log.debug(f"Classification complete. Found {len(classified_files_info)} potential assets.")
if final_primary_asset_name and extra_files_to_associate: # Enhanced logging for the content of classified_files_info
log.debug(f"Associating {len(extra_files_to_associate)} extra file(s) with primary asset '{final_primary_asset_name}'") boucle_chunky_data = {
for file_path_str, filename in extra_files_to_associate: key: val for key, val in classified_files_info.items()
if not any(f['file_path'] == file_path_str for f in temp_grouped_files[final_primary_asset_name]): if 'BoucleChunky001' in key or any('BoucleChunky001' in (f_info.get('file_path','')) for f_info in val)
temp_grouped_files[final_primary_asset_name].append({ }
'file_path': file_path_str, import json # Make sure json is imported if not already at top of file
'item_type': "EXTRA", log.info(f"DEBUG_ROO: Final classified_files_info for BoucleChunky001 (content): \n{json.dumps(boucle_chunky_data, indent=2)}")
'asset_name': final_primary_asset_name return dict(classified_files_info)
})
else:
log.debug(f"Skipping duplicate association of extra file: {filename}")
elif extra_files_to_associate:
pass
log.debug(f"Classification complete. Found {len(temp_grouped_files)} potential assets.")
return dict(temp_grouped_files)
class RuleBasedPredictionHandler(BasePredictionHandler): class RuleBasedPredictionHandler(BasePredictionHandler):
@@ -367,7 +392,8 @@ class RuleBasedPredictionHandler(BasePredictionHandler):
source_rule = SourceRule( source_rule = SourceRule(
input_path=input_source_identifier, input_path=input_source_identifier,
supplier_identifier=supplier_identifier, supplier_identifier=supplier_identifier,
preset_name=preset_name # Use the internal display name from the config object
preset_name=config.internal_display_preset_name
) )
asset_rules = [] asset_rules = []
file_type_definitions = config._core_settings.get('FILE_TYPE_DEFINITIONS', {}) file_type_definitions = config._core_settings.get('FILE_TYPE_DEFINITIONS', {})
@@ -463,23 +489,22 @@ class RuleBasedPredictionHandler(BasePredictionHandler):
base_item_type = file_info['item_type'] base_item_type = file_info['item_type']
target_asset_name_override = file_info['asset_name'] target_asset_name_override = file_info['asset_name']
final_item_type = base_item_type final_item_type = base_item_type
if not base_item_type.startswith("MAP_") and base_item_type not in ["FILE_IGNORE", "EXTRA", "MODEL"]: # The classification logic now returns the final item_type directly,
final_item_type = f"MAP_{base_item_type}" # including "FILE_IGNORE" and correctly prioritized MAP_ types.
# No need for the old MAP_ prefixing logic here.
if file_type_definitions and final_item_type not in file_type_definitions and base_item_type not in ["FILE_IGNORE", "EXTRA"]: # Validate the final_item_type against definitions, unless it's EXTRA or FILE_IGNORE
log.warning(f"Predicted ItemType '{base_item_type}' (checked as '{final_item_type}') for file '{file_info['file_path']}' is not in FILE_TYPE_DEFINITIONS. Setting to FILE_IGNORE.") if final_item_type not in ["EXTRA", "FILE_IGNORE"] and file_type_definitions and final_item_type not in file_type_definitions:
log.warning(f"Predicted ItemType '{final_item_type}' for file '{file_info['file_path']}' is not in FILE_TYPE_DEFINITIONS. Setting to FILE_IGNORE.")
final_item_type = "FILE_IGNORE" final_item_type = "FILE_IGNORE"
# is_gloss_source_value = file_info.get('is_gloss_source', False) # Removed
file_rule = FileRule( file_rule = FileRule(
file_path=file_info['file_path'], file_path=file_info['file_path'],
item_type=final_item_type, item_type=final_item_type,
item_type_override=final_item_type, item_type_override=final_item_type, # item_type_override defaults to item_type
target_asset_name_override=target_asset_name_override, target_asset_name_override=target_asset_name_override,
output_format_override=None, output_format_override=None,
# is_gloss_source=is_gloss_source_value if isinstance(is_gloss_source_value, bool) else False, # Removed
resolution_override=None, resolution_override=None,
channel_merge_instructions={}, channel_merge_instructions={},
) )
@@ -489,6 +514,18 @@ class RuleBasedPredictionHandler(BasePredictionHandler):
source_rule.assets = asset_rules source_rule.assets = asset_rules
source_rules_list.append(source_rule) source_rules_list.append(source_rule)
# DEBUG: Log the structure of the source_rule being emitted
if source_rule and source_rule.assets:
for asset_r_idx, asset_r in enumerate(source_rule.assets):
log.info(f"DEBUG_ROO_EMIT: Source '{input_source_identifier}', Asset {asset_r_idx} ('{asset_r.asset_name}') has {len(asset_r.files)} FileRules.")
for fr_idx, fr in enumerate(asset_r.files):
log.info(f"DEBUG_ROO_EMIT: FR {fr_idx}: Path='{fr.file_path}', Type='{fr.item_type}', TargetAsset='{fr.target_asset_name_override}'")
elif source_rule:
log.info(f"DEBUG_ROO_EMIT: Emitting SourceRule for {input_source_identifier} but it has no assets.")
else:
log.info(f"DEBUG_ROO_EMIT: Attempting to emit for {input_source_identifier}, but source_rule object is None.")
# END DEBUG
except Exception as e: except Exception as e:
log.exception(f"Error building rule hierarchy for source '{input_source_identifier}': {e}") log.exception(f"Error building rule hierarchy for source '{input_source_identifier}': {e}")
raise RuntimeError(f"Error building rule hierarchy: {e}") from e raise RuntimeError(f"Error building rule hierarchy: {e}") from e

View File

@@ -36,8 +36,8 @@ class PresetEditorWidget(QWidget):
# Signal emitted when presets list changes (saved, deleted, new) # Signal emitted when presets list changes (saved, deleted, new)
presets_changed_signal = Signal() presets_changed_signal = Signal()
# Signal emitted when the selected preset (or LLM/Placeholder) changes # Signal emitted when the selected preset (or LLM/Placeholder) changes
# Emits: mode ("preset", "llm", "placeholder"), preset_name (str or None) # Emits: mode ("preset", "llm", "placeholder"), display_name (str or None), file_path (Path or None)
preset_selection_changed_signal = Signal(str, str) preset_selection_changed_signal = Signal(str, str, Path)
def __init__(self, parent=None): def __init__(self, parent=None):
super().__init__(parent) super().__init__(parent)
@@ -296,8 +296,22 @@ class PresetEditorWidget(QWidget):
log.warning(msg) log.warning(msg)
else: else:
for preset_path in presets: for preset_path in presets:
item = QListWidgetItem(preset_path.stem) preset_display_name = preset_path.stem # Fallback
item.setData(Qt.ItemDataRole.UserRole, preset_path) try:
with open(preset_path, 'r', encoding='utf-8') as f:
preset_content = json.load(f)
internal_name = preset_content.get("preset_name")
if internal_name and isinstance(internal_name, str) and internal_name.strip():
preset_display_name = internal_name.strip()
else:
log.warning(f"Preset file {preset_path.name} is missing 'preset_name' or it's empty. Using filename stem '{preset_path.stem}' as display name.")
except json.JSONDecodeError:
log.error(f"Failed to parse JSON from {preset_path.name}. Using filename stem '{preset_path.stem}' as display name.")
except Exception as e:
log.error(f"Error reading {preset_path.name}: {e}. Using filename stem '{preset_path.stem}' as display name.")
item = QListWidgetItem(preset_display_name)
item.setData(Qt.ItemDataRole.UserRole, preset_path) # Store the path for loading
self.editor_preset_list.addItem(item) self.editor_preset_list.addItem(item)
log.info(f"Loaded {len(presets)} presets into editor list.") log.info(f"Loaded {len(presets)} presets into editor list.")
@@ -525,7 +539,8 @@ class PresetEditorWidget(QWidget):
log.debug(f"PresetEditor: currentItemChanged signal triggered. current: {current_item.text() if current_item else 'None'}") log.debug(f"PresetEditor: currentItemChanged signal triggered. current: {current_item.text() if current_item else 'None'}")
mode = "placeholder" mode = "placeholder"
preset_name = None display_name_to_emit = None # Changed from preset_name
file_path_to_emit = None # New variable for Path
# Check for unsaved changes before proceeding # Check for unsaved changes before proceeding
if self.check_unsaved_changes(): if self.check_unsaved_changes():
@@ -540,41 +555,53 @@ class PresetEditorWidget(QWidget):
# Determine mode and preset name based on selection # Determine mode and preset name based on selection
if current_item: if current_item:
item_data = current_item.data(Qt.ItemDataRole.UserRole) item_data = current_item.data(Qt.ItemDataRole.UserRole)
current_display_text = current_item.text() # This is the internal name from populate_presets
if item_data == "__PLACEHOLDER__": if item_data == "__PLACEHOLDER__":
log.debug("Placeholder item selected.") log.debug("Placeholder item selected.")
self._clear_editor() self._clear_editor()
self._set_editor_enabled(False) self._set_editor_enabled(False)
mode = "placeholder" mode = "placeholder"
display_name_to_emit = None
file_path_to_emit = None
self._last_valid_preset_name = None # Clear last valid name self._last_valid_preset_name = None # Clear last valid name
elif item_data == "__LLM__": elif item_data == "__LLM__":
log.debug("LLM Interpretation item selected.") log.debug("LLM Interpretation item selected.")
self._clear_editor() self._clear_editor()
self._set_editor_enabled(False) self._set_editor_enabled(False)
mode = "llm" mode = "llm"
# Keep _last_valid_preset_name as it was display_name_to_emit = None # LLM mode has no specific preset display name
elif isinstance(item_data, Path): file_path_to_emit = None
log.debug(f"Loading preset for editing: {current_item.text()}") # Keep _last_valid_preset_name as it was (it should be the display name)
preset_path = item_data elif isinstance(item_data, Path): # item_data is the Path object for a preset
self._load_preset_for_editing(preset_path) log.debug(f"Loading preset for editing: {current_display_text}")
self._last_valid_preset_name = preset_path.stem preset_file_path_obj = item_data
self._load_preset_for_editing(preset_file_path_obj)
# _last_valid_preset_name should store the display name for delegate use
self._last_valid_preset_name = current_display_text
mode = "preset" mode = "preset"
preset_name = self._last_valid_preset_name display_name_to_emit = current_display_text
else: file_path_to_emit = preset_file_path_obj
else: # Should not happen if list is populated correctly
log.error(f"Invalid data type for preset path: {type(item_data)}. Clearing editor.") log.error(f"Invalid data type for preset path: {type(item_data)}. Clearing editor.")
self._clear_editor() self._clear_editor()
self._set_editor_enabled(False) self._set_editor_enabled(False)
mode = "placeholder" # Treat as placeholder on error mode = "placeholder"
display_name_to_emit = None
file_path_to_emit = None
self._last_valid_preset_name = None self._last_valid_preset_name = None
else: else: # No current_item (e.g., list cleared)
log.debug("No preset selected. Clearing editor.") log.debug("No preset selected. Clearing editor.")
self._clear_editor() self._clear_editor()
self._set_editor_enabled(False) self._set_editor_enabled(False)
mode = "placeholder" mode = "placeholder"
display_name_to_emit = None
file_path_to_emit = None
self._last_valid_preset_name = None self._last_valid_preset_name = None
# Emit the signal regardless of what was selected # Emit the signal with all three arguments
log.debug(f"Emitting preset_selection_changed_signal: mode='{mode}', preset_name='{preset_name}'") log.debug(f"Emitting preset_selection_changed_signal: mode='{mode}', display_name='{display_name_to_emit}', file_path='{file_path_to_emit}'")
self.preset_selection_changed_signal.emit(mode, preset_name) self.preset_selection_changed_signal.emit(mode, display_name_to_emit, file_path_to_emit)
def _gather_editor_data(self) -> dict: def _gather_editor_data(self) -> dict:
"""Gathers data from all editor UI widgets and returns a dictionary.""" """Gathers data from all editor UI widgets and returns a dictionary."""
@@ -757,22 +784,25 @@ class PresetEditorWidget(QWidget):
# --- Public Access Methods for MainWindow --- # --- Public Access Methods for MainWindow ---
def get_selected_preset_mode(self) -> tuple[str, str | None]: def get_selected_preset_mode(self) -> tuple[str, str | None, Path | None]:
""" """
Returns the current selection mode and preset name (if applicable). Returns the current selection mode, display name, and file path for loading.
Returns: tuple(mode_string, preset_name_string_or_None) Returns: tuple(mode_string, display_name_string_or_None, file_path_or_None)
mode_string can be "preset", "llm", "placeholder" mode_string can be "preset", "llm", "placeholder"
""" """
current_item = self.editor_preset_list.currentItem() current_item = self.editor_preset_list.currentItem()
if current_item: if current_item:
item_data = current_item.data(Qt.ItemDataRole.UserRole) item_data = current_item.data(Qt.ItemDataRole.UserRole)
display_text = current_item.text() # This is now the internal name
if item_data == "__PLACEHOLDER__": if item_data == "__PLACEHOLDER__":
return "placeholder", None return "placeholder", None, None
elif item_data == "__LLM__": elif item_data == "__LLM__":
return "llm", None return "llm", None, None # LLM mode doesn't have a specific preset file path
elif isinstance(item_data, Path): elif isinstance(item_data, Path):
return "preset", item_data.stem # For a preset, display_text is the internal name, item_data is the Path
return "placeholder", None # Default or if no item selected return "preset", display_text, item_data # Return internal name and path
return "placeholder", None, None # Default or if no item selected
def get_last_valid_preset_name(self) -> str | None: def get_last_valid_preset_name(self) -> str | None:
""" """

204
main.py
View File

@@ -15,11 +15,12 @@ from typing import List, Dict, Tuple, Optional
# --- Utility Imports --- # --- Utility Imports ---
from utils.hash_utils import calculate_sha256 from utils.hash_utils import calculate_sha256
from utils.path_utils import get_next_incrementing_value from utils.path_utils import get_next_incrementing_value
from utils import app_setup_utils # Import the new utility module
# --- Qt Imports for Application Structure --- # --- Qt Imports for Application Structure ---
from PySide6.QtCore import QObject, Slot, QThreadPool, QRunnable, Signal from PySide6.QtCore import QObject, Slot, QThreadPool, QRunnable, Signal
from PySide6.QtCore import Qt from PySide6.QtCore import Qt
from PySide6.QtWidgets import QApplication from PySide6.QtWidgets import QApplication, QDialog # Import QDialog for the setup dialog
# --- Backend Imports --- # --- Backend Imports ---
# Add current directory to sys.path for direct execution # Add current directory to sys.path for direct execution
@@ -45,6 +46,10 @@ try:
from gui.main_window import MainWindow from gui.main_window import MainWindow
print("DEBUG: Successfully imported MainWindow.") print("DEBUG: Successfully imported MainWindow.")
print("DEBUG: Attempting to import FirstTimeSetupDialog...")
from gui.first_time_setup_dialog import FirstTimeSetupDialog # Import the setup dialog
print("DEBUG: Successfully imported FirstTimeSetupDialog.")
print("DEBUG: Attempting to import prepare_processing_workspace...") print("DEBUG: Attempting to import prepare_processing_workspace...")
from utils.workspace_utils import prepare_processing_workspace from utils.workspace_utils import prepare_processing_workspace
print("DEBUG: Successfully imported prepare_processing_workspace.") print("DEBUG: Successfully imported prepare_processing_workspace.")
@@ -300,8 +305,9 @@ class App(QObject):
# Signal emitted when all queued processing tasks are complete # Signal emitted when all queued processing tasks are complete
all_tasks_finished = Signal(int, int, int) # processed_count, skipped_count, failed_count (Placeholder counts for now) all_tasks_finished = Signal(int, int, int) # processed_count, skipped_count, failed_count (Placeholder counts for now)
def __init__(self): def __init__(self, user_config_path: str):
super().__init__() super().__init__()
self.user_config_path = user_config_path # Store the determined user config path
self.config_obj = None self.config_obj = None
self.processing_engine = None self.processing_engine = None
self.main_window = None self.main_window = None
@@ -310,34 +316,19 @@ class App(QObject):
self._task_results = {"processed": 0, "skipped": 0, "failed": 0} self._task_results = {"processed": 0, "skipped": 0, "failed": 0}
log.info(f"Maximum threads for pool: {self.thread_pool.maxThreadCount()}") log.info(f"Maximum threads for pool: {self.thread_pool.maxThreadCount()}")
self._load_config() self._load_config(self.user_config_path) # Pass the user config path
self._init_engine() self._init_engine()
self._init_gui() self._init_gui()
def _load_config(self): def _load_config(self, user_config_path: str):
"""Loads the base configuration using a default preset.""" """Loads the base configuration using the determined user config path."""
# The actual preset name comes from the GUI request later, but the engine
# needs an initial valid configuration object.
try: try:
# Find the first available preset to use as a default # Initialize Configuration with the determined user config path
preset_dir = Path(__file__).parent / "Presets" # The Configuration class is responsible for finding presets and other configs
default_preset_name = None self.config_obj = Configuration(base_dir_user_config=user_config_path)
if preset_dir.is_dir(): log.info(f"Base configuration loaded using user config path: '{user_config_path}'.")
presets = sorted([f.stem for f in preset_dir.glob("*.json") if f.is_file() and not f.name.startswith('_')])
if presets:
default_preset_name = presets[0]
log.info(f"Using first available preset as default for initial config: '{default_preset_name}'")
if not default_preset_name:
# Fallback or raise error if no presets found
log.error("No presets found in the 'Presets' directory. Cannot initialize default configuration.")
# Option 1: Raise an error
raise ConfigurationError("No presets found to load default configuration.")
self.config_obj = Configuration(preset_name=default_preset_name)
log.info(f"Base configuration loaded using default preset '{default_preset_name}'.")
except ConfigurationError as e: except ConfigurationError as e:
log.error(f"Fatal: Failed to load base configuration using default preset: {e}") log.error(f"Fatal: Failed to load base configuration using user config path '{user_config_path}': {e}")
# In a real app, show this error to the user before exiting # In a real app, show this error to the user before exiting
sys.exit(1) sys.exit(1)
except Exception as e: except Exception as e:
@@ -401,120 +392,6 @@ class App(QObject):
log.debug(f"Initialized active task count to: {self._active_tasks_count}") log.debug(f"Initialized active task count to: {self._active_tasks_count}")
# Update GUI progress bar/status via MainPanelWidget # Update GUI progress bar/status via MainPanelWidget
self.main_window.main_panel_widget.progress_bar.setMaximum(len(source_rules))
self.main_window.main_panel_widget.progress_bar.setValue(0)
self.main_window.main_panel_widget.progress_bar.setFormat(f"0/{len(source_rules)} tasks")
# --- Get paths needed for ProcessingTask ---
try:
# Get output_dir from processing_settings passed from autotest.py
output_base_path_str = processing_settings.get("output_dir")
log.info(f"APP_DEBUG: Received output_dir in processing_settings: {output_base_path_str}")
if not output_base_path_str:
log.error("Cannot queue tasks: Output directory path is empty in processing_settings.")
# self.main_window.statusBar().showMessage("Error: Output directory cannot be empty.", 5000) # GUI specific
return
output_base_path = Path(output_base_path_str)
# Basic validation - check if it's likely a valid path structure (doesn't guarantee existence/writability here)
if not output_base_path.is_absolute():
# Or attempt to resolve relative to workspace? For now, require absolute from GUI.
log.warning(f"Output path '{output_base_path}' is not absolute. Processing might fail if relative path is not handled correctly by engine.")
# Consider resolving: output_base_path = Path.cwd() / output_base_path # If relative paths are allowed
# Define workspace path (assuming main.py is in the project root)
workspace_path = Path(__file__).parent.resolve()
log.debug(f"Using Workspace Path: {workspace_path}")
log.debug(f"Using Output Base Path: {output_base_path}")
except Exception as e:
log.exception(f"Error getting/validating paths for processing task: {e}")
self.main_window.statusBar().showMessage(f"Error preparing paths: {e}", 5000)
return
# --- End Get paths ---
# Set max threads based on GUI setting
worker_count = processing_settings.get('workers', 1)
self.thread_pool.setMaxThreadCount(worker_count)
log.info(f"Set thread pool max workers to: {worker_count}")
# Queue tasks in the thread pool
log.debug("DEBUG: Entering task queuing loop.")
for i, rule in enumerate(source_rules):
if isinstance(rule, SourceRule):
log.info(f"DEBUG Task {i+1}: Rule Input='{rule.input_path}', Supplier ID='{getattr(rule, 'supplier_identifier', 'Not Set')}', Preset='{getattr(rule, 'preset_name', 'Not Set')}'")
log.debug(f"DEBUG: Preparing to queue task {i+1}/{len(source_rules)} for rule: {rule.input_path}")
# --- Create a new Configuration and Engine instance for this specific task ---
task_engine = None
try:
# Get preset name from the rule, fallback to app's default if missing
preset_name_for_task = getattr(rule, 'preset_name', None)
if not preset_name_for_task:
log.warning(f"Task {i+1} (Rule: {rule.input_path}): SourceRule missing preset_name. Falling back to default preset '{self.config_obj.preset_name}'.")
preset_name_for_task = self.config_obj.preset_name
task_config = Configuration(preset_name=preset_name_for_task)
task_engine = ProcessingEngine(task_config)
log.debug(f"Task {i+1}: Created new ProcessingEngine instance with preset '{preset_name_for_task}'.")
except ConfigurationError as config_err:
log.error(f"Task {i+1} (Rule: {rule.input_path}): Failed to load configuration for preset '{preset_name_for_task}': {config_err}. Skipping task.")
self._active_tasks_count -= 1 # Decrement count as this task won't run
self._task_results["failed"] += 1
# Optionally update GUI status for this specific rule
self.main_window.update_file_status(str(rule.input_path), "failed", f"Config Error: {config_err}")
continue # Skip to the next rule
except Exception as engine_err:
log.exception(f"Task {i+1} (Rule: {rule.input_path}): Failed to initialize ProcessingEngine for preset '{preset_name_for_task}': {engine_err}. Skipping task.")
self._active_tasks_count -= 1 # Decrement count
self._task_results["failed"] += 1
self.main_window.update_file_status(str(rule.input_path), "failed", f"Engine Init Error: {engine_err}")
continue # Skip to the next rule
if task_engine is None: # Should not happen if exceptions are caught, but safety check
log.error(f"Task {i+1} (Rule: {rule.input_path}): Engine is None after initialization attempt. Skipping task.")
self._active_tasks_count -= 1 # Decrement count
self._task_results["failed"] += 1
self.main_window.update_file_status(str(rule.input_path), "failed", "Engine initialization failed (unknown reason).")
continue # Skip to the next rule
# --- End Engine Instantiation ---
task = ProcessingTask(
engine=task_engine,
rule=rule,
workspace_path=workspace_path,
output_base_path=output_base_path # This is Path(output_base_path_str)
)
log.info(f"APP_DEBUG: Passing to ProcessingTask: output_base_path = {output_base_path}")
task.signals.finished.connect(self._on_task_finished)
log.debug(f"DEBUG: Calling thread_pool.start() for task {i+1}")
self.thread_pool.start(task)
log.debug(f"DEBUG: Returned from thread_pool.start() for task {i+1}")
else:
log.warning(f"Skipping invalid item (index {i}) in rule list: {type(rule)}")
log.info(f"Queued {len(source_rules)} processing tasks (finished loop).")
# GUI status already updated in MainWindow when signal was emitted
# --- Slot to handle completion of individual tasks ---
@Slot(str, str, object)
def _on_task_finished(self, rule_input_path, status, result_or_error):
"""Handles the 'finished' signal from a ProcessingTask."""
log.info(f"Task finished signal received for {rule_input_path}. Status: {status}")
self._active_tasks_count -= 1
log.debug(f"Active tasks remaining: {self._active_tasks_count}")
# Update overall results (basic counts for now)
if status == "processed":
self._task_results["processed"] += 1
elif status == "skipped": # Assuming engine might return 'skipped' status eventually
self._task_results["skipped"] += 1
else: # Count all other statuses (failed_preparation, failed_processing) as failed
self._task_results["failed"] += 1
# Update progress bar via MainPanelWidget
total_tasks = self.main_window.main_panel_widget.progress_bar.maximum() total_tasks = self.main_window.main_panel_widget.progress_bar.maximum()
completed_tasks = total_tasks - self._active_tasks_count completed_tasks = total_tasks - self._active_tasks_count
self.main_window.main_panel_widget.update_progress_bar(completed_tasks, total_tasks) # Use MainPanelWidget's method self.main_window.main_panel_widget.update_progress_bar(completed_tasks, total_tasks) # Use MainPanelWidget's method
@@ -558,9 +435,56 @@ if __name__ == "__main__":
log.info("No required CLI arguments detected, starting GUI mode.") log.info("No required CLI arguments detected, starting GUI mode.")
# --- Run the GUI Application --- # --- Run the GUI Application ---
try: try:
user_config_path = app_setup_utils.read_saved_user_config_path()
log.debug(f"Read saved user config path: {user_config_path}")
first_run_needed = False
if user_config_path is None or not user_config_path.strip():
log.info("No saved user config path found. First run setup needed.")
first_run_needed = True
else:
user_config_dir = Path(user_config_path)
marker_file = app_setup_utils.get_first_run_marker_file(user_config_path)
if not user_config_dir.is_dir():
log.warning(f"Saved user config directory does not exist: {user_config_path}. First run setup needed.")
first_run_needed = True
elif not Path(marker_file).is_file():
log.warning(f"First run marker file not found in {user_config_path}. First run setup needed.")
first_run_needed = True
else:
log.info(f"Saved user config path found and valid: {user_config_path}. Marker file exists.")
qt_app = None
if first_run_needed:
log.info("Initiating first-time setup dialog.")
# Need a QApplication instance to show the dialog
qt_app = QApplication.instance()
if qt_app is None:
qt_app = QApplication(sys.argv) qt_app = QApplication(sys.argv)
app_instance = App() dialog = FirstTimeSetupDialog()
if dialog.exec() == QDialog.Accepted:
user_config_path = dialog.get_chosen_path()
log.info(f"First-time setup completed. Chosen path: {user_config_path}")
# The dialog should have already saved the path and created the marker file
else:
log.info("First-time setup cancelled by user. Exiting application.")
sys.exit(0) # Exit gracefully
# If qt_app was created for the dialog, reuse it. Otherwise, create it now.
if qt_app is None:
qt_app = QApplication.instance()
if qt_app is None:
qt_app = QApplication(sys.argv)
# Ensure user_config_path is set before initializing App
if not user_config_path or not Path(user_config_path).is_dir():
log.error(f"Fatal: User config path is invalid or not set after setup: {user_config_path}. Cannot proceed.")
sys.exit(1)
app_instance = App(user_config_path) # Pass the determined path
app_instance.run() app_instance.run()
sys.exit(qt_app.exec()) sys.exit(qt_app.exec())

66
utils/app_setup_utils.py Normal file
View File

@@ -0,0 +1,66 @@
import os
import sys
import platform
def get_app_data_dir():
"""
Gets the OS-specific application data directory for Asset Processor.
Uses standard library methods as appdirs is not available.
"""
app_name = "AssetProcessor"
if platform.system() == "Windows":
# On Windows, use APPDATA environment variable
app_data_dir = os.path.join(os.environ.get("APPDATA", "~"), app_name)
elif platform.system() == "Darwin":
# On macOS, use ~/Library/Application Support
app_data_dir = os.path.join("~", "Library", "Application Support", app_name)
else:
# On Linux and other Unix-like systems, use ~/.config
app_data_dir = os.path.join("~", ".config", app_name)
# Expand the user home directory symbol if present
return os.path.expanduser(app_data_dir)
def get_persistent_config_path_file():
"""
Gets the full path to the file storing the user's chosen config directory.
"""
app_data_dir = get_app_data_dir()
# Ensure the app data directory exists
os.makedirs(app_data_dir, exist_ok=True)
return os.path.join(app_data_dir, "asset_processor_user_root.txt")
def read_saved_user_config_path():
"""
Reads the saved user config path from the persistent file.
Returns the path string or None if the file doesn't exist or is empty.
"""
path_file = get_persistent_config_path_file()
if os.path.exists(path_file):
try:
with open(path_file, "r", encoding="utf-8") as f:
saved_path = f.read().strip()
if saved_path:
return saved_path
except IOError:
# Handle potential file reading errors
pass
return None
def save_user_config_path(user_config_path):
"""
Saves the user's chosen config path to the persistent file.
"""
path_file = get_persistent_config_path_file()
try:
with open(path_file, "w", encoding="utf-8") as f:
f.write(user_config_path)
except IOError:
# Handle potential file writing errors
print(f"Error saving user config path to {path_file}", file=sys.stderr)
def get_first_run_marker_file(user_config_path):
"""
Gets the full path to the first-run marker file within the user config directory.
"""
return os.path.join(user_config_path, ".first_run_complete")