normalize.py
scripts/normalize.py
#!/usr/bin/env python3
"""Turn an OPERA Cloud JSON export into standard rows.
Usage: python3 scripts/normalize.py <opera.json> <out_dir>
Writes to <out_dir>:
days.csv date, occupancy, rooms_available, source_file, occupancy_row, availability_row
groups.csv block, date, allocated, picked_up
maintenance.csv room, status, created
meta.json business_date, property_id, source_file, inventory, blocks_listed, warnings
The script reads only the sections it needs. It never reads reservations, financial
details or memberships, and a final check makes sure no guest ID reached the output.
Unknown sections or values stop the run: a person must approve a new mapping first.
Exit codes: 0 done, 1 error, 3 unknown format (stop and ask).
"""
import csv
import json
import os
import re
import sys
KNOWN_SECTIONS = {
"metadata", "reservations", "ratePlans", "groups", "roomTypes", "availability",
"roomAvailability", "occupancy", "prices", "roomStatus", "maintenance",
"unavailableRooms", "operationalRooms", "financialDetails", "memberships",
"groupGrid", "rentalCodes", "functionSpaces", "functionSpaceAvailability",
"events", "bookingPace",
}
NEEDED = ["metadata", "occupancy", "availability", "groups", "groupGrid", "maintenance"]
KNOWN_ALLOCATIONS = {"Initial", "Actual", "Pickup", "Available", "Rates", "Released"}
GUEST_ID = re.compile(r"GUEST-\d+")
class UnknownFormat(Exception):
"""The export does not match what this script knows. A person decides what to do."""
def write_csv(path: str, header: list[str], rows: list[list]) -> None:
with open(path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(header)
writer.writerows(rows)
def normalize(data: dict, source_file: str, out_dir: str) -> list[str]:
unknown = sorted(set(data) - KNOWN_SECTIONS)
if unknown:
raise UnknownFormat(f"unknown sections: {', '.join(unknown)}")
missing = [s for s in NEEDED if s not in data]
if missing:
raise UnknownFormat(f"missing sections: {', '.join(missing)}")
bad = sorted({r["allocation"] for r in data["groupGrid"]} - KNOWN_ALLOCATIONS)
if bad:
raise UnknownFormat(f"unknown group allocation values: {', '.join(bad)}")
meta = data["metadata"]
warnings = []
hotels = {r["hotel"] for r in data["occupancy"]}
if len(hotels) != 1:
raise UnknownFormat(f"expected one hotel in occupancy, found {len(hotels)}")
property_id = hotels.pop()
inventory = meta.get("occupancyScaled", {}).get("inventory")
# days.csv: occupancy per date, rooms available from the hotel-level availability rows.
available = {r["startDate"]: (i, r["availableRooms"]) for i, r in enumerate(data["availability"])}
days = []
for i, row in enumerate(data["occupancy"]):
date, occ = row["date"], row["occupancy"]
if date in available:
index, rooms = available[date]
avail_ref = f"availability[{index}]"
elif inventory:
rooms = round(inventory * (1 - occ))
avail_ref = "derived from inventory"
else:
rooms, avail_ref = "", "missing"
warnings.append(f"rooms_available is missing for {date}: no availability row and no inventory.")
days.append([date, occ, rooms, source_file, f"occupancy[{i}]", avail_ref])
days.sort(key=lambda r: r[0])
# groups.csv: one row per block and date. Allocated = Actual, picked up = Pickup.
grid = {}
for row in data["groupGrid"]:
entry = grid.setdefault((row["block"], row["date"]), {"allocated": "", "picked_up": ""})
if row["allocation"] == "Actual":
entry["allocated"] = row["rooms"]
elif row["allocation"] == "Pickup":
entry["picked_up"] = row["rooms"]
groups = [[b, d, v["allocated"], v["picked_up"]] for (b, d), v in sorted(grid.items())]
# maintenance.csv: room, status, created. No other field.
maintenance = [[r["room"], r["status"], r["createdDate"]] for r in data["maintenance"]]
os.makedirs(out_dir, exist_ok=True)
write_csv(os.path.join(out_dir, "days.csv"),
["date", "occupancy", "rooms_available", "source_file", "occupancy_row", "availability_row"], days)
write_csv(os.path.join(out_dir, "groups.csv"), ["block", "date", "allocated", "picked_up"], groups)
write_csv(os.path.join(out_dir, "maintenance.csv"), ["room", "status", "created"], maintenance)
with open(os.path.join(out_dir, "meta.json"), "w", encoding="utf-8") as f:
json.dump({
"business_date": meta["businessDate"],
"property_id": property_id,
"source_file": source_file,
"inventory": inventory,
"blocks_listed": len(data["groups"]),
"warnings": warnings,
}, f, indent=2)
f.write("\n")
# Last check: no guest ID may be in any file we wrote.
for name in ("days.csv", "groups.csv", "maintenance.csv", "meta.json"):
with open(os.path.join(out_dir, name), encoding="utf-8") as f:
if GUEST_ID.search(f.read()):
raise RuntimeError(f"a guest ID reached {name}")
return warnings
def main(argv: list[str]) -> int:
if len(argv) != 3:
print("Usage: python3 scripts/normalize.py <opera.json> <out_dir>", file=sys.stderr)
return 1
src, out_dir = argv[1], argv[2]
try:
with open(src, encoding="utf-8") as f:
data = json.load(f)
warnings = normalize(data, os.path.basename(src), out_dir)
except (OSError, json.JSONDecodeError) as err:
print(f"Could not read or write files: {err}", file=sys.stderr)
return 1
except UnknownFormat as err:
print(f"Unknown format, stopped: {err}. Propose a mapping and wait for approval.", file=sys.stderr)
return 3
except (KeyError, TypeError, RuntimeError) as err:
print(f"Unexpected export content: {err!r}", file=sys.stderr)
return 1
print(f"Wrote days.csv, groups.csv, maintenance.csv and meta.json to {out_dir}")
for w in warnings:
print(f"Warning: {w}")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))