# This file is part of beets. # # Permission is hereby granted, free of charge, to any person obtaining # a copy of this software and associated documentation files (the # "Software"), to deal in the Software without restriction, including # without limitation the rights to use, copy, modify, merge, publish, # distribute, sublicense, and/or sell copies of the Software, and to # permit persons to whom the Software is furnished to do so, subject to # the following conditions: # # The above copyright notice and this permission notice shall be # included in all copies or substantial portions of the Software. """Adds support for ipfs. Requires go-ipfs and a running ipfs daemon""" import os import shutil import subprocess import tempfile from beets import config, library, ui, util from beets.plugins import BeetsPlugin class IPFSPlugin(BeetsPlugin): def __init__(self): super().__init__() self.config.add( { "auto": True, "nocopy": False, } ) if self.config["auto"]: self.import_stages = [self.auto_add] def commands(self): cmd = ui.Subcommand("ipfs", help="interact with ipfs") cmd.parser.add_option( "-a", "--add", dest="add", action="store_true", help="Add to ipfs" ) cmd.parser.add_option( "-g", "--get", dest="get", action="store_true", help="Get from ipfs" ) cmd.parser.add_option( "-p", "--publish", dest="publish", action="store_true", help="Publish local library to ipfs", ) cmd.parser.add_option( "-i", "--import", dest="_import", action="store_true", help="Import remote library from ipfs", ) cmd.parser.add_option( "-l", "--list", dest="_list", action="store_true", help="Query imported libraries", ) cmd.parser.add_option( "-m", "--play", dest="play", action="store_true", help="Play music from remote libraries", ) def func(lib, opts, args): if opts.add: for album in lib.albums(args): if len(album.items()) == 0: self._log.info( "{} does not contain items, aborting", album ) self.ipfs_add(album) album.store() if opts.get: self.ipfs_get(lib, args) if opts.publish: self.ipfs_publish(lib) if opts._import: self.ipfs_import(lib, args) if opts._list: self.ipfs_list(lib, args) if opts.play: self.ipfs_play(lib, opts, args) cmd.func = func return [cmd] def auto_add(self, session, task): if task.is_album: if self.ipfs_add(task.album): task.album.store() def ipfs_play(self, lib, opts, args): from beetsplug.play import PlayPlugin jlib = self.get_remote_lib(lib) player = PlayPlugin() config["play"]["relative_to"] = None player.album = True player.play_music(jlib, player, args) def ipfs_add(self, album): try: album_dir = album.item_dir() except AttributeError: return False try: if album.ipfs: self._log.debug("{} already added", album_dir) # Already added to ipfs return False except AttributeError: pass self._log.info("Adding {} to ipfs", album_dir) if self.config["nocopy"]: cmd = "ipfs add --nocopy -q -r".split() else: cmd = "ipfs add -q -r".split() cmd.append(album_dir) try: output = util.command_output(cmd).stdout.split() except (OSError, subprocess.CalledProcessError) as exc: self._log.error("Failed to add {}, error: {}", album_dir, exc) return False length = len(output) for linenr, line in enumerate(output): line = line.strip() if linenr == length - 1: # last printed line is the album hash self._log.info("album: {}", line) album.ipfs = line else: try: item = album.items()[linenr] self._log.info("item: {}", line) item.ipfs = line item.store() except IndexError: # if there's non music files in the to-add folder they'll # get ignored here pass return True def ipfs_get(self, lib, query): query = query[0] # Check if query is a hash # TODO: generalize to other hashes; probably use a multihash # implementation if query.startswith("Qm") and len(query) == 46: self.ipfs_get_from_hash(lib, query) else: albums = self.query(lib, query) for album in albums: self.ipfs_get_from_hash(lib, album.ipfs) def ipfs_get_from_hash(self, lib, _hash): try: cmd = "ipfs get".split() cmd.append(_hash) util.command_output(cmd) except (OSError, subprocess.CalledProcessError) as err: self._log.error( "Failed to get {} from ipfs.\n{.output}", _hash, err ) return False self._log.info("Getting {} from ipfs", _hash) imp = ui.commands.TerminalImportSession( lib, loghandler=None, query=None, paths=[_hash] ) imp.run() # This uses a relative path, hence we cannot use util.syspath(_hash, # prefix=True). However, that should be fine since the hash will not # exceed MAX_PATH. shutil.rmtree(util.syspath(_hash, prefix=False)) def ipfs_publish(self, lib): with tempfile.NamedTemporaryFile() as tmp: self.ipfs_added_albums(lib, tmp.name) try: if self.config["nocopy"]: cmd = "ipfs add --nocopy -q ".split() else: cmd = "ipfs add -q ".split() cmd.append(tmp.name) output = util.command_output(cmd).stdout except (OSError, subprocess.CalledProcessError) as err: msg = f"Failed to publish library. Error: {err}" self._log.error(msg) return False self._log.info("hash of library: {}", output) def ipfs_import(self, lib, args): _hash = args[0] if len(args) > 1: lib_name = args[1] else: lib_name = _hash lib_root = os.path.dirname(lib.path) remote_libs = os.path.join(lib_root, b"remotes") if not os.path.exists(remote_libs): try: os.makedirs(remote_libs) except OSError as e: msg = f"Could not create {remote_libs}. Error: {e}" self._log.error(msg) return False path = os.path.join(remote_libs, lib_name.encode() + b".db") if not os.path.exists(path): cmd = f"ipfs get {_hash} -o".split() cmd.append(path) try: util.command_output(cmd) except (OSError, subprocess.CalledProcessError): self._log.error("Could not import {}", _hash) return False # add all albums from remotes into a combined library jpath = os.path.join(remote_libs, b"joined.db") jlib = library.Library(jpath) nlib = library.Library(path) for album in nlib.albums(): if not self.already_added(album, jlib): new_album = [] for item in album.items(): item.id = None new_album.append(item) added_album = jlib.add_album(new_album) added_album.ipfs = album.ipfs added_album.store() def already_added(self, check, jlib): for jalbum in jlib.albums(): if jalbum.mb_albumid == check.mb_albumid: return True return False def ipfs_list(self, lib, args): fmt = config["format_album"].get() try: albums = self.query(lib, args) except OSError: ui.print_("No imported libraries yet.") return for album in albums: ui.print_(format(album, fmt), " : ", album.ipfs.decode()) def query(self, lib, args): rlib = self.get_remote_lib(lib) albums = rlib.albums(args) return albums def get_remote_lib(self, lib): lib_root = os.path.dirname(lib.path) remote_libs = os.path.join(lib_root, b"remotes") path = os.path.join(remote_libs, b"joined.db") if not os.path.isfile(path): raise OSError return library.Library(path) def ipfs_added_albums(self, rlib, tmpname): """Returns a new library with only albums/items added to ipfs""" tmplib = library.Library(tmpname) for album in rlib.albums(): try: if album.ipfs: self.create_new_album(album, tmplib) except AttributeError: pass return tmplib def create_new_album(self, album, tmplib): items = [] for item in album.items(): try: if not item.ipfs: break except AttributeError: pass item_path = os.fsdecode(os.path.basename(item.path)) # Clear current path from item item.path = f"/ipfs/{album.ipfs}/{item_path}" item.id = None items.append(item) if len(items) < 1: return False self._log.info("Adding '{}' to temporary library", album) new_album = tmplib.add_album(items) new_album.ipfs = album.ipfs new_album.store(inherit=False)