Koha9/PlexPlaylistSync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement Regex Rules backend functionality for path mapping

Browse Source 
- Add CompiledRegexRules dataclass for all four processing stages
- Update _compile_regex_rules to support both legacy (pattern/replacement)
  and new (search/replace) field names with proper empty string handling
- Add _compile_path_mapping_rules helper function
- Update _write_results to apply post-processing rules:
  - local_result.m3u8 with local_post rules
  - remote_result.m3u8 with remote_post rules
  - base_next.m3u8 unprocessed (normalized sync result)
- Update merge_playlists and _sync_single_playlist to pass compiled_rules
- Update sync_all_playlists to implement full processing flow:
  1. Detect REGEX mode from path_mapping config
  2. Apply local_pre rules to local playlists before sync
  3. Apply remote_pre rules to remote playlists before sync
  4. Perform sync/merge
  5. Apply post rules to results for respective outputs

Co-authored-by: Koha9 <36852125+Koha9@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2025-12-02 20:08:06 +00:00
parent f9dbe733c3
commit fbb5bb55c7
1 changed files with 114 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/utils/playlist_merge.py
+114 -19
View File
@@ -40,6 +40,15 @@ class PlaylistSyncResult:
output_dir: str
@dataclass
class CompiledRegexRules:
"""Holds compiled regex rules for all four processing stages."""
local_pre: list[tuple[re.Pattern[str], str]]
local_post: list[tuple[re.Pattern[str], str]]
remote_pre: list[tuple[re.Pattern[str], str]]
remote_post: list[tuple[re.Pattern[str], str]]
def load_paths(text: str) -> list[str]:
"""Normalize playlist text into a list of absolute paths.
@@ -72,12 +81,21 @@ def save_paths(paths: Sequence[str]) -> str:
def _compile_regex_rules(rules: Sequence[dict[str, str]]) -> list[tuple[re.Pattern[str], str]]:
"""Compile regex rules into pattern/replacement pairs.
Supports both legacy format (pattern/replacement) and new format (search/replace).
"""
compiled: list[tuple[re.Pattern[str], str]] = []
for rule in rules:
pattern = rule.get("pattern")
# Support both legacy (pattern/replacement) and new (search/replace) field names
# Use explicit None checks to allow empty strings as valid values
pattern = rule.get("pattern") if rule.get("pattern") is not None else rule.get("search")
if not pattern:
continue
replacement = rule.get("replacement", "")
# For replacement, empty string is a valid value (for deletion)
replacement = rule.get("replacement") if rule.get("replacement") is not None else rule.get("replace")
if replacement is None:
replacement = ""
try:
compiled.append((re.compile(pattern), replacement))
except re.error as exc:
@@ -234,9 +252,31 @@ def _merge_chunks(
return chunks
def _write_results(merged_lines: Sequence[str], folder: str) -> None:
_save_playlist_to_folder("local_result.m3u8", merged_lines, folder)
_save_playlist_to_folder("remote_result.m3u8", merged_lines, folder)
def _write_results(
merged_lines: Sequence[str],
folder: str,
compiled_rules: CompiledRegexRules | None = None
) -> None:
"""Write sync results to the test folder.
If compiled_rules is provided with post-processing rules:
- local_result.m3u8: merged_lines processed with local_post rules
- remote_result.m3u8: merged_lines processed with remote_post rules
- base_next.m3u8: unprocessed merged_lines (normalized sync result)
"""
# Apply post-processing regex rules if provided
if compiled_rules and compiled_rules.local_post:
local_lines = _apply_compiled_rules_to_paths(merged_lines, compiled_rules.local_post)
else:
local_lines = list(merged_lines)
if compiled_rules and compiled_rules.remote_post:
remote_lines = _apply_compiled_rules_to_paths(merged_lines, compiled_rules.remote_post)
else:
remote_lines = list(merged_lines)
_save_playlist_to_folder("local_result.m3u8", local_lines, folder)
_save_playlist_to_folder("remote_result.m3u8", remote_lines, folder)
_save_playlist_to_folder("base_next.m3u8", merged_lines, folder)
@@ -379,12 +419,16 @@ def merge_playlists(
remote_text: str,
strategy: ConflictResolutionStrategy = ConflictResolutionStrategy.LOCAL_PRIORITY,
test_folder: str = TEST_PLAYLIST_DIR,
compiled_rules: CompiledRegexRules | None = None,
) -> MergeResult:
"""Merge playlists using diff3 and resolve conflicts per strategy.
The base, local, and remote normalized playlists are saved into ``test_folder``
for inspection. The merged playlist is also stored twice to simulate the
versions intended for local save and cloud upload.
If compiled_rules is provided, post-processing regex rules will be applied
to the results before writing.
"""
base_paths, local_paths, remote_paths = _normalize_inputs(
@@ -420,7 +464,7 @@ def merge_playlists(
merged_lines, base_paths, local_paths, remote_paths
)
_write_results(merged_lines, test_folder)
_write_results(merged_lines, test_folder, compiled_rules)
return MergeResult(merged_paths=merged_lines, conflicts=conflicts)
@@ -517,6 +561,7 @@ def _sync_single_playlist(
remote_text: str,
playlist_folder: str,
remote_present: bool,
compiled_rules: CompiledRegexRules | None = None,
) -> PlaylistSyncResult:
local_present = local_text is not None
local_text = local_text or ""
@@ -535,7 +580,7 @@ def _sync_single_playlist(
base_text, local_text, remote_text, playlist_folder
)
merged_lines = list(local_paths)
_write_results(merged_lines, playlist_folder)
_write_results(merged_lines, playlist_folder, compiled_rules)
return PlaylistSyncResult(playlist, merged_lines, [], "synced", playlist_folder)
if mode == SyncMode.REMOTE_FORCE:
@@ -547,7 +592,7 @@ def _sync_single_playlist(
base_text, local_text, remote_text, playlist_folder
)
merged_lines = list(remote_paths)
_write_results(merged_lines, playlist_folder)
_write_results(merged_lines, playlist_folder, compiled_rules)
return PlaylistSyncResult(playlist, merged_lines, [], "synced", playlist_folder)
if mode not in (SyncMode.MERGE_LOCAL_PRIMARY, SyncMode.MERGE_REMOTE_PRIMARY):
@@ -565,6 +610,7 @@ def _sync_single_playlist(
remote_text=remote_text,
strategy=merge_strategy,
test_folder=playlist_folder,
compiled_rules=compiled_rules,
)
if not merge_result.merged_paths and (not local_present or not remote_present):
@@ -578,13 +624,48 @@ def _sync_single_playlist(
)
def _compile_path_mapping_rules(path_mapping: dict) -> CompiledRegexRules:
"""Compile regex rules from path_mapping config for all four processing stages."""
regex_config = path_mapping.get("regex", {})
return CompiledRegexRules(
local_pre=_compile_regex_rules(regex_config.get("local_pre", [])),
local_post=_compile_regex_rules(regex_config.get("local_post", [])),
remote_pre=_compile_regex_rules(regex_config.get("remote_pre", [])),
remote_post=_compile_regex_rules(regex_config.get("remote_post", [])),
)
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."""
"""Synchronize all playlists that can be matched by name.
When path_mapping mode is REGEX, the following processing flow is applied:
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
4. local_post rules are applied to results before writing to local_result.m3u8
5. remote_post rules are applied to results before writing to remote_result.m3u8
"""
server_config.load()
compiled_rules = _compile_regex_rules(server_config.path_rules)
# Check if we should use the new path_mapping REGEX mode
path_mapping = server_config.path_mapping
use_regex_mode = path_mapping.get("mode") == "REGEX"
# Compile rules based on the mode
compiled_rules: CompiledRegexRules | None = None
legacy_compiled_rules: list[tuple[re.Pattern[str], str]] = []
if use_regex_mode:
compiled_rules = _compile_path_mapping_rules(path_mapping)
logger.info("Using REGEX mode for path mapping with 4 rule groups")
else:
# Use legacy path_rules for backward compatibility
legacy_compiled_rules = _compile_regex_rules(server_config.path_rules)
logger.info("Using legacy path_rules for preprocessing")
_ensure_test_dir(test_folder)
logger.info(f"Syncing playlists to test folder: {test_folder}")
local_playlists = _load_local_playlists(local_dir)
@@ -613,16 +694,29 @@ 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
if use_regex_mode and compiled_rules:
# Apply pre-processing rules for REGEX 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(
local_text, [], compiled_rules.local_pre
)
if remote_text and compiled_rules.remote_pre:
remote_text = preprocess_playlist_text(
remote_text, [], compiled_rules.remote_pre
)
else:
# Use legacy preprocessing for all texts
base_text = preprocess_playlist_text(
base_text, server_config.path_rules, legacy_compiled_rules
)
remote_text = preprocess_playlist_text(
remote_text, server_config.path_rules, legacy_compiled_rules
)
if local_text is not None:
local_text = preprocess_playlist_text(
local_text, server_config.path_rules, legacy_compiled_rules
)
# Treat missing remote text as absent playlist.
result = _sync_single_playlist(
@@ -633,6 +727,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,
)
results.append(result)