tegwick
04145a59d9
Implement RMGR-WP-0004 T01 (flavor, required files, anti-patterns from canon) and T08 (prefix ownership registry plus uniqueness scan). rmgr conform and rmgr prefix-uniqueness are detection only.
112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
"""Fleet workplan-prefix ownership and uniqueness (RMGR-WP-0004-T08)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from repo_manager.standards import FLAVOR_MARKER_PREFIX, workplan_ids
|
|
|
|
DEFAULT_REGISTRY = Path(__file__).resolve().parents[2] / "config" / "workplan-prefix-registry.yaml"
|
|
|
|
|
|
def load_registry(path: Path | None = None) -> dict[str, Any]:
|
|
target = path or DEFAULT_REGISTRY
|
|
data = yaml.safe_load(target.read_text(encoding="utf-8")) or {}
|
|
if not isinstance(data, dict):
|
|
return {}
|
|
return data
|
|
|
|
|
|
def _skip_dir(name: str) -> bool:
|
|
return name.startswith(".") or name in {
|
|
"node_modules",
|
|
"__pycache__",
|
|
".venv",
|
|
"venv",
|
|
"archive",
|
|
}
|
|
|
|
|
|
def iter_repo_roots(root: Path) -> list[Path]:
|
|
root = root.resolve()
|
|
if (root / "workplans").is_dir() and (root / ".git").exists():
|
|
return [root]
|
|
found: list[Path] = []
|
|
if not root.is_dir():
|
|
return found
|
|
for child in sorted(root.iterdir()):
|
|
if not child.is_dir() or _skip_dir(child.name):
|
|
continue
|
|
if (child / "workplans").is_dir():
|
|
found.append(child)
|
|
return found
|
|
|
|
|
|
def scan_prefixes(root: Path, *, registry_path: Path | None = None) -> dict[str, Any]:
|
|
"""Detect prefix sharing, identifier reuse, flavor-derived prefixes, number reuse.
|
|
|
|
Detection only. Does not rewrite files or remediations.
|
|
"""
|
|
registry = load_registry(registry_path)
|
|
owners = {str(k): str(v) for k, v in (registry.get("owners") or {}).items()}
|
|
retired = {str(item.get("prefix")) for item in registry.get("retired") or [] if item.get("prefix")}
|
|
|
|
by_prefix: dict[str, set[str]] = defaultdict(set)
|
|
by_id: dict[str, list[tuple[str, str]]] = defaultdict(list)
|
|
numbers: dict[tuple[str, str], list[str]] = defaultdict(list)
|
|
|
|
repos_scanned = []
|
|
for repo in iter_repo_roots(root):
|
|
repos_scanned.append(repo.name)
|
|
for rel, prefix, number in workplan_ids(repo):
|
|
by_prefix[prefix].add(repo.name)
|
|
ident = f"{prefix}-{number}"
|
|
by_id[ident].append((repo.name, rel))
|
|
numbers[(repo.name, prefix)].append(number)
|
|
|
|
shared = {
|
|
prefix: sorted(repos)
|
|
for prefix, repos in sorted(by_prefix.items())
|
|
if len(repos) > 1
|
|
}
|
|
reused = {
|
|
ident: [{"repo": repo, "path": path} for repo, path in entries]
|
|
for ident, entries in sorted(by_id.items())
|
|
if len({repo for repo, _path in entries}) > 1 or len(entries) > 1
|
|
}
|
|
flavor_derived = sorted(repo for repo in by_prefix.get(FLAVOR_MARKER_PREFIX, []))
|
|
retired_in_use = {
|
|
prefix: sorted(repos)
|
|
for prefix, repos in by_prefix.items()
|
|
if prefix in retired
|
|
}
|
|
owner_mismatch = {
|
|
prefix: {"declared_owner": owners[prefix], "seen_in": sorted(repos)}
|
|
for prefix, repos in by_prefix.items()
|
|
if prefix in owners and any(repo != owners[prefix] for repo in repos)
|
|
}
|
|
|
|
reused_numbers = []
|
|
for (repo, prefix), nums in numbers.items():
|
|
seen: set[str] = set()
|
|
for num in nums:
|
|
if num in seen:
|
|
reused_numbers.append({"repo": repo, "prefix": prefix, "number": num})
|
|
seen.add(num)
|
|
|
|
return {
|
|
"ok": not shared and not flavor_derived and not reused_numbers and not reused,
|
|
"root": str(root.resolve()),
|
|
"repos_scanned": repos_scanned,
|
|
"shared_prefixes": shared,
|
|
"reused_identifiers": reused,
|
|
"flavor_derived_prefix": flavor_derived,
|
|
"retired_in_use": retired_in_use,
|
|
"owner_mismatch": owner_mismatch,
|
|
"reused_numbers_in_repo": reused_numbers,
|
|
"detection_only": True,
|
|
}
|