from __future__ import annotations from difflib import SequenceMatcher from typing import Callable, Iterator, Literal, Mapping, Sequence, Union, cast from src.enums import ServiceType __all__ = ["get_playlist_id", "compare_sequences"] SERVICE_TYPE_ID_MAP = {ServiceType.spotify: lambda p_id: p_id.split(":")[-1]} InsertActionType = Literal["insert"] DeleteActionType = Literal["delete"] ReplaceActionType = Literal["replace"] ActionType = Union[InsertActionType, DeleteActionType, ReplaceActionType] INSERT_ACTION: InsertActionType = "insert" DELETE_ACTION: DeleteActionType = "delete" REPLACE_ACTION: ReplaceActionType = "replace" OPERATION_OFFSET_MAP: Mapping[ActionType, Callable[[int, int, int, int], int]] = { INSERT_ACTION: lambda t1, t2, s1, s2: s2 - s1, DELETE_ACTION: lambda t1, t2, s1, s2: -(t2 - t1), REPLACE_ACTION: lambda t1, t2, s1, s2: s2 - s1 - (t2 - t1), } def get_playlist_id(service_type: ServiceType, playlist_id: str) -> str: if not playlist_id: return playlist_id return SERVICE_TYPE_ID_MAP.get(service_type, lambda p_id: p_id)(playlist_id) def compare_sequences(source: Sequence, target: Sequence) -> Iterator[tuple[ActionType, int, int, int, int]]: squeeze = SequenceMatcher(a=target, b=source) offset = 0 for action, t1, t2, s1, s2 in squeeze.get_opcodes(): action = cast(ActionType, action) if action not in OPERATION_OFFSET_MAP: continue yield action, t1 + offset, t2 + offset, s1, s2 offset += OPERATION_OFFSET_MAP[action](t1, t2, s1, s2)