cli: switch to click

This commit is contained in:
Torrie Fischer 2023-07-01 13:39:46 +02:00
parent 55c5bc489c
commit 1bea5f5a6d
11 changed files with 362 additions and 352 deletions

1
.gitignore vendored
View File

@ -1,2 +1,3 @@
*.swp
Pipfile.lock
build

View File

@ -6,6 +6,9 @@ name = "pypi"
[packages]
pyyaml = "*"
semantic-version = "*"
click = "*"
minecraft-package-manager = {editable = true, path = "."}
columnar = "*"
[dev-packages]

View File

@ -9,4 +9,4 @@ name = "minecraft-package-manager"
version = "0.0.1"
[project.scripts]
mpm = "mpm.main:main"
mpm = "mpm.commands:mpm"

View File

@ -0,0 +1,16 @@
from . import server, repo, config
from mpm.config import Config
import click
@click.group()
@click.option("--config", type=click.Path(readable=True, writable=False,
dir_okay=False))
@click.pass_context
def mpm(ctx, config):
ctx.ensure_object(dict)
ctx.obj['config'] = Config(config)
pass
mpm.add_command(config.config)
mpm.add_command(server.server)
mpm.add_command(repo.repo)

View File

@ -0,0 +1,14 @@
import click
import sys
import yaml
from mpm.config import Dumper
@click.group(help='Configure MPM')
def config():
pass
@config.command(help='Show the current configuration')
@click.pass_context
def show(ctx):
sys.stdout.write("# {}\n".format(ctx.obj['config'].path))
yaml.dump(ctx.obj['config'].config, sys.stdout, Dumper=Dumper)

77
src/mpm/commands/repo.py Normal file
View File

@ -0,0 +1,77 @@
import click
import os
from mpm.model import Plugin
from mpm.config import RepoParamType
import pathlib
from columnar import columnar
@click.group(help='Add and import plugins to repositories')
def repo():
pass
@repo.command('list', help='List configured repositories')
@click.pass_context
def list_(ctx):
for repo in ctx.obj['config'].repositories():
click.echo("{} ({})".format(click.style(repo.name,bold=True), repo.path))
rows = []
for plugin in sorted(repo.plugins()):
rows.append([
click.style(plugin.name, fg='green'),
click.style(plugin.version, fg='yellow')
])
if len(rows) > 0:
click.echo(columnar(rows, ['Plugin', 'Version'], no_borders=True))
else:
click.echo("No plugins found.")
for badFile in sorted(repo.badFiles()):
click.echo('\tWARNING: Unknown file {}'.format(badFile))
@repo.command(help='Add a new repository')
@click.pass_context
@click.argument("name")
@click.argument("path", type=click.Path(path_type=pathlib.Path))
def add(ctx, name, path):
path.mkdir(parents=True, exist_ok=True)
ctx.obj['config'].add_repository(name, path)
ctx.obj['config'].save()
click.echo("Added repository {} from {}".format(name, path))
@repo.command('import', help='Import a plugin to a repository')
@click.argument('repo', type=RepoParamType())
@click.argument('paths', type=click.Path(exists=True, readable=True,
path_type=pathlib.Path), nargs=-1)
def import_(repo, paths):
plugins = []
searchPaths = list(paths)
for path in searchPaths:
try:
if path.is_dir():
searchPaths += list(path.glob("*"))
continue
except PermissionError:
click.echo("Cannot read {}".format(path))
continue
try:
plugins.append(Plugin(path))
except:
click.echo("Bad plugin filename {}".format(path))
if len(plugins) == 0:
click.echo("No plugins found.")
return
click.echo('Found the following plugins:')
for plugin in plugins:
click.echo("\t{} {}".format(click.style(plugin.name, fg='green'),
click.style(plugin.version, fg='yellow')))
click.echo("Import plugins into {} ({})? [y/N] ".format(repo.name,
repo.path), nl=False)
answer = click.getchar().lower()
if answer == "y":
with click.progressbar(plugins, label='Importing') as bar:
for plugin in bar:
repo.importPlugin(plugin)
click.echo("Imported!")
else:
click.echo("Cancelled.")

