221 lines
11 KiB
Python
221 lines
11 KiB
Python
"""Explicit native-XML replay for an isolated local portal, never an OHIP adapter."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, replace
|
|
from datetime import date
|
|
import hashlib
|
|
from http.cookies import SimpleCookie
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
from urllib.parse import urlsplit
|
|
|
|
from arr_web.arr_download_handoff import outcome_from_handoff
|
|
from arr_web.arr_downloads import validate_request_id
|
|
from arr_web.contracts import MAX_UPLOAD_BYTES, PortalError, Response, failure, validate_xml_payload
|
|
from integrations.ohip import processing_handoff as handoff
|
|
from integrations.ohip.audit_arr_capture import protected_read
|
|
from integrations.ohip.capture_job import atomic_json, fingerprint, private_directory
|
|
from integrations.ohip.compare_report_xml import _parse as parse_native_report
|
|
|
|
|
|
VERSION = "local-native-xml-replay/v1"
|
|
COOKIE = "arr_xml_replay_session"
|
|
PROJECT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def document(path: Path) -> dict:
|
|
from integrations.ohip.collect_arr_source import strict_json
|
|
return strict_json(protected_read(path, 65536))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NativeXMLSnapshot:
|
|
root: Path
|
|
manifest: dict
|
|
|
|
@classmethod
|
|
def create(cls, root: Path, source: Path, expected_sha256: str, report_date: str):
|
|
if root.resolve().is_relative_to(PROJECT):
|
|
raise ValueError("replay_data_must_be_outside_repository")
|
|
# Source is read exactly once, bounded and without following a symlink.
|
|
fd = os.open(source, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
|
with os.fdopen(fd, "rb") as handle:
|
|
if not stat.S_ISREG(os.fstat(handle.fileno()).st_mode):
|
|
raise ValueError("source_must_be_regular_file")
|
|
raw = handle.read(MAX_UPLOAD_BYTES + 1)
|
|
validate_xml_payload(raw)
|
|
if hashlib.sha256(raw).hexdigest() != expected_sha256:
|
|
raise ValueError("source_hash_mismatch")
|
|
report = parse_native_report(raw)
|
|
if report.arrival_date != report_date:
|
|
raise ValueError("source_date_mismatch")
|
|
private_directory(root)
|
|
manifest = {"version": VERSION, "source_kind": "native_xml_replay",
|
|
"report_date": report_date, "hotel_id": report.hotel,
|
|
"source_sha256": expected_sha256, "source_bytes": len(raw),
|
|
"source_records": len(report.rows), "oracle_calls": 0,
|
|
"source_mapping_verified": False, "report_equivalence_verified": False}
|
|
fd = os.open(root / "source.xml", os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
|
with os.fdopen(fd, "wb") as handle:
|
|
handle.write(raw)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
atomic_json(root / "source.json", manifest, replace=False)
|
|
return cls(root, manifest)
|
|
|
|
@classmethod
|
|
def load(cls, root: Path):
|
|
private_directory(root)
|
|
instance = cls(root, document(root / "source.json"))
|
|
instance.payload()
|
|
return instance
|
|
|
|
@property
|
|
def day(self):
|
|
return date.fromisoformat(self.manifest["report_date"])
|
|
|
|
def payload(self) -> bytes:
|
|
if (fingerprint(document(self.root / "source.json")) != fingerprint(self.manifest)
|
|
or self.manifest.get("version") != VERSION):
|
|
raise ValueError("replay_source_identity_changed")
|
|
raw = protected_read(self.root / "source.xml", MAX_UPLOAD_BYTES)
|
|
if (len(raw) != self.manifest["source_bytes"] or
|
|
hashlib.sha256(raw).hexdigest() != self.manifest["source_sha256"]):
|
|
raise ValueError("replay_source_changed")
|
|
report = parse_native_report(raw)
|
|
if (report.hotel != self.manifest["hotel_id"] or report.arrival_date != self.manifest["report_date"]
|
|
or len(report.rows) != self.manifest["source_records"]):
|
|
raise ValueError("replay_source_context_changed")
|
|
return raw
|
|
|
|
|
|
class NativeXMLReplayExecutor:
|
|
def __init__(self, root, snapshot, policy, object_store, repository, service, processor=None):
|
|
self.root, self.snapshot, self.policy = root, snapshot, policy
|
|
self.store, self.repository, self.service, self.processor = object_store, repository, service, processor
|
|
|
|
def execute(self, *, request_id, from_date, to_date, report_stage):
|
|
validate_request_id(request_id)
|
|
if type(from_date) is not date or from_date != to_date or from_date != self.snapshot.day:
|
|
raise ValueError("replay_requires_exact_source_date")
|
|
payload = self.snapshot.payload()
|
|
binding = handoff.CaptureBinding("localxml-" + request_id, self.snapshot.manifest["hotel_id"],
|
|
from_date.isoformat(), fingerprint(self.snapshot.manifest), VERSION)
|
|
report_stage("processing")
|
|
prepared = handoff.prepare(self.root, binding, "ARR.XML", payload, self.policy, processor=self.processor)
|
|
receipt = handoff.deliver(Path(prepared["directory"]), prepared["manifest_sha256"],
|
|
object_store=self.store, repository=self.repository, service=self.service,
|
|
expected_binding=binding, expected_policy=self.policy)
|
|
return outcome_from_handoff(receipt, job_id=binding.job_id, report_date=from_date,
|
|
manifest_sha256=prepared["manifest_sha256"])
|
|
|
|
|
|
class SourceDateDownloads:
|
|
"""Reject missing fixture dates before queueing a task, retaining the existing API."""
|
|
def __init__(self, queue, snapshot):
|
|
self.queue, self.snapshot = queue, snapshot
|
|
|
|
@property
|
|
def ready(self):
|
|
return self.queue.ready
|
|
|
|
def latest(self):
|
|
return self.queue.latest()
|
|
|
|
def get(self, request_id):
|
|
return self.queue.get(request_id)
|
|
|
|
def retry(self, request_id):
|
|
return self.queue.retry(request_id)
|
|
|
|
def create(self, report_date, request_id):
|
|
if report_date != self.snapshot.day.isoformat():
|
|
raise PortalError("REPLAY_DATE_NOT_AVAILABLE", "本机重放仅提供已导入 XML 的报表日期。", 409)
|
|
return self.queue.create(report_date, request_id)
|
|
|
|
|
|
class LocalReplayPortal:
|
|
"""Visible simulation label and a separate cookie; underlying auth/CSRF stay active."""
|
|
allow_xml_upload = False
|
|
cookie_name = COOKIE
|
|
environment = "local-xml-replay"
|
|
source_kind = "native_xml_replay"
|
|
filename_prefix = "LOCAL-REPLAY-"
|
|
banner_title = "本机 XML 重放 · 模拟联调"
|
|
banner_action = "点击下载将重放已导入的 XML，使用独立测试数据库。"
|
|
page_title = "本机模拟"
|
|
|
|
def __init__(self, application, snapshot, port):
|
|
self.app, self.snapshot = application, snapshot
|
|
self.authority = f"127.0.0.1:{port}"
|
|
|
|
def handle(self, method, target, headers, body=b"", client_id="direct"):
|
|
headers = {k.lower(): v for k, v in headers.items()}
|
|
if headers.get("host") != self.authority:
|
|
return self._error("REPLAY_HOST_REJECTED", 403)
|
|
if headers.get("origin") not in (None, f"http://{self.authority}"):
|
|
return self._error("REPLAY_ORIGIN_REJECTED", 403)
|
|
path = urlsplit(target).path
|
|
if method == "GET" and path == "/healthz":
|
|
ready = self._health_ready()
|
|
return Response(200 if ready else 503, (self.environment + " ready\n").encode() if ready else b"unavailable\n",
|
|
"text/plain; charset=utf-8", {"X-ARR-Environment": self.environment, "Cache-Control": "no-store"})
|
|
if path.startswith("/api/public/") or path in {"/h5", "/h5.html"}:
|
|
return self._error("REPLAY_ROUTE_DISABLED", 404)
|
|
allowed = (self.allow_xml_upload and path == "/api/jobs") or path in {"/api/login", "/api/logout", "/api/arr-downloads"} or bool(
|
|
re.fullmatch(r"/api/arr-downloads/[a-f0-9]{32}/retry", path)) or bool(
|
|
re.fullmatch(r"/api/jobs/[^/]+/review/(?:items/[0-9]+|finalize|cancel)", path))
|
|
if method != "GET" and not allowed:
|
|
return self._error("REPLAY_ROUTE_DISABLED", 405)
|
|
cookie = SimpleCookie()
|
|
try:
|
|
cookie.load(headers.get("cookie", ""))
|
|
except Exception:
|
|
cookie = SimpleCookie()
|
|
token = cookie.get(self.cookie_name)
|
|
headers["cookie"] = f"arr_session={token.value}" if token else ""
|
|
response = self.app.handle(method, target, headers, body, client_id)
|
|
response_headers = dict(response.headers)
|
|
if "Set-Cookie" in response_headers:
|
|
response_headers["Set-Cookie"] = response_headers["Set-Cookie"].replace("arr_session=", self.cookie_name + "=", 1)
|
|
if "Content-Disposition" in response_headers:
|
|
response_headers["Content-Disposition"] = response_headers["Content-Disposition"].replace(
|
|
'filename="arr-report.xlsx"', f'filename="{self.filename_prefix}arr-report.xlsx"').replace(
|
|
"filename*=UTF-8''", "filename*=UTF-8''" + self.filename_prefix, 1)
|
|
response_headers["X-ARR-Environment"] = self.environment
|
|
response = replace(response, headers=response_headers)
|
|
if path == "/api/arr-downloads" and method == "GET" and response.status == 200:
|
|
payload = json.loads(response.body)
|
|
payload["data"].update(self._download_context())
|
|
response = replace(response, body=json.dumps(payload).encode())
|
|
if response.status == 200 and response.content_type.startswith("text/html"):
|
|
html = response.body.decode().replace("</head>", '<link rel="stylesheet" href="/assets/local-replay.css" /></head>')
|
|
html = html.replace("<body>", "<body>" + self._banner_html(), 1).replace("<title>ARR", f"<title>{self.page_title} · ARR", 1)
|
|
response = replace(response, body=html.encode())
|
|
if method == "GET" and path == "/assets/local-replay.css":
|
|
return Response(200, b'.local-replay-banner{padding:14px 20px;background:#fff1ce;color:#473500;'
|
|
b'border-bottom:2px solid #d2aa35;font:14px/1.6 system-ui;position:sticky;top:0;z-index:100}',
|
|
"text/css; charset=utf-8", {"Cache-Control": "no-store"})
|
|
return response
|
|
|
|
def _health_ready(self):
|
|
return self.app._health.database_ready and self.app._arr_downloads.ready
|
|
|
|
def _download_context(self):
|
|
return dict(default_date=self.snapshot.day.isoformat(), environment=self.environment,
|
|
source_kind=self.source_kind, oracle_connected=False,
|
|
source_records=self.snapshot.manifest["source_records"])
|
|
|
|
def _banner_html(self):
|
|
return (f'<aside class="local-replay-banner" role="note"><strong>{self.banner_title}</strong>'
|
|
f' · 数据日期 {self.snapshot.day.isoformat()} · {self.banner_action}'
|
|
'未连接 Oracle，处理结果仅用于测试。</aside>')
|
|
|
|
@staticmethod
|
|
def _error(code, status):
|
|
return Response.json(status, failure(PortalError(code, "本机重放入口不接受此请求。", status)))
|