91 lines
2.7 KiB
Python
91 lines
2.7 KiB
Python
"""Helper functions for dealing with package files."""
|
|
|
|
import logging
|
|
import re
|
|
import shutil
|
|
|
|
from pathlib import Path
|
|
|
|
# ========== Constants ==========
|
|
# Some versions might have a colon (:)
|
|
# Match any pkgfile of any name with the .pkg.tar.* extension
|
|
PKGREGEX = r"^[\:\w.+/-]+\.pkg\.tar(\.\w+)?$"
|
|
# Match any sigfile of any name with the .pkg.tar.*.sig extension
|
|
SIGREGEX = r"^[\:\w.+/-]+\.pkg\.tar(\.\w+)?\.sig$"
|
|
|
|
|
|
# ========== Logging Setup ===========
|
|
syslog = logging.getLogger(__name__)
|
|
|
|
|
|
# ========== Functions ==========
|
|
def get(query=None, *, directory=None, signatures_only=False):
|
|
"""Retrieve all package-related files from the filesystem.
|
|
|
|
:param query: names of package files to search for
|
|
:type query: str
|
|
:param directory: a directory to search in
|
|
:type directory: str or path-like object
|
|
:param signatures_only: include only signature files
|
|
:type signatures_only: bool
|
|
:yields: paths of package files
|
|
"""
|
|
path = Path.cwd() if directory is None else Path(directory)
|
|
filequery = f"*{query}*" if query else "*"
|
|
|
|
if signatures_only:
|
|
yield from _filter_by_regex(SIGREGEX, path.glob(filequery))
|
|
else:
|
|
yield from _filter_by_regex(PKGREGEX, path.glob(filequery))
|
|
|
|
|
|
def filter(iterable, *, signatures_only=False):
|
|
"""Retrieve package files from a predefined iterable.
|
|
|
|
:param iterable: any iterable
|
|
:type iterable: anything that returns str or path-like objects
|
|
:param signatures_only: include only signature files
|
|
:type signatures_only: bool
|
|
:yields: paths of package files
|
|
"""
|
|
if signatures_only:
|
|
yield from _filter_by_regex(SIGREGEX, sorted(set(iterable)))
|
|
else:
|
|
yield from _filter_by_regex(PKGREGEX, sorted(set(iterable)))
|
|
|
|
|
|
def add(pkgfile, cachedir):
|
|
"""Add package file to the repository cache directory.
|
|
|
|
:param pkg: path of package to add
|
|
:type pkg: path-like object
|
|
:param cachedir: cache directory to move package to
|
|
:type cachedir: path-like object
|
|
"""
|
|
syslog.info("Adding %s to %s", pkgfile, cachedir)
|
|
shutil.move(pkgfile, cachedir / pkgfile.name)
|
|
|
|
|
|
def delete(pkg):
|
|
"""Remove package file.
|
|
|
|
:param pkg: path of package to remove
|
|
:type pkg: path-like object
|
|
"""
|
|
pkg.unlink()
|
|
syslog.info("Removed %s", pkg)
|
|
|
|
|
|
def _filter_by_regex(regex_expression, iterable):
|
|
"""Filter by regular expression.
|
|
|
|
:param regex_expression: the expression to filter by
|
|
:type regex_expression: str
|
|
:param iterable: iterable to filter through
|
|
:type iterable: str or path-like objects
|
|
:yields: pathlib.Path objects to package files
|
|
"""
|
|
for item in iterable:
|
|
if re.match(regex_expression, str(item)):
|
|
yield Path(item)
|