150
src/mpm/commands/server.py Normal file
View File

@ -0,0 +1,150 @@
import click
from mpm.model import *
import pathlib
from mpm.config import ServerParamType
@click.group(help='Add, remove, and synchronize Servers')
def server():
pass
@server.command(help='Add a new server')
@click.pass_context
@click.argument('name')
@click.argument('path', type=click.Path(dir_okay=True, exists=True))
def add(ctx, name, path):
ctx.obj['config'].add_server(name, path)
ctx.obj['config'].save()
click.echo("Added server {} in {}".format(name, path))
@server.command('list', help='List configured servers')
@click.pass_context
def list_(ctx):
for server in ctx.obj['config'].servers():
click.echo('{} ({}):'.format(server.name, server.path))
serverPath = pathlib.Path(server.path)
if not serverPath.exists():
click.echo(click.style("\tServer path does not exist", fg='red'))
continue
pluginPath = pathlib.Path(server.path+'/plugins')
if not pluginPath.exists():
click.echo(click.style("\tServer plugin directory does not exist", fg='red'))
continue
outdatedLinks = []
missing = []
installed = []
unmanaged = []
conflicts = []
for state in sorted(server.pluginStates(ctx.obj['config'].repositories())):
if isinstance(state, OutdatedSymlink):
outdatedLinks.append(state)
elif isinstance(state, Installed):
installed.append(state)
elif isinstance(state, MissingVersions):
missing.append(state)
elif isinstance(state, UnmanagedFile):
unmanaged.append(state)
elif isinstance(state, SymlinkConflict):
conflicts.append(state)
click.echo("Installed plugins:")
for state in sorted(installed):
click.echo("\t{} {}: {}".format(state.plugin.name, state.plugin.versionSpec, state.currentVersion))
click.echo("Oudated symlinks:")
for state in sorted(outdatedLinks):
click.echo("\t{} {}: Current: {} Wanted: {}".format(state.plugin.name, state.plugin.versionSpec, state.currentVersion, state.wantedVersion))
click.echo("Missing plugins:")
for state in sorted(missing):
click.echo("\t{}: {}".format(state.plugin.name, state.plugin.versionSpec))
click.echo("Unmanaged files:")
for state in sorted(unmanaged):
click.echo("\t{}".format(state.filename))
click.echo("Symlink Conflicts:")
for state in sorted(conflicts):
click.echo("\t{}.jar".format(state.plugin.name))
@server.command(help='Add a plugin to a server')
@click.pass_context
@click.argument('server', type=ServerParamType())
@click.argument('plugins', nargs=-1)
def add_plugin(ctx, server, plugins):
pluginQueue = []
for pluginSpec in plugins:
if os.path.exists(pluginSpec):
plugin = Plugin(pluginSpec)
pluginSpec = PluginSpec(plugin.name, str(plugin.version))
else:
allVersions = []
for repo in ctx.obj['config'].repositories():
allVersions += repo.versionsForPlugin(pluginSpec)
pluginSpec = PluginSpec(pluginSpec, list(reversed(sorted(allVersions)))[0])
pluginQueue.append(pluginSpec)
click.echo("Resolved plugin list:")
for pluginSpec in pluginQueue:
click.echo("\t{} {}".format(click.style(pluginSpec.name, fg='green'),
click.style(pluginSpec.versionSpec,
fg='yellow')))
click.echo("Add these plugins to server {}? [y/N] ".format(server.name),
nl=False)
answer = click.getchar().lower()
if answer == "y":
for pluginSpec in pluginQueue:
try:
server.add_plugin(pluginSpec)
except KeyError:
click.echo("{} already added!".format(pluginSpec))
ctx.obj['config'].update_server(server.name, server.config)
ctx.obj['config'].save()
click.echo("Added!")
else:
click.echo("Cancelled.")
@server.command(help="Synchronize a server's plugins")
@click.pass_context
def sync(ctx):
for server in ctx.obj['config'].servers():
click.echo('{} ({}):'.format(server.name, server.path))
outdatedLinks = []
available = []
missing = []
for state in sorted(server.pluginStates(ctx.obj['config'].repositories())):
if isinstance(state, OutdatedSymlink):
outdatedLinks.append(state)
elif isinstance(state, Available):
available.append(state)
elif isinstance(state, MissingVersions):
missing.append(state)
click.echo("Plugins to update:")
for state in sorted(outdatedLinks):
click.echo("\t{} {}: Current: {} Wanted: {}".format(state.plugin.name, state.plugin.versionSpec, state.currentVersion, state.wantedVersion))
click.echo("New plugins to install:")
for state in sorted(available):
click.echo("\t{}: {}".format(state.plugin.name, state.plugin.version))
click.echo("Missing plugins:")
for state in sorted(missing):
click.echo("\t{} {}".format(click.style(state.plugin.name, fg='red'),
click.style(state.plugin.versionSpec,
fg='yellow')))
if len(outdatedLinks) > 0 or len(available) > 0:
click.echo("Apply changes? [y/N]", nl=False)
answer = click.getchar().lower()
if answer == "y":
with click.progressbar(length=len(available)+len(outdatedLinks),
label='Synchronizing') as bar:
i = 0
for state in available:
server.installVersion(state.plugin)
server.updateSymlinkForPlugin(state.plugin, state.plugin.version)
i += 1
bar.update(i)
for state in outdatedLinks:
server.updateSymlinkForPlugin(state.plugin, state.wantedVersion)
i += 1
bar.update(i)
else:
click.echo("Not applying changes.")
else:
click.echo("No changes to apply.")

