Skills › Scientific Agent Skills › geopandas › vector_inventory.py
vector_inventory.py
140 lines · skills/geopandas/scripts/vector_inventory.py at 5f2bc03
1#!/usr/bin/env python3
2"""Create a redacted, metadata-only inventory of one local vector dataset."""
4from __future__ import annotations
6import argparse
7import hashlib
8import sys
10from _common import (
11 ABSOLUTE_MAX_FEATURES,
12 DEFAULT_MAX_FEATURES,
13 DEFAULT_MAX_INPUT_BYTES,
14 PINNED_STACK,
15 bounded_limit,
16 checked_input_file,
17 emit_json,
18 fail_json,
19 inspect_local_vector,
20 native_versions,
21 package_versions,
22 positive_int,
23 sha256_file,
24)
26TOOL = "vector_inventory"
29def build_parser() -> argparse.ArgumentParser:
30 parser = argparse.ArgumentParser(
31 description=(
32 "Inventory one allowlisted local vector file without loading feature "
33 "coordinates or emitting paths, layer names, fields, IDs, or bounds."
34 ),
35 epilog=(
36 "URLs, GDAL VSI paths, archives, compression, symlinks, traversal, "
37 "and non-allowlisted extensions are rejected. Native GDAL/GEOS/PROJ "
38 "parsers remain a trust boundary; inspect only vetted local files."
39 ),
40 )
41 parser.add_argument("input", help="Local vector or GeoParquet file")
42 parser.add_argument(
43 "--root",
44 default=".",
45 help="Existing local root that must contain the input (default: .)",
46 )
47 parser.add_argument(
48 "--layer",
49 help="Exact local layer name; required for multi-layer GDAL datasets",
50 )
51 parser.add_argument(
52 "--max-input-bytes",
53 type=positive_int,
54 default=DEFAULT_MAX_INPUT_BYTES,
55 help=f"Maximum primary-file bytes (default: {DEFAULT_MAX_INPUT_BYTES})",
56 )
57 parser.add_argument(
58 "--max-features",
59 type=positive_int,
60 default=DEFAULT_MAX_FEATURES,
61 help=(
62 "Downstream feature-load threshold to compare with declared count "
63 f"(default: {DEFAULT_MAX_FEATURES}; no features are loaded)"
64 ),
65 )
66 return parser
69def inventory(args: argparse.Namespace) -> dict:
70 max_features = bounded_limit(
71 args.max_features,
72 name="max_features",
73 maximum=ABSOLUTE_MAX_FEATURES,
74 )
75 path = checked_input_file(
76 args.input,
77 root=args.root,
78 max_bytes=args.max_input_bytes,
79 )
80 technical = inspect_local_vector(path, layer=args.layer)
81 installed = package_versions()
82 declared = technical.get("declared_features")
83 warnings = [
84 "No feature values, coordinates, bounds, paths, layer names, or field names were emitted.",
85 "GDAL/GEOS/PROJ and binary wheels are native-code trust boundaries.",
86 "Driver metadata is not proof that every feature or field will parse safely.",
87 ]
88 if path.suffix.casefold() == ".shp":
89 warnings.append(
90 "The hash covers only the .shp primary file, not required sidecars."
91 )
92 if declared is None:
93 warnings.append(
94 "The driver did not provide a cheap feature count; downstream reads must enforce limit+1."
95 )
96 return {
97 "ok": True,
98 "tool": TOOL,
99 "source": {
100 "basename_sha256": hashlib.sha256(path.name.encode("utf-8")).hexdigest(),
101 "suffix": path.suffix.casefold(),
102 "primary_file_bytes": path.stat().st_size,
103 "primary_file_sha256": sha256_file(path),
104 "path_emitted": False,
105 },
106 "technical_inventory": technical,
107 "resource_limits": {
108 "max_input_bytes": args.max_input_bytes,
109 "max_features_for_downstream_load": max_features,
110 "declared_within_feature_limit": (
111 None if declared is None else declared <= max_features
112 ),
113 "feature_data_loaded": False,
114 },
115 "stack": {
116 "packages": installed,
117 "native": native_versions(),
118 "matches_pinned_snapshot": {
119 name: installed.get(name) == expected
120 for name, expected in PINNED_STACK.items()
121 },
122 },
123 "network_accessed": False,
124 "coordinates_emitted": False,
125 "identifiers_emitted": False,
126 "warnings": warnings,
127 }
130def main(argv: list[str] | None = None) -> int:
131 args = build_parser().parse_args(argv)
132 try:
133 emit_json(inventory(args))
134 return 0
135 except Exception as exc: # noqa: BLE001 - errors are redacted at CLI boundary
136 return fail_json(TOOL, exc)
139if __name__ == "__main__":
140 sys.exit(main())