Atlas - commands.py
Home / npkg-testing / tools / npkg Lines: 1 | Size: 7118 bytes [Download] [Show on GitHub] [Search similar files] [Raw] [Raw (proxy)][FILE BEGIN]1from pathlib import Path 2import fnmatch 3 4from .archive import command_context, ensure_package_archive, extract_archive_into_root, render_command, run_shell 5from .console import fail, info, ok, style, warn 6from .install_db import load_installed_rows, uninstall_from_manifest, write_install_manifest 7from .metadata import discover_packages, select_package 8from .paths import workspace_root 9 10 11def resolve_package_selectors(packages, selectors): 12 root = workspace_root() 13 matched: list = [] 14 seen = set() 15 had_selector_miss = False 16 17 for selector in selectors: 18 selector = selector.strip() 19 if not selector: 20 continue 21 22 wildcard = any(token in selector for token in "*?[") 23 local_matches = [] 24 25 if wildcard: 26 for pkg in packages.values(): 27 rel_dir = str(pkg.package_dir.relative_to(root)) 28 if fnmatch.fnmatch(pkg.name, selector) or fnmatch.fnmatch(rel_dir, selector): 29 local_matches.append(pkg) 30 else: 31 try: 32 local_matches.append(select_package(packages, selector)) 33 except KeyError: 34 normalized_selector = selector.strip().strip("/") 35 for pkg in packages.values(): 36 rel_dir = str(pkg.package_dir.relative_to(root)) 37 if rel_dir == normalized_selector or rel_dir.startswith(f"{normalized_selector}/"): 38 local_matches.append(pkg) 39 40 if not local_matches: 41 normalized_selector = selector.strip().strip("/") 42 selector_exists = (root / normalized_selector).exists() if normalized_selector else False 43 if wildcard or not selector_exists: 44 warn(f"no packages matched selector: {selector}") 45 had_selector_miss = True 46 continue 47 48 for pkg in local_matches: 49 if pkg.name in seen: 50 continue 51 seen.add(pkg.name) 52 matched.append(pkg) 53 54 return matched, had_selector_miss 55 56 57def cmd_list(_args) -> int: 58 packages = discover_packages() 59 print(style("NPKG packages", bold=True)) 60 if not packages: 61 warn("no packages found") 62 return 0 63 64 for pkg in packages.values(): 65 flags = [] 66 if pkg.build_command: 67 flags.append("build") 68 if pkg.package_command: 69 flags.append("install") 70 if pkg.clean_command: 71 flags.append("clean") 72 if not flags: 73 flags.append("meta-only") 74 75 rel_dir = pkg.package_dir.relative_to(workspace_root()) 76 print( 77 f" {style(pkg.name, bold=True)} {pkg.version} " 78 f"[{', '.join(flags)}] {rel_dir}" 79 ) 80 if pkg.description: 81 print(f" {pkg.description}") 82 83 return 0 84 85 86def cmd_build(args) -> int: 87 packages = discover_packages() 88 matched, had_selector_miss = resolve_package_selectors(packages, args.packages) 89 if not matched: 90 return 2 91 92 had_failures = False 93 for pkg in matched: 94 if not pkg.build_command: 95 warn(f"skipping {pkg.name}: build.command is not defined") 96 had_failures = True 97 continue 98 99 command = render_command(pkg.build_command, command_context(pkg)) 100 info(f"building {pkg.name} with: {command}") 101 try: 102 run_shell(command, cwd=workspace_root()) 103 except RuntimeError as error: 104 warn(f"failed to build {pkg.name}: {error}") 105 had_failures = True 106 continue 107 108 ok(f"built {pkg.name}") 109 110 return 1 if (had_failures or had_selector_miss) else 0 111 112 113def cmd_package(args) -> int: 114 packages = discover_packages() 115 matched, had_selector_miss = resolve_package_selectors(packages, args.packages) 116 if not matched: 117 return 2 118 119 had_failures = False 120 for pkg in matched: 121 try: 122 archive_path = ensure_package_archive(pkg) 123 except RuntimeError as error: 124 warn(f"failed to package {pkg.name}: {error}") 125 had_failures = True 126 continue 127 128 ok(f"package archive ready: {archive_path}") 129 130 return 1 if (had_failures or had_selector_miss) else 0 131 132 133def cmd_install(args) -> int: 134 packages = discover_packages() 135 try: 136 pkg = select_package(packages, args.package) 137 except KeyError: 138 fail(f"package not found: {args.package}") 139 return 2 140 141 install_root = Path(args.root).expanduser().resolve() 142 try: 143 archive_path = ensure_package_archive(pkg) 144 except RuntimeError as error: 145 fail(str(error)) 146 return 2 147 148 install_root.mkdir(parents=True, exist_ok=True) 149 info(f"installing {pkg.name} into {install_root}") 150 try: 151 member_names = extract_archive_into_root(archive_path, install_root) 152 except RuntimeError as error: 153 fail(str(error)) 154 return 1 155 156 write_install_manifest(pkg, install_root, member_names) 157 ok(f"installed {pkg.name} -> {install_root}") 158 return 0 159 160 161def cmd_clean(args) -> int: 162 packages = discover_packages() 163 root = workspace_root() 164 matched, had_selector_miss = resolve_package_selectors(packages, args.packages) 165 if not matched: 166 return 2 167 168 had_failures = False 169 for pkg in matched: 170 if not pkg.clean_command: 171 warn(f"skipping {pkg.name}: clean.command is not defined") 172 had_failures = True 173 continue 174 175 command = render_command(pkg.clean_command, command_context(pkg)) 176 info(f"cleaning {pkg.name} with: {command}") 177 try: 178 run_shell(command, cwd=root) 179 except RuntimeError as error: 180 warn(f"failed to clean {pkg.name}: {error}") 181 had_failures = True 182 continue 183 184 ok(f"cleaned {pkg.name}") 185 186 return 1 if (had_failures or had_selector_miss) else 0 187 188 189def cmd_uninstall(args) -> int: 190 packages = discover_packages() 191 pkg_name = args.package 192 if pkg_name in packages: 193 pkg_name = packages[pkg_name].name 194 else: 195 try: 196 pkg_name = select_package(packages, args.package).name 197 except KeyError: 198 pkg_name = args.package 199 200 install_root = Path(args.root).expanduser().resolve() 201 info(f"uninstalling {pkg_name} from {install_root}") 202 try: 203 uninstall_from_manifest(pkg_name, install_root) 204 except RuntimeError as error: 205 fail(str(error)) 206 return 2 207 208 ok(f"uninstalled {pkg_name} from {install_root}") 209 return 0 210 211 212def cmd_installed(args) -> int: 213 install_root = Path(args.root).expanduser().resolve() 214 rows = load_installed_rows(install_root) 215 216 print(style(f"Installed packages in {install_root}", bold=True)) 217 if not rows: 218 warn("no installed packages found") 219 return 0 220 221 for row in rows: 222 if row.get("error"): 223 warn(f"skipping invalid manifest {row['manifest'].name}: {row['error']}") 224 continue 225 226 print( 227 f" {style(row['package'], bold=True)} {row['version']} " 228 f"({row['path_count']} paths)" 229 ) 230 231 return 0 232[FILE END](C) 2025 0x4248 (C) 2025 4248 Media and 4248 Systems, All part of 0x4248 See LICENCE files for more information. Not all files are by 0x4248 always check Licencing.