85
src/mpm/config.py Normal file
View File

@ -0,0 +1,85 @@
import os
import yaml
import click
from mpm.repo import Repo
from mpm.server import Server
from mpm.model import *
try:
from yaml import CLoader as Loader, CDumper as Dumper
except ImportError:
from yaml import Loader, Dumper
DEFAULT_CONFIG = """
repositories: {}
servers: {}
"""
class RepoParamType(click.ParamType):
name = "repo-name"
def convert(self, value, param, ctx):
if isinstance(value, Repo):
return value
return ctx.obj['config'].repository(value)
class ServerParamType(click.ParamType):
name = "server-name"
def convert(self, value, param, ctx):
if isinstance(value, Repo):
return value
return ctx.obj['config'].server(value)
class Config():
def __init__(self, path):
if path is None:
path = os.path.expanduser('~/mpm.yaml')
self.path = path
with open(path, 'r') as fd:
self.yaml = yaml.load(fd, Loader=Loader)
self.config = yaml.load(DEFAULT_CONFIG, Loader=Loader)
if isinstance(self.yaml, dict):
self.config.update(self.yaml)
def repository(self, name):
return Repo(name, self.config['repositories'][name])
def repositories(self):
return [Repo(name, c) for (name, c) in self.config['repositories'].items()]
def update_repository(self, name, config):
self.config['repositories'][name] = config
def add_repository(self, name, path):
if name in self.config['repositories']:
raise ValueError('Repository already exists')
self.update_repository(name, {
'path': str(path)
})
def servers(self):
return [Server(name, c) for (name, c) in self.config['servers'].items()]
def server(self, name):
return Server(name, self.config['servers'][name])
def update_server(self, server, config):
self.config['servers'][server] = config
def add_server(self, name, path):
if name in self.config['servers']:
raise ValueError("Server already exists")
self.update_server(name, {
'path': path,
'plugins': [],
'inherit': []
})
def save(self):
stream = open(self.path, 'w')
yaml.dump(self.config, stream, Dumper=Dumper)
stream.close()

View File

