Implement Simple Mapping backend functionality

- Add _compile_simple_mapping_rules() that generates four rule sets from mapping pairs
- Each mapping uses UUID as unique mapping_id with special markers (__MAPPING__uuid__)
- local_pre: local_path → mapping_id
- remote_pre: cloud_path → mapping_id
- local_post: mapping_id → local_path
- remote_post: mapping_id → cloud_path
- Add UUID validation to prevent injection attacks
- Update sync_all_playlists() to detect and use SIMPLE mode

Co-authored-by: Koha9 <36852125+Koha9@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2025-12-02 21:15:14 +00:00
parent fbb5bb55c7
commit fbafe75fae
+113 -8
View File
@@ -635,12 +635,110 @@ def _compile_path_mapping_rules(path_mapping: dict) -> CompiledRegexRules:
)
def _compile_simple_mapping_rules(simple_mappings: list[dict]) -> CompiledRegexRules:
"""Compile simple mapping pairs into four rule groups using UUID-based mapping_ids.
Each simple mapping has:
- id: UUID used as the mapping_id (unique identifier to prevent conflicts)
- search: Local path prefix
- replace: Cloud path prefix
This generates four rule sets:
- local_pre: Replace local path (search) with mapping_id
- remote_pre: Replace cloud path (replace) with mapping_id
- local_post: Replace mapping_id with local path (search)
- remote_post: Replace mapping_id with cloud path (replace)
The mapping_id is wrapped with special markers to prevent conflicts with actual paths.
"""
local_pre_rules: list[dict[str, str]] = []
local_post_rules: list[dict[str, str]] = []
remote_pre_rules: list[dict[str, str]] = []
remote_post_rules: list[dict[str, str]] = []
# UUID pattern for validation (accepts standard UUID format with or without hyphens)
uuid_pattern = re.compile(r'^[a-fA-F0-9]{8}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{12}$')
for mapping in simple_mappings:
# Get the mapping values
mapping_id = mapping.get("id")
local_path = mapping.get("search", "") # Local path is stored in 'search' field
cloud_path = mapping.get("replace", "") # Cloud path is stored in 'replace' field
# Validate mapping_id is a proper UUID to prevent injection attacks
if not mapping_id or not isinstance(mapping_id, str):
logger.warning(f"Skipping mapping with missing or invalid id: {mapping}")
continue
if not uuid_pattern.match(mapping_id):
logger.warning(f"Skipping mapping with non-UUID id format: {mapping_id}")
continue
# Paths must be non-empty strings
if not local_path or not isinstance(local_path, str):
logger.warning(f"Skipping mapping with missing local path: {mapping}")
continue
if not cloud_path or not isinstance(cloud_path, str):
logger.warning(f"Skipping mapping with missing cloud path: {mapping}")
continue
# Create a unique placeholder using the validated UUID
# Using special markers to prevent conflicts with actual paths
placeholder = f"__MAPPING__{mapping_id}__"
# Pre-processing rules (use re.escape to treat paths as literal strings)
# local_pre: Replace local path with placeholder
local_pre_rules.append({
"pattern": re.escape(local_path),
"replacement": placeholder
})
# remote_pre: Replace cloud path with placeholder
remote_pre_rules.append({
"pattern": re.escape(cloud_path),
"replacement": placeholder
})
# Post-processing rules
# local_post: Replace placeholder with local path
local_post_rules.append({
"pattern": re.escape(placeholder),
"replacement": local_path
})
# remote_post: Replace placeholder with cloud path
remote_post_rules.append({
"pattern": re.escape(placeholder),
"replacement": cloud_path
})
logger.info(f"Compiled {len(local_pre_rules)} simple mapping pairs into rules")
return CompiledRegexRules(
local_pre=_compile_regex_rules(local_pre_rules),
local_post=_compile_regex_rules(local_post_rules),
remote_pre=_compile_regex_rules(remote_pre_rules),
remote_post=_compile_regex_rules(remote_post_rules),
)
def sync_all_playlists(
local_dir: str, mode: SyncMode, test_folder: str = TEST_PLAYLIST_DIR
) -> list[PlaylistSyncResult]:
"""Synchronize all playlists that can be matched by name.
When path_mapping mode is REGEX, the following processing flow is applied:
Path mapping modes:
- SIMPLE: Uses UUID-based mapping_ids to convert between local and cloud paths
- local_pre: local_path -> mapping_id
- remote_pre: cloud_path -> mapping_id
- local_post: mapping_id -> local_path
- remote_post: mapping_id -> cloud_path
- REGEX: Uses custom regex rules for each processing stage
- local_pre, local_post, remote_pre, remote_post rules are applied directly
Processing flow:
1. local_pre rules are applied to local playlists before sync
2. remote_pre rules are applied to remote playlists before sync
3. Sync/merge is performed
@@ -650,17 +748,24 @@ def sync_all_playlists(
server_config.load()
# Check if we should use the new path_mapping REGEX mode
# Get path_mapping configuration
path_mapping = server_config.path_mapping
use_regex_mode = path_mapping.get("mode") == "REGEX"
mapping_mode = path_mapping.get("mode", "SIMPLE")
# Compile rules based on the mode
compiled_rules: CompiledRegexRules | None = None
legacy_compiled_rules: list[tuple[re.Pattern[str], str]] = []
if use_regex_mode:
if mapping_mode == "REGEX":
compiled_rules = _compile_path_mapping_rules(path_mapping)
logger.info("Using REGEX mode for path mapping with 4 rule groups")
elif mapping_mode == "SIMPLE":
simple_mappings = path_mapping.get("simple", [])
if simple_mappings:
compiled_rules = _compile_simple_mapping_rules(simple_mappings)
logger.info(f"Using SIMPLE mode for path mapping with {len(simple_mappings)} mapping pairs")
else:
logger.info("SIMPLE mode with no mappings - no path transformations will be applied")
else:
# Use legacy path_rules for backward compatibility
legacy_compiled_rules = _compile_regex_rules(server_config.path_rules)
@@ -694,8 +799,8 @@ def sync_all_playlists(
remote_text = snapshot_remote_text
remote_present = bool(remote_text.strip()) or remote_exists
if use_regex_mode and compiled_rules:
# Apply pre-processing rules for REGEX mode
if compiled_rules:
# Apply pre-processing rules for REGEX or SIMPLE mode
# base_text doesn't need pre-processing as it's the normalized state
if local_text is not None and compiled_rules.local_pre:
local_text = preprocess_playlist_text(
@@ -705,7 +810,7 @@ def sync_all_playlists(
remote_text = preprocess_playlist_text(
remote_text, [], compiled_rules.remote_pre
)
else:
elif legacy_compiled_rules:
# Use legacy preprocessing for all texts
base_text = preprocess_playlist_text(
base_text, server_config.path_rules, legacy_compiled_rules
@@ -727,7 +832,7 @@ def sync_all_playlists(
remote_text=remote_text,
playlist_folder=playlist_folder,
remote_present=remote_present,
compiled_rules=compiled_rules if use_regex_mode else None,
compiled_rules=compiled_rules,
)
results.append(result)