GeezerVison is showing github.com — winpython/srcreq_manifest.py at master · winpython/winpython

Expand file tree

srcreq_manifest.py

Copy path

More file actions

Latest commit

History

286 lines (248 loc) · 12.4 KB

srcreq_manifest.py

Copy path

File metadata and controls

286 lines (248 loc) · 12.4 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

"""Inventory the local wheelhouse, and say which files nothing needs any more.

python srcreq_manifest.py # report + write both lists

python srcreq_manifest.py --cycle winpython/portable/cycle_2026_04

python srcreq_manifest.py --delete dead # actually remove a bucket

The wheelhouse is the index a local build resolves against

(`pip install --no-index --find-links=...`), so what is in it decides what gets

built. Left alone it only grows: versions superseded cycles ago, and packages

no requirement file mentions any longer.

What a build still needs is not a guess -- the shipped lockfiles say it exactly.

A (name, version) pair is `used` when a lockfile installs it, `superseded` when

the project is still in use but that version is not, and `dead` when no

lockfile and no requirement or constraint file names the project at all.

Deleting is safe because the lockfiles are in git: any past distribution can be

refetched with `pip download --require-hashes -r <its pylock>`. Nothing is

deleted without --delete, which names the bucket explicitly.

Hashes come from the lockfiles rather than from the files, so this reads no

package bytes. The build's web-vs-local double lock is what makes that sound:

it already proves the local copies match what PyPI serves.

"""

from __future__ import annotations

import argparse

import re

import sys

import tomllib

from collections import defaultdict

from datetime import date

from pathlib import Path

WHEEL = ".whl"

SDIST = (".tar.gz", ".zip", ".tar.bz2")

REQUIREMENT_FILES = ("constraints.txt", "requirements_slim.txt", "requirements_slimf.txt",

"dot_requirements.txt", "requirements_whl.txt", "mandatory_requirements.txt")

# A cycle accumulates lockfiles as it converges: several Python patch levels

# (3_13_13_1, 3_13_14_0, 3_13_15_0) and several release levels (b0, b1, ... then

# none once it ships). Only the newest per flavor counts, so rank rather than

# exclude -- a cycle still in progress has nothing but levelled files.

LOCK_NAME = re.compile(

r"^pylock\.64-(?P<ver>\d+(?:_\d+)+?)(?P<flavor>[a-z]+?)(?:b(?P<level>\d+))?"

r"(?P<wheels>_wheels)?\.toml$"

)

# What a prune may take. `pending` is deliberately absent: a version newer than

# anything locked is a candidate staged for the next cycle, not leftovers.

PRUNABLE = ("superseded", "dead", "unparsed")

REPORTED = ("used", "pending", *PRUNABLE)

def version_key(text: str):

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

try:

from packaging.version import InvalidVersion, Version

try:

return (1, Version(text))

except InvalidVersion:

pass

except ImportError:

pass

return (0, tuple(int(p) if p.isdigit() else p for p in re.split(r"[._-]", text)))

def normalize(name: str) -> str:

"""PEP 503 normalized project name."""

return re.sub(r"[-_.]+", "-", name).lower()

def parse_filename(path: Path, known: set[str]) -> tuple[str, str] | None:

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

Wheel names are unambiguous. Sdist names are not -- a project name may

contain the same '-' that separates it from the version -- so the longest

known project name prefixing the file wins, and only when nothing matches

do we fall back to splitting at the first '-' before a digit.

"""

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)

return (normalize(match.group(1)), match.group(2)) if match else None

def newest_locks(cycle_dirs: list[Path]) -> list[Path]:

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

Ranked by version then release level, with an absent level ranking highest

because that is the one a cycle ships.

"""

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)

return sorted(path for _, path in best.values())

def read_locks(cycle_dirs: list[Path]) -> tuple[dict, dict, list[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)

return wanted, from_sdist, locks

def read_declared(repo: Path) -> set[str]:

"""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))

return names

def gigabytes(paths) -> float:

return sum(p.stat().st_size for p in paths) / 2 ** 30

def main(argv: list[str]) -> int:

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}")

return 0

if __name__ == "__main__":

raise SystemExit(main(sys.argv[1:]))