@ -1,267 +0,0 @@
#!/bin/env python
import argparse
import pathlib
import sys
import yaml
import os
from mpm.repo import Repo
from mpm.server import Server
from mpm.model import *
try:
from yaml import CLoader as Loader, CDumper as Dumper
except ImportError:
from yaml import Loader, Dumper
DEFAULT_CONFIG = """
repositories: {}
servers: {}
"""
class Config():
def __init__(self, path):
if path is None:
path = os.path.expanduser('~/mpm.yaml')
self.path = path
with open(path, 'r') as fd:
self.yaml = yaml.load(fd, Loader=Loader)
self.config = yaml.load(DEFAULT_CONFIG, Loader=Loader)
if isinstance(self.yaml, dict):
self.config.update(self.yaml)
def repository(self, name):
return Repo(name, self.config['repositories'][name])
def repositories(self):
return [Repo(name, c) for (name, c) in self.config['repositories'].items()]
def update_repository(self, name, config):
self.config['repositories'][name] = config
def add_repository(self, name, path):
if name in self.config['repositories']:
raise ValueError('Repository already exists')
self.update_repository(name, {
'path': path
})
def servers(self):
return [Server(name, c) for (name, c) in self.config['servers'].items()]
def server(self, name):
return Server(name, self.config['servers'][name])
def update_server(self, server, config):
self.config['servers'][server] = config
def add_server(self, name, path):
if name in self.config['servers']:
raise ValueError("Server already exists")
self.update_server(name, {
'path': path,
'plugins': [],
'inherit': []
})
def save(self):
stream = open(self.path, 'w')
yaml.dump(self.config, stream, Dumper=Dumper)
stream.close()
def do_repo_add(args, config):
if not os.path.exists(args.path):
os.makedirs(args.path)
config.add_repository(args.name, args.path)
config.save()
print("Added repository {}".format(args.path))
def do_repo_list(args, config):
for repo in config.repositories():
print("{} ({})".format(repo.name, repo.path))
for plugin in sorted(repo.plugins()):
print('\t', plugin.name, '\t', plugin.version)
for badFile in sorted(repo.badFiles()):
print('\tWARNING: Unknown file', badFile)
def do_repo_import(args, config):
repo = config.repository(args.name)
plugins = []
for path in args.path:
try:
plugins.append(Plugin(path))
except:
print("Bad plugin filename {}".format(path))
if len(plugins) == 0:
print("No plugins found.")
print('Found the following plugins:')
for plugin in plugins:
print("\t{} {}".format(plugin.name, plugin.version))
print("Import plugins into {}? [y/N]".format(repo.name))
answer = input().lower()
if answer == "y":
for plugin in plugins:
repo.importPlugin(plugin)
print("Imported!")
else:
print("Cancelled.")
def do_server_add(args, config):
config.add_server(args.name, args.path)
config.save()
print("Added server {} in {}".format(args.name, args.path))
def do_server_list(args, config):
for server in config.servers():
print('{} ({}):'.format(server.name, server.path))
outdatedLinks = []
missing = []
installed = []
unmanaged = []
conflicts = []
for state in sorted(server.pluginStates(config.repositories())):
if isinstance(state, OutdatedSymlink):
outdatedLinks.append(state)
elif isinstance(state, Installed):
installed.append(state)
elif isinstance(state, MissingVersions):
missing.append(state)
elif isinstance(state, UnmanagedFile):
unmanaged.append(state)
elif isinstance(state, SymlinkConflict):
conflicts.append(state)
print("Installed plugins:")
for state in sorted(installed):
print("\t{} {}: {}".format(state.plugin.name, state.plugin.versionSpec, state.currentVersion))
print("Oudated symlinks:")
for state in sorted(outdatedLinks):
print("\t{} {}: Current: {} Wanted: {}".format(state.plugin.name, state.plugin.versionSpec, state.currentVersion, state.wantedVersion))
print("Missing plugins:")
for state in sorted(missing):
print("\t{}: {}".format(state.plugin.name, state.plugin.versionSpec))
print("Unmanaged files:")
for state in sorted(unmanaged):
print("\t{}".format(state.filename))
print("Symlink Conflicts:")
for state in sorted(conflicts):
print("\t{}.jar".format(state.plugin.name))
def do_server_add_plugin(args, config):
server = config.server(args.server)
plugins = []
for pluginSpec in args.plugin:
if os.path.exists(pluginSpec):
plugin = Plugin(pluginSpec)
pluginSpec = PluginSpec(plugin.name, str(plugin.version))
else:
allVersions = []
for repo in config.repositories():
allVersions += repo.versionsForPlugin(pluginSpec)
pluginSpec = PluginSpec(pluginSpec, list(reversed(sorted(allVersions)))[0])
plugins.append(pluginSpec)
print("Added {} to {}".format(pluginSpec, server.name))
for pluginSpec in plugins:
print("\t{} {}".format(pluginSpec.name, pluginSpec.versionSpec))
print("Add these plugins to server {}? [y/N]".format(server.name))
answer = input().lower()
if answer == "y":
for pluginSpec in plugins:
server.add_plugin(pluginSpec)
config.update_server(server.name, server.config)
config.save()
print("Added!")
else:
print("Cancelled.")
def do_server_sync(args, config):
for server in config.servers():
print('{} ({}):'.format(server.name, server.path))
outdatedLinks = []
available = []
for state in sorted(server.pluginStates(config.repositories())):
if isinstance(state, OutdatedSymlink):
outdatedLinks.append(state)
elif isinstance(state, Available):
available.append(state)
print("Plugins to update:")
for state in sorted(outdatedLinks):
print("\t{} {}: Current: {} Wanted: {}".format(state.plugin.name, state.plugin.versionSpec, state.currentVersion, state.wantedVersion))
print("New plugins to install:")
for state in sorted(available):
print("\t{}: {}".format(state.plugin.name, state.plugin.version))
if len(outdatedLinks) > 0 or len(available) > 0:
print("Apply changes? [y/N]")
answer = input().lower()
if answer == "y":
for state in available:
server.installVersion(state.plugin)
server.updateSymlinkForPlugin(state.plugin, state.plugin.version)
print("Installed {} {}".format(state.plugin.name, state.plugin.version))
for state in outdatedLinks:
server.updateSymlinkForPlugin(state.plugin, state.wantedVersion)
print("Updated {} to {}".format(state.plugin.name, state.wantedVersion))
else:
print("Not applying changes.")
else:
print("No changes to apply.")
def main():
parser = argparse.ArgumentParser(description='Paper Plugin Sync')
parser.add_argument('--config', dest='config_path', type=Config)
subparsers = parser.add_subparsers()
repos = subparsers.add_parser('repo')
repo_sub = repos.add_subparsers()
repo_add = repo_sub.add_parser('add')
repo_add.add_argument('name', help='Name of the repository')
repo_add.add_argument('path', help='Where to add a repository or create a new one')
repo_add.set_defaults(func=do_repo_add)
repo_list = repo_sub.add_parser('list')
repo_list.set_defaults(func=do_repo_list)
repo_import = repo_sub.add_parser('import')
repo_import.add_argument('name', help='Name of the repository')
repo_import.add_argument('path', nargs="+", help='Path of the plugin to import')
repo_import.set_defaults(func=do_repo_import)
servers = subparsers.add_parser('server')
server_sub = servers.add_subparsers()
server_add = server_sub.add_parser('add')
server_add.add_argument('name', help='Name for the server')
server_add.add_argument('path', help='Path to your server\'s root directory')
server_add.set_defaults(func=do_server_add)
server_list = server_sub.add_parser('list')
server_list.set_defaults(func=do_server_list)
server_add_plugin = server_sub.add_parser('add-plugin')
server_add_plugin.add_argument('server', help='Name of server to modify')
server_add_plugin.add_argument('plugin', nargs='+', help='Plugin file or spec to install')
server_add_plugin.set_defaults(func=do_server_add_plugin)
server_sync = server_sub.add_parser('sync')
server_sync.set_defaults(func=do_server_sync)
args = parser.parse_args()
config = Config(args.config_path)
if 'func' not in args:
parser.print_usage()
else:
args.func(args, config)
if __name__ == "__main__":
main()

