"""Sortable version key, tolerating anything unparseable."""

try:

from packaging.version import InvalidVersion, Version

try:

return (1, Version(text))

except InvalidVersion:

pass

except ImportError:

pass

"""(normalized name, version) for a wheel or sdist file name, else None.

name = path.name

if name.endswith(WHEEL):

parts = name[: -len(WHEEL)].split("-")

return (normalize(parts[0]), parts[1]) if len(parts) >= 3 else None

for suffix in SDIST:

if name.endswith(suffix):

stem = name[: -len(suffix)]

break

else:

return None

candidates = [k for k in known if normalize(stem).startswith(k + "-")]

if candidates:

best = max(candidates, key=len)

return best, stem[len(best) + 1:]

match = re.match(r"^(.*?)-(\d.*)$", stem)

"""The newest lockfile per (cycle, Python minor, flavor).

best: dict[tuple, tuple[tuple, Path]] = {}

for directory in cycle_dirs:

for path in directory.glob("pylock.*.toml"):

match = LOCK_NAME.match(path.name)

if match is None:

continue

parts = tuple(int(n) for n in match["ver"].split("_"))

level = float(match["level"]) if match["level"] else float("inf")

key = (directory, parts[:2], match["flavor"], match["wheels"] or "")

rank = (parts, level)

if key not in best or rank > best[key][0]:

best[key] = (rank, path)

"""Artifacts the shipped lockfiles install, and which were built from sdist."""

locks = newest_locks(cycle_dirs)

wanted: dict[tuple[str, str], set[str]] = defaultdict(set)

from_sdist: dict[tuple[str, str], set[str]] = defaultdict(set)

for lock in locks:

data = tomllib.loads(lock.read_text(encoding="utf-8"))

for pkg in data.get("packages", []):

key = (normalize(pkg["name"]), pkg["version"])

for artifact in pkg.get("wheels", []):

if digest := artifact.get("hashes", {}).get("sha256"):

wanted[key].add(digest)

if sdist := pkg.get("sdist"):

if digest := sdist.get("hashes", {}).get("sha256"):

wanted[key].add(digest)

from_sdist[key].add(lock.name)

"""Project names any current requirement or constraint file mentions."""

names: set[str] = set()

for filename in REQUIREMENT_FILES:

path = repo / filename

if not path.is_file():

continue

for line in path.read_text(encoding="utf-8", errors="replace").splitlines():

if line := line.split("#")[0].strip():

if name := re.split(r"[<>=!~;\[ ]", line)[0].strip():

names.add(normalize(name))

parser = argparse.ArgumentParser(description=__doc__,

formatter_class=argparse.RawDescriptionHelpFormatter)

parser.add_argument("--srcreq", type=Path, default=Path(r"C:\WinP\packages.srcreq"),

help="the wheelhouse to inventory")

parser.add_argument("--cycle", type=Path, action="append", dest="cycles",

help="cycle directory to read lockfiles from; repeatable, "

"defaults to the two newest under winpython/portable")

parser.add_argument("--repo", type=Path, default=Path("."))

parser.add_argument("--out", type=Path, default=Path("."))

parser.add_argument("--delete", choices=PRUNABLE, action="append", dest="delete",

help="remove the files in this bucket; repeatable")

args = parser.parse_args(argv)

cycles = args.cycles or sorted((args.repo / "winpython/portable").glob("cycle_*"))[-2:]

if not args.srcreq.is_dir():

print(f"no such wheelhouse: {args.srcreq}", file=sys.stderr)

return 2

missing = [d for d in cycles if not d.is_dir()]

if missing:

print(f"no such cycle directory: {missing[0]}", file=sys.stderr)

return 2

wanted, from_sdist, locks = read_locks(cycles)

if not wanted:

print(f"no lockfiles found under {', '.join(map(str, cycles))}", file=sys.stderr)

return 2

live = {name for name, _ in wanted} | read_declared(args.repo)

newest_locked: dict[str, tuple] = {}

for name, version in wanted:

key = version_key(version)

if name not in newest_locked or key > newest_locked[name]:

newest_locked[name] = key

files = sorted(p for p in args.srcreq.iterdir() if p.is_file())

