1556 lines
48 KiB
Python
1556 lines
48 KiB
Python
#!/usr/bin/env python3
|
|
#
|
|
# Copyright (c) 2020 - Odd Strabo <oddstr13@openshell.no>
|
|
#
|
|
# This Source Code Form is subject to the terms of the Mozilla Public
|
|
# License, v. 2.0. If a copy of the MPL was not distributed with this
|
|
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
|
#
|
|
from __future__ import annotations
|
|
|
|
import configparser
|
|
import datetime
|
|
import hashlib
|
|
import importlib.metadata
|
|
import itertools
|
|
import json
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import urllib.parse
|
|
import uuid
|
|
import xml.etree.ElementTree as ET
|
|
import zipfile
|
|
from collections.abc import Callable
|
|
from collections.abc import Generator
|
|
from collections.abc import Iterable
|
|
from collections.abc import Iterator
|
|
from enum import Enum
|
|
from functools import total_ordering
|
|
from typing import Any
|
|
from typing import cast
|
|
from typing import TypedDict
|
|
|
|
import click
|
|
import click_log
|
|
import requests
|
|
import tabulate
|
|
import yaml
|
|
from click.core import Context
|
|
from click.core import Parameter
|
|
from slugify import slugify
|
|
|
|
if sys.version_info < (3, 11):
|
|
def file_digest(fileobj, digest, /, *, _bufsize=2**18): # type: ignore[no-untyped-def]
|
|
"""Hash the contents of a file-like object. Returns a digest object.
|
|
|
|
*fileobj* must be a file-like object opened for reading in binary mode.
|
|
It accepts file objects from open(), io.BytesIO(), and SocketIO objects.
|
|
The function may bypass Python's I/O and use the file descriptor *fileno*
|
|
directly.
|
|
|
|
*digest* must either be a hash algorithm name as a *str*, a hash
|
|
constructor, or a callable that returns a hash object.
|
|
"""
|
|
# On Linux we could use AF_ALG sockets and sendfile() to archive zero-copy
|
|
# hashing with hardware acceleration.
|
|
if isinstance(digest, str):
|
|
digestobj = hashlib.new(digest)
|
|
else:
|
|
digestobj = digest()
|
|
|
|
if hasattr(fileobj, 'getbuffer'):
|
|
# io.BytesIO object, use zero-copy buffer
|
|
digestobj.update(fileobj.getbuffer())
|
|
return digestobj
|
|
|
|
# Only binary files implement readinto().
|
|
if not (hasattr(fileobj, 'readinto') and hasattr(fileobj, 'readable') and fileobj.readable()):
|
|
raise ValueError(f"'{fileobj!r}' is not a file-like object in binary reading mode.")
|
|
|
|
# binary file, socket.SocketIO object
|
|
# Note: socket I/O uses different syscalls than file I/O.
|
|
buf = bytearray(_bufsize) # Reusable buffer to reduce allocations.
|
|
view = memoryview(buf)
|
|
while True:
|
|
size = fileobj.readinto(buf)
|
|
if size == 0:
|
|
break # EOF
|
|
digestobj.update(view[:size])
|
|
|
|
return digestobj
|
|
|
|
class StrEnum(str, Enum):
|
|
"""
|
|
Enum where members are also (and must be) strings
|
|
"""
|
|
|
|
def __new__(cls, *values: Any) -> Any:
|
|
'values must already be of type `str`'
|
|
if len(values) > 3:
|
|
raise TypeError(f"too many arguments for str(): {values!r}")
|
|
if len(values) == 1:
|
|
# it must be a string
|
|
if not isinstance(values[0], str):
|
|
raise TypeError(f"{values[0]!r} is not a string")
|
|
if len(values) >= 2:
|
|
# check that encoding argument is a string
|
|
if not isinstance(values[1], str):
|
|
raise TypeError(f"encoding must be a string, not {values[1]!r}")
|
|
if len(values) == 3:
|
|
# check that errors argument is a string
|
|
if not isinstance(values[2], str):
|
|
raise TypeError('errors must be a string, not %r' % (values[2]))
|
|
value = str(*values)
|
|
member = str.__new__(cls, value)
|
|
member._value_ = value
|
|
return member
|
|
|
|
@staticmethod
|
|
def _generate_next_value_(name: str, start: int, count: int, last_values: Any) -> str: # dead: disable
|
|
"""
|
|
Return the lower-cased version of the member name.
|
|
"""
|
|
return name.lower()
|
|
|
|
def __str__(self) -> str:
|
|
return self.value
|
|
else:
|
|
from enum import StrEnum
|
|
from hashlib import file_digest
|
|
|
|
|
|
try:
|
|
from ._version import __version__
|
|
except ImportError:
|
|
__version__ = importlib.metadata.version(
|
|
importlib.metadata.packages_distributions().get('jprm', ['traitorous_jprm'])[0],
|
|
)
|
|
|
|
logger = logging.getLogger('jprm')
|
|
click_log.basic_config(logger)
|
|
|
|
JSON_METADATA_FILE = 'meta.json'
|
|
DEFAULT_IMAGE_FILE = 'image.png'
|
|
DEFAULT_FRAMEWORK = 'net9.0'
|
|
CONFIG_LOCATIONS = [
|
|
'build.yaml',
|
|
'meta.yaml',
|
|
'jprm.yaml',
|
|
'.jprm.yaml',
|
|
'.ci/jprm.yaml',
|
|
'.github/jprm.yaml',
|
|
'.gitlab/jprm.yaml',
|
|
]
|
|
|
|
jprm_config = pathlib.Path.home() / '.config/jprm'
|
|
|
|
|
|
class _ManifestVersion(TypedDict, total=False):
|
|
repositoryName: str
|
|
repositoryUrl: str
|
|
|
|
|
|
class ManifestVersion(_ManifestVersion):
|
|
version: str
|
|
changelog: str
|
|
targetAbi: str
|
|
sourceUrl: str
|
|
checksum: str
|
|
timestamp: str
|
|
|
|
|
|
class _Manifest(TypedDict, total=False):
|
|
imageUrl: str
|
|
|
|
|
|
class Manifest(_Manifest):
|
|
name: str
|
|
description: str
|
|
overview: str
|
|
owner: str
|
|
category: str
|
|
guid: str
|
|
versions: list[ManifestVersion]
|
|
|
|
|
|
class PluginStatus(StrEnum):
|
|
Restart = 'Restart' # dead: disable
|
|
Active = 'Active' # dead: disable
|
|
Disabled = 'Disabled' # dead: disable
|
|
NotSupported = 'NotSupported' # dead: disable
|
|
Malfunctioned = 'Malfunctioned' # dead: disable
|
|
Superseded = 'Superseded' # dead: disable
|
|
Superceded = 'Superceded' # dead: disable
|
|
Deleted = 'Deleted' # dead: disable
|
|
|
|
|
|
class _PluginManifest(TypedDict, total=False):
|
|
status: PluginStatus
|
|
autoUpdate: bool
|
|
imagePath: str
|
|
assemblies: list[str]
|
|
|
|
|
|
class PluginManifest(_PluginManifest):
|
|
category: str
|
|
changelog: str
|
|
description: str
|
|
guid: str
|
|
name: str
|
|
overview: str
|
|
owner: str
|
|
targetAbi: str
|
|
timestamp: str
|
|
version: str
|
|
|
|
|
|
def checksum_file(path: pathlib.Path) -> hashlib._Hash:
|
|
with path.open('rb') as f:
|
|
digest = file_digest(f, lambda: hashlib.md5(usedforsecurity=False))
|
|
return digest
|
|
|
|
|
|
def zip_path(zip_file: pathlib.Path, path: pathlib.Path) -> None:
|
|
zip_file.parent.mkdir(exist_ok=True, parents=True)
|
|
if zip_file.is_dir():
|
|
zip_file.rmdir()
|
|
zip_file.unlink(missing_ok=True)
|
|
with zipfile.ZipFile(zip_file, 'w', zipfile.ZIP_DEFLATED) as z:
|
|
root: pathlib.Path | str
|
|
for root, dirs, files in os.walk(path, topdown=True):
|
|
root = pathlib.Path(root)
|
|
for d in (*files, *dirs):
|
|
filename = root / d
|
|
arcname = filename.relative_to(path)
|
|
|
|
z.write(filename, arcname)
|
|
|
|
|
|
def load_build_config(manifest_file_name: pathlib.Path) -> dict[str, Any] | None:
|
|
"""
|
|
Read in an arbitrary YAML manifest and return it
|
|
"""
|
|
try:
|
|
with manifest_file_name.open(encoding='utf-8') as manifest_file:
|
|
cfg = yaml.load(manifest_file, Loader=yaml.SafeLoader)
|
|
except (yaml.YAMLError, OSError) as e:
|
|
logger.error(f"Failed to load YAML manifest {manifest_file_name}: {e}")
|
|
return None
|
|
return cfg
|
|
|
|
|
|
def get_config(path: pathlib.Path) -> dict[str, Any] | None:
|
|
for config_file in CONFIG_LOCATIONS:
|
|
build_cfg = load_build_config(path / config_file)
|
|
if build_cfg is not None:
|
|
return build_cfg
|
|
logger.warning('Failed to locate config file.')
|
|
return None
|
|
|
|
|
|
@total_ordering
|
|
class Version:
|
|
version_re = re.compile(r'^(?P<major>[0-9]+)(\.(?P<minor>[0-9]+))?(\.(?P<build>[0-9]+))?(\.(?P<revision>[0-9]+))?$')
|
|
|
|
major = None
|
|
minor = None
|
|
build = None
|
|
revision = None
|
|
|
|
def __init__(self, version: Any) -> None:
|
|
if isinstance(version, Version):
|
|
self.major = version.major
|
|
self.minor = version.minor
|
|
self.build = version.build
|
|
self.revision = version.revision
|
|
|
|
elif isinstance(version, str):
|
|
match = self.version_re.match(version)
|
|
if not match:
|
|
raise ValueError(version)
|
|
|
|
gd = match.groupdict()
|
|
self.major = int(gd.get('major', '0')) if gd.get('major') else None
|
|
self.minor = int(gd.get('minor', '0')) if gd.get('minor') else None
|
|
self.build = int(gd.get('build', '0')) if gd.get('build') else None
|
|
self.revision = int(gd.get('revision', '0')) if gd.get('revision') else None
|
|
|
|
elif isinstance(version, int):
|
|
self.major = version
|
|
|
|
else:
|
|
raise TypeError(version)
|
|
|
|
def full(self) -> str:
|
|
return '{major}.{minor}.{build}.{revision}'.format(
|
|
major=self.major or 0,
|
|
minor=self.minor or 0,
|
|
build=self.build or 0,
|
|
revision=self.revision or 0,
|
|
)
|
|
|
|
def __str__(self) -> str:
|
|
if self.revision is not None:
|
|
return '{major}.{minor}.{build}.{revision}'.format(**self)
|
|
if self.build is not None:
|
|
return '{major}.{minor}.{build}'.format(**self)
|
|
if self.minor is not None:
|
|
return '{major}.{minor}'.format(**self)
|
|
else:
|
|
return '{major}'.format(**self)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<{self.__class__.__name__}({repr(str(self))})>"
|
|
|
|
def __iter__(self) -> Iterator[int | None]:
|
|
return iter(self.values())
|
|
|
|
def __getitem__(self, key: str) -> int | None:
|
|
if key in ('major', 0):
|
|
return self.major
|
|
|
|
if key in ('minor', 1):
|
|
return self.minor
|
|
|
|
if key in ('build', 2):
|
|
return self.build
|
|
|
|
if key in ('revision', 3):
|
|
return self.revision
|
|
|
|
raise KeyError
|
|
|
|
def __setitem__(self, key: object, value: int | None) -> None:
|
|
if key not in self:
|
|
raise KeyError(key)
|
|
|
|
if value is not None:
|
|
value = int(value)
|
|
|
|
if key in ('major', 0):
|
|
self.major = value
|
|
if value is None:
|
|
self.minor = None
|
|
self.build = None
|
|
self.revision = None
|
|
|
|
if key in ('minor', 1):
|
|
self.minor = value
|
|
if value is None:
|
|
self.build = None
|
|
self.revision = None
|
|
|
|
if key in ('build', 2):
|
|
self.build = value
|
|
if value is None:
|
|
self.revision = None
|
|
|
|
if key in ('revision', 3):
|
|
self.revision = value
|
|
|
|
def __delitem__(self, key: str) -> None:
|
|
self[key] = None
|
|
|
|
def __len__(self) -> int:
|
|
return 4
|
|
|
|
def __contains__(self, key: Any) -> bool:
|
|
return key in ('major', 'minor', 'build', 'revision', 0, 1, 2, 3)
|
|
|
|
def keys(self) -> tuple[str, str, str, str]: # dead: disable
|
|
return ('major', 'minor', 'build', 'revision')
|
|
|
|
def values(self) -> tuple[int | None, int | None, int | None, int | None]:
|
|
return (self.major, self.minor, self.build, self.revision)
|
|
|
|
def items(
|
|
self,
|
|
) -> tuple[tuple[str, int | None], tuple[str, int | None], tuple[str, int | None], tuple[str, int | None]]:
|
|
return (
|
|
('major', self.major),
|
|
('minor', self.minor),
|
|
('build', self.build),
|
|
('revision', self.revision),
|
|
)
|
|
|
|
def get(self, key: str, default: bool | None = None) -> int | None:
|
|
if key not in self:
|
|
logger.warning(f"Accessing non-existant key `{key}` of `{self!r}`")
|
|
return default
|
|
|
|
return self[key]
|
|
|
|
@staticmethod
|
|
def _hasattrs(obj: Any, *names: str) -> bool:
|
|
for name in names:
|
|
if not hasattr(obj, name):
|
|
return False
|
|
return True
|
|
|
|
def __eq__(self, other: Any) -> bool:
|
|
if self._hasattrs(other, 'major', 'minor', 'build', 'revision'):
|
|
return (
|
|
self.major or 0,
|
|
self.minor or 0,
|
|
self.build or 0,
|
|
self.revision or 0,
|
|
) == (
|
|
other.major or 0,
|
|
other.minor or 0,
|
|
other.build or 0,
|
|
other.revision or 0,
|
|
)
|
|
|
|
return NotImplemented
|
|
|
|
def __lt__(self, other: Any) -> bool:
|
|
if self._hasattrs(other, 'major', 'minor', 'build', 'revision'):
|
|
return (
|
|
self.major or 0,
|
|
self.minor or 0,
|
|
self.build or 0,
|
|
self.revision or 0,
|
|
) < (
|
|
other.major or 0,
|
|
other.minor or 0,
|
|
other.build or 0,
|
|
other.revision or 0,
|
|
)
|
|
|
|
return NotImplemented
|
|
|
|
def pop(self, k: Any, d: Callable[..., KeyError] = KeyError) -> None:
|
|
raise NotImplementedError
|
|
|
|
|
|
def get_remote_url(repo_path: pathlib.Path, remote_name: str = 'origin') -> urllib.parse.ParseResult | None:
|
|
config_path = repo_path / '.git' / 'config'
|
|
config = configparser.ConfigParser()
|
|
config.read(config_path)
|
|
|
|
section_name = f'remote "{remote_name}"'
|
|
if config.has_section(section_name) and config.has_option(section_name, 'url'):
|
|
return urllib.parse.urlparse(config.get(section_name, 'url'))
|
|
return None
|
|
|
|
|
|
def get_repo_owner(repo_path: pathlib.Path) -> tuple[str, str, str]:
|
|
url = get_remote_url(repo_path)
|
|
if not url or not url.hostname:
|
|
return '', '', ''
|
|
|
|
owner, _, project = url.path.removesuffix('.git').strip('/').partition('/')
|
|
return url.hostname, owner, project
|
|
|
|
|
|
def find_git(repo_path: pathlib.Path) -> pathlib.Path | None:
|
|
for path in itertools.chain((repo_path, repo_path.parent, repo_path.parent.parent), repo_path.glob('*/')):
|
|
if (path / '.git/config').is_file():
|
|
return path
|
|
return None
|
|
|
|
|
|
def get_jellyfin_dependencies(targetAbi: Version, cache_dir: pathlib.Path) -> list[ET.Element]:
|
|
if targetAbi < Version('10.9'):
|
|
raise ValueError('Unable to get jellyfin dependencies before version 10.9')
|
|
abi_path = (cache_dir / str(targetAbi))
|
|
dep_path = abi_path.with_suffix('.Directory.Packages.props')
|
|
if dep_path.exists() and datetime.datetime.fromtimestamp(dep_path.stat().st_mtime) > datetime.datetime.today() - datetime.timedelta(days=1):
|
|
return ET.fromstring(dep_path.read_text(encoding='utf-8')).findall('.//PackageVersion')
|
|
|
|
etag_path = abi_path.with_suffix('.etag')
|
|
etag_path.parent.mkdir(exist_ok=True, parents=True)
|
|
etag_path.touch()
|
|
etag = abi_path.with_suffix('.etag').read_text(encoding='utf-8')
|
|
try:
|
|
jellyfin_version = targetAbi.full().rpartition('.')[0]
|
|
resp = requests.get(
|
|
f'https://github.com/jellyfin/jellyfin/raw/refs/tags/v{jellyfin_version}/Directory.Packages.props',
|
|
timeout=5,
|
|
headers={
|
|
'User-Agent': f'Traitorous JPRM/{__version__}',
|
|
'If-None-Match': etag,
|
|
},
|
|
)
|
|
except Exception:
|
|
return ET.fromstring(dep_path.read_text(encoding='utf-8')).findall('.//PackageVersion')
|
|
if resp.status_code not in (requests.codes.ok, requests.codes.not_modified):
|
|
raise Exception('Failed to download jellyfin dependencies', resp.request.url, resp.status_code, resp.text)
|
|
|
|
if resp.status_code == requests.codes.ok:
|
|
dep_path.write_bytes(resp.content)
|
|
etag_path.write_text(resp.headers['Etag'].strip('"'))
|
|
|
|
return ET.fromstring(dep_path.read_text(encoding='utf-8')).findall('.//PackageVersion')
|
|
|
|
|
|
def build_dep_graph(target_section: dict) -> dict[str, tuple[list[str], list[str]]]:
|
|
deps: dict[str, list[str]] = {}
|
|
files: dict[str, list[str]] = {}
|
|
s_dep: dict[str, tuple[list[str], list[str]]] = {}
|
|
|
|
for lib_key, lib_val in target_section.items():
|
|
name, _, version = lib_key.partition('/')
|
|
if name in deps:
|
|
raise SystemError(f"duplicate {name!r} in {json.dumps(deps)}")
|
|
deps[name] = list(lib_val.get('dependencies', {}))
|
|
|
|
dll_names = [
|
|
os.path.basename(path)
|
|
for path in lib_val.get('runtime', {})
|
|
if path.casefold().endswith('.dll')
|
|
]
|
|
files.setdefault(name, []).extend(dll_names)
|
|
s_dep[name.casefold()] = (deps[name], dll_names)
|
|
|
|
return s_dep
|
|
|
|
|
|
def prune_exclusive_deps(assets_file: pathlib.Path, packages: Iterable[str], tfm: str, publish_dir: pathlib.Path) -> list[pathlib.Path]:
|
|
"""
|
|
Deletes every DLL/PDB reachable from packages's dependency closure
|
|
(including packages itself) out of publish_dir.
|
|
|
|
Returns the list of files that were deleted.
|
|
"""
|
|
with assets_file.open(encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
targets = data.get('targets', {})
|
|
if not targets:
|
|
return []
|
|
|
|
target_section = targets.get(tfm, {})
|
|
if not target_section:
|
|
return []
|
|
s_dep = build_dep_graph(target_section)
|
|
|
|
def get_all_children(*start: str) -> set[str]:
|
|
visited: set[str] = set()
|
|
stack = [x.casefold() for x in start]
|
|
while stack:
|
|
cur = stack.pop()
|
|
if cur in visited:
|
|
continue
|
|
visited.add(cur)
|
|
for child in s_dep.get(cur, ([], []))[0]:
|
|
child_l = child.casefold()
|
|
if child_l not in visited:
|
|
stack.append(child_l)
|
|
return visited
|
|
|
|
to_remove = get_all_children(*packages)
|
|
|
|
deleted: list[pathlib.Path] = []
|
|
|
|
def delete(path: pathlib.Path) -> None:
|
|
if path.exists():
|
|
path.unlink()
|
|
deleted.append(path)
|
|
for pkg in to_remove:
|
|
for dll in s_dep.get(pkg, ([], []))[1]:
|
|
dll_path = publish_dir / dll
|
|
delete(dll_path)
|
|
delete(dll_path.with_suffix('.pdb'))
|
|
projectName = data.get('project', {}).get('restore', {}).get('projectName', '')
|
|
if projectName:
|
|
# Remove files that we don't want in a release zip
|
|
delete(publish_dir / f'{projectName}.deps.json')
|
|
delete(publish_dir / f'{projectName}.pdb')
|
|
delete(publish_dir / f'{projectName}.xml')
|
|
|
|
return deleted
|
|
|
|
|
|
def build_plugin(
|
|
*,
|
|
path: pathlib.Path,
|
|
build_cfg: dict[str, Any],
|
|
output: pathlib.Path,
|
|
version: str | None,
|
|
dotnet_config: str,
|
|
dotnet_framework: str | None,
|
|
max_cpu_count: int = 1,
|
|
) -> None:
|
|
if version is None:
|
|
version = build_cfg['version']
|
|
|
|
if version is None:
|
|
raise ValueError('No version found')
|
|
v = Version(version)
|
|
|
|
if dotnet_framework is None:
|
|
if 'framework' not in build_cfg:
|
|
logger.warning(f"`framework` is not specified in build manifest, defaulting to `{DEFAULT_FRAMEWORK}`.")
|
|
logger.warning('The default target framework may change in the future.')
|
|
dotnet_framework = build_cfg.get('framework', DEFAULT_FRAMEWORK)
|
|
assert dotnet_framework
|
|
|
|
logger.debug(
|
|
{
|
|
'dotnet_config': dotnet_config,
|
|
'dotnet_framework': dotnet_framework,
|
|
'output': output,
|
|
'max_cpu_count': max_cpu_count,
|
|
'version': v.full(),
|
|
},
|
|
)
|
|
|
|
projects: list[pathlib.Path] = []
|
|
|
|
sln_file = None
|
|
manual_projects = []
|
|
for fn in path.iterdir():
|
|
if fn.suffix in ('.sln', '.slnx'):
|
|
if sln_file:
|
|
raise Exception('Multiple solution files found')
|
|
sln_file = path / fn
|
|
if fn.suffix == '.csproj':
|
|
manual_projects.append(path / fn)
|
|
|
|
if sln_file is not None:
|
|
projects.extend(solution_get_projects(sln_file))
|
|
else:
|
|
projects.extend(manual_projects)
|
|
|
|
dbp_file = path / 'Directory.Build.props'
|
|
if os.path.exists(dbp_file):
|
|
projects.append(dbp_file)
|
|
jellyfin_modules: set[str] = set()
|
|
asset_paths = []
|
|
asset_file_command = (
|
|
'dotnet', 'msbuild', '-getProperty:ProjectAssetsFile',
|
|
)
|
|
for project in projects:
|
|
with project.open('r+', encoding='utf-8') as f:
|
|
xml = ET.parse(f)
|
|
set_project_version(xml.getroot(), version=v)
|
|
set_project_framework(xml.getroot(), framework=dotnet_framework)
|
|
jellyfin_modules.update(
|
|
mangle_jellyfin_dependencies(
|
|
xml.getroot(), jellyfin_version=build_cfg['targetAbi'],
|
|
),
|
|
)
|
|
xml.write(project, encoding='utf-8', xml_declaration=False)
|
|
f.seek(0, os.SEEK_END)
|
|
f.write('\n')
|
|
if project.suffix != '.csproj':
|
|
continue
|
|
cmd = subprocess.run(asset_file_command, capture_output=True, cwd=project.parent)
|
|
if cmd.returncode != 0:
|
|
logger.error('Failed when running: %s in cwd: %s', asset_file_command, project)
|
|
logger.info(cmd.stdout.decode())
|
|
logger.error(cmd.stderr.decode())
|
|
raise SystemExit(1)
|
|
asset_paths.append(pathlib.Path(cmd.stdout.decode().strip()))
|
|
if not asset_paths[-1].absolute().is_relative_to(path.absolute()):
|
|
raise SystemError(
|
|
f"ProjectAssetsFile is in a weird location: {asset_paths[-1]}: expected it to be in {path}",
|
|
)
|
|
|
|
clean_command = (
|
|
'dotnet',
|
|
'clean',
|
|
f"--configuration={dotnet_config}",
|
|
f"--framework={dotnet_framework}",
|
|
)
|
|
cmd = subprocess.run(clean_command, capture_output=True, cwd=path)
|
|
if cmd.returncode != 0:
|
|
logger.info(cmd.stdout.decode())
|
|
logger.error(cmd.stderr.decode())
|
|
raise SystemExit(1)
|
|
|
|
restore_command = (
|
|
'dotnet',
|
|
'restore',
|
|
'--no-cache',
|
|
)
|
|
cmd = subprocess.run(restore_command, capture_output=True, cwd=path)
|
|
if cmd.returncode != 0:
|
|
logger.info(cmd.stdout.decode())
|
|
logger.error(cmd.stderr.decode())
|
|
raise SystemExit(1)
|
|
|
|
build_command = (
|
|
'dotnet',
|
|
'publish',
|
|
'--nologo',
|
|
'--no-restore',
|
|
f"--configuration={dotnet_config}",
|
|
f"--framework={dotnet_framework}",
|
|
f"-p:PublishDir={output}",
|
|
f"-p:Version={v.full()}",
|
|
f"-maxcpucount:{max_cpu_count}",
|
|
)
|
|
cmd = subprocess.run(build_command, capture_output=True, cwd=path)
|
|
if cmd.returncode != 0:
|
|
logger.info(cmd.stdout.decode())
|
|
logger.error(cmd.stderr.decode())
|
|
raise SystemExit(1)
|
|
|
|
logger.info(cmd.stdout.decode())
|
|
if 'assets' in build_cfg:
|
|
return
|
|
targetAbi = Version(build_cfg['targetAbi'])
|
|
jellyfin_deps = [x.attrib for x in get_jellyfin_dependencies(targetAbi, jprm_config)]
|
|
# TODO: do potential version matching with dotnet list/.csproj files
|
|
jellyfin_modules.update(x['Include'] for x in jellyfin_deps)
|
|
for assets_path in asset_paths:
|
|
pruned_files = prune_exclusive_deps(assets_path, jellyfin_modules, dotnet_framework, output)
|
|
for file in pruned_files:
|
|
logger.debug('Pruned unneeded dependency: %s', file.relative_to(output))
|
|
|
|
|
|
def _read_md5sum(file: pathlib.Path) -> tuple[str, bool, str]:
|
|
with file.open() as f:
|
|
line = f.readline()
|
|
|
|
checksum, _, filename = line.partition(' ')
|
|
if len(checksum) != hashlib.md5().digest_size * 2:
|
|
raise ValueError('Invalid checksum line')
|
|
|
|
binary = filename[0]
|
|
if binary not in ' *':
|
|
raise ValueError('Invalid checksum line')
|
|
|
|
filename = filename[1:]
|
|
if not filename:
|
|
raise ValueError('Invalid checksum line')
|
|
return checksum, binary == '*', filename
|
|
|
|
|
|
def package_plugin(
|
|
*,
|
|
path: pathlib.Path,
|
|
build_cfg: dict[str, Any],
|
|
version: str | None = None,
|
|
binary_path: pathlib.Path = pathlib.Path('./bin'),
|
|
output: pathlib.Path = pathlib.Path('./artifacts'),
|
|
) -> pathlib.Path:
|
|
logger.debug('Packaging plugin: %s', build_cfg)
|
|
if version is None:
|
|
version = build_cfg['version']
|
|
|
|
if version is not None:
|
|
version = Version(version).full()
|
|
|
|
slug = slugify(build_cfg['name'])
|
|
|
|
output_file = f"{slug}_{version}.zip"
|
|
output_path = output / output_file
|
|
|
|
with tempfile.TemporaryDirectory() as tempdir:
|
|
temp_path = pathlib.Path(tempdir)
|
|
for artifact in build_cfg.get('artifacts', binary_path.iterdir()):
|
|
artifact = pathlib.Path(artifact)
|
|
artifact_path = binary_path / artifact.name
|
|
artifact_temp_path = temp_path / artifact.name
|
|
|
|
artifact_temp_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
shutil.copyfile(artifact_path, artifact_temp_path)
|
|
|
|
image_path = path / build_cfg.get('image', DEFAULT_IMAGE_FILE)
|
|
if image_path is not None:
|
|
try:
|
|
image_temp_path = temp_path / image_path.name
|
|
shutil.copyfile(image_path, image_temp_path)
|
|
|
|
build_cfg['image'] = image_path.name
|
|
except OSError:
|
|
logger.warning('Unable to include image %s in build', image_path)
|
|
|
|
meta = generate_metadata(build_cfg, version=version, image_path='' if image_path is None else image_path.name)
|
|
meta_tempfile = temp_path / JSON_METADATA_FILE
|
|
with meta_tempfile.open('w', encoding='utf-8') as fh:
|
|
json.dump(meta, fh, sort_keys=True, indent=4)
|
|
|
|
try:
|
|
zip_path(output_path, temp_path)
|
|
except FileNotFoundError as e:
|
|
logger.error('Failed to zip plugin: %s', e)
|
|
raise SystemExit(1)
|
|
|
|
md5 = checksum_file(output_path)
|
|
md5_path = output_path.with_name(output_path.name + '.md5sum')
|
|
md5_path.write_bytes(md5.hexdigest().encode() + b' *' + output_file.encode() + b'\n')
|
|
|
|
shutil.move(meta_tempfile, f"{output_path}.{JSON_METADATA_FILE}")
|
|
|
|
return output_path
|
|
|
|
|
|
def generate_metadata(
|
|
build_cfg: dict[str, Any], version: str | None = None, build_date: str | None = None,
|
|
image_path: str = '',
|
|
) -> PluginManifest:
|
|
|
|
if version is None:
|
|
version = build_cfg['version']
|
|
|
|
if version is not None:
|
|
version = Version(version).full()
|
|
if version is None:
|
|
raise ValueError('No version found for plugin')
|
|
assert version
|
|
|
|
if build_date is None:
|
|
build_date = datetime.datetime.now(datetime.timezone.utc).isoformat(timespec='seconds')
|
|
|
|
meta = PluginManifest(
|
|
guid=str(uuid.UUID(build_cfg['guid'])),
|
|
name=build_cfg['name'],
|
|
description=build_cfg['description'],
|
|
overview=build_cfg['overview'],
|
|
owner=build_cfg['owner'],
|
|
category=build_cfg['category'],
|
|
version=version,
|
|
changelog=build_cfg['changelog'],
|
|
targetAbi=build_cfg['targetAbi'],
|
|
timestamp=build_date,
|
|
imagePath=image_path,
|
|
)
|
|
|
|
return meta
|
|
|
|
|
|
def generate_plugin_manifest(
|
|
manifest_path: pathlib.Path,
|
|
plugin_path: pathlib.Path,
|
|
repo_url: str,
|
|
plugin_url: str = '',
|
|
image_url: str = '',
|
|
) -> Manifest:
|
|
meta_path = plugin_path.with_name(plugin_path.name + '.' + JSON_METADATA_FILE)
|
|
meta = None
|
|
try:
|
|
with meta_path.open(encoding='utf-8') as fh:
|
|
meta = json.load(fh)
|
|
logger.info(f"Read meta from `{meta_path}`")
|
|
logger.debug(meta)
|
|
except (json.JSONDecodeError, OSError):
|
|
...
|
|
|
|
if meta is None:
|
|
with zipfile.ZipFile(plugin_path, 'r') as zf:
|
|
if JSON_METADATA_FILE in zf.namelist():
|
|
with zf.open(JSON_METADATA_FILE, 'r') as fh:
|
|
meta = json.load(fh)
|
|
logger.info(f"Read meta from `{plugin_path}:{JSON_METADATA_FILE}`")
|
|
logger.debug(meta)
|
|
|
|
if meta is None:
|
|
raise ValueError('Metadata not provided')
|
|
|
|
# TODO: Validate .md5sum file??
|
|
try:
|
|
checksum, _, _ = _read_md5sum(plugin_path.with_name(plugin_path.name + '.md5sum'))
|
|
except (OSError, ValueError):
|
|
checksum = checksum_file(plugin_path).hexdigest()
|
|
|
|
if not repo_url and not plugin_url:
|
|
logger.warning('repo and plugin url not provided, provide at least one.')
|
|
|
|
slug = slugify(meta['name'])
|
|
|
|
hostname = ''
|
|
owner = ''
|
|
project = ''
|
|
git_repo_path = find_git(manifest_path)
|
|
if git_repo_path is not None:
|
|
hostname, owner, project = get_repo_owner(git_repo_path)
|
|
|
|
source_url = repo_url.format(
|
|
slug=slug,
|
|
version=meta['version'],
|
|
hostname=hostname,
|
|
owner=owner,
|
|
project=project,
|
|
)
|
|
if plugin_url:
|
|
logger.info(f"Plugin url `{plugin_url}` overrides the autogenerated `{source_url}`.")
|
|
source_url = plugin_url
|
|
|
|
manifest = Manifest(
|
|
guid=str(uuid.UUID(meta['guid'])),
|
|
name=meta['name'],
|
|
description=meta['description'],
|
|
overview=meta['overview'],
|
|
owner=meta['owner'],
|
|
category=meta['category'],
|
|
versions=[
|
|
ManifestVersion(
|
|
version=meta['version'],
|
|
changelog=meta['changelog'],
|
|
targetAbi=meta['targetAbi'],
|
|
sourceUrl=source_url,
|
|
checksum=checksum,
|
|
timestamp=meta['timestamp'],
|
|
),
|
|
],
|
|
)
|
|
if image_url:
|
|
manifest['imageUrl'] = image_url
|
|
|
|
return manifest
|
|
|
|
|
|
def update_plugin_manifest(old: Manifest, new: Manifest) -> None:
|
|
|
|
new_version_numbers = [x['version'] for x in new['versions']]
|
|
|
|
# Fix old versions and remove replaced versions
|
|
for v in old['versions'].copy():
|
|
if v['version'] in new_version_numbers:
|
|
old['versions'].remove(v)
|
|
logger.warning('replacing %s: %s', v['version'], v)
|
|
continue
|
|
# Upgrade old incomplete version numbers - Jellyfin is not a fan of those.
|
|
v['version'] = Version(v['version']).full()
|
|
|
|
# Add the old versions to the current manifest
|
|
new['versions'].extend(old['versions'])
|
|
|
|
# Update old dictionary with new data
|
|
old.update(new)
|
|
old['versions'].sort(key=lambda v: Version(v['version']), reverse=True)
|
|
|
|
|
|
def get_plugin_from_manifest(
|
|
repo_manifest: list[Manifest], plugin: str | uuid.UUID | None,
|
|
) -> Manifest | None:
|
|
if plugin is None:
|
|
return None
|
|
|
|
if isinstance(plugin, uuid.UUID):
|
|
plugin = str(plugin)
|
|
else:
|
|
try:
|
|
plugin = str(uuid.UUID(plugin))
|
|
except ValueError:
|
|
pass
|
|
|
|
for item in repo_manifest:
|
|
if plugin in (item.get('name'), item.get('guid'), slugify(item.get('name'))):
|
|
return item
|
|
|
|
return None
|
|
|
|
|
|
def set_project_version(xml: ET.Element, version: Version) -> tuple[str | None, str | None, str | None]:
|
|
logger.info(f"Setting project version to {version.full()}")
|
|
|
|
versions = xml.findall('.//Version')
|
|
file_versions = xml.findall('.//FileVersion')
|
|
assembly_versions = xml.findall('.//AssemblyVersion')
|
|
if len(versions) > 1 or len(file_versions) > 1 or len(assembly_versions) > 1:
|
|
logger.error('Found multiple instances of the version tag(s), bailing.')
|
|
return (None, None, None)
|
|
|
|
old_version = None
|
|
if versions:
|
|
old_version = (versions[0].text or '').strip() or None
|
|
logger.debug(f"Old version: {old_version}")
|
|
versions[0].text = version.full()
|
|
|
|
old_file_version = None
|
|
if file_versions:
|
|
old_file_version = (file_versions[0].text or '').strip() or None
|
|
logger.debug(f"Old file version: {old_file_version}")
|
|
file_versions[0].text = version.full()
|
|
|
|
old_assembly_version = None
|
|
if assembly_versions:
|
|
old_assembly_version = (assembly_versions[0].text or '').strip() or None
|
|
logger.debug(f"Old assembly version: {old_assembly_version}")
|
|
assembly_versions[0].text = version.full()
|
|
|
|
return (old_version, old_file_version, old_assembly_version)
|
|
|
|
|
|
def set_project_framework(xml: ET.Element, framework: str) -> str | None:
|
|
logger.info(f"Setting project framework to {framework}")
|
|
|
|
frameworks = xml.findall('.//TargetFrameWork')
|
|
if len(frameworks) > 1:
|
|
logger.error('Found multiple instances of the TargetFramework tag, bailing.')
|
|
return None
|
|
|
|
old_framework = None
|
|
if frameworks:
|
|
old_framework = (frameworks[0].text or '').strip()
|
|
logger.debug(f"Old framework: {old_framework}")
|
|
frameworks[0].text = framework.strip()
|
|
|
|
return old_framework
|
|
|
|
|
|
def mangle_jellyfin_dependencies(xml: ET.Element, jellyfin_version: str) -> set[str]:
|
|
logger.info(f"Setting Jellyfin version to {jellyfin_version}")
|
|
|
|
jf_modules = set()
|
|
for x in xml.findall('.//PackageReference'):
|
|
# TODO: are there plugins that I need to account for here. EG Jellyfin.Plugin.FileTransformation.PluginInterface
|
|
if x.attrib['Include'].startswith('Jellyfin.'):
|
|
jf_modules.add(x.attrib['Include'])
|
|
x.attrib['Version'] = jellyfin_version
|
|
if 'ExcludeAssets' not in x.attrib:
|
|
x.attrib['ExcludeAssets'] = 'runtime'
|
|
|
|
return jf_modules
|
|
|
|
|
|
def solution_get_projects(sln_file: pathlib.Path) -> Generator[pathlib.Path]:
|
|
sln_path = sln_file.parent
|
|
if sln_file.suffix == '.sln':
|
|
yield from solution_get_deprecated_projects(sln_file)
|
|
return
|
|
|
|
with sln_file.open(encoding='utf-8') as fh:
|
|
sln = ET.parse(fh)
|
|
|
|
for project in sln.findall('.//Project'):
|
|
yield sln_path / pathlib.Path(project.attrib['Path'])
|
|
|
|
|
|
_solution_file_project_re = re.compile(
|
|
r'\s*Project\("[^"]*"\)\s*=\s*"(?P<project_name>[^"]*)",\s*"(?P<project_file>[^"]+proj)",\s*"[^"]*"\s*',
|
|
)
|
|
|
|
|
|
def solution_get_deprecated_projects(sln_file: pathlib.Path) -> Generator[pathlib.Path]:
|
|
with sln_file.open(encoding='utf-8') as fh:
|
|
data = fh.read()
|
|
|
|
sln_dir = sln_file.parent
|
|
matches = _solution_file_project_re.finditer(data)
|
|
for match in matches:
|
|
gd = match.groupdict()
|
|
project_file = sln_dir / gd.get('project_file', 'UNKNOWN').replace('\\', os.path.sep)
|
|
yield project_file
|
|
|
|
|
|
class RepoPathParam(click.ParamType[pathlib.Path]):
|
|
name = 'manifest_path'
|
|
|
|
def __init__(self, should_exist: bool | None = None) -> None:
|
|
self.should_exist = should_exist
|
|
|
|
def convert(self, value: str, param: Parameter | None, ctx: Context | None) -> pathlib.Path: # dead: disable
|
|
if value is None:
|
|
self.fail('--repo-path is required')
|
|
path = pathlib.Path(value)
|
|
if path.suffix != '.json':
|
|
path = path / 'manifest.json'
|
|
|
|
if self.should_exist is not None:
|
|
does_exist = path.exists()
|
|
if self.should_exist and not does_exist:
|
|
self.fail(f"Can not find repository at `{value}`. Try initializing the repo first.")
|
|
elif not self.should_exist and does_exist:
|
|
self.fail(f"There is already an existing repository at `{value}`.", param, ctx)
|
|
|
|
parent = path.parent
|
|
if not parent:
|
|
self.fail(f"The directory `{parent}` does not exist.", param, ctx)
|
|
|
|
return path.absolute()
|
|
|
|
|
|
class ZipFileParam(click.ParamType[pathlib.Path]):
|
|
name = 'zipfile'
|
|
|
|
def convert(self, value: str, param: Parameter | None, ctx: Context | None) -> pathlib.Path: # dead: disable
|
|
return pathlib.Path(value).absolute()
|
|
|
|
def validate(self, value: str, param: Parameter | None, ctx: Context | None) -> bool: # dead: disable
|
|
if not os.path.exists(value):
|
|
self.fail(f"No such file: `{value}`")
|
|
|
|
if not zipfile.is_zipfile(value):
|
|
self.fail(f"`{value}` is not a zip file.")
|
|
|
|
return True
|
|
|
|
|
|
@click.group()
|
|
@click.version_option(version=__version__, prog_name='Traitorous Jellyfin Plugin Repository Manager')
|
|
@click_log.simple_verbosity_option(logger)
|
|
def cli() -> None:
|
|
pass # Command grouping
|
|
|
|
|
|
@cli.group('plugin')
|
|
def cli_plugin() -> None:
|
|
pass # Command grouping
|
|
|
|
|
|
@cli_plugin.command('build')
|
|
@click.argument(
|
|
'path',
|
|
nargs=1,
|
|
required=False,
|
|
default='.',
|
|
type=click.Path(
|
|
path_type=pathlib.Path,
|
|
resolve_path=True,
|
|
exists=True,
|
|
file_okay=False,
|
|
dir_okay=True,
|
|
writable=True,
|
|
),
|
|
)
|
|
@click.option(
|
|
'--output',
|
|
'-o',
|
|
default=pathlib.Path('./artifacts'),
|
|
type=click.Path(
|
|
path_type=pathlib.Path,
|
|
resolve_path=True,
|
|
exists=False,
|
|
file_okay=False,
|
|
dir_okay=True,
|
|
writable=True,
|
|
),
|
|
help='Path to dotnet build directory',
|
|
show_default=True,
|
|
)
|
|
@click.option(
|
|
'--version',
|
|
'-v',
|
|
default=None,
|
|
help='Plugin version',
|
|
)
|
|
@click.option(
|
|
'--dotnet-configuration',
|
|
default='Release',
|
|
help='Dotnet configuration',
|
|
show_default=True,
|
|
)
|
|
@click.option(
|
|
'--dotnet-framework',
|
|
default=DEFAULT_FRAMEWORK,
|
|
help='Dotnet framework',
|
|
show_default=True,
|
|
)
|
|
@click.option(
|
|
'--max-cpu-count',
|
|
default=1,
|
|
type=int,
|
|
help='Max number of cores to use during build',
|
|
show_default=True,
|
|
)
|
|
@click.option(
|
|
'--changelog',
|
|
'-c',
|
|
default=None,
|
|
help='The changelog to use for this build. Use @path/to/file.txt to read from a file',
|
|
)
|
|
def cli_plugin_build( # dead: disable
|
|
path: pathlib.Path,
|
|
output: pathlib.Path,
|
|
dotnet_configuration: str,
|
|
dotnet_framework: str | None,
|
|
changelog: str | None,
|
|
max_cpu_count: int,
|
|
version: str | None,
|
|
) -> None:
|
|
build_cfg = get_config(path)
|
|
if build_cfg is None:
|
|
raise click.UsageError(f"No build config found in `{path}`")
|
|
|
|
if changelog is not None:
|
|
if changelog and changelog[0] == '@':
|
|
changelog = pathlib.Path(changelog[1:]).read_text(encoding='utf-8')
|
|
build_cfg['changelog'] = changelog
|
|
|
|
with tempfile.TemporaryDirectory() as bintemp:
|
|
bin_path = pathlib.Path(bintemp)
|
|
build_plugin(
|
|
path=path,
|
|
output=bin_path,
|
|
build_cfg=build_cfg,
|
|
dotnet_config=dotnet_configuration,
|
|
dotnet_framework=dotnet_framework,
|
|
version=version,
|
|
max_cpu_count=max_cpu_count,
|
|
)
|
|
logger.debug('Output files:')
|
|
for s in bin_path.iterdir():
|
|
logger.debug('%s', s)
|
|
filename = package_plugin(path=path, build_cfg=build_cfg, version=version, binary_path=bin_path, output=output)
|
|
click.echo(filename)
|
|
|
|
|
|
@cli.group('repo')
|
|
def cli_repo() -> None:
|
|
pass # Command grouping
|
|
|
|
|
|
@cli_repo.command('add')
|
|
@click.argument(
|
|
'manifest_path',
|
|
nargs=1,
|
|
required=True,
|
|
type=RepoPathParam(),
|
|
)
|
|
@click.argument(
|
|
'plugins',
|
|
nargs=-1,
|
|
required=True,
|
|
type=ZipFileParam(),
|
|
)
|
|
@click.option(
|
|
'--url',
|
|
'-u',
|
|
default='https://{hostname}/{owner}/{project}/releases/download/{version}/{slug}_{version}.zip',
|
|
help='Repository url format string {slug} and {version} will be replaced automatically',
|
|
)
|
|
@click.option(
|
|
'plugin_urls',
|
|
'--plugin-url',
|
|
'-U',
|
|
default=[],
|
|
help='Full URL of the plugin zip file',
|
|
multiple=True,
|
|
)
|
|
def cli_repo_add(manifest_path: pathlib.Path, plugins: list[pathlib.Path], url: str, plugin_urls: list[str], build_cfg: dict[str, Any] | None = None) -> None:
|
|
try:
|
|
with manifest_path.open(encoding='utf-8') as fh:
|
|
logger.debug(f"Reading repo manifest from {manifest_path}")
|
|
repo_manifest = cast(list[Manifest], json.load(fh))
|
|
except OSError:
|
|
repo_manifest = []
|
|
|
|
if not build_cfg:
|
|
build_cfg = get_config(pathlib.Path('.')) or {}
|
|
|
|
if plugin_urls and len(plugin_urls) != len(plugins):
|
|
logger.error(
|
|
"When plugin url is specified, the number of times it's specified must match the number of plugins.",
|
|
)
|
|
raise SystemExit(1)
|
|
|
|
for i, plugin_file in enumerate(plugins):
|
|
logger.info(f"Processing {plugin_file}")
|
|
|
|
plugin_url = ''
|
|
if len(plugin_urls) > i:
|
|
plugin_url = plugin_urls[i]
|
|
|
|
repo_dir = manifest_path.parent
|
|
plugin_manifest = generate_plugin_manifest(
|
|
repo_dir, plugin_file, repo_url=url, plugin_url=plugin_url, image_url=build_cfg.get('imageUrl', ''),
|
|
)
|
|
logger.debug(plugin_manifest)
|
|
|
|
# TODO: Add support for separate repo file path
|
|
name = plugin_manifest['name']
|
|
version = plugin_manifest['versions'][0]['version']
|
|
guid = uuid.UUID(plugin_manifest['guid'])
|
|
|
|
logger.info(
|
|
'Adding {plugin} version {version} to {repo}'.format(
|
|
plugin=name,
|
|
version=version,
|
|
repo=manifest_path,
|
|
),
|
|
)
|
|
|
|
# TODO: re-implement installing plugin/image/manifest into a specific place
|
|
|
|
updated = False
|
|
for p_manifest in repo_manifest:
|
|
if uuid.UUID(p_manifest.get('guid')) == guid:
|
|
update_plugin_manifest(p_manifest, plugin_manifest)
|
|
updated = True
|
|
break
|
|
|
|
if not updated:
|
|
repo_manifest.append(plugin_manifest)
|
|
|
|
tmpfile = manifest_path.with_suffix('.tmp')
|
|
with tmpfile.open('w', encoding='utf-8') as fh:
|
|
logging.debug(f"Writing repo manifest to {tmpfile}")
|
|
json.dump(repo_manifest, fh, indent=4)
|
|
|
|
logging.debug(f"Renaming {tmpfile} to {manifest_path}")
|
|
tmpfile.rename(manifest_path)
|
|
|
|
|
|
@cli_repo.command('list')
|
|
@click.argument(
|
|
'manifest_path',
|
|
nargs=1,
|
|
required=True,
|
|
type=RepoPathParam(should_exist=True),
|
|
)
|
|
@click.argument(
|
|
'plugin',
|
|
nargs=1,
|
|
required=False,
|
|
default=None,
|
|
)
|
|
def cli_repo_list(manifest_path: pathlib.Path, plugin: str | None) -> None: # dead: disable
|
|
try:
|
|
with manifest_path.open(encoding='utf-8') as fh:
|
|
logger.debug(f"Reading repo manifest from {manifest_path}")
|
|
repo_manifest: list[Manifest] = json.load(fh)
|
|
except FileNotFoundError:
|
|
repo_manifest = []
|
|
|
|
if plugin is not None:
|
|
try:
|
|
plugin = str(uuid.UUID(plugin))
|
|
except ValueError:
|
|
pass
|
|
|
|
plugin_found = False
|
|
for item in repo_manifest:
|
|
if plugin in (item.get('name'), item.get('guid'), slugify(item.get('name'))):
|
|
for v in item.get('versions', []):
|
|
click.echo(v.get('version'))
|
|
plugin_found = True
|
|
if not plugin_found:
|
|
raise click.UsageError(f"PLUGIN `{plugin}` not found in `{manifest_path}`")
|
|
return
|
|
table = []
|
|
for item in repo_manifest:
|
|
name = item.get('name')
|
|
guid = item.get('guid')
|
|
versions = sorted(
|
|
[release.get('version', '0.0') for release in item.get('versions', [])],
|
|
key=lambda rel: Version(rel),
|
|
reverse=True,
|
|
)
|
|
|
|
version = ''
|
|
if versions:
|
|
version = versions[0]
|
|
|
|
table.append([name, version, slugify(name), guid])
|
|
|
|
if table:
|
|
click.echo(
|
|
tabulate.tabulate(
|
|
table,
|
|
headers=('NAME', 'VERSION', 'SLUG', 'GUID'),
|
|
tablefmt='plain',
|
|
colalign=('left', 'right', 'left'),
|
|
),
|
|
)
|
|
|
|
|
|
@cli_repo.command('remove')
|
|
@click.argument(
|
|
'manifest_path',
|
|
nargs=1,
|
|
required=True,
|
|
type=RepoPathParam(should_exist=True),
|
|
)
|
|
@click.argument(
|
|
'plugin',
|
|
nargs=1,
|
|
required=True,
|
|
default=None,
|
|
)
|
|
@click.argument(
|
|
'version',
|
|
nargs=1,
|
|
required=False,
|
|
default=None,
|
|
type=Version,
|
|
)
|
|
def cli_repo_remove(manifest_path: pathlib.Path, plugin: str | None, version: Version | None) -> None: # dead: disable
|
|
try:
|
|
with manifest_path.open(encoding='utf-8') as fh:
|
|
logger.debug(f"Reading repo manifest from {manifest_path}")
|
|
repo_manifest: list[Manifest] = json.load(fh)
|
|
except FileNotFoundError:
|
|
repo_manifest = []
|
|
|
|
plugin_manifest = get_plugin_from_manifest(repo_manifest, plugin)
|
|
if plugin_manifest is None:
|
|
raise click.UsageError(f"PLUGIN `{plugin}` not found in `{manifest_path}`")
|
|
|
|
if version is None:
|
|
logger.warning(f"Removing plugin {plugin_manifest.get('name')})")
|
|
repo_manifest.remove(plugin_manifest)
|
|
click.echo(f"removed {plugin_manifest.get('guid')}")
|
|
else:
|
|
version_str = version.full()
|
|
for release in list(plugin_manifest.get('versions', [])):
|
|
if release.get('version') == version_str:
|
|
logger.warning(f"Removing version {version} of plugin {plugin_manifest.get('name')})")
|
|
plugin_manifest['versions'].remove(release)
|
|
click.echo(f"removed {plugin_manifest.get('guid')} {version_str}")
|
|
|
|
tmp_file = manifest_path.with_suffix('.tmp')
|
|
with tmp_file.open('w', encoding='utf-8') as fh:
|
|
logging.debug(f"Writing repo manifest to {tmp_file}")
|
|
json.dump(repo_manifest, fh, indent=4)
|
|
logging.debug(f"Renaming {tmp_file} to {manifest_path}")
|
|
tmp_file.rename(manifest_path)
|
|
|
|
|
|
def get_last_version(manifest_path: pathlib.Path, build_cfg: dict[str, Any]) -> Version:
|
|
try:
|
|
with manifest_path.open(encoding='utf-8') as fh:
|
|
logger.debug(f"Reading repo manifest from {manifest_path}")
|
|
repo_manifest: list[Manifest] = json.load(fh)
|
|
except FileNotFoundError:
|
|
repo_manifest = []
|
|
|
|
plugin_manifest = get_plugin_from_manifest(repo_manifest, build_cfg['guid'])
|
|
if plugin_manifest is None or not plugin_manifest['versions']:
|
|
return Version('0.0.0.0')
|
|
|
|
plugin_manifest['versions'].sort(key=lambda v: Version(v['version']), reverse=True)
|
|
return Version(Version(plugin_manifest['versions'][0]['version']).full())
|
|
|
|
|
|
@cli_repo.command('build')
|
|
@click.pass_context
|
|
@click.argument(
|
|
'manifest_path',
|
|
nargs=1,
|
|
required=True,
|
|
type=RepoPathParam(),
|
|
)
|
|
@click.argument(
|
|
'path',
|
|
nargs=1,
|
|
required=False,
|
|
default='.',
|
|
# help='path to dotnet solution to build',
|
|
type=click.Path(
|
|
path_type=pathlib.Path,
|
|
resolve_path=True,
|
|
exists=True,
|
|
file_okay=False,
|
|
dir_okay=True,
|
|
writable=True,
|
|
),
|
|
)
|
|
@click.option(
|
|
'--output',
|
|
'-o',
|
|
default=pathlib.Path('./artifacts'),
|
|
type=click.Path(
|
|
path_type=pathlib.Path,
|
|
resolve_path=True,
|
|
exists=False,
|
|
file_okay=False,
|
|
dir_okay=True,
|
|
writable=True,
|
|
),
|
|
help='Where to output build artifacts',
|
|
show_default=True,
|
|
)
|
|
@click.option(
|
|
'--version',
|
|
'-v',
|
|
default=None,
|
|
help='Set the plugin version to build, defaults to incrementing the revision',
|
|
)
|
|
@click.option(
|
|
'--changelog',
|
|
'-c',
|
|
default=None,
|
|
help='The changelog to use for this build. Use @path/to/file.txt to read from a file',
|
|
)
|
|
@click.option(
|
|
'--dotnet-configuration',
|
|
default='Release',
|
|
help='Dotnet configuration',
|
|
show_default=True,
|
|
)
|
|
@click.option(
|
|
'--dotnet-framework',
|
|
default=DEFAULT_FRAMEWORK,
|
|
help='Dotnet framework',
|
|
show_default=True,
|
|
)
|
|
@click.option(
|
|
'--max-cpu-count',
|
|
default=1,
|
|
type=int,
|
|
help='Max number of cores to use during build',
|
|
show_default=True,
|
|
)
|
|
@click.option(
|
|
'--url',
|
|
'-u',
|
|
default='https://{hostname}/{owner}/{project}/releases/download/{version}/{slug}_{version}.zip',
|
|
help='Repository url format string {hostname}, {owner}, {project}, {slug} and {version} will be replaced automatically',
|
|
show_default=True,
|
|
)
|
|
@click.option(
|
|
'plugin_url',
|
|
'--plugin-url',
|
|
'-U',
|
|
default=[],
|
|
help='Full URL of the plugin zip file',
|
|
)
|
|
def cli_repo_build( # dead: disable
|
|
ctx: click.Context,
|
|
manifest_path: pathlib.Path,
|
|
path: pathlib.Path,
|
|
output: pathlib.Path,
|
|
dotnet_configuration: str,
|
|
dotnet_framework: str | None,
|
|
max_cpu_count: int,
|
|
version: str | None,
|
|
changelog: str | None,
|
|
url: str,
|
|
plugin_url: list[str],
|
|
) -> None:
|
|
build_cfg = get_config(path)
|
|
if build_cfg is None:
|
|
raise click.UsageError(f"No build config found in `{path}`")
|
|
if changelog is not None:
|
|
if changelog and changelog[0] == '@':
|
|
changelog = pathlib.Path(changelog[1:]).read_text(encoding='utf-8')
|
|
build_cfg['changelog'] = changelog
|
|
|
|
if version is None:
|
|
v = get_last_version(manifest_path, build_cfg)
|
|
assert v.revision
|
|
v.revision += 1
|
|
version = v.full()
|
|
|
|
with tempfile.TemporaryDirectory() as bintemp:
|
|
bin_path = pathlib.Path(bintemp)
|
|
build_plugin(
|
|
path=path,
|
|
output=bin_path,
|
|
build_cfg=build_cfg,
|
|
dotnet_config=dotnet_configuration,
|
|
dotnet_framework=dotnet_framework,
|
|
version=version,
|
|
max_cpu_count=max_cpu_count,
|
|
)
|
|
logger.debug('Output files:')
|
|
for s in bin_path.iterdir():
|
|
logger.debug('%s', s)
|
|
filename = package_plugin(path=path, build_cfg=build_cfg, version=version, binary_path=bin_path, output=output)
|
|
|
|
ctx.invoke(
|
|
cli_repo_add,
|
|
manifest_path=manifest_path,
|
|
plugins=[filename],
|
|
plugin_urls=[plugin_url],
|
|
url=url,
|
|
build_cfg=build_cfg,
|
|
)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
cli()
|