View File

@ -1,73 +0,0 @@
#!/bin/env python
import yaml
import os
import sys
from model import *
try:
from yaml import CLoader as Loader, CDumper as Dumper
except ImportError:
from yaml import Loader, Dumper
conf = yaml.load(open('plugins.yml', 'r'), Loader=Loader)
for (serverName,server) in conf['servers'].items():
changeset = []
if len(sys.argv) > 1 and serverName not in sys.argv[1:]:
continue
if 'pluginPath' not in server:
continue
if not os.path.exists(server['pluginPath']):
print("Missing plugin path for {}: {}".format(serverName, server['pluginPath']))
else:
print("=== Updating server {}".format(serverName))
pluginSpecs = {}
for inherited in server.get('inherit', ()):
for inheritedPlugin in conf['servers'][inherited]['plugins']:
pluginSpecs[inheritedPlugin['name']] = PluginSpec(inheritedPlugin['name'], str(inheritedPlugin.get('version', '*')))
for pluginConf in server.get('plugins', ()):
pluginSpecs[pluginConf['name']] = PluginSpec(pluginConf['name'], str(pluginConf.get('version', '*')))
repo = Repo(server['pluginPath'], pluginSpecs.values())
outdatedLinks = []
missing = []
installed = []
unmanaged = []
conflicts = []
for state in repo.pluginStates():
if isinstance(state, OutdatedSymlink):
outdatedLinks.append(state)
elif isinstance(state, Installed):
installed.append(state)
elif isinstance(state, MissingVersions):
missing.append(state)
elif isinstance(state, UnmanagedFile):
unmanaged.append(state)
elif isinstance(state, SymlinkConflict):
conflicts.append(state)
print("Installed plugins:")
for state in sorted(installed):
print("\t{} {}: {}".format(state.plugin.name, state.plugin.versionSpec, state.currentVersion))
print("Oudated symlinks:")
for state in sorted(outdatedLinks):
print("\t{} {}: Current: {} Wanted: {}".format(state.plugin.name, state.plugin.versionSpec, state.currentVersion, state.wantedVersion))
print("Missing plugins:")
for state in sorted(missing):
print("\t{}: {}".format(state.plugin.name, state.plugin.versionSpec))
print("Unmanaged files:")
for state in sorted(unmanaged):
print("\t{}".format(state.filename))
print("Symlink Conflicts:")
for state in sorted(conflicts):
print("\t{}.jar".format(state.plugin.name))
if len(outdatedLinks) > 0:
print("Apply changes? [y/N]")
answer = input().lower()
if answer == "y":
for state in outdatedLinks:
repo.updateSymlinkForPlugin(state.plugin, state.wantedVersion)
else:
print("Not applying changes.")