buckets: dict[str, list[Path]] = defaultdict(list)

labels: dict[Path, tuple[str, str]] = {}

for path in files:

parsed = parse_filename(path, live)

if parsed is None:

buckets["unparsed"].append(path)

continue

labels[path] = parsed

name, version = parsed

if (name, version) in wanted:

bucket = "used"

elif name not in live:

bucket = "dead"

elif name in newest_locked and version_key(version) > newest_locked[name]:

bucket = "pending"

else:

bucket = "superseded"

buckets[bucket].append(path)

on_disk = set(labels.values())

absent = sorted(set(wanted) - on_disk)

today = date.today().isoformat()

manifest = args.out / "srcreq_manifest.txt"

sdist_note = [f"# {n}=={v} built from sdist by {', '.join(sorted(w))}"

for (n, v), w in sorted(from_sdist.items())]

manifest.write_text("\n".join([

f"# packages.srcreq manifest, generated {today}",

f"# from {len(locks)} shipped lockfiles in {', '.join(d.as_posix() for d in cycles)}",

f"# {len(wanted)} (name, version) pairs; regenerate with srcreq_manifest.py",

"#",

"# An inventory and a verification list, not a one-command restore: one",

"# pip download only fetches artifacts matching the interpreter running it,",

"# so a wheelhouse serving several Pythons needs one pass per target, each",

"# against that flavor's own pylock:",

"#",

"# pip download --dest <folder> --no-deps --require-hashes -r <pylock.toml>",

"#",

*(["# Every line installs from a wheel, with these exceptions:", *sdist_note]

if sdist_note else ["# No lockfile here builds anything from an sdist."]),

"",

*(f"{name}=={version} " +

" ".join(f"--hash=sha256:{h}" for h in sorted(wanted[(name, version)])).rstrip()

for name, version in sorted(wanted)),

]) + "\n", encoding="utf-8", newline="\n")

prune = args.out / "srcreq_prune.txt"

reasons = {

"superseded": "older version of a package still in use",

"dead": "no lockfile and no requirement file names this project",

"pending": "newer than anything locked -- staged for a coming cycle, NOT listed",

"unparsed": "not a wheel or sdist name -- look before removing",

}

lines = [

f"# packages.srcreq prune candidates, generated {today}",

f"# wheelhouse holds {len(files)} files, {gigabytes(files):.1f} GB",

f"# {len(buckets['used'])} files are installed by a shipped lockfile and are NOT listed",

"#",

"# Safe to remove: every pylock is in git, so any past distribution refetches",

"# with pip download --require-hashes. Remove with --delete <bucket>.",

]

for bucket in PRUNABLE:

if entries := buckets[bucket]:

lines += ["", f"# --- {bucket}: {reasons[bucket]}",

f"# {len(entries)} files, {gigabytes(entries):.2f} GB", ""]

lines += [p.name for p in sorted(entries, key=lambda p: (labels.get(p, ("", ""))[0], p.name))]

prune.write_text("\n".join(lines) + "\n", encoding="utf-8", newline="\n")

print(f"lockfiles read : {len(locks)} from {', '.join(d.name for d in cycles)}")

print(f"pairs to keep : {len(wanted)}"

+ (f" ** {len(absent)} NOT on disk **" if absent else " all present on disk"))

for name, version in absent[:10]:

print(f" missing {name}=={version}")

print(f"wheelhouse : {len(files)} files, {gigabytes(files):.1f} GB")

for bucket in REPORTED:

entries = buckets[bucket]

print(f" {bucket:<13}: {len(entries):>5} files, {gigabytes(entries):>5.2f} GB")

reclaimable = [p for b in PRUNABLE for p in buckets[b]]

if reclaimable:

print(f"reclaimable : {gigabytes(reclaimable):.1f} GB "

f"({gigabytes(reclaimable) / max(gigabytes(files), 1e-9):.0%})")

for bucket in args.delete or []:

entries = buckets[bucket]

freed = gigabytes(entries)

for path in entries:

path.unlink()

print(f"deleted {len(entries)} files from {bucket}, {freed:.2f} GB freed")

print(f"\nwrote {manifest}\n {prune}")