rpki/scripts/cir/json_to_vaps_csv.py

51 lines
1.4 KiB
Python
Executable File

#!/usr/bin/env python3
from __future__ import annotations
import argparse
import csv
import json
from pathlib import Path
def normalize_asn(value: str | int) -> str:
text = str(value).strip().upper()
if text.startswith("AS"):
text = text[2:]
return f"AS{int(text)}"
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--input", required=True, type=Path)
parser.add_argument("--csv-out", required=True, type=Path)
args = parser.parse_args()
obj = json.loads(args.input.read_text(encoding="utf-8"))
rows: list[tuple[str, str, str]] = []
for aspa in obj.get("aspas", []):
providers = sorted(
{normalize_asn(item) for item in aspa.get("providers", [])},
key=lambda s: int(s[2:]),
)
rows.append(
(
normalize_asn(aspa["customer"]),
";".join(providers),
str(aspa.get("ta", "")).strip().lower(),
)
)
rows.sort(key=lambda row: (int(row[0][2:]), row[1], row[2]))
args.csv_out.parent.mkdir(parents=True, exist_ok=True)
with args.csv_out.open("w", encoding="utf-8", newline="") as fh:
writer = csv.writer(fh)
writer.writerow(["Customer ASN", "Providers", "Trust Anchor"])
writer.writerows(rows)
print(args.csv_out)
return 0
if __name__ == "__main__":
raise SystemExit(main())