"""The 3-way merge of a YouTube-linked playlist, as a pure function: what each side gains and loses so neither loses an item, the merged order, and the reorder moves YouTube needs to reach it. No I/O — `sync/playlists.py` reads both sides, calls `merge_plan`, and writes the outcome.""" from dataclasses import dataclass def _reorder_moves(current: list[str], desired: list[str]) -> int: """Count the playlistItems.update moves an insertion-sort would need to turn `current` order into `desired` order (both already the same set of video ids). Items already in place are skipped; this matches what apply_push actually executes.""" model = list(current) moves = 0 for i, want in enumerate(desired): if i < len(model) and model[i] == want: continue j = model.index(want, i) model.pop(j) model.insert(i, want) moves += 1 return moves @dataclass class MergePlan: """The 3-way merge of a linked playlist: what each side gains and loses so neither side loses an item, plus the merged order. `final` is the id set both sides end up with.""" final: list[str] # merged order (local order, YouTube-only adds appended) add_to_yt: list[str] # local-adds to push out to YouTube add_to_local: list[str] # YouTube-adds to pull into Siftlode remove_from_yt: list[str] # items deleted locally that are still on YouTube remove_from_local: list[str] # items deleted on YouTube that are still local reorders: int # estimated YouTube playlistItems.update moves to reach `final` baseline_known: bool def merge_plan( baseline: list[str] | None, local: list[str], youtube: list[str], unavailable: set[str] | None = None, ) -> MergePlan: """Pure 3-way merge. Classify each id against the last-synced `baseline`: - in local but not youtube, not deleted → add to YouTube (a local-add) - in youtube but not local, not deleted → add to Siftlode (a YouTube-add) - in baseline but gone from local → a genuine local delete → propagate to YouTube - in baseline but gone from youtube → a genuine YouTube delete → propagate to Siftlode With NO baseline (first-ever merge) we can't tell an add from a delete, so we UNION both sides (never delete) — the lossless default. Merged order: local order first, YouTube-only additions appended in their YouTube order. `unavailable` = known un-ingestable-but-live YouTube ids: dropped from the YouTube side up front so they're never re-classified as a YouTube-add (they can't ingest) nor as a local delete (they were never local) — otherwise such a video would make the playlist never converge. They stay in place on YouTube, untouched by this merge.""" un = set(unavailable or ()) L = list(dict.fromkeys(local)) Y = [v for v in dict.fromkeys(youtube) if v not in un] Lset, Yset = set(L), set(Y) if baseline is None: add_to_yt = [v for v in L if v not in Yset] add_to_local = [v for v in Y if v not in Lset] remove_from_yt: list[str] = [] remove_from_local: list[str] = [] final = L + [v for v in Y if v not in Lset] else: B = set(baseline) deleted_local = B - Lset # was synced, removed on our side deleted_yt = B - Yset # was synced, removed on YouTube's side gone = deleted_local | deleted_yt remove_from_yt = [v for v in Y if v in deleted_local] remove_from_local = [v for v in L if v in deleted_yt] add_to_yt = [v for v in L if v not in Yset and v not in gone] add_to_local = [v for v in Y if v not in Lset and v not in gone] final = [v for v in L if v not in gone] + [v for v in Y if v not in Lset and v not in gone] final_set = set(final) # On YouTube, after membership reconciliation the order is current-kept + inserts appended; # the reorder pass then moves items into `final`. after_membership = [v for v in Y if v in final_set] + add_to_yt reorders = _reorder_moves(after_membership, final) return MergePlan( final=final, add_to_yt=add_to_yt, add_to_local=add_to_local, remove_from_yt=remove_from_yt, remove_from_local=remove_from_local, reorders=reorders, baseline_known=baseline is not None, )