| 1 | """Optional hosted publishing for rendered HTML artifacts.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | from collections.abc import Mapping |
| 7 | from typing import Any, Callable |
| 8 | from urllib.error import HTTPError, URLError |
| 9 | from urllib.request import Request, urlopen |
| 10 | |
| 11 | |
| 12 | DEFAULT_ENDPOINT = "https://api.ht-ml.app/v1/sites" |
| 13 | |
| 14 | |
| 15 | class HtmlPublishError(RuntimeError): |
| 16 | """Raised when the hosted HTML publish endpoint rejects the artifact.""" |
| 17 | |
| 18 | |
| 19 | class HtmlPublishBatchResult(dict[str, dict[str, Any]]): |
| 20 | """Successful document publishes plus an optional later failure.""" |
| 21 | |
| 22 | def __init__(self) -> None: |
| 23 | super().__init__() |
| 24 | self.error: HtmlPublishError | None = None |
| 25 | |
| 26 | |
| 27 | def publish_html( |
| 28 | html_content: str, |
| 29 | *, |
| 30 | password: str | None = None, |
| 31 | endpoint: str = DEFAULT_ENDPOINT, |
| 32 | opener: Callable[..., Any] | None = None, |
| 33 | timeout: int = 30, |
| 34 | ) -> dict[str, Any]: |
| 35 | """Publish a single HTML document and return the provider response.""" |
| 36 | if not html_content.strip(): |
| 37 | raise HtmlPublishError("HTML content is empty") |
| 38 | |
| 39 | payload: dict[str, str] = {"html_content": html_content} |
| 40 | if password is not None: |
| 41 | payload["password"] = password |
| 42 | |
| 43 | request = Request( |
| 44 | endpoint, |
| 45 | data=json.dumps(payload).encode("utf-8"), |
| 46 | headers={"Content-Type": "application/json", "Accept": "application/json"}, |
| 47 | method="POST", |
| 48 | ) |
| 49 | open_fn = opener or urlopen |
| 50 | try: |
| 51 | with open_fn(request, timeout=timeout) as response: |
| 52 | body = response.read().decode("utf-8") |
| 53 | except HTTPError as exc: |
| 54 | detail = exc.read().decode("utf-8", errors="replace") |
| 55 | raise HtmlPublishError(_error_message(exc.code, detail)) from exc |
| 56 | except URLError as exc: |
| 57 | raise HtmlPublishError(str(exc.reason)) from exc |
| 58 | except OSError as exc: |
| 59 | raise HtmlPublishError(str(exc)) from exc |
| 60 | |
| 61 | try: |
| 62 | result = json.loads(body) |
| 63 | except json.JSONDecodeError as exc: |
| 64 | raise HtmlPublishError("publish endpoint returned non-JSON response") from exc |
| 65 | if not isinstance(result, dict): |
| 66 | raise HtmlPublishError("publish endpoint returned unexpected JSON response") |
| 67 | |
| 68 | url = result.get("url") |
| 69 | if not isinstance(url, str) or not url.startswith("https://"): |
| 70 | raise HtmlPublishError("publish endpoint response did not include a valid url") |
| 71 | return result |
| 72 | |
| 73 | |
| 74 | def publish_html_documents( |
| 75 | documents: Mapping[str, str], |
| 76 | *, |
| 77 | password: str | None = None, |
| 78 | endpoint: str = DEFAULT_ENDPOINT, |
| 79 | opener: Callable[..., Any] | None = None, |
| 80 | timeout: int = 30, |
| 81 | ) -> HtmlPublishBatchResult: |
| 82 | """Publish a named set of documents, preserving caller order in results.""" |
| 83 | results = HtmlPublishBatchResult() |
| 84 | for name, content in documents.items(): |
| 85 | try: |
| 86 | results[name] = publish_html( |
| 87 | content, |
| 88 | password=password, |
| 89 | endpoint=endpoint, |
| 90 | opener=opener, |
| 91 | timeout=timeout, |
| 92 | ) |
| 93 | except HtmlPublishError as exc: |
| 94 | results.error = exc |
| 95 | break |
| 96 | return results |
| 97 | |
| 98 | |
| 99 | def _error_message(status: int, detail: str) -> str: |
| 100 | try: |
| 101 | payload = json.loads(detail) |
| 102 | except json.JSONDecodeError: |
| 103 | payload = {} |
| 104 | message = payload.get("message") if isinstance(payload, dict) else None |
| 105 | if message: |
| 106 | return f"{status}: {message}" |
| 107 | return f"{status}: {detail.strip() or 'publish failed'}" |
| 108 |