View File

@ -1,5 +1,6 @@
from mpm.model import *
import shutil
import pathlib
class Server:
def __init__(self, name, config):
@ -70,17 +71,18 @@ class Server:
return pluginSemver
def versionsForPlugin(self, pluginName, repos):
plugins = os.listdir(os.path.join(self.pluginPath, 'versions'))
for pluginJar in plugins:
if pluginJar.startswith(pluginName):
prefix = pluginName + '-'
suffix = '.jar'
jarVersion = pluginJar[len(prefix):len(pluginJar)-len(suffix)]
try:
pluginSemver = Version.coerce(jarVersion)
except ValueError:
pluginSemver = jarVersion
yield pluginSemver
if os.path.exists(os.path.join(self.pluginPath, 'versions')):
plugins = os.listdir(os.path.join(self.pluginPath, 'versions'))
for pluginJar in plugins:
if pluginJar.startswith(pluginName):
prefix = pluginName + '-'
suffix = '.jar'
jarVersion = pluginJar[len(prefix):len(pluginJar)-len(suffix)]
try:
pluginSemver = Version.coerce(jarVersion)
except ValueError:
pluginSemver = jarVersion
yield pluginSemver
def updateSymlinkForPlugin(self, plugin, version):
pluginFilename = os.path.join(self.pluginPath, 'versions/{}-{}.jar'.format(plugin.name, version))
@ -92,5 +94,7 @@ class Server:
os.symlink(linkDst, pluginSymlink)
def installVersion(self, plugin):
if not os.path.exists(os.path.join(self.pluginPath, 'versions')):
os.path.mkdir(os.path.join(self.pluginPath, 'versions'))
dest = os.path.join(self.pluginPath, 'versions/{}-{}.jar'.format(plugin.name, plugin.version))
shutil.copyfile(plugin.path, dest)