253 lines
9.7 KiB
Python
253 lines
9.7 KiB
Python
"""Read-only native RES_DETAIL comparison; never an ingestion/acceptance gate."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections import Counter, defaultdict
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
import xml.etree.ElementTree as ET
|
|
|
|
|
|
MAX_BYTES = 64 * 1024 * 1024
|
|
MAX_DEPTH = 128
|
|
DETAIL_LIMIT = 50
|
|
SAFE_FIELDS = frozenset("""
|
|
RESV_NAME_ID RESORT ROWNUM ROOM ARRIVAL TRUNC_BEGIN DEPARTURE BEGIN_DATE END_DATE
|
|
ARRIVAL_TIME ARRIVAL_TIME1 RATE_CODE EFFECTIVE_RATE_AMOUNT CURRENCY_CODE
|
|
NO_OF_ROOMS ADULTS CHILDREN NIGHTS RESV_STATUS IS_SHARED_YN COMPANY_NAME
|
|
LIST_G_COMMENT_RESV_NAME_ID LIST_G_COMMENT_NAME_ID LIST_G_DEPT_ID
|
|
CF_DISPLAY_RECORD_01 CF_NO_OF_ROOMS
|
|
""".split())
|
|
|
|
|
|
class InputError(ValueError):
|
|
"""Only static error codes, never input values or parser messages."""
|
|
|
|
|
|
def _read(path: Path) -> bytes:
|
|
try:
|
|
if not path.is_file() or path.stat().st_size > MAX_BYTES:
|
|
raise InputError("INPUT_NOT_REGULAR_FILE_OR_TOO_LARGE")
|
|
with path.open("rb") as source:
|
|
raw = source.read(MAX_BYTES + 1)
|
|
except OSError:
|
|
raise InputError("INPUT_UNREADABLE") from None
|
|
if len(raw) > MAX_BYTES:
|
|
raise InputError("INPUT_TOO_LARGE")
|
|
return raw
|
|
|
|
|
|
def _required(node: ET.Element, field: str) -> str:
|
|
fields = node.findall(field)
|
|
if len(fields) != 1 or len(fields[0]) or not (fields[0].text or "").strip():
|
|
raise InputError("REPORT_CONTEXT_OR_ID_MISSING_OR_AMBIGUOUS")
|
|
return fields[0].text.strip()
|
|
|
|
|
|
def _text(node: ET.Element) -> str:
|
|
# Ignore formatting between elements, but retain leaf spaces and mixed content.
|
|
text = node.text or ""
|
|
if len(node) and not text.strip():
|
|
text = ""
|
|
return text
|
|
|
|
|
|
def _digest(node: ET.Element, excluded: frozenset[int] = frozenset()) -> str:
|
|
children = [
|
|
[_digest(child, excluded), child.tail if (child.tail or "").strip() else ""]
|
|
for child in node if id(child) not in excluded
|
|
]
|
|
tag = node.tag if isinstance(node.tag, str) else "#processing-instruction"
|
|
value = [tag, sorted(node.attrib.items()), _text(node), children]
|
|
return hashlib.sha256(json.dumps(value, ensure_ascii=True).encode("ascii")).hexdigest()
|
|
|
|
|
|
@dataclass
|
|
class Report:
|
|
raw_sha256: str
|
|
hotel: str
|
|
arrival_date: str
|
|
root: ET.Element
|
|
rows: list[ET.Element]
|
|
identities: list[str]
|
|
|
|
def summary(self) -> dict:
|
|
return {
|
|
"sha256": self.raw_sha256,
|
|
"hotel_code": self.hotel,
|
|
"arrival_date": self.arrival_date,
|
|
"source_records": len(self.rows),
|
|
"distinct_reservation_ids": len(set(self.identities)),
|
|
"repeated_id_occurrences": len(self.rows) - len(set(self.identities)),
|
|
}
|
|
|
|
|
|
def _parse(raw: bytes) -> Report:
|
|
try:
|
|
text = raw.decode("utf-8-sig")
|
|
except UnicodeError:
|
|
raise InputError("UTF8_REQUIRED") from None
|
|
if "\x00" in text:
|
|
raise InputError("UTF8_REQUIRED")
|
|
if re.search(r"<!\s*(?:DOCTYPE|ENTITY)\b", text, re.IGNORECASE):
|
|
raise InputError("DTD_OR_ENTITY_FORBIDDEN")
|
|
declaration = re.match(r"<\?xml\s+[^?]*\?>", text)
|
|
if declaration:
|
|
encoding = re.search(r"encoding\s*=\s*['\"]([^'\"]+)['\"]", declaration[0])
|
|
if encoding and encoding[1].lower() not in {"utf-8", "utf8"}:
|
|
raise InputError("UTF8_REQUIRED")
|
|
try:
|
|
root = ET.fromstring(text, parser=ET.XMLParser(target=ET.TreeBuilder(insert_pis=True)))
|
|
except (ET.ParseError, ValueError):
|
|
raise InputError("INVALID_XML") from None
|
|
stack = [(root, 1)]
|
|
while stack:
|
|
node, depth = stack.pop()
|
|
if depth > MAX_DEPTH:
|
|
raise InputError("XML_TOO_DEEP")
|
|
stack.extend((child, depth + 1) for child in node)
|
|
if root.tag != "RES_DETAIL":
|
|
raise InputError("RES_DETAIL_REQUIRED")
|
|
groups = root.findall("./LIST_G_GROUP_BY1/G_GROUP_BY1")
|
|
if not groups:
|
|
raise InputError("ARRIVAL_GROUP_REQUIRED")
|
|
dates, rows = set(), []
|
|
for group in groups:
|
|
sort_date = _required(group, "GROUPBY1_SORT_COL")
|
|
display_date = _required(group, "GROUPBY1_COL")
|
|
try:
|
|
if not re.fullmatch(r"[0-9]{8}", sort_date):
|
|
raise ValueError
|
|
day = datetime.strptime(sort_date, "%Y%m%d").date()
|
|
if day.strftime("%d-%m-%y") != display_date:
|
|
raise ValueError
|
|
except ValueError:
|
|
raise InputError("INVALID_OR_CONFLICTING_GROUP_DATE") from None
|
|
dates.add(day.isoformat())
|
|
lists = group.findall("LIST_G_RESERVATION")
|
|
if len(lists) != 1:
|
|
raise InputError("RESERVATION_LIST_REQUIRED")
|
|
rows.extend(lists[0].findall("G_RESERVATION"))
|
|
if len(dates) != 1:
|
|
raise InputError("MIXED_GROUP_DATES")
|
|
if not rows or len(rows) != sum(1 for _ in root.iter("G_RESERVATION")):
|
|
raise InputError("EMPTY_OR_UNEXPECTED_RESERVATION_LAYOUT")
|
|
hotels = {_required(row, "RESORT") for row in rows}
|
|
if len(hotels) != 1:
|
|
raise InputError("MIXED_HOTELS")
|
|
hotel = hotels.pop()
|
|
if not re.fullmatch(r"[A-Za-z0-9_-]{1,32}", hotel):
|
|
raise InputError("INVALID_HOTEL_CODE")
|
|
return Report(
|
|
hashlib.sha256(raw).hexdigest(), hotel, dates.pop(), root, rows,
|
|
[_required(row, "RESV_NAME_ID") for row in rows],
|
|
)
|
|
|
|
|
|
def _occurrences(report: Report) -> dict:
|
|
seen = Counter()
|
|
occurrences = {}
|
|
for position, (identity, row) in enumerate(zip(report.identities, report.rows), 1):
|
|
seen[identity] += 1
|
|
occurrences[(identity, seen[identity])] = (position, row)
|
|
return occurrences
|
|
|
|
|
|
def _changed_fields(before: ET.Element, after: ET.Element) -> list[str]:
|
|
def fields(row):
|
|
result = defaultdict(list)
|
|
for field in row:
|
|
result[field.tag].append(_digest(field))
|
|
return result
|
|
|
|
left, right = fields(before), fields(after)
|
|
changed = {
|
|
name if name in SAFE_FIELDS else "OTHER_FIELD"
|
|
for name in left.keys() | right.keys() if left[name] != right[name]
|
|
}
|
|
if [child.tag for child in before] != [child.tag for child in after]:
|
|
changed.add("FIELD_ORDER_OR_STRUCTURE")
|
|
if before.attrib != after.attrib or _text(before) != _text(after):
|
|
changed.add("RECORD_ATTRIBUTES_OR_TEXT")
|
|
return sorted(changed or {"OTHER_CONTENT"})
|
|
|
|
|
|
def compare_files(baseline: Path, candidate: Path, expected_sha256: str) -> tuple[dict, int]:
|
|
"""Return a privacy-minimized result and exit code: match 0/diff 1/context 2/input 3."""
|
|
side = "baseline"
|
|
try:
|
|
if not re.fullmatch(r"[a-fA-F0-9]{64}", expected_sha256):
|
|
raise InputError("INVALID_BASELINE_SHA256")
|
|
raw = _read(baseline)
|
|
if hashlib.sha256(raw).hexdigest() != expected_sha256.lower():
|
|
raise InputError("BASELINE_SHA256_MISMATCH")
|
|
left = _parse(raw)
|
|
side = "candidate"
|
|
right = _parse(_read(candidate))
|
|
except InputError as error:
|
|
return {"status": "invalid_input", "input": side, "error": str(error)}, 3
|
|
|
|
result = {"baseline": left.summary(), "candidate": right.summary()}
|
|
if (left.hotel, left.arrival_date) != (right.hotel, right.arrival_date):
|
|
return {**result, "status": "not_comparable", "reason": "HOTEL_OR_GROUP_DATE_DIFFERS"}, 2
|
|
|
|
left_rows, right_rows = _occurrences(left), _occurrences(right)
|
|
missing = [entry[0] for key, entry in left_rows.items() if key not in right_rows]
|
|
additional = [entry[0] for key, entry in right_rows.items() if key not in left_rows]
|
|
changed = []
|
|
for key, (position, row) in left_rows.items():
|
|
if key in right_rows:
|
|
other_position, other = right_rows[key]
|
|
if _digest(row) != _digest(other):
|
|
changed.append({
|
|
"baseline_position": position, "candidate_position": other_position,
|
|
"fields": _changed_fields(row, other),
|
|
})
|
|
same = _digest(left.root) == _digest(right.root)
|
|
metadata_same = _digest(left.root, frozenset(map(id, left.rows))) == _digest(
|
|
right.root, frozenset(map(id, right.rows))
|
|
)
|
|
result.update({
|
|
"status": "match" if same else "different",
|
|
"byte_identical": left.raw_sha256 == right.raw_sha256,
|
|
"xml_content_identical": same,
|
|
"report_metadata_identical": metadata_same,
|
|
"identity_sequence_identical": left.identities == right.identities,
|
|
"record_content_sequence_identical": list(map(_digest, left.rows)) == list(map(_digest, right.rows)),
|
|
"missing_occurrences": len(missing), "additional_occurrences": len(additional),
|
|
"changed_occurrences": len(changed),
|
|
"missing_baseline_positions": missing[:DETAIL_LIMIT],
|
|
"additional_candidate_positions": additional[:DETAIL_LIMIT],
|
|
"changed_records": changed[:DETAIL_LIMIT],
|
|
"details_truncated": any(len(items) > DETAIL_LIMIT for items in (missing, additional, changed)),
|
|
"scope": "XML_CONTENT_ONLY_NOT_BUSINESS_ACCEPTANCE",
|
|
})
|
|
return result, 0 if same else 1
|
|
|
|
|
|
class _Parser(argparse.ArgumentParser):
|
|
def error(self, message):
|
|
print(json.dumps({"status": "invalid_input", "error": "INVALID_ARGUMENTS"}))
|
|
raise SystemExit(3)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = _Parser(description=__doc__)
|
|
parser.add_argument("--baseline", type=Path, required=True)
|
|
parser.add_argument("--candidate", type=Path, required=True)
|
|
parser.add_argument("--baseline-sha256", required=True)
|
|
args = parser.parse_args(argv)
|
|
result, code = compare_files(args.baseline, args.candidate, args.baseline_sha256)
|
|
print(json.dumps(result, ensure_ascii=True, indent=2))
|
|
return code
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|