Implement typed template library and rendering
This commit is contained in:
@@ -0,0 +1,674 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from collections.abc import Mapping, Sequence
|
||||
from html import escape
|
||||
from html.parser import HTMLParser
|
||||
|
||||
from sqlalchemy import or_, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_core.auth import ApiPrincipal
|
||||
from govoplan_core.core.concurrency import claim_revision
|
||||
from govoplan_core.core.templates import (
|
||||
TemplateCompatibility,
|
||||
TemplateCompatibilityError,
|
||||
TemplateFieldRequirement,
|
||||
TemplateNotFoundError,
|
||||
TemplateOutputProfile,
|
||||
TemplateRef,
|
||||
TemplateRevisionRef,
|
||||
)
|
||||
from govoplan_core.security.time import utc_now
|
||||
from govoplan_templates.backend.db.models import TemplateDefinition, TemplateRevision
|
||||
from govoplan_templates.backend.schemas import TemplateCreateRequest, TemplateRevisionPayload, TemplateUpdateRequest
|
||||
|
||||
|
||||
READ_SCOPE = "templates:template:read"
|
||||
WRITE_SCOPE = "templates:template:write"
|
||||
PUBLISH_SCOPE = "templates:template:publish"
|
||||
RENDER_SCOPE = "templates:template:render"
|
||||
ADMIN_SCOPE = "templates:template:admin"
|
||||
|
||||
_TOKEN_PATTERN = re.compile(r"{{\s*([A-Za-z_][A-Za-z0-9_.-]*)\s*}}")
|
||||
|
||||
|
||||
def visible_templates_statement(principal: ApiPrincipal, *, include_deleted: bool = False):
|
||||
statement = select(TemplateDefinition).where(
|
||||
TemplateDefinition.tenant_id == principal.tenant_id
|
||||
)
|
||||
if not principal.has(ADMIN_SCOPE):
|
||||
statement = statement.where(
|
||||
or_(
|
||||
TemplateDefinition.scope_type == "tenant",
|
||||
(
|
||||
(TemplateDefinition.scope_type == "user")
|
||||
& (TemplateDefinition.scope_id == principal.account_id)
|
||||
),
|
||||
(
|
||||
(TemplateDefinition.scope_type == "group")
|
||||
& TemplateDefinition.scope_id.in_(tuple(principal.group_ids) or ("",))
|
||||
),
|
||||
)
|
||||
)
|
||||
if not include_deleted:
|
||||
statement = statement.where(TemplateDefinition.deleted_at.is_(None))
|
||||
return statement
|
||||
|
||||
|
||||
def list_templates(
|
||||
session: Session,
|
||||
principal: ApiPrincipal,
|
||||
*,
|
||||
query: str = "",
|
||||
usage: str | None = None,
|
||||
template_type: str | None = None,
|
||||
locale: str | None = None,
|
||||
include_deleted: bool = False,
|
||||
limit: int = 200,
|
||||
) -> list[TemplateDefinition]:
|
||||
statement = visible_templates_statement(principal, include_deleted=include_deleted)
|
||||
normalized_query = query.strip()
|
||||
if normalized_query:
|
||||
pattern = f"%{normalized_query}%"
|
||||
statement = statement.where(
|
||||
or_(
|
||||
TemplateDefinition.name.ilike(pattern),
|
||||
TemplateDefinition.slug.ilike(pattern),
|
||||
TemplateDefinition.description.ilike(pattern),
|
||||
)
|
||||
)
|
||||
if template_type:
|
||||
statement = statement.where(TemplateDefinition.template_type == template_type)
|
||||
rows = list(
|
||||
session.scalars(
|
||||
statement.order_by(TemplateDefinition.name, TemplateDefinition.id).limit(
|
||||
max(1, min(limit, 500))
|
||||
)
|
||||
)
|
||||
)
|
||||
if not usage and not locale:
|
||||
return rows
|
||||
filtered: list[TemplateDefinition] = []
|
||||
for item in rows:
|
||||
revision = get_template_revision(session, item, published_preferred=True)
|
||||
if usage and usage.strip().lower() not in revision.usages:
|
||||
continue
|
||||
if locale and revision.locale.lower() != locale.lower():
|
||||
continue
|
||||
filtered.append(item)
|
||||
return filtered
|
||||
|
||||
|
||||
def get_template(
|
||||
session: Session,
|
||||
principal: ApiPrincipal,
|
||||
template_id: str,
|
||||
*,
|
||||
include_deleted: bool = False,
|
||||
) -> TemplateDefinition:
|
||||
item = session.scalar(
|
||||
visible_templates_statement(principal, include_deleted=include_deleted).where(
|
||||
TemplateDefinition.id == template_id
|
||||
)
|
||||
)
|
||||
if item is None:
|
||||
raise TemplateNotFoundError("Template not found.")
|
||||
return item
|
||||
|
||||
|
||||
def get_template_revision(
|
||||
session: Session,
|
||||
definition: TemplateDefinition,
|
||||
*,
|
||||
revision: int | None = None,
|
||||
published_preferred: bool = False,
|
||||
) -> TemplateRevision:
|
||||
if revision is not None:
|
||||
revision_number = revision
|
||||
elif published_preferred and definition.published_revision_id:
|
||||
published = session.get(TemplateRevision, definition.published_revision_id)
|
||||
if published is not None and published.template_id == definition.id:
|
||||
return published
|
||||
revision_number = definition.current_revision
|
||||
else:
|
||||
revision_number = definition.current_revision
|
||||
item = session.scalar(
|
||||
select(TemplateRevision).where(
|
||||
TemplateRevision.template_id == definition.id,
|
||||
TemplateRevision.tenant_id == definition.tenant_id,
|
||||
TemplateRevision.revision == revision_number,
|
||||
)
|
||||
)
|
||||
if item is None:
|
||||
raise TemplateNotFoundError("Template revision not found.")
|
||||
return item
|
||||
|
||||
|
||||
def list_template_revisions(
|
||||
session: Session,
|
||||
definition: TemplateDefinition,
|
||||
) -> list[TemplateRevision]:
|
||||
return list(
|
||||
session.scalars(
|
||||
select(TemplateRevision)
|
||||
.where(
|
||||
TemplateRevision.tenant_id == definition.tenant_id,
|
||||
TemplateRevision.template_id == definition.id,
|
||||
)
|
||||
.order_by(TemplateRevision.revision.desc())
|
||||
.limit(500)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def create_template(
|
||||
session: Session,
|
||||
principal: ApiPrincipal,
|
||||
payload: TemplateCreateRequest,
|
||||
) -> tuple[TemplateDefinition, TemplateRevision]:
|
||||
_ensure_requested_scope(principal, payload.scope_type, payload.scope_id)
|
||||
slug = _slug(payload.slug or payload.name)
|
||||
_ensure_unique_slug(
|
||||
session,
|
||||
principal,
|
||||
slug=slug,
|
||||
scope_type=payload.scope_type,
|
||||
scope_id=payload.scope_id,
|
||||
)
|
||||
definition = TemplateDefinition(
|
||||
tenant_id=principal.tenant_id,
|
||||
scope_type=payload.scope_type,
|
||||
scope_id=payload.scope_id,
|
||||
name=payload.name.strip(),
|
||||
slug=slug,
|
||||
description=_text(payload.description),
|
||||
template_type=payload.template_type,
|
||||
status="draft",
|
||||
current_revision_id="pending",
|
||||
current_revision=1,
|
||||
resource_revision=1,
|
||||
created_by_account_id=principal.account_id,
|
||||
updated_by_account_id=principal.account_id,
|
||||
metadata_={},
|
||||
)
|
||||
session.add(definition)
|
||||
session.flush()
|
||||
revision = _create_revision(
|
||||
session,
|
||||
principal,
|
||||
definition=definition,
|
||||
revision_number=1,
|
||||
payload=payload,
|
||||
)
|
||||
definition.current_revision_id = revision.id
|
||||
session.flush()
|
||||
return definition, revision
|
||||
|
||||
|
||||
def update_template(
|
||||
session: Session,
|
||||
principal: ApiPrincipal,
|
||||
definition: TemplateDefinition,
|
||||
payload: TemplateUpdateRequest,
|
||||
) -> tuple[TemplateDefinition, TemplateRevision]:
|
||||
_ensure_mutable_scope(principal, definition)
|
||||
_ensure_requested_scope(principal, payload.scope_type, payload.scope_id)
|
||||
slug = _slug(payload.slug or payload.name)
|
||||
_ensure_unique_slug(
|
||||
session,
|
||||
principal,
|
||||
slug=slug,
|
||||
scope_type=payload.scope_type,
|
||||
scope_id=payload.scope_id,
|
||||
exclude_id=definition.id,
|
||||
)
|
||||
next_resource_revision = claim_revision(
|
||||
session,
|
||||
model=TemplateDefinition,
|
||||
filters=(
|
||||
TemplateDefinition.id == definition.id,
|
||||
TemplateDefinition.tenant_id == definition.tenant_id,
|
||||
TemplateDefinition.deleted_at.is_(None),
|
||||
),
|
||||
revision_attribute="resource_revision",
|
||||
expected_revision=payload.base_revision,
|
||||
resource_type="template_definition",
|
||||
resource_id=definition.id,
|
||||
refresh_path=f"/api/v1/templates/{definition.id}",
|
||||
)
|
||||
definition.resource_revision = next_resource_revision
|
||||
definition.scope_type = payload.scope_type
|
||||
definition.scope_id = payload.scope_id
|
||||
definition.name = payload.name.strip()
|
||||
definition.slug = slug
|
||||
definition.description = _text(payload.description)
|
||||
definition.template_type = payload.template_type
|
||||
definition.current_revision += 1
|
||||
definition.current_revision_id = "pending"
|
||||
definition.status = "draft"
|
||||
definition.updated_by_account_id = principal.account_id
|
||||
revision = _create_revision(
|
||||
session,
|
||||
principal,
|
||||
definition=definition,
|
||||
revision_number=definition.current_revision,
|
||||
payload=payload,
|
||||
)
|
||||
definition.current_revision_id = revision.id
|
||||
session.flush()
|
||||
return definition, revision
|
||||
|
||||
|
||||
def publish_template(
|
||||
session: Session,
|
||||
principal: ApiPrincipal,
|
||||
definition: TemplateDefinition,
|
||||
*,
|
||||
revision: int | None,
|
||||
base_revision: int,
|
||||
) -> tuple[TemplateDefinition, TemplateRevision]:
|
||||
_ensure_mutable_scope(principal, definition)
|
||||
next_resource_revision = claim_revision(
|
||||
session,
|
||||
model=TemplateDefinition,
|
||||
filters=(
|
||||
TemplateDefinition.id == definition.id,
|
||||
TemplateDefinition.tenant_id == definition.tenant_id,
|
||||
TemplateDefinition.deleted_at.is_(None),
|
||||
),
|
||||
revision_attribute="resource_revision",
|
||||
expected_revision=base_revision,
|
||||
resource_type="template_definition",
|
||||
resource_id=definition.id,
|
||||
refresh_path=f"/api/v1/templates/{definition.id}",
|
||||
)
|
||||
item_revision = get_template_revision(session, definition, revision=revision)
|
||||
now = utc_now()
|
||||
item_revision.published_at = now
|
||||
item_revision.published_by_account_id = principal.account_id
|
||||
definition.published_revision_id = item_revision.id
|
||||
definition.status = "active"
|
||||
definition.resource_revision = next_resource_revision
|
||||
definition.updated_by_account_id = principal.account_id
|
||||
session.add(item_revision)
|
||||
session.add(definition)
|
||||
session.flush()
|
||||
return definition, item_revision
|
||||
|
||||
|
||||
def delete_template(
|
||||
session: Session,
|
||||
principal: ApiPrincipal,
|
||||
definition: TemplateDefinition,
|
||||
*,
|
||||
base_revision: int,
|
||||
) -> TemplateDefinition:
|
||||
_ensure_mutable_scope(principal, definition)
|
||||
next_resource_revision = claim_revision(
|
||||
session,
|
||||
model=TemplateDefinition,
|
||||
filters=(
|
||||
TemplateDefinition.id == definition.id,
|
||||
TemplateDefinition.tenant_id == definition.tenant_id,
|
||||
TemplateDefinition.deleted_at.is_(None),
|
||||
),
|
||||
revision_attribute="resource_revision",
|
||||
expected_revision=base_revision,
|
||||
resource_type="template_definition",
|
||||
resource_id=definition.id,
|
||||
refresh_path="/api/v1/templates",
|
||||
)
|
||||
definition.resource_revision = next_resource_revision
|
||||
definition.status = "deleted"
|
||||
definition.deleted_at = utc_now()
|
||||
definition.updated_by_account_id = principal.account_id
|
||||
session.flush()
|
||||
return definition
|
||||
|
||||
|
||||
def compatibility(
|
||||
revision: TemplateRevision,
|
||||
*,
|
||||
usage: str | None,
|
||||
output_format: str | None,
|
||||
available_fields: Mapping[str, str] | Sequence[str],
|
||||
) -> TemplateCompatibility:
|
||||
field_types = (
|
||||
{str(key): str(value) for key, value in available_fields.items()}
|
||||
if isinstance(available_fields, Mapping)
|
||||
else {str(item): "unknown" for item in available_fields}
|
||||
)
|
||||
normalized_usage = usage.strip().lower() if usage else None
|
||||
diagnostics: list[dict[str, object]] = []
|
||||
missing: list[str] = []
|
||||
incompatible: list[str] = []
|
||||
if normalized_usage and normalized_usage not in revision.usages:
|
||||
diagnostics.append(
|
||||
{
|
||||
"code": "template.usage_incompatible",
|
||||
"severity": "error",
|
||||
"message": f"This template is not published for {normalized_usage}.",
|
||||
}
|
||||
)
|
||||
formats = {str(item.get("output_format")) for item in revision.output_profiles}
|
||||
if output_format and output_format not in formats:
|
||||
diagnostics.append(
|
||||
{
|
||||
"code": "template.output_format_incompatible",
|
||||
"severity": "error",
|
||||
"message": f"This template does not provide {output_format} output.",
|
||||
}
|
||||
)
|
||||
for requirement in revision.required_fields:
|
||||
path = str(requirement.get("path") or "")
|
||||
if not path or not bool(requirement.get("required", True)):
|
||||
continue
|
||||
if path not in field_types:
|
||||
missing.append(path)
|
||||
continue
|
||||
actual = field_types[path]
|
||||
expected = str(requirement.get("value_type") or "string")
|
||||
if actual not in {"unknown", expected} and not (
|
||||
expected == "number" and actual in {"integer", "number"}
|
||||
):
|
||||
incompatible.append(path)
|
||||
if missing:
|
||||
diagnostics.append(
|
||||
{
|
||||
"code": "template.required_fields_missing",
|
||||
"severity": "error",
|
||||
"message": f"Missing required fields: {', '.join(sorted(missing))}.",
|
||||
}
|
||||
)
|
||||
if incompatible:
|
||||
diagnostics.append(
|
||||
{
|
||||
"code": "template.field_types_incompatible",
|
||||
"severity": "error",
|
||||
"message": f"Fields have incompatible types: {', '.join(sorted(incompatible))}.",
|
||||
}
|
||||
)
|
||||
return TemplateCompatibility(
|
||||
compatible=not diagnostics,
|
||||
template_id=revision.template_id,
|
||||
revision_id=revision.id,
|
||||
usage=normalized_usage,
|
||||
output_format=output_format,
|
||||
missing_fields=tuple(sorted(missing)),
|
||||
incompatible_fields=tuple(sorted(incompatible)),
|
||||
diagnostics=tuple(diagnostics),
|
||||
)
|
||||
|
||||
|
||||
def template_ref(
|
||||
definition: TemplateDefinition,
|
||||
revision: TemplateRevision,
|
||||
*,
|
||||
read_only: bool = False,
|
||||
) -> TemplateRef:
|
||||
return TemplateRef(
|
||||
id=definition.id,
|
||||
tenant_id=definition.tenant_id,
|
||||
name=definition.name,
|
||||
slug=definition.slug,
|
||||
template_type=definition.template_type, # type: ignore[arg-type]
|
||||
status=definition.status,
|
||||
current_revision=definition.current_revision,
|
||||
current_revision_id=definition.current_revision_id,
|
||||
published_revision_id=definition.published_revision_id,
|
||||
description=definition.description,
|
||||
scope_type=definition.scope_type,
|
||||
scope_id=definition.scope_id,
|
||||
read_only=read_only,
|
||||
updated_at=definition.updated_at,
|
||||
revision=revision_ref(revision),
|
||||
metadata=dict(definition.metadata_ or {}),
|
||||
)
|
||||
|
||||
|
||||
def revision_ref(revision: TemplateRevision) -> TemplateRevisionRef:
|
||||
return TemplateRevisionRef(
|
||||
id=revision.id,
|
||||
template_id=revision.template_id,
|
||||
revision=revision.revision,
|
||||
definition_hash=revision.definition_hash,
|
||||
template_type=revision.template_type, # type: ignore[arg-type]
|
||||
usages=tuple(revision.usages or []),
|
||||
locale=revision.locale,
|
||||
required_fields=tuple(
|
||||
TemplateFieldRequirement(**item) for item in revision.required_fields
|
||||
),
|
||||
output_profiles=tuple(
|
||||
TemplateOutputProfile(**item) for item in revision.output_profiles
|
||||
),
|
||||
published_at=revision.published_at,
|
||||
provenance={
|
||||
"module": "templates",
|
||||
"created_by_account_id": revision.created_by_account_id,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def referenced_fields(revision: TemplateRevision) -> tuple[str, ...]:
|
||||
content = f"{revision.content_text or ''}\n{revision.content_html or ''}"
|
||||
return tuple(sorted(set(_TOKEN_PATTERN.findall(content))))
|
||||
|
||||
|
||||
def _create_revision(
|
||||
session: Session,
|
||||
principal: ApiPrincipal,
|
||||
*,
|
||||
definition: TemplateDefinition,
|
||||
revision_number: int,
|
||||
payload: TemplateRevisionPayload,
|
||||
) -> TemplateRevision:
|
||||
output_profiles = [item.model_dump(mode="json") for item in payload.output_profiles]
|
||||
if not output_profiles:
|
||||
output_profiles = _default_output_profiles(payload.template_type)
|
||||
sanitized_html = sanitize_template_html(payload.content_html)
|
||||
definition_payload = {
|
||||
"template_type": payload.template_type,
|
||||
"usages": payload.usages,
|
||||
"locale": payload.locale,
|
||||
"required_fields": [item.model_dump(mode="json") for item in payload.required_fields],
|
||||
"output_profiles": output_profiles,
|
||||
"content_text": payload.content_text,
|
||||
"content_html": sanitized_html,
|
||||
"layout": payload.layout,
|
||||
"metadata": payload.metadata,
|
||||
}
|
||||
revision = TemplateRevision(
|
||||
tenant_id=principal.tenant_id,
|
||||
template_id=definition.id,
|
||||
revision=revision_number,
|
||||
definition_hash=_canonical_hash(definition_payload),
|
||||
template_type=payload.template_type,
|
||||
usages=list(payload.usages),
|
||||
locale=payload.locale,
|
||||
required_fields=definition_payload["required_fields"],
|
||||
output_profiles=output_profiles,
|
||||
content_text=payload.content_text,
|
||||
content_html=sanitized_html,
|
||||
layout=dict(payload.layout),
|
||||
metadata_=dict(payload.metadata),
|
||||
created_by_account_id=principal.account_id,
|
||||
)
|
||||
session.add(revision)
|
||||
session.flush()
|
||||
return revision
|
||||
|
||||
|
||||
def _default_output_profiles(template_type: str) -> list[dict[str, object]]:
|
||||
media = "A4"
|
||||
if template_type == "envelope":
|
||||
media = "DL"
|
||||
return [
|
||||
{
|
||||
"id": "print-html",
|
||||
"label": "Printable HTML",
|
||||
"output_format": "html",
|
||||
"media_type": "text/html",
|
||||
"channel": "print",
|
||||
"capabilities": ["browser_print", template_type],
|
||||
"page": {"size": media},
|
||||
},
|
||||
{
|
||||
"id": "plain-text",
|
||||
"label": "Plain text",
|
||||
"output_format": "text",
|
||||
"media_type": "text/plain",
|
||||
"channel": "download",
|
||||
"capabilities": ["deterministic_text"],
|
||||
"page": {},
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def _ensure_unique_slug(
|
||||
session: Session,
|
||||
principal: ApiPrincipal,
|
||||
*,
|
||||
slug: str,
|
||||
scope_type: str,
|
||||
scope_id: str | None,
|
||||
exclude_id: str | None = None,
|
||||
) -> None:
|
||||
statement = select(TemplateDefinition.id).where(
|
||||
TemplateDefinition.tenant_id == principal.tenant_id,
|
||||
TemplateDefinition.scope_type == scope_type,
|
||||
TemplateDefinition.slug == slug,
|
||||
TemplateDefinition.deleted_at.is_(None),
|
||||
)
|
||||
if scope_id is None:
|
||||
statement = statement.where(TemplateDefinition.scope_id.is_(None))
|
||||
else:
|
||||
statement = statement.where(TemplateDefinition.scope_id == scope_id)
|
||||
if exclude_id:
|
||||
statement = statement.where(TemplateDefinition.id != exclude_id)
|
||||
if session.scalar(statement) is not None:
|
||||
raise TemplateCompatibilityError("A template with this slug already exists in the selected scope.")
|
||||
|
||||
|
||||
def _ensure_requested_scope(
|
||||
principal: ApiPrincipal,
|
||||
scope_type: str,
|
||||
scope_id: str | None,
|
||||
) -> None:
|
||||
if principal.has(ADMIN_SCOPE):
|
||||
return
|
||||
if scope_type == "tenant":
|
||||
return
|
||||
if scope_type == "user" and scope_id == principal.account_id:
|
||||
return
|
||||
if scope_type == "group" and scope_id in principal.group_ids:
|
||||
return
|
||||
raise TemplateCompatibilityError("The requested template scope is not writable by this principal.")
|
||||
|
||||
|
||||
def _ensure_mutable_scope(principal: ApiPrincipal, definition: TemplateDefinition) -> None:
|
||||
_ensure_requested_scope(principal, definition.scope_type, definition.scope_id)
|
||||
|
||||
|
||||
def _slug(value: str) -> str:
|
||||
normalized = re.sub(r"[^a-z0-9_.-]+", "-", value.strip().lower()).strip("-.")
|
||||
if not normalized:
|
||||
raise TemplateCompatibilityError("Template slug cannot be empty.")
|
||||
return normalized[:160]
|
||||
|
||||
|
||||
def _text(value: str | None) -> str | None:
|
||||
normalized = str(value or "").strip()
|
||||
return normalized or None
|
||||
|
||||
|
||||
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()
|
||||
|
||||
|
||||
class _TemplateHtmlSanitizer(HTMLParser):
|
||||
allowed_tags = {
|
||||
"a", "blockquote", "br", "code", "div", "em", "h1", "h2", "h3",
|
||||
"h4", "h5", "h6", "hr", "li", "ol", "p", "pre", "span", "strong",
|
||||
"table", "tbody", "td", "th", "thead", "tr", "u", "ul",
|
||||
}
|
||||
void_tags = {"br", "hr"}
|
||||
blocked_tags = {"script", "style", "iframe", "object", "embed", "svg", "math"}
|
||||
allowed_attrs = {"a": {"href", "title"}, "td": {"colspan", "rowspan"}, "th": {"colspan", "rowspan"}}
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__(convert_charrefs=True)
|
||||
self.parts: list[str] = []
|
||||
self.blocked_depth = 0
|
||||
|
||||
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
||||
tag = tag.lower()
|
||||
if tag in self.blocked_tags:
|
||||
self.blocked_depth += 1
|
||||
return
|
||||
if self.blocked_depth or tag not in self.allowed_tags:
|
||||
return
|
||||
clean_attrs: list[str] = []
|
||||
for name, raw_value in attrs:
|
||||
name = name.lower()
|
||||
value = str(raw_value or "")
|
||||
if name not in self.allowed_attrs.get(tag, set()):
|
||||
continue
|
||||
if name == "href" and not _safe_href(value):
|
||||
continue
|
||||
clean_attrs.append(f' {name}="{escape(value, quote=True)}"')
|
||||
self.parts.append(f"<{tag}{''.join(clean_attrs)}>")
|
||||
|
||||
def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
||||
self.handle_starttag(tag, attrs)
|
||||
|
||||
def handle_endtag(self, tag: str) -> None:
|
||||
tag = tag.lower()
|
||||
if tag in self.blocked_tags:
|
||||
self.blocked_depth = max(0, self.blocked_depth - 1)
|
||||
return
|
||||
if not self.blocked_depth and tag in self.allowed_tags and tag not in self.void_tags:
|
||||
self.parts.append(f"</{tag}>")
|
||||
|
||||
def handle_data(self, data: str) -> None:
|
||||
if not self.blocked_depth:
|
||||
self.parts.append(escape(data, quote=False))
|
||||
|
||||
|
||||
def sanitize_template_html(value: str | None) -> str | None:
|
||||
if not value or not value.strip():
|
||||
return None
|
||||
parser = _TemplateHtmlSanitizer()
|
||||
parser.feed(value)
|
||||
parser.close()
|
||||
return "".join(parser.parts).strip() or None
|
||||
|
||||
|
||||
def _safe_href(value: str) -> bool:
|
||||
normalized = value.strip().lower()
|
||||
return normalized.startswith(("https://", "http://", "mailto:", "#", "/"))
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ADMIN_SCOPE",
|
||||
"PUBLISH_SCOPE",
|
||||
"READ_SCOPE",
|
||||
"RENDER_SCOPE",
|
||||
"WRITE_SCOPE",
|
||||
"compatibility",
|
||||
"create_template",
|
||||
"delete_template",
|
||||
"get_template",
|
||||
"get_template_revision",
|
||||
"list_template_revisions",
|
||||
"list_templates",
|
||||
"publish_template",
|
||||
"referenced_fields",
|
||||
"revision_ref",
|
||||
"sanitize_template_html",
|
||||
"template_ref",
|
||||
"update_template",
|
||||
]
|
||||
Reference in New Issue
Block a user