784 lines
27 KiB
Python
784 lines
27 KiB
Python
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import re
|
|
from collections.abc import Mapping, Sequence
|
|
from html import escape
|
|
from html.parser import HTMLParser
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.files import (
|
|
CAPABILITY_FILES_ARTIFACT_STORE,
|
|
ManagedArtifactStore,
|
|
ManagedArtifactWriteRequest,
|
|
)
|
|
from govoplan_core.core.templates import (
|
|
TemplateArtifactRef,
|
|
TemplateCompatibilityError,
|
|
TemplateRenderError,
|
|
TemplateRenderRequest,
|
|
TemplateRenderResult,
|
|
)
|
|
from govoplan_templates.backend.db.models import (
|
|
TemplateDefinition,
|
|
TemplateRender,
|
|
TemplateRevision,
|
|
)
|
|
from govoplan_templates.backend.service import (
|
|
ADMIN_SCOPE,
|
|
RENDER_SCOPE,
|
|
compatibility,
|
|
get_template,
|
|
get_template_revision,
|
|
)
|
|
|
|
|
|
RENDERER_VERSION = "templates-html-1"
|
|
MAX_OUTPUT_BYTES = 5 * 1024 * 1024
|
|
MAX_ITEMS = 5_000
|
|
_TOKEN_PATTERN = re.compile(r"{{\s*([A-Za-z_][A-Za-z0-9_.-]*)\s*}}")
|
|
|
|
|
|
class SqlTemplateRenderer:
|
|
def __init__(self, registry: object | None = None) -> None:
|
|
self.registry = registry
|
|
|
|
def render(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
request: TemplateRenderRequest,
|
|
) -> TemplateRenderResult:
|
|
if not isinstance(session, Session):
|
|
raise TypeError("Template rendering requires a SQLAlchemy session.")
|
|
if not isinstance(principal, ApiPrincipal):
|
|
raise TypeError("Template rendering requires an API principal.")
|
|
if not principal.has(RENDER_SCOPE) and not principal.has("templates:template:admin"):
|
|
raise PermissionError(f"Template rendering requires {RENDER_SCOPE}.")
|
|
return render_template(
|
|
session,
|
|
principal,
|
|
registry=self.registry,
|
|
request=request,
|
|
)
|
|
|
|
|
|
def render_template(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
registry: object | None,
|
|
request: TemplateRenderRequest,
|
|
) -> TemplateRenderResult:
|
|
definition = get_template(session, principal, request.template_id)
|
|
revision = get_template_revision(
|
|
session,
|
|
definition,
|
|
revision=request.revision,
|
|
published_preferred=request.mode == "final" and request.revision is None,
|
|
)
|
|
if request.mode == "final" and revision.published_at is None:
|
|
raise TemplateCompatibilityError(
|
|
"Final output requires a published template revision."
|
|
)
|
|
if len(request.items) > MAX_ITEMS:
|
|
raise TemplateRenderError(f"Template renders are limited to {MAX_ITEMS} items.")
|
|
items = tuple(request.items) or ({},)
|
|
diagnostics = _validate_render_inputs(revision, request, items)
|
|
blocking = [item for item in diagnostics if item.get("severity") == "error"]
|
|
if blocking:
|
|
raise TemplateCompatibilityError(
|
|
"; ".join(str(item.get("message") or "Template input is incompatible.") for item in blocking)
|
|
)
|
|
input_hash = _canonical_hash(
|
|
{
|
|
"usage": request.usage,
|
|
"locale": request.locale,
|
|
"output_format": request.output_format,
|
|
"profile_id": request.profile_id,
|
|
"parameters": request.parameters,
|
|
"items": items,
|
|
"input_snapshot": request.input_snapshot,
|
|
}
|
|
)
|
|
existing = _idempotent_render(
|
|
session,
|
|
principal,
|
|
request=request,
|
|
revision=revision,
|
|
input_hash=input_hash,
|
|
)
|
|
if existing is not None:
|
|
return render_result(existing)
|
|
|
|
payload, content_type, page_count = _render_payload(
|
|
definition,
|
|
revision,
|
|
request=request,
|
|
items=items,
|
|
)
|
|
if len(payload) > MAX_OUTPUT_BYTES:
|
|
raise TemplateRenderError(
|
|
f"Rendered output exceeds the {MAX_OUTPUT_BYTES} byte bounded-download limit."
|
|
)
|
|
output_sha256 = hashlib.sha256(payload).hexdigest()
|
|
filename = _output_filename(definition, revision, request.output_format)
|
|
artifact = _persist_artifact(
|
|
registry,
|
|
session,
|
|
principal,
|
|
request=request,
|
|
definition=definition,
|
|
revision=revision,
|
|
payload=payload,
|
|
filename=filename,
|
|
content_type=content_type,
|
|
input_hash=input_hash,
|
|
output_sha256=output_sha256,
|
|
diagnostics=diagnostics,
|
|
)
|
|
row = TemplateRender(
|
|
tenant_id=principal.tenant_id,
|
|
template_id=definition.id,
|
|
revision_id=revision.id,
|
|
revision_number=revision.revision,
|
|
mode=request.mode,
|
|
usage=request.usage,
|
|
output_format=request.output_format,
|
|
content_type=content_type,
|
|
filename=filename,
|
|
idempotency_key=request.idempotency_key,
|
|
template_hash=revision.definition_hash,
|
|
input_hash=input_hash,
|
|
renderer_version=RENDERER_VERSION,
|
|
output_sha256=output_sha256,
|
|
output_size_bytes=len(payload),
|
|
item_count=len(items),
|
|
page_count=page_count,
|
|
diagnostics=diagnostics,
|
|
input_snapshot=dict(request.input_snapshot),
|
|
artifact_ref=dataclasses.asdict(artifact) if artifact else None,
|
|
payload=None if artifact is not None else payload,
|
|
created_by_account_id=principal.account_id,
|
|
)
|
|
session.add(row)
|
|
session.flush()
|
|
if artifact is None:
|
|
artifact = TemplateArtifactRef(
|
|
kind="bounded_download",
|
|
filename=filename,
|
|
content_type=content_type,
|
|
size_bytes=len(payload),
|
|
sha256=output_sha256,
|
|
download_path=f"/api/v1/templates/renders/{row.id}/download",
|
|
provenance={"module": "templates", "bounded": True},
|
|
)
|
|
row.artifact_ref = dataclasses.asdict(artifact)
|
|
session.add(row)
|
|
session.flush()
|
|
return render_result(row)
|
|
|
|
|
|
def get_render_for_principal(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
render_id: str,
|
|
) -> TemplateRender:
|
|
row = session.scalar(
|
|
select(TemplateRender).where(
|
|
TemplateRender.id == render_id,
|
|
TemplateRender.tenant_id == principal.tenant_id,
|
|
)
|
|
)
|
|
if row is None:
|
|
raise TemplateRenderError("Template render not found.")
|
|
get_template(session, principal, row.template_id)
|
|
if (
|
|
row.created_by_account_id != principal.account_id
|
|
and not principal.has(ADMIN_SCOPE)
|
|
):
|
|
# Render payloads may contain recipient-specific or otherwise
|
|
# confidential data. Do not reveal whether another actor's render
|
|
# exists to ordinary template readers.
|
|
raise TemplateRenderError("Template render not found.")
|
|
return row
|
|
|
|
|
|
def list_renders(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
template_id: str | None = None,
|
|
limit: int = 100,
|
|
) -> list[TemplateRender]:
|
|
statement = select(TemplateRender).where(
|
|
TemplateRender.tenant_id == principal.tenant_id
|
|
)
|
|
if not principal.has(ADMIN_SCOPE):
|
|
statement = statement.where(
|
|
TemplateRender.created_by_account_id == principal.account_id
|
|
)
|
|
if template_id:
|
|
get_template(session, principal, template_id)
|
|
statement = statement.where(TemplateRender.template_id == template_id)
|
|
rows = list(
|
|
session.scalars(
|
|
statement.order_by(TemplateRender.created_at.desc()).limit(
|
|
max(1, min(limit, 500))
|
|
)
|
|
)
|
|
)
|
|
visible_template_ids = {
|
|
row.template_id
|
|
for row in rows
|
|
if _template_visible(session, principal, row.template_id)
|
|
}
|
|
return [row for row in rows if row.template_id in visible_template_ids]
|
|
|
|
|
|
def render_result(row: TemplateRender) -> TemplateRenderResult:
|
|
artifact = (
|
|
TemplateArtifactRef(**row.artifact_ref)
|
|
if isinstance(row.artifact_ref, dict)
|
|
else None
|
|
)
|
|
return TemplateRenderResult(
|
|
render_id=row.id,
|
|
template_id=row.template_id,
|
|
revision_id=row.revision_id,
|
|
revision=row.revision_number,
|
|
template_hash=row.template_hash,
|
|
input_hash=row.input_hash,
|
|
renderer_version=row.renderer_version,
|
|
output_format=row.output_format, # type: ignore[arg-type]
|
|
content_type=row.content_type,
|
|
filename=row.filename,
|
|
item_count=row.item_count,
|
|
page_count=row.page_count,
|
|
output_sha256=row.output_sha256,
|
|
output_size_bytes=row.output_size_bytes,
|
|
diagnostics=tuple(row.diagnostics or []),
|
|
artifact=artifact,
|
|
generated_at=row.created_at,
|
|
payload=row.payload,
|
|
)
|
|
|
|
|
|
def _validate_render_inputs(
|
|
revision: TemplateRevision,
|
|
request: TemplateRenderRequest,
|
|
items: Sequence[Mapping[str, object]],
|
|
) -> list[dict[str, object]]:
|
|
diagnostics: list[dict[str, object]] = []
|
|
available_fields = _available_field_types(request.parameters, items)
|
|
contract = compatibility(
|
|
revision,
|
|
usage=request.usage,
|
|
output_format=request.output_format,
|
|
available_fields=available_fields,
|
|
)
|
|
diagnostics.extend(dict(item) for item in contract.diagnostics)
|
|
if request.profile_id:
|
|
profile = next(
|
|
(
|
|
item
|
|
for item in revision.output_profiles
|
|
if isinstance(item, dict) and item.get("id") == request.profile_id
|
|
),
|
|
None,
|
|
)
|
|
if profile is None:
|
|
diagnostics.append(
|
|
{
|
|
"code": "template.output_profile_missing",
|
|
"severity": "error",
|
|
"message": f"Output profile {request.profile_id} is not defined by this revision.",
|
|
}
|
|
)
|
|
elif profile.get("output_format") != request.output_format:
|
|
diagnostics.append(
|
|
{
|
|
"code": "template.output_profile_format_mismatch",
|
|
"severity": "error",
|
|
"message": (
|
|
f"Output profile {request.profile_id} does not provide "
|
|
f"{request.output_format} output."
|
|
),
|
|
}
|
|
)
|
|
if request.locale and revision.locale.lower() != request.locale.lower():
|
|
diagnostics.append(
|
|
{
|
|
"code": "template.locale_mismatch",
|
|
"severity": "warning",
|
|
"message": f"Requested locale {request.locale} uses template locale {revision.locale}.",
|
|
}
|
|
)
|
|
for index, item in enumerate(items):
|
|
context = _render_context(request.parameters, item, index)
|
|
for requirement in revision.required_fields:
|
|
if not bool(requirement.get("required", True)):
|
|
continue
|
|
path = str(requirement.get("path") or "")
|
|
value, present = _resolve_path(context, path)
|
|
if not present or value in (None, ""):
|
|
diagnostics.append(
|
|
{
|
|
"code": "template.item_required_field_missing",
|
|
"severity": "error",
|
|
"message": f"Item {index + 1} is missing required field {path}.",
|
|
"item_index": index,
|
|
"field": path,
|
|
}
|
|
)
|
|
continue
|
|
expected = str(requirement.get("value_type") or "string")
|
|
if not _value_matches_type(value, expected):
|
|
diagnostics.append(
|
|
{
|
|
"code": "template.item_field_type_invalid",
|
|
"severity": "error",
|
|
"message": f"Item {index + 1} field {path} is not {expected}.",
|
|
"item_index": index,
|
|
"field": path,
|
|
}
|
|
)
|
|
return _unique_diagnostics(diagnostics)
|
|
|
|
|
|
def _render_payload(
|
|
definition: TemplateDefinition,
|
|
revision: TemplateRevision,
|
|
*,
|
|
request: TemplateRenderRequest,
|
|
items: Sequence[Mapping[str, object]],
|
|
) -> tuple[bytes, str, int]:
|
|
if request.output_format == "text":
|
|
body = revision.content_text or _html_to_text(revision.content_html or "")
|
|
separator = "\n\n---\n\n" if revision.template_type != "list_layout" else "\n"
|
|
rendered = _render_items(body, request.parameters, items, html=False, separator=separator)
|
|
payload = separator.join(rendered).encode("utf-8")
|
|
return payload, "text/plain; charset=utf-8", _page_count(revision, len(items))
|
|
|
|
body = revision.content_html or f"<pre>{escape(revision.content_text or '')}</pre>"
|
|
rendered = _render_items(body, request.parameters, items, html=True)
|
|
page_count = _page_count(revision, len(items))
|
|
document = _html_document(definition, revision, rendered)
|
|
payload = document.encode("utf-8")
|
|
if len(payload) > MAX_OUTPUT_BYTES:
|
|
_output_limit_exceeded()
|
|
return payload, "text/html; charset=utf-8", page_count
|
|
|
|
|
|
def _output_limit_exceeded() -> None:
|
|
raise TemplateRenderError(
|
|
f"Rendered output exceeds the {MAX_OUTPUT_BYTES} byte bounded-download limit."
|
|
)
|
|
|
|
|
|
def _render_items(
|
|
body: str,
|
|
parameters: Mapping[str, object],
|
|
items: Sequence[Mapping[str, object]],
|
|
*,
|
|
html: bool,
|
|
separator: str = "",
|
|
) -> list[str]:
|
|
# Reject as soon as the same existing output budget is exhausted; never
|
|
# build thousands of oversized documents and only then measure the join.
|
|
remaining = MAX_OUTPUT_BYTES
|
|
separator_bytes = len(separator.encode("utf-8"))
|
|
rendered: list[str] = []
|
|
for index, item in enumerate(items):
|
|
if index:
|
|
remaining -= separator_bytes
|
|
if remaining < 0:
|
|
_output_limit_exceeded()
|
|
value = _substitute(body, _render_context(parameters, item, index), html=html, max_bytes=remaining)
|
|
remaining -= len(value.encode("utf-8"))
|
|
rendered.append(value)
|
|
return rendered
|
|
|
|
|
|
def _html_document(
|
|
definition: TemplateDefinition,
|
|
revision: TemplateRevision,
|
|
rendered: Sequence[str],
|
|
) -> str:
|
|
page_size = _page_size(revision.layout.get("page_size") or _profile_page_size(revision))
|
|
margin = _millimetres(revision.layout.get("margin_mm"), 15.0, minimum=0, maximum=60)
|
|
template_type = revision.template_type
|
|
if template_type == "label_sheet":
|
|
columns = _integer(revision.layout.get("columns"), 3, minimum=1, maximum=12)
|
|
rows = _integer(revision.layout.get("rows"), 8, minimum=1, maximum=30)
|
|
gap = _millimetres(revision.layout.get("gap_mm"), 2.0, minimum=0, maximum=20)
|
|
per_page = columns * rows
|
|
pages = []
|
|
for start in range(0, len(rendered), per_page):
|
|
labels = "".join(f'<section class="template-label">{item}</section>' for item in rendered[start:start + per_page])
|
|
pages.append(f'<main class="template-page template-label-sheet">{labels}</main>')
|
|
body = "".join(pages)
|
|
type_css = (
|
|
f".template-label-sheet{{display:grid;grid-template-columns:repeat({columns},minmax(0,1fr));"
|
|
f"grid-template-rows:repeat({rows},minmax(0,1fr));gap:{gap}mm;}}"
|
|
".template-label{overflow:hidden;border:0.2mm solid #c9c9c9;padding:2mm;}"
|
|
)
|
|
elif template_type == "list_layout":
|
|
body = f'<main class="template-page template-list">{"".join(rendered)}</main>'
|
|
type_css = ".template-list>*{break-inside:avoid;}"
|
|
else:
|
|
body = "".join(f'<main class="template-page">{item}</main>' for item in rendered)
|
|
type_css = ""
|
|
return (
|
|
"<!doctype html><html><head><meta charset=\"utf-8\">"
|
|
f"<title>{escape(definition.name)}</title><style>"
|
|
f"@page{{size:{page_size};margin:{margin}mm;}}"
|
|
"*{box-sizing:border-box;}html,body{margin:0;padding:0;color:#171717;background:#fff;"
|
|
"font-family:Arial,Helvetica,sans-serif;font-size:10pt;line-height:1.35;}"
|
|
".template-page{break-after:page;min-height:1px;}"
|
|
".template-page:last-child{break-after:auto;}table{border-collapse:collapse;width:100%;}"
|
|
"th,td{padding:1.5mm;text-align:left;vertical-align:top;}"
|
|
f"{type_css}</style></head><body>{body}</body></html>"
|
|
)
|
|
|
|
|
|
def _persist_artifact(
|
|
registry: object | None,
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
request: TemplateRenderRequest,
|
|
definition: TemplateDefinition,
|
|
revision: TemplateRevision,
|
|
payload: bytes,
|
|
filename: str,
|
|
content_type: str,
|
|
input_hash: str,
|
|
output_sha256: str,
|
|
diagnostics: list[dict[str, object]],
|
|
) -> TemplateArtifactRef | None:
|
|
if not request.persist_to_files:
|
|
return None
|
|
if (
|
|
registry is None
|
|
or not hasattr(registry, "has_capability")
|
|
or not registry.has_capability(CAPABILITY_FILES_ARTIFACT_STORE)
|
|
):
|
|
diagnostics.append(
|
|
{
|
|
"code": "template.files_unavailable",
|
|
"severity": "warning",
|
|
"message": "Files artifact storage is unavailable; using a bounded Templates download.",
|
|
}
|
|
)
|
|
return None
|
|
capability = registry.capability(CAPABILITY_FILES_ARTIFACT_STORE)
|
|
if not isinstance(capability, ManagedArtifactStore):
|
|
diagnostics.append(
|
|
{
|
|
"code": "template.files_contract_invalid",
|
|
"severity": "warning",
|
|
"message": "Files artifact storage has an incompatible contract; using a bounded Templates download.",
|
|
}
|
|
)
|
|
return None
|
|
try:
|
|
stored = capability.store_artifact(
|
|
session,
|
|
principal,
|
|
request=ManagedArtifactWriteRequest(
|
|
filename=filename,
|
|
payload=payload,
|
|
content_type=content_type,
|
|
folder="Generated/Templates",
|
|
description=f"Rendered from template {definition.name} revision {revision.revision}.",
|
|
idempotency_key=request.idempotency_key,
|
|
metadata={
|
|
"producer_module": "templates",
|
|
"template_id": definition.id,
|
|
"template_revision_id": revision.id,
|
|
"template_hash": revision.definition_hash,
|
|
"input_hash": input_hash,
|
|
"output_sha256": output_sha256,
|
|
},
|
|
),
|
|
)
|
|
except (PermissionError, RuntimeError, ValueError) as exc:
|
|
diagnostics.append(
|
|
{
|
|
"code": "template.files_store_failed",
|
|
"severity": "warning",
|
|
"message": "Managed Files persistence was not permitted or available; using a bounded Templates download.",
|
|
"error_type": type(exc).__name__,
|
|
}
|
|
)
|
|
return None
|
|
return TemplateArtifactRef(
|
|
kind="managed_file",
|
|
filename=stored.filename,
|
|
content_type=stored.content_type,
|
|
size_bytes=stored.size_bytes,
|
|
sha256=stored.sha256,
|
|
file_asset_id=stored.file_asset_id,
|
|
file_version_id=stored.file_version_id,
|
|
download_path=f"/api/v1/files/{stored.file_asset_id}/download",
|
|
provenance=dict(stored.provenance),
|
|
)
|
|
|
|
|
|
def _idempotent_render(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
request: TemplateRenderRequest,
|
|
revision: TemplateRevision,
|
|
input_hash: str,
|
|
) -> TemplateRender | None:
|
|
if not request.idempotency_key:
|
|
return None
|
|
existing = session.scalar(
|
|
select(TemplateRender).where(
|
|
TemplateRender.tenant_id == principal.tenant_id,
|
|
TemplateRender.idempotency_key == request.idempotency_key,
|
|
)
|
|
)
|
|
if existing is None:
|
|
return None
|
|
if (
|
|
existing.template_id != revision.template_id
|
|
or existing.revision_id != revision.id
|
|
or existing.input_hash != input_hash
|
|
or existing.output_format != request.output_format
|
|
or existing.mode != request.mode
|
|
):
|
|
raise TemplateRenderError(
|
|
"The render idempotency key was already used for different input."
|
|
)
|
|
return existing
|
|
|
|
|
|
def _render_context(
|
|
parameters: Mapping[str, object],
|
|
item: Mapping[str, object],
|
|
index: int,
|
|
) -> dict[str, object]:
|
|
return {
|
|
**dict(parameters),
|
|
**dict(item),
|
|
"parameters": dict(parameters),
|
|
"item": dict(item),
|
|
"recipient": dict(item),
|
|
"index": index + 1,
|
|
}
|
|
|
|
|
|
def _substitute(template: str, context: Mapping[str, object], *, html: bool, max_bytes: int | None = None) -> str:
|
|
remaining = MAX_OUTPUT_BYTES if max_bytes is None else max_bytes
|
|
pieces: list[str] = []
|
|
|
|
def append(value: str) -> None:
|
|
nonlocal remaining
|
|
# Character count is a cheap lower bound before allocating UTF-8 bytes.
|
|
if len(value) > remaining:
|
|
_output_limit_exceeded()
|
|
remaining -= len(value.encode("utf-8"))
|
|
if remaining < 0:
|
|
_output_limit_exceeded()
|
|
pieces.append(value)
|
|
|
|
previous = 0
|
|
for match in _TOKEN_PATTERN.finditer(template):
|
|
append(template[previous:match.start()])
|
|
value, present = _resolve_path(context, match.group(1))
|
|
if present and value is not None:
|
|
rendered = _display_value(value)
|
|
if len(rendered) > remaining:
|
|
_output_limit_exceeded()
|
|
append(escape(rendered, quote=True) if html else rendered)
|
|
previous = match.end()
|
|
append(template[previous:])
|
|
return "".join(pieces)
|
|
|
|
|
|
def _resolve_path(context: Mapping[str, object], path: str) -> tuple[object | None, bool]:
|
|
if path in context:
|
|
return context[path], True
|
|
current: object = context
|
|
for part in path.split("."):
|
|
if not isinstance(current, Mapping) or part not in current:
|
|
return None, False
|
|
current = current[part]
|
|
return current, True
|
|
|
|
|
|
def _available_field_types(
|
|
parameters: Mapping[str, object],
|
|
items: Sequence[Mapping[str, object]],
|
|
) -> dict[str, str]:
|
|
result: dict[str, str] = {}
|
|
for prefix, value in (("parameters", parameters),):
|
|
_flatten_types(value, prefix, result)
|
|
for item in items:
|
|
_flatten_types(item, "", result)
|
|
_flatten_types(item, "item", result)
|
|
_flatten_types(item, "recipient", result)
|
|
return result
|
|
|
|
|
|
def _flatten_types(value: object, prefix: str, result: dict[str, str]) -> None:
|
|
if isinstance(value, Mapping):
|
|
if prefix:
|
|
result.setdefault(prefix, "object")
|
|
for key, item in value.items():
|
|
path = f"{prefix}.{key}" if prefix else str(key)
|
|
_flatten_types(item, path, result)
|
|
return
|
|
result.setdefault(prefix, _value_type(value))
|
|
|
|
|
|
def _value_type(value: object) -> str:
|
|
if isinstance(value, bool):
|
|
return "boolean"
|
|
if isinstance(value, int):
|
|
return "integer"
|
|
if isinstance(value, float):
|
|
return "number"
|
|
if isinstance(value, Mapping):
|
|
return "object"
|
|
if isinstance(value, (list, tuple)):
|
|
return "array"
|
|
return "string"
|
|
|
|
|
|
def _value_matches_type(value: object, expected: str) -> bool:
|
|
actual = _value_type(value)
|
|
if expected == "number":
|
|
return actual in {"integer", "number"}
|
|
if expected in {"date", "datetime"}:
|
|
return isinstance(value, str) and bool(value.strip())
|
|
return actual == expected
|
|
|
|
|
|
def _display_value(value: object) -> str:
|
|
if isinstance(value, (dict, list, tuple)):
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
if isinstance(value, bool):
|
|
return "true" if value else "false"
|
|
return str(value)
|
|
|
|
|
|
def _canonical_hash(value: object) -> str:
|
|
payload = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
|
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _page_count(revision: TemplateRevision, item_count: int) -> int:
|
|
if revision.template_type == "label_sheet":
|
|
columns = _integer(revision.layout.get("columns"), 3, minimum=1, maximum=12)
|
|
rows = _integer(revision.layout.get("rows"), 8, minimum=1, maximum=30)
|
|
return max(1, math.ceil(item_count / (columns * rows)))
|
|
if revision.template_type == "list_layout":
|
|
return 1
|
|
return max(1, item_count)
|
|
|
|
|
|
def _output_filename(
|
|
definition: TemplateDefinition,
|
|
revision: TemplateRevision,
|
|
output_format: str,
|
|
) -> str:
|
|
extension = "html" if output_format == "html" else "txt"
|
|
return f"{definition.slug}-r{revision.revision}.{extension}"
|
|
|
|
|
|
def _profile_page_size(revision: TemplateRevision) -> object:
|
|
for profile in revision.output_profiles:
|
|
page = profile.get("page") if isinstance(profile, dict) else None
|
|
if isinstance(page, dict) and page.get("size"):
|
|
return page["size"]
|
|
return "A4"
|
|
|
|
|
|
def _page_size(value: object) -> str:
|
|
normalized = str(value or "A4").upper()
|
|
return normalized if normalized in {"A3", "A4", "A5", "LETTER", "LEGAL", "DL"} else "A4"
|
|
|
|
|
|
def _integer(value: object, fallback: int, *, minimum: int, maximum: int) -> int:
|
|
try:
|
|
number = int(value)
|
|
except (TypeError, ValueError):
|
|
number = fallback
|
|
return max(minimum, min(maximum, number))
|
|
|
|
|
|
def _millimetres(value: object, fallback: float, *, minimum: float, maximum: float) -> str:
|
|
try:
|
|
number = float(value)
|
|
except (TypeError, ValueError):
|
|
number = fallback
|
|
number = max(minimum, min(maximum, number))
|
|
return f"{number:.2f}".rstrip("0").rstrip(".")
|
|
|
|
|
|
class _PlainTextExtractor(HTMLParser):
|
|
block_tags = {"br", "div", "h1", "h2", "h3", "h4", "h5", "h6", "li", "p", "tr"}
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__(convert_charrefs=True)
|
|
self.parts: list[str] = []
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
del attrs
|
|
if tag in self.block_tags:
|
|
self.parts.append("\n")
|
|
|
|
def handle_data(self, data: str) -> None:
|
|
self.parts.append(data)
|
|
|
|
|
|
def _html_to_text(value: str) -> str:
|
|
parser = _PlainTextExtractor()
|
|
parser.feed(value)
|
|
parser.close()
|
|
return "\n".join(line.strip() for line in "".join(parser.parts).splitlines() if line.strip())
|
|
|
|
|
|
def _unique_diagnostics(items: Sequence[dict[str, object]]) -> list[dict[str, object]]:
|
|
seen: set[str] = set()
|
|
result: list[dict[str, object]] = []
|
|
for item in items:
|
|
key = json.dumps(item, sort_keys=True, default=str)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
result.append(item)
|
|
return result
|
|
|
|
|
|
def _template_visible(session: Session, principal: ApiPrincipal, template_id: str) -> bool:
|
|
try:
|
|
get_template(session, principal, template_id)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
__all__ = [
|
|
"MAX_ITEMS",
|
|
"MAX_OUTPUT_BYTES",
|
|
"RENDERER_VERSION",
|
|
"SqlTemplateRenderer",
|
|
"get_render_for_principal",
|
|
"list_renders",
|
|
"render_result",
|
|
"render_template",
|
|
]
|