97 lines
2.3 KiB
Python
97 lines
2.3 KiB
Python
import re
|
|
from argparse import ArgumentParser
|
|
|
|
import os
|
|
from glob import iglob
|
|
from typing import List
|
|
from pathlib import Path
|
|
|
|
from dataclasses import dataclass
|
|
from productmd.common import parse_nvra
|
|
|
|
|
|
@dataclass
|
|
class Package:
|
|
nvra: dict
|
|
path: Path
|
|
|
|
|
|
def search_rpms(top_dir: Path) -> List[Package]:
|
|
"""
|
|
Search for all *.rpm files recursively
|
|
in given top directory
|
|
Returns:
|
|
list: list of paths
|
|
"""
|
|
return [Package(
|
|
nvra=parse_nvra(Path(path).stem),
|
|
path=Path(path),
|
|
) for path in iglob(str(top_dir.joinpath('**/*.rpm')), recursive=True)]
|
|
|
|
|
|
def is_excluded_package(
|
|
package: Package,
|
|
excluded_packages: List[str],
|
|
) -> bool:
|
|
package_key = f'{package.nvra["name"]}.{package.nvra["arch"]}'
|
|
return any(
|
|
re.search(
|
|
f'^{excluded_pkg}$',
|
|
package_key,
|
|
) or excluded_pkg in (package.nvra['name'], package_key)
|
|
for excluded_pkg in excluded_packages
|
|
)
|
|
|
|
|
|
def copy_rpms(
|
|
packages: List[Package],
|
|
target_top_dir: Path,
|
|
excluded_packages: List[str],
|
|
):
|
|
"""
|
|
Search synced repos for rpms and prepare
|
|
koji-like structure for pungi
|
|
|
|
Instead of repos, use following structure:
|
|
# ls /mnt/koji/
|
|
i686/ noarch/ x86_64/
|
|
Returns:
|
|
Nothing:
|
|
"""
|
|
for package in packages:
|
|
if is_excluded_package(package, excluded_packages):
|
|
continue
|
|
target_arch_dir = target_top_dir.joinpath(package.nvra['arch'])
|
|
target_file = target_arch_dir.joinpath(package.path.name)
|
|
os.makedirs(target_arch_dir, exist_ok=True)
|
|
|
|
if not target_file.exists():
|
|
try:
|
|
os.link(package.path, target_file)
|
|
except OSError:
|
|
# hardlink failed, try symlinking
|
|
package.path.symlink_to(target_file)
|
|
|
|
|
|
def cli_main():
|
|
parser = ArgumentParser()
|
|
parser.add_argument('-p', '--path', required=True, type=Path)
|
|
parser.add_argument('-t', '--target', required=True, type=Path)
|
|
parser.add_argument(
|
|
'-e',
|
|
'--excluded-packages',
|
|
required=False,
|
|
nargs='+',
|
|
type=str,
|
|
default=[],
|
|
)
|
|
|
|
namespace = parser.parse_args()
|
|
|
|
rpms = search_rpms(namespace.path)
|
|
copy_rpms(rpms, namespace.target, namespace.excluded_packages)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
cli_main()
|