70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
"""Read-only manifest collector for a 1C SQL configuration storage.
|
|
|
|
The REST endpoint caps a single storage.files.list response. GUID-prefix
|
|
partitioning keeps every request below that cap and makes the manifest
|
|
reproducible without SQL credentials in source control.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import urllib.request
|
|
from collections import Counter
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
PREFIXES = [*"0123456789abcdef", "root"]
|
|
|
|
|
|
def rpc(url: str, method: str, payload: dict) -> dict:
|
|
request = urllib.request.Request(
|
|
url.rstrip("/") + "/rpc",
|
|
data=json.dumps({"method": method, "payload": payload}).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
with urllib.request.urlopen(request, timeout=180) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--url", required=True)
|
|
parser.add_argument("--base-id", required=True)
|
|
parser.add_argument("--table", default="Config")
|
|
parser.add_argument("--output", required=True, type=Path)
|
|
args = parser.parse_args()
|
|
|
|
files: list[dict] = []
|
|
partitions: list[dict] = []
|
|
for prefix in PREFIXES:
|
|
result = rpc(args.url, "storage.files.list", {
|
|
"base_id": args.base_id, "table": args.table, "prefix": prefix,
|
|
"limit": 5000, "diagnostic": True,
|
|
})
|
|
rows = result.get("files") if result.get("status") == "ok" else None
|
|
if not isinstance(rows, list):
|
|
raise RuntimeError(f"{prefix}: {result}")
|
|
if len(rows) >= 5000:
|
|
raise RuntimeError(f"{prefix}: response reached the cap; split this prefix further")
|
|
files.extend(rows)
|
|
partitions.append({"prefix": prefix, "files": len(rows)})
|
|
|
|
suffixes = Counter(Path(str(row.get("FileName") or "")).suffix or "<descriptor>" for row in files)
|
|
manifest = {
|
|
"schema": "onec_config_storage_manifest.v1",
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"source": {"base_id": args.base_id, "table": args.table},
|
|
"partitions": partitions,
|
|
"counts": {"files": len(files), "compressed_bytes": sum(int(row.get("Bytes") or 0) for row in files), "suffixes": dict(sorted(suffixes.items()))},
|
|
"files": files,
|
|
}
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|