"""Inventory loading and validation.""" from __future__ import annotations import re from pathlib import Path import yaml from .models import UPDATE_CLASSES, Coupling, Item VALID_CATEGORIES = { "video", "audio", "rf", "lighting", "plugin", "os", "network", "other", } VALID_CRITICALITY = {"showstopper", "high", "normal", "low"} VALID_STATUS = {"confirmed", "unconfirmed"} class InventoryError(ValueError): pass def resolve_paths(path: Path) -> list[Path]: """Inventory can be split across files. ``inventory.yaml`` holds what the user actually confirmed; sibling ``inventory-*.yaml`` files hold proposed or optional additions. Keeping them separate means a whole tranche of guesses can be deleted in one go without touching the verified list. """ path = Path(path) if path.is_dir(): found = sorted(path.glob("inventory*.yaml")) if not found: raise InventoryError(f"no inventory*.yaml in {path}") return found if not path.exists(): raise InventoryError(f"inventory not found: {path}") extra = sorted( p for p in path.parent.glob("inventory-*.yaml") if p != path ) return [path] + extra def load(path: Path) -> list[Item]: """Load every inventory file and return the merged item list.""" items: list[Item] = [] seen: set[str] = set() for f in resolve_paths(path): items.extend(_load_one(f, seen)) return items def _load_one(path: Path, seen: set[str]) -> list[Item]: with open(path, "r", encoding="utf-8") as fh: raw = yaml.safe_load(fh) or {} entries = raw.get("items") if not isinstance(entries, list): raise InventoryError(f"{path.name} must contain a top-level 'items:' list") defaults = raw.get("defaults", {}) or {} items: list[Item] = [] for i, entry in enumerate(entries): if not isinstance(entry, dict): raise InventoryError(f"{path.name} items[{i}] is not a mapping") merged = {**defaults, **entry} item_id = str(merged.get("id") or "").strip() if not item_id: raise InventoryError(f"{path.name} items[{i}] missing required 'id'") if item_id in seen: raise InventoryError( f"duplicate item id {item_id!r} in {path.name} -- it is " f"already defined in another inventory file" ) seen.add(item_id) category = str(merged.get("category", "other")).lower() if category not in VALID_CATEGORIES: raise InventoryError( f"{item_id}: category {category!r} not one of {sorted(VALID_CATEGORIES)}" ) criticality = str(merged.get("criticality", "normal")).lower() if criticality not in VALID_CRITICALITY: raise InventoryError( f"{item_id}: criticality {criticality!r} not one of {sorted(VALID_CRITICALITY)}" ) source = merged.get("source") or {"kind": "manual"} if not isinstance(source, dict): raise InventoryError(f"{item_id}: 'source' must be a mapping") status = str(merged.get("status", "confirmed")).lower() if status not in VALID_STATUS: raise InventoryError( f"{item_id}: status {status!r} not one of {sorted(VALID_STATUS)}" ) update_class = str(merged.get("update_class", "normal")).lower() if update_class not in UPDATE_CLASSES: raise InventoryError( f"{item_id}: update_class {update_class!r} not one of " f"{sorted(UPDATE_CLASSES)}" ) coupled: list[Coupling] = [] for c in merged.get("coupled_with") or []: if isinstance(c, str): raise InventoryError( f"{item_id}: coupled_with entries must be mappings with " f"'id' and 'relation', got the bare string {c!r}" ) if not isinstance(c, dict) or "id" not in c or "relation" not in c: raise InventoryError( f"{item_id}: each coupled_with entry needs 'id' and " f"'relation'" ) significant = str(c.get("significant", "major")).lower() if significant not in ("major", "minor", "any"): raise InventoryError( f"{item_id}: coupled_with 'significant' must be " f"major, minor or any (got {significant!r})" ) coupled.append(Coupling( id=str(c["id"]), relation=str(c["relation"]), note=c.get("note"), significant=significant, auto=bool(c.get("auto", False)), )) installed = merged.get("installed") items.append( Item( id=item_id, name=str(merged.get("name") or item_id), vendor=str(merged.get("vendor") or "unknown"), category=category, installed=str(installed) if installed is not None else None, criticality=criticality, source=source, risk_queries=[str(q) for q in (merged.get("risk_queries") or [])], github_repo=merged.get("github_repo"), check_url=merged.get("check_url"), notes=merged.get("notes"), tags=[str(t) for t in (merged.get("tags") or [])], freeze=bool(merged.get("freeze", False)), status=status, coupled_with=coupled, update_group=merged.get("update_group"), update_class=update_class, expires=str(merged["expires"]) if merged.get("expires") else None, version_hint=merged.get("version_hint"), source_file=path.name, ) ) return items _ID_LINE = re.compile(r"^(\s*)-\s+id:\s*(\S+)\s*$") _INSTALLED_LINE = re.compile(r"^(\s*)installed:.*$") _NEW_ITEM = re.compile(r"^\s*-\s+id:\s*\S+") def save_installed(path: Path, item_id: str, version: str) -> None: """Set one item's ``installed`` value by editing that single line. Deliberately a text edit rather than a YAML load-and-dump: round-tripping through PyYAML discards every comment in the file, and the comments in inventory.yaml (which vendors cannot be scraped, which products still need identifying, why a given item is marked showstopper) are a large part of its value. """ # The item may live in any of the inventory files. target: Path | None = None lines: list[str] = [] start = None indent = "" for candidate in resolve_paths(Path(path)): candidate_lines = candidate.read_text(encoding="utf-8").splitlines( keepends=True) for i, line in enumerate(candidate_lines): m = _ID_LINE.match(line.rstrip("\n")) if m and m.group(2).strip("'\"") == item_id: target, lines, start = candidate, candidate_lines, i indent = m.group(1) + " " break if target is not None: break if target is None or start is None: raise InventoryError(f"no such item id: {item_id}") path = target # Quoted so values like 10.2 or 7.16 are not read back as floats. replacement = f"installed: '{version}'" # Walk to the end of this item's block. end = len(lines) for j in range(start + 1, len(lines)): if _NEW_ITEM.match(lines[j]): end = j break for j in range(start + 1, end): im = _INSTALLED_LINE.match(lines[j].rstrip("\n")) if im: lines[j] = f"{im.group(1)}{replacement}\n" break else: # No installed: key on this item -- insert one right after the id. lines.insert(start + 1, f"{indent}{replacement}\n") path.write_text("".join(lines), encoding="utf-8") def by_id(items: list[Item]) -> dict[str, Item]: return {i.id: i for i in items} def load_file_ids(path: Path) -> list[str]: """Ids declared in one inventory file, for reporting which file an item came from.""" with open(path, "r", encoding="utf-8") as fh: raw = yaml.safe_load(fh) or {} return [str(e.get("id")) for e in (raw.get("items") or []) if isinstance(e, dict) and e.get("id")]