Jan Lindemann
2e4afca5e3
All checks were successful
CI / Packaging - Kali Linux (pull_request) Successful in 5m30s
CI / Packaging - OpenSUSE Tumbleweed (pull_request) Successful in 4m35s
CI / Packaging test (pull_request) Successful in 0s
CI / Packaging - Kali Linux (push) Successful in 5m7s
CI / Packaging - OpenSUSE Tumbleweed (push) Successful in 4m36s
CI / Packaging test (push) Successful in 0s
get_project_refs() deduplicates the walk results by scanning the result list for each appended element, so the cost grows quadratically with the number of projects, which is noticeable when walking a dev tree of 320 projects. Track the already-seen projects in a set, and append an element only when it is new. The resulting list is unchanged, and the membership test is O(1) instead of O(n). Signed-off-by: Jan Lindemann <jan@janware.com> Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v0.86.1
629 lines
22 KiB
Python
629 lines
22 KiB
Python
#
|
|
# This source code file is a merge of various build tools and a horrible mess.
|
|
#
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
|
|
from enum import Enum, auto
|
|
from functools import cache
|
|
from typing import TYPE_CHECKING, override
|
|
|
|
from .lib.Distro import Distro
|
|
from .lib.ExecApp import ExecApp as Base
|
|
from .lib.log import DEBUG, ERR, log
|
|
from .lib.ProjectConf import ProjectConf
|
|
from .lib.version.Dependency import Dependency
|
|
|
|
if TYPE_CHECKING:
|
|
import argparse
|
|
|
|
from argparse import ArgumentParser
|
|
from typing import Iterable, TypeAlias
|
|
|
|
from .lib.PackageFilter import PackageFilter
|
|
|
|
# Meaning of pkg.requires.xxx variables
|
|
# build: needs to be built and installed before this can be built
|
|
# devel: needs to be installed before this-devel can be installed,
|
|
# i.e. before _other_ packages can be built against this
|
|
# run: needs to be installed before this-run can be installed,
|
|
# i.e. before this and other packages can run with this
|
|
|
|
# --------------------------------------------------------------------- Helpers
|
|
|
|
class Scope(Enum):
|
|
Self = auto()
|
|
One = auto()
|
|
Subtree = auto()
|
|
|
|
Graph: TypeAlias = dict[str, set[str]]
|
|
|
|
# ----------------------------------------------------------------- class App
|
|
|
|
class App(Base):
|
|
|
|
def __format_topdir(self, path: None | str, fmt: str) -> str | None:
|
|
if path is None:
|
|
return None
|
|
match fmt:
|
|
case 'unaltered':
|
|
return path
|
|
case 'relative':
|
|
return os.path.relpath(path)
|
|
case 'absolute':
|
|
return os.path.abspath(path)
|
|
case _:
|
|
m = re.search(r'^make:(\S+)$', fmt)
|
|
if m is None:
|
|
raise Exception(
|
|
f'Can\'t interpret "{fmt}" as valid topdir reference, '
|
|
'expecting "absolute", "relative", "unaltered", '
|
|
'or "make:<variable-name>"'
|
|
)
|
|
return '$(' + m.group(1) + ')'
|
|
|
|
@property
|
|
def __topdir(self) -> str:
|
|
if self.___topdir is None:
|
|
raise Exception('Tried to access undefined top directory')
|
|
return self.___topdir
|
|
|
|
@property
|
|
def __pretty_topdir(self) -> str:
|
|
if self.___pretty_topdir is None:
|
|
raise Exception('Tried to access undefined pretty top directory')
|
|
return self.___pretty_topdir
|
|
|
|
def __proj_dir(
|
|
self,
|
|
name: str,
|
|
pretty: bool,
|
|
projs_roots: Iterable[str | None] | None = None,
|
|
) -> str | None:
|
|
if name == self.__top_name:
|
|
if pretty:
|
|
return self.__pretty_topdir
|
|
return self.__topdir
|
|
for d in projs_roots or [self.__projs_root, self.___opt_root]:
|
|
if d is None:
|
|
continue
|
|
ret = d + '/' + name
|
|
if os.path.exists(ret):
|
|
return ret
|
|
if os.path.exists(f'/usr/share/doc/packages/{name}/VERSION'):
|
|
# The package exists but does not have a dedicated project directory
|
|
return None
|
|
raise Exception('No project path found for module "{}"'.format(name))
|
|
|
|
@cache
|
|
def __is_installed(self, name: str, devel: bool) -> bool:
|
|
# devel: the project is in the dev tree or the -devel
|
|
# package is installed: make/project.conf is present.
|
|
# run: the -run package is installed: a VERSION file is
|
|
# present in the project directory or
|
|
# /usr/share/doc/packages.
|
|
if devel and name == self.__top_name:
|
|
# the topdir is the project's own buildable checkout
|
|
if os.path.exists(f'{self.__topdir}/make/project.conf'):
|
|
return True
|
|
search, file = (
|
|
(self.__projs_root, self.___opt_root),
|
|
'/make/project.conf'
|
|
) if devel else (
|
|
(self.__projs_root, '/usr/share/doc/packages'),
|
|
'/VERSION'
|
|
)
|
|
for root in search:
|
|
if root is not None and os.path.exists(f'{root}/{name}{file}'):
|
|
return True
|
|
return False
|
|
|
|
def __find_dir(
|
|
self,
|
|
name: str,
|
|
search_subdirs: list[str] | None = None,
|
|
search_absdirs: list[str] | None = None,
|
|
pretty: bool = True,
|
|
projs_roots: Iterable[str | None] | None = None,
|
|
) -> str | None:
|
|
if search_subdirs is None:
|
|
search_subdirs = []
|
|
if search_absdirs is None:
|
|
search_absdirs = []
|
|
|
|
def __format_relpath(path: str) -> str:
|
|
if path.startswith('./'):
|
|
return path[2:]
|
|
if path.endswith('/.'):
|
|
return path[:-2]
|
|
return path
|
|
|
|
def __relpath(target: str, base: str) -> str:
|
|
return __format_relpath(os.path.relpath(target, base))
|
|
|
|
def __format_pd(name: str, pd: str, pretty: bool) -> str | None:
|
|
if not pretty:
|
|
return pd
|
|
if self.__topdir_fmt == 'absolute':
|
|
return str(os.path.abspath(pd))
|
|
if self.__topdir_fmt == 'unaltered':
|
|
return pd
|
|
if self.__topdir_fmt == 'relative':
|
|
return __relpath(pd, self.__topdir)
|
|
if self.__topdir_fmt.startswith('make:'):
|
|
relpath = __relpath(pd, self.__topdir)
|
|
var = self.__topdir_fmt.split(':')[1]
|
|
return __format_relpath(f'$({var})/{relpath}')
|
|
if name == self.__top_name:
|
|
return self.__pretty_topdir
|
|
raise NotImplementedError(
|
|
f'Tried to pretty-format directory {pd}, not implemented'
|
|
)
|
|
|
|
pd = self.__proj_dir(name, pretty = False, projs_roots = projs_roots)
|
|
if pd is None:
|
|
return None
|
|
if not search_subdirs and not search_absdirs:
|
|
return __format_pd(name, pd, pretty)
|
|
for sd in search_subdirs:
|
|
path = pd + '/' + sd
|
|
if os.path.isdir(path):
|
|
ret = __format_pd(name, pd, pretty)
|
|
assert ret is not None
|
|
if sd and sd[0] != '/':
|
|
if ret == '.':
|
|
ret = ''
|
|
else:
|
|
ret += '/'
|
|
ret += sd
|
|
return ret
|
|
for ret in search_absdirs:
|
|
if os.path.isdir(ret):
|
|
return ret
|
|
return None
|
|
|
|
def __read_project_conf(self, project_dir: str) -> ProjectConf:
|
|
return ProjectConf.read(project_dir + '/make/project.conf')
|
|
|
|
@cache
|
|
def __get_project_conf(self, project: str) -> ProjectConf | None:
|
|
pd = self.__proj_dir(project, False)
|
|
if pd is None:
|
|
raise Exception(f'Failed to find directory of project {project}')
|
|
try:
|
|
return self.__read_project_conf(pd)
|
|
except FileNotFoundError:
|
|
return None
|
|
|
|
def read_dep_edges(
|
|
self,
|
|
name: str,
|
|
sections: list[str],
|
|
key: str,
|
|
) -> list[str]:
|
|
"""The deduplicated dependency values of name for the given
|
|
sections and key (flavour), in section and value order.
|
|
"""
|
|
ret: list[str] = []
|
|
for section in sections:
|
|
vals = self.get_value(name, section, key)
|
|
log(DEBUG, f'name={name}, section={section}, key={key}, deps={vals}')
|
|
if not vals:
|
|
continue
|
|
for val in vals.split(','):
|
|
val = val.strip()
|
|
if (len(val)) and (val not in ret):
|
|
ret.append(val)
|
|
return ret
|
|
|
|
def walk_project_deps(
|
|
self,
|
|
buf: list[str],
|
|
visited: set[str],
|
|
spec: str,
|
|
sections: list[str],
|
|
key: str,
|
|
add_self: bool,
|
|
scope: Scope,
|
|
names_only: bool,
|
|
*,
|
|
check_installed: bool = True,
|
|
exclude: set[str] = set(),
|
|
recurse: bool = True,
|
|
) -> None:
|
|
"""Walk the dependency edges of sections and key (flavour)
|
|
starting at spec, appending the visited specs to buf in
|
|
postorder; buf and visited are updated in place.
|
|
|
|
check_installed applies the installed-package check, which
|
|
skips the dependencies of an installed run package and fails
|
|
on an unmet dependency. exclude names specs that are neither
|
|
traversed nor appended. recurse = False stops the walk at the
|
|
starting spec.
|
|
"""
|
|
dep = Dependency(spec)
|
|
name = dep.base_name
|
|
mod = dep.full_name
|
|
if names_only:
|
|
spec = name
|
|
if spec in buf:
|
|
return
|
|
if spec in exclude:
|
|
return
|
|
if spec in visited:
|
|
if add_self:
|
|
buf.append(spec)
|
|
return
|
|
visited.add(spec)
|
|
if check_installed:
|
|
needed_subpackage = 'devel' if mod.endswith(('-dev', '-devel')) else 'run'
|
|
if not self.is_installed(name, devel = True):
|
|
if not mod.endswith(('-dev', '-devel')):
|
|
if self.is_installed(name, devel = False):
|
|
return
|
|
raise Exception(
|
|
f'Unmet dependency on {mod}: the -{needed_subpackage} package '
|
|
f'of project {name} is not installed'
|
|
)
|
|
vals_list = self.read_dep_edges(name, sections, key)
|
|
match scope:
|
|
case Scope.Self:
|
|
buf += vals_list
|
|
case Scope.One | Scope.Subtree:
|
|
if recurse:
|
|
subscope = scope.Self if scope == Scope.One else scope
|
|
for val in vals_list:
|
|
self.walk_project_deps(
|
|
buf,
|
|
visited,
|
|
val,
|
|
sections,
|
|
key,
|
|
add_self = True,
|
|
scope = subscope,
|
|
names_only = names_only,
|
|
check_installed = check_installed,
|
|
exclude = exclude,
|
|
recurse = recurse,
|
|
)
|
|
if add_self:
|
|
buf.append(spec)
|
|
|
|
def __read_dep_graph(
|
|
self,
|
|
projects: list[str],
|
|
sections: str | list[str],
|
|
graph: Graph,
|
|
) -> None:
|
|
if isinstance(sections, str):
|
|
sections = [sections]
|
|
for project in projects:
|
|
if project in graph:
|
|
continue
|
|
deps = self.get_project_refs(
|
|
[project],
|
|
['pkg.requires.jw'],
|
|
sections,
|
|
scope = Scope.One,
|
|
add_self = False,
|
|
names_only = True,
|
|
)
|
|
graph[project] = set(deps)
|
|
for dep in deps:
|
|
self.__read_dep_graph([dep], sections, graph)
|
|
|
|
def __flip_dep_graph(self, graph: Graph) -> Graph:
|
|
ret: Graph = {}
|
|
for project, deps in graph.items():
|
|
for d in deps:
|
|
if d not in ret:
|
|
ret[d] = set()
|
|
ret[d].add(project)
|
|
return ret
|
|
|
|
def __find_circular_deps_recursive(
|
|
self,
|
|
project: str,
|
|
graph: Graph,
|
|
unvisited: list[str],
|
|
stack: list[str],
|
|
) -> list[str] | None:
|
|
if project in stack:
|
|
log(DEBUG, 'found circular dependency at project', project)
|
|
idx = stack.index(project)
|
|
return stack[idx:] + [project]
|
|
if project not in unvisited:
|
|
return None
|
|
stack.append(project)
|
|
if project in graph:
|
|
for dep in graph[project]:
|
|
cycle = self.__find_circular_deps_recursive(
|
|
dep, graph, unvisited, stack
|
|
)
|
|
if cycle is not None:
|
|
return cycle
|
|
unvisited.remove(project)
|
|
stack.pop()
|
|
return None
|
|
|
|
def __find_circular_deps(self, projects: list[str],
|
|
flavours: list[str]) -> list[str]:
|
|
graph: Graph = {}
|
|
self.__read_dep_graph(projects, flavours, graph)
|
|
unvisited = list(graph.keys())
|
|
flipped = self.__flip_dep_graph(graph)
|
|
while unvisited:
|
|
project = unvisited[0]
|
|
log(DEBUG, 'Checking circular dependency of', project)
|
|
cycle = self.__find_circular_deps_recursive(project, flipped, unvisited, [])
|
|
if cycle is not None:
|
|
# An edge a -> b in the flipped graph means that b
|
|
# depends on a, so reverse to report the cycle in the
|
|
# original direction
|
|
cycle = list(reversed(cycle))
|
|
log(DEBUG, f'Found circular dependency: {" -> ".join(cycle)}')
|
|
return cycle
|
|
return []
|
|
|
|
def __init__(self, distro: Distro | None = None) -> None:
|
|
|
|
super().__init__(
|
|
description = 'jw-pkg swiss army knife', modules = ['jw.pkg.cmds']
|
|
)
|
|
|
|
# -- Members without default values
|
|
self.__top_name: str | None = None
|
|
self.__distro = distro
|
|
self.___topdir: str | None = None
|
|
self.___pretty_topdir: str | None = None
|
|
|
|
# -- Members with default values
|
|
self.__topdir_fmt = 'absolute'
|
|
self.__projs_root: str | None = None
|
|
self.___opt_root = '/opt'
|
|
self.__pretty_projs_root = None
|
|
|
|
async def __init_async(self) -> None:
|
|
if self.__distro is None:
|
|
pkg_filter_str = self.args.pkg_filter
|
|
if pkg_filter_str is None:
|
|
pkg_filter_str = os.getenv('JW_DEFAULT_PKG_FILTER')
|
|
pkg_filter: PackageFilter | None = None
|
|
if pkg_filter_str is not None:
|
|
from .lib.PackageFilter import PackageFilterString
|
|
|
|
pkg_filter = PackageFilterString(pkg_filter_str)
|
|
self.__distro = await Distro.instantiate(
|
|
ec = self.exec_context,
|
|
id = self.args.distro_id,
|
|
default_pkg_filter = pkg_filter,
|
|
)
|
|
|
|
@override
|
|
def _add_arguments(self, parser: ArgumentParser) -> None:
|
|
super()._add_arguments(parser)
|
|
parser.add_argument('-t', '--topdir', default = None, help = 'Project Path')
|
|
parser.add_argument(
|
|
'--topdir-format',
|
|
default = 'absolute',
|
|
help = (
|
|
'Output references to topdir as one of "make:<var-name>", '
|
|
'"unaltered", "relative", "absolute". Absolute topdir by default'
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
'-p',
|
|
'--prefix',
|
|
default = None,
|
|
help = 'Parent directory of project source directories',
|
|
)
|
|
parser.add_argument(
|
|
'--distro-id',
|
|
default = None,
|
|
help = 'Distribution ID (default is taken from /etc/os-release)',
|
|
)
|
|
parser.add_argument(
|
|
'--pkg-filter',
|
|
help = 'Default filter for all distribution package-related operations',
|
|
)
|
|
|
|
@override
|
|
async def _run(self, args: argparse.Namespace) -> None:
|
|
self.___topdir = args.topdir
|
|
self.___pretty_topdir = self.__format_topdir(self.___topdir, args.topdir_format)
|
|
self.__topdir_fmt = args.topdir_format
|
|
if self.___topdir is not None:
|
|
try:
|
|
conf = self.__read_project_conf(self.__topdir)
|
|
self.__top_name = conf.get_str_or_none('build', 'name')
|
|
except FileNotFoundError:
|
|
pass
|
|
if not self.__top_name:
|
|
self.__top_name = re.sub(
|
|
'-[0-9.-]*$',
|
|
'',
|
|
os.path.basename(os.path.realpath(self.___topdir))
|
|
)
|
|
if args.prefix is not None:
|
|
self.__projs_root = args.prefix
|
|
self.__pretty_projs_root = args.prefix
|
|
await self.__init_async()
|
|
await super()._run(args)
|
|
|
|
@property
|
|
def top_name(self) -> str | None:
|
|
return self.__top_name
|
|
|
|
@property
|
|
def projs_root(self) -> str:
|
|
if self.__projs_root is None:
|
|
raise Exception('Tried to get unknown projects root directory')
|
|
return self.__projs_root
|
|
|
|
@property
|
|
def distro(self) -> Distro:
|
|
if self.__distro is None:
|
|
raise Exception('No distro object')
|
|
return self.__distro
|
|
|
|
def is_installed(
|
|
self,
|
|
name: str,
|
|
devel: bool = False,
|
|
) -> bool:
|
|
"""True if the project is installed: for devel,
|
|
make/project.conf is present in the dev tree or /opt; for
|
|
run, a VERSION file is present in the project directory
|
|
or /usr/share/doc/packages.
|
|
"""
|
|
return self.__is_installed(name, devel)
|
|
|
|
def find_dir(
|
|
self,
|
|
name: str,
|
|
search_subdirs: list[str] | None = None,
|
|
search_absdirs: list[str] | None = None,
|
|
pretty: bool = True,
|
|
throw: bool = False,
|
|
projs_roots: Iterable[str | None] | None = None,
|
|
) -> str | None:
|
|
ret = self.__find_dir(name, search_subdirs, search_absdirs, pretty, projs_roots)
|
|
if ret is not None:
|
|
return ret
|
|
if not throw:
|
|
return None
|
|
msg = f'Failed to find directory for "{name}":'
|
|
log(ERR, msg)
|
|
for search_name, search in [
|
|
('subdirs', search_subdirs),
|
|
('absdirs', search_absdirs),
|
|
]:
|
|
if search:
|
|
log(ERR, f'Searched {search_name}:')
|
|
for d in search:
|
|
log(ERR, f' - {d}')
|
|
raise FileNotFoundError(msg)
|
|
|
|
# TODO: add support for customizing this in project.conf
|
|
def htdocs_dir(self, project: str) -> str | None:
|
|
return self.find_dir(
|
|
project,
|
|
['/src/html/htdocs', '/tools/html/htdocs', '/htdocs'],
|
|
['/srv/www/proj/' + project],
|
|
)
|
|
|
|
# TODO: add support for customizing this in project.conf
|
|
def tmpl_dir(self, name: str) -> str | None:
|
|
return self.find_dir(name, ['/tmpl'], ['/opt/' + name + '/share/tmpl'])
|
|
|
|
def strip_module_from_spec(self, mod: str) -> str:
|
|
return Dependency(mod).base_name
|
|
|
|
@cache
|
|
def get_value(self, project: str, section: str, key: str) -> str | None:
|
|
ret: str | None
|
|
if section == 'version':
|
|
proj_dir = self.__proj_dir(project, pretty = False)
|
|
if proj_dir is None:
|
|
raise Exception(f"Can't get project directory for {project}")
|
|
proj_version_dirs = [proj_dir]
|
|
if proj_dir != self.___topdir:
|
|
proj_version_dirs.append('/usr/share/doc/packages/' + project)
|
|
for d in proj_version_dirs:
|
|
version_path = d + '/VERSION'
|
|
try:
|
|
with open(version_path) as fd:
|
|
ret = fd.read().replace('\n', '').replace('-dev', '')
|
|
return ret
|
|
except EnvironmentError:
|
|
log(DEBUG, f'Ignoring unreadable file "{version_path}"')
|
|
continue
|
|
raise Exception(f'No version file found for project "{project}"')
|
|
proj_conf = self.__get_project_conf(project)
|
|
if proj_conf is None:
|
|
return None
|
|
ret = proj_conf.get_str_or_none(section, key)
|
|
log(
|
|
DEBUG,
|
|
'Lookup %s -> %s / [%s%s] -> "%s"' %
|
|
(self.__top_name, project, section, '.' + key if key else '', ret),
|
|
)
|
|
return ret
|
|
|
|
@cache
|
|
def get_version(self, project: str) -> str:
|
|
ret = self.get_value(project, 'version', '')
|
|
if ret is None:
|
|
raise Exception(f"Can't get version of project {project}")
|
|
return ret
|
|
|
|
def get_values(self, projects: list[str], sections: list[str],
|
|
keys: list[str]) -> list[str]:
|
|
"""
|
|
Collect a list of values from a list of given projects, sections and
|
|
keys, maintaining order
|
|
"""
|
|
ret: list[str] = []
|
|
for p in projects:
|
|
for section in sections:
|
|
for key in keys:
|
|
vals = self.get_value(p, section, key)
|
|
if vals:
|
|
for val in vals.split(','):
|
|
stripped = val.strip()
|
|
if stripped:
|
|
ret.append(stripped)
|
|
return list(dict.fromkeys(ret)) # Remove duplicates, keep ordering
|
|
|
|
def get_project_refs(
|
|
self,
|
|
projects: list[str],
|
|
sections: list[str],
|
|
keys: str | list[str],
|
|
scope: Scope = Scope.One,
|
|
add_self: bool = False,
|
|
names_only: bool = False,
|
|
) -> list[str]:
|
|
if isinstance(keys, str):
|
|
keys = [keys]
|
|
ret: list[str] = []
|
|
seen: set[str] = set()
|
|
for key in keys:
|
|
visited: set[str] = set()
|
|
for name in projects:
|
|
rr: list[str] = []
|
|
self.walk_project_deps(
|
|
rr, visited, name, sections, key, add_self, scope, names_only
|
|
)
|
|
for m in rr:
|
|
if m not in seen:
|
|
seen.add(m)
|
|
ret.append(m)
|
|
return ret
|
|
|
|
def get_libname(self, spec: str) -> str | None:
|
|
project_name = self.strip_module_from_spec(spec)
|
|
ret = self.get_value(project_name, 'build', 'libname')
|
|
if ret == 'none':
|
|
return None
|
|
if ret is None:
|
|
return project_name
|
|
return ret
|
|
|
|
def is_excluded_from_build(self, project: str) -> str | None:
|
|
log(DEBUG, 'checking if project ' + project + ' is excluded from build')
|
|
exclude = self.get_value(project, 'build', 'exclude')
|
|
if exclude is None:
|
|
return None
|
|
exclude_arr = re.split(r'[, ]+', exclude)
|
|
cascade = self.distro.os_cascade + ['all']
|
|
intersection = [x for x in cascade if x in set(exclude_arr)]
|
|
if intersection:
|
|
return ', '.join(intersection)
|
|
return None
|
|
|
|
def find_circular_deps(self, projects: list[str], flavours: list[str]) -> list[str]:
|
|
return self.__find_circular_deps(projects, flavours)
|