187 lines
10 KiB
Python
187 lines
10 KiB
Python
"""Bounded, archived reads for the ARR structured-data source (no Oracle writes)."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from collections import OrderedDict
|
|
from decimal import Decimal
|
|
import json
|
|
import re
|
|
import time
|
|
import urllib.error
|
|
from urllib.parse import urlencode
|
|
|
|
from . import collect_arr_source as base
|
|
from . import profile_summary, rate_info
|
|
|
|
|
|
FETCH = tuple(value for value in base.FETCH if value != "Traces")
|
|
OPERATIONS = (base.SEARCH, base.DETAIL, rate_info.POST, profile_summary.OPERATION,
|
|
"getProfile", "getBlock", "getRoomCalendar", "getPackage")
|
|
require = base.require
|
|
|
|
|
|
def json_bytes(value) -> bytes:
|
|
"""Represent Decimal as exact text in our output, retain original HTTP bytes separately."""
|
|
def decimal_text(item):
|
|
if isinstance(item, Decimal) and item.is_finite():
|
|
return str(item)
|
|
raise TypeError("unsupported_json_value")
|
|
return (json.dumps(value, ensure_ascii=False, allow_nan=False, sort_keys=True,
|
|
separators=(",", ":"), default=decimal_text) + "\n").encode("utf-8")
|
|
|
|
|
|
def document(raw: bytes) -> dict:
|
|
base.strict_json(raw) # duplicate keys / non-JSON constants / depth failures
|
|
return json.loads(raw, parse_float=Decimal)
|
|
|
|
|
|
def internal_id(value) -> str:
|
|
require(type(value) is str and re.fullmatch(r"[A-Za-z0-9_-]{1,128}", value), "invalid_related_id")
|
|
return value
|
|
|
|
|
|
def typed_id(values, kind: str) -> str:
|
|
require(type(values) is list and all(type(v) is dict for v in values), "invalid_related_ids")
|
|
matches = [v.get("id") for v in values if v.get("type") == kind]
|
|
require(len(matches) == 1, "ambiguous_related_id")
|
|
return internal_id(matches[0])
|
|
|
|
|
|
@dataclass(frozen=True, repr=False)
|
|
class ReadResult:
|
|
data: dict
|
|
raw: bytes
|
|
reference: str
|
|
request: dict
|
|
|
|
|
|
class DataReader:
|
|
"""One date/hotel context, fixed operations, shared related-object cache.
|
|
|
|
The injected transport is the only I/O seam; production uses base.HTTPTransport.
|
|
This class never reads a credential file or permits caller-controlled URLs.
|
|
"""
|
|
|
|
def __init__(self, archive, options, transport, *, key="", sleep=time.sleep, max_requests=25000,
|
|
source_kind="test_transport"):
|
|
options.validate()
|
|
require(type(max_requests) is int and 1 <= max_requests <= 100000, "invalid_request_limit")
|
|
require(source_kind in {"ohip_platform", "test_transport"}, "invalid_source_kind")
|
|
self.archive, self.options, self.hotel_id = archive, options, options.hotel_id
|
|
self.transport, self._key, self.sleep = transport, key, sleep
|
|
self.max_requests, self.request_count = max_requests, 0
|
|
self.source_kind = source_kind
|
|
self.cache, self.used = OrderedDict(), []
|
|
self.cache_bytes = 0
|
|
|
|
def read(self, operation, *, body=None, identity=None):
|
|
# Duck-typed seam for the existing strict reservation pagination code.
|
|
return self.query(operation, body=body, identity=identity).data
|
|
|
|
def _request(self, operation, identity, body, params):
|
|
require(operation in OPERATIONS, "operation_not_allowed")
|
|
day = self.options.arrival_date
|
|
if operation == base.SEARCH:
|
|
require(identity is None and params is None and type(body) is dict, "invalid_search_request")
|
|
require(set(body) == {"arrivalStartDate", "arrivalEndDate", "limit", "offset", "orderBy", "sortOrder"}
|
|
and body["arrivalStartDate"] == body["arrivalEndDate"] == day
|
|
and body["limit"] == self.options.page_size and type(body["limit"]) is int
|
|
and type(body["offset"]) is int and 0 <= body["offset"] < self.options.max_pages * self.options.page_size
|
|
and body["orderBy"] == ["ConfirmationNo"] and body["sortOrder"] == ["Asc"],
|
|
"invalid_search_request")
|
|
return "POST", "/api/v1/reservations/searches", body
|
|
require(body is None, "unexpected_query_body")
|
|
if operation == "getRoomCalendar":
|
|
require(identity is None and type(params) is dict, "invalid_calendar_request")
|
|
allowed = {"startDate", "endDate", "includeRoomMoveHistory", "showRoomMoveSegments", "pageIndex", "recordsPerPage"}
|
|
require(set(params) <= allowed and params.get("startDate") == params.get("endDate") == day
|
|
and params.get("includeRoomMoveHistory") == "true"
|
|
and params.get("showRoomMoveSegments") == "true", "invalid_calendar_request")
|
|
for key in ("pageIndex", "recordsPerPage"):
|
|
if key in params:
|
|
require(type(params[key]) is int and 0 <= params[key] <= 10000
|
|
and (key != "recordsPerPage" or params[key] > 0), "invalid_calendar_request")
|
|
return "GET", "/api/v1/reservations/room-calendar?" + urlencode(params), None
|
|
identity = internal_id(identity)
|
|
if operation == "getPackage":
|
|
require(type(params) is dict and set(params) == {"productCode", "reservationTimeSpanStartDate", "reservationTimeSpanEndDate"},
|
|
"invalid_package_request")
|
|
require(type(params["productCode"]) is str and 0 < len(params["productCode"]) <= 20
|
|
and params["reservationTimeSpanStartDate"] == day, "invalid_package_request")
|
|
from datetime import date
|
|
try:
|
|
end = date.fromisoformat(params["reservationTimeSpanEndDate"])
|
|
require(end.isoformat() == params["reservationTimeSpanEndDate"] and end >= date.fromisoformat(day),
|
|
"invalid_package_request")
|
|
except (ValueError, TypeError):
|
|
raise base.CollectionError("invalid_package_request") from None
|
|
query = list(params.items()) + [("fetchInstructions", "Primary"), ("fetchInstructions", "Schedule")]
|
|
return "GET", f"/api/v1/reservations/{identity}/packages?" + urlencode(query), None
|
|
require(params is None, "unexpected_query_parameters")
|
|
if operation == base.DETAIL:
|
|
return "GET", f"/api/v1/reservations/{identity}?" + urlencode([("fetchInstructions", f) for f in FETCH]), None
|
|
if operation == rate_info.POST:
|
|
method, path, _payload, value = rate_info.request(identity, day)
|
|
return method, path, value
|
|
if operation == profile_summary.OPERATION:
|
|
method, path, payload = profile_summary.request(identity)
|
|
return method, path, base.strict_json(payload)
|
|
if operation == "getProfile":
|
|
return "GET", f"/api/v1/profiles/{identity}", None
|
|
return "GET", f"/api/v1/blocks/{identity}?fetchInstructions=Block", None
|
|
|
|
def query(self, operation, *, identity=None, body=None, params=None) -> ReadResult:
|
|
method, path, body = self._request(operation, identity, body, params)
|
|
cache_key = (operation, path, json_bytes(body))
|
|
# Every reservation search is real, including the end-of-batch recheck.
|
|
if cache_key in self.cache:
|
|
result = self.cache[cache_key]
|
|
self.cache.move_to_end(cache_key)
|
|
self.used.append(result.reference)
|
|
return result
|
|
for attempt in range(1, 4):
|
|
require(self.request_count < self.max_requests, "request_limit_exceeded")
|
|
self.request_count += 1
|
|
label = f"data-request-{self.request_count:06d}"
|
|
request = {"operation_id": operation, "method": method, "path": path, "body": body,
|
|
"attempt": attempt, "started_at": base.utc_now()}
|
|
self.archive.write(label + ".json", request)
|
|
try:
|
|
status, headers, raw = self.transport(method, path, None if body is None else json_bytes(body))
|
|
except (urllib.error.URLError, TimeoutError, ConnectionError):
|
|
self.archive.write(label + ".meta.json", {"error": "transport_failure"})
|
|
if attempt == 3:
|
|
raise base.CollectionError("transport_retry_exhausted") from None
|
|
self.sleep(float(2 ** (attempt - 1)))
|
|
continue
|
|
require(type(raw) is bytes and len(raw) <= base.MAX_RESPONSE_BYTES, "response_too_large")
|
|
require(not self._key or self._key.encode() not in raw, "secret_in_response")
|
|
headers = {k.lower(): v for k, v in headers.items()}
|
|
self.archive.write(label + ".response.bin", raw)
|
|
self.archive.write(label + ".meta.json", {"http_status": status,
|
|
"edge_request_id": headers.get("x-request-id"), "ended_at": base.utc_now()})
|
|
if status in base.RETRY_STATUSES and attempt < 3:
|
|
self.sleep(base.retry_delay(headers, attempt))
|
|
continue
|
|
require(status == 200, "http_permission_denied" if status in (401, 403) else "http_failure")
|
|
envelope = document(raw)
|
|
require(envelope.get("operation_id") == operation, "operation_mismatch")
|
|
require(envelope.get("hotel_id") == self.hotel_id, "hotel_mismatch")
|
|
require(type(envelope.get("oracle_request_id")) is str and bool(envelope["oracle_request_id"].strip()),
|
|
"missing_oracle_request_id")
|
|
require(type(envelope.get("data")) is dict, "invalid_data_envelope")
|
|
base.check_warnings(envelope)
|
|
result = ReadResult(envelope["data"], raw, label + ".response.bin", request)
|
|
self.used.append(result.reference)
|
|
# Reservation details/day prices are used once, and calendars once
|
|
# per page. Bound reusable related-object caching in both bytes and
|
|
# entries so a large day cannot retain every raw response in RAM.
|
|
if operation in (profile_summary.OPERATION, "getProfile", "getBlock", "getPackage"):
|
|
while self.cache and (len(self.cache) >= 128 or self.cache_bytes + len(raw) > 32 * 1024 * 1024):
|
|
_key, evicted = self.cache.popitem(last=False)
|
|
self.cache_bytes -= len(evicted.raw)
|
|
self.cache[cache_key] = result
|
|
self.cache_bytes += len(raw)
|
|
return result
|
|
raise base.CollectionError("retry_exhausted")
|