Add regex-based path preprocessing for playlist sync

app/utils/playlist_merge.py

@@ -1,4 +1,5 @@
 import os
+import re
 from collections import Counter
 from dataclasses import dataclass
 from enum import Enum
@@ -70,6 +71,60 @@ def save_paths(paths: Sequence[str]) -> str:
     return "#EXTM3U\n" + "\n".join(paths) + "\n"
 
 
+def _compile_regex_rules(rules: Sequence[dict[str, str]]) -> list[tuple[re.Pattern[str], str]]:
+    compiled: list[tuple[re.Pattern[str], str]] = []
+    for rule in rules:
+        pattern = rule.get("pattern")
+        if not pattern:
+            continue
+        replacement = rule.get("replacement", "")
+        try:
+            compiled.append((re.compile(pattern), replacement))
+        except re.error as exc:
+            logger.warning(f"Skipping invalid regex '{pattern}': {exc}")
+    return compiled
+
+
+def _apply_compiled_rules_to_paths(
+    paths: Sequence[str], compiled_rules: Sequence[tuple[re.Pattern[str], str]]
+) -> list[str]:
+    if not compiled_rules:
+        return list(paths)
+
+    updated: list[str] = []
+    for path in paths:
+        new_path = path
+        for pattern, replacement in compiled_rules:
+            new_path = pattern.sub(replacement, new_path)
+        updated.append(new_path)
+    return updated
+
+
+def apply_regex_rules_to_paths(
+    paths: Sequence[str], rules: Sequence[dict[str, str]]
+) -> list[str]:
+    """Apply regex replacement rules to each path in order."""
+
+    compiled_rules = _compile_regex_rules(rules)
+    return _apply_compiled_rules_to_paths(paths, compiled_rules)
+
+
+def preprocess_playlist_text(
+    text: str,
+    rules: Sequence[dict[str, str]],
+    compiled_rules: Sequence[tuple[re.Pattern[str], str]] | None = None,
+) -> str:
+    """Normalize playlist text and apply regex replacements."""
+
+    if not text:
+        text = "#EXTM3U\n"
+
+    paths = load_paths(text)
+    compiled = compiled_rules if compiled_rules is not None else _compile_regex_rules(rules)
+    replaced_paths = _apply_compiled_rules_to_paths(paths, compiled)
+    return save_paths(replaced_paths)
+
+
 @dataclass
 class MergeChunk:
     type: Literal["normal", "conflict"]
@@ -514,6 +569,8 @@ def sync_all_playlists(
 ) -> list[PlaylistSyncResult]:
     """Synchronize all playlists that can be matched by name."""
 
+    server_config.load()
+    compiled_rules = _compile_regex_rules(server_config.path_rules)
     _ensure_test_dir(test_folder)
     local_playlists = _load_local_playlists(local_dir)
     remote_playlists = _fetch_remote_playlists()
@@ -541,6 +598,17 @@ def sync_all_playlists(
             remote_text = snapshot_remote_text
             remote_present = bool(remote_text.strip()) or remote_exists
 
+        base_text = preprocess_playlist_text(
+            base_text, server_config.path_rules, compiled_rules
+        )
+        remote_text = preprocess_playlist_text(
+            remote_text, server_config.path_rules, compiled_rules
+        )
+        if local_text is not None:
+            local_text = preprocess_playlist_text(
+                local_text, server_config.path_rules, compiled_rules
+            )
+
         # Treat missing remote text as absent playlist.
         result = _sync_single_playlist(
             playlist=playlist,