feat: prepare ARR for controlled public deployment

arr_web/contracts.py

@@ -0,0 +1,97 @@
+"""Small, dependency-free contracts shared by the ARR web adapters."""
+
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass
+from typing import Any, Dict, Mapping, Optional
+
+
+API_VERSION = "1.0"
+MAX_UPLOAD_BYTES = 25 * 1024 * 1024
+MONTH_RE = re.compile(r"^(\d{4})-(\d{2})$")
+SAFE_FILENAME_RE = re.compile(r"^[^/\\\x00-\x1f\x7f]{1,255}$")
+
+
+class PortalError(RuntimeError):
+    """A public-safe portal failure with an HTTP status."""
+
+    def __init__(self, code: str, safe_message: str, status: int = 400):
+        super().__init__(safe_message)
+        self.code = code
+        self.safe_message = safe_message
+        self.status = status
+
+
+def validate_month(value: str) -> str:
+    match = MONTH_RE.fullmatch(value)
+    if not match or not 1 <= int(match.group(2)) <= 12:
+        raise PortalError("MONTH_INVALID", "month must use YYYY-MM")
+    return value
+
+
+def validate_upload_filename(value: object) -> str:
+    if (
+        not isinstance(value, str)
+        or SAFE_FILENAME_RE.fullmatch(value) is None
+        or not value.lower().endswith(".xml")
+        or value in {".", ".."}
+    ):
+        raise PortalError("UPLOAD_FILENAME_INVALID", "请选择 XML 文件")
+    return value
+
+
+def validate_xml_payload(value: bytes) -> None:
+    if not value:
+        raise PortalError("UPLOAD_EMPTY", "XML 文件为空")
+    if len(value) > MAX_UPLOAD_BYTES:
+        raise PortalError("UPLOAD_TOO_LARGE", "XML 文件超过 25 MB")
+    sample = value[:4096]
+    if sample.startswith(b"\xef\xbb\xbf"):
+        sample = sample[3:]
+    if not sample.lstrip().startswith(b"<"):
+        raise PortalError("UPLOAD_CONTENT_INVALID", "文件内容不是 XML")
+
+
+@dataclass(frozen=True)
+class Response:
+    status: int
+    body: bytes
+    content_type: str
+    headers: Mapping[str, str]
+
+    @classmethod
+    def json(
+        cls,
+        status: int,
+        payload: Mapping[str, Any],
+        headers: Optional[Mapping[str, str]] = None,
+    ) -> "Response":
+        body = (
+            json.dumps(
+                payload,
+                ensure_ascii=False,
+                allow_nan=False,
+                separators=(",", ":"),
+            )
+            + "\n"
+        ).encode("utf-8")
+        response_headers = {"Cache-Control": "no-store"}
+        if headers:
+            response_headers.update(headers)
+        return cls(status, body, "application/json; charset=utf-8", response_headers)
+
+
+def success(data: Any, **extra: Any) -> Dict[str, Any]:
+    payload: Dict[str, Any] = {"ok": True, "api_version": API_VERSION, "data": data}
+    payload.update(extra)
+    return payload
+
+
+def failure(error: PortalError) -> Dict[str, Any]:
+    return {
+        "ok": False,
+        "api_version": API_VERSION,
+        "error": {"code": error.code, "message": error.safe_message},
+    }