subimporter/migrater.py at master · peguerosdc/subimporter · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

from utils import stringifySong
from datetime import datetime

import logging
logger = logging.getLogger(__name__)

class Migrater(object):

    """Migrater"""
    def __init__(self, migrateFrom, migrateTo, mock=False):
        # Store clients
        self.source = migrateFrom
        self.target = migrateTo
        self.isMocked = mock
        logger.info(f"MOCK MIGRATION? {mock}")

    def migratePlaylists(self, interactive=True):
        # Wrapper to gather input from the user
        def should(text):
            value = input(text)
            return value.strip().lower() == 'y'
        # Core to migrate each playlist
        def migratePlaylist(playlist):
            # Create a new playlist in the target server with a unique name
            if not self.isMocked:
                targetPlaylist = self.target.createPlaylist(f"{playlist['name']} at {int(datetime.now().timestamp())}")
                logger.info(f"Target playlist: '{targetPlaylist['name']}'")
            # Add each song in this playlist to the new targetPlaylist
            playlist = self.source.getPlaylist(playlist["id"])
            for song in playlist['entry']:
                try:
                    matchSong = self.target.findClosestMatchToSong(song)
                    if matchSong:
                        logger.info(f"Migrating {stringifySong(song)} as {stringifySong(matchSong)}")
                        if not self.isMocked:
                            self.target.addSongToPlaylist(targetPlaylist['id'], matchSong['id'])
                    else:
                        logger.warning(f"No match to {stringifySong(song)}")
                except Exception as e:
                    logger.exception(f"Unable to migrate song {stringifySong(song)}")
        # Perform playlsts migration
        try:
            # Get all playlists
            playlists = self.source.getPlaylists()
            for playlist in playlists:
                # Ask if this playlist should be migrated only if running on interactive mode
                if not interactive or should(f"Migrate '{playlist['name']}' with {playlist['songCount']} songs Y/[N]? "):
                    logger.info(f"Migrating playlist '{playlist['name']} ({playlist['songCount']})'")
                    try:
                        migratePlaylist(playlist)
                    except Exception as e:
                        logger.exception(f"Unable to migrate playlist '{playlist['name']} ({playlist['songCount']})'")
        except Exception as e:
            logger.exception("Unable to migrate playlists")

    def migrateStarred(self):
        # TODO: support albums and artists too
        try:
            songs = self.source.getStarredSongs()
            # Migrate each song
            logger.info(f"{len(songs)} starred songs to migrate")
            for song in songs:
                try:
                    matchSong = self.target.findClosestMatchToSong(song)
                    if matchSong:
                        logger.info(f"Migrating {stringifySong(song)} as {stringifySong(matchSong)}")
                        if not self.isMocked:
                            self.target.starSong(matchSong['id'])
                    else:
                        logger.warning(f"No match to {stringifySong(song)}")
                except Exception as e:
                    logger.exception(f"Unable to star song {stringifySong(song)}")
        except Exception as e:
            logger.exception("Unable to migrate starred songs")