113 lines
3.5 KiB
Python
113 lines
3.5 KiB
Python
"""
|
|
Template variable resolution and DOCX preview using docxtpl.
|
|
|
|
MVP features:
|
|
- Resolve variables from explicit context, FormVariable, ReportVariable
|
|
- Built-in variables (dates)
|
|
- Render DOCX using docxtpl when mime_type is docx; otherwise return bytes as-is
|
|
- Return unresolved tokens list
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import re
|
|
from datetime import date, datetime
|
|
from typing import Any, Dict, List, Tuple
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.additional import FormVariable, ReportVariable
|
|
|
|
try:
|
|
from docxtpl import DocxTemplate
|
|
DOCXTPL_AVAILABLE = True
|
|
except Exception:
|
|
DOCXTPL_AVAILABLE = False
|
|
|
|
|
|
TOKEN_PATTERN = re.compile(r"\{\{\s*([a-zA-Z0-9_\.]+)\s*\}\}")
|
|
|
|
|
|
def extract_tokens_from_bytes(content: bytes) -> List[str]:
|
|
# Prefer docxtpl-based extraction for DOCX if available
|
|
if DOCXTPL_AVAILABLE:
|
|
try:
|
|
buf = io.BytesIO(content)
|
|
tpl = DocxTemplate(buf)
|
|
# jinja2 analysis for undeclared template variables
|
|
vars_set = tpl.get_undeclared_template_variables({})
|
|
return sorted({str(v) for v in vars_set})
|
|
except Exception:
|
|
pass
|
|
# Fallback: naive regex over decoded text
|
|
try:
|
|
text = content.decode("utf-8", errors="ignore")
|
|
except Exception:
|
|
text = ""
|
|
return sorted({m.group(1) for m in TOKEN_PATTERN.finditer(text)})
|
|
|
|
|
|
def build_context(payload_context: Dict[str, Any]) -> Dict[str, Any]:
|
|
# Built-ins
|
|
today = date.today()
|
|
builtins = {
|
|
"TODAY": today.strftime("%B %d, %Y"),
|
|
"TODAY_ISO": today.isoformat(),
|
|
"NOW": datetime.utcnow().isoformat() + "Z",
|
|
}
|
|
merged = {**builtins}
|
|
# Normalize keys to support both FOO and foo
|
|
for k, v in payload_context.items():
|
|
merged[k] = v
|
|
if isinstance(k, str):
|
|
merged.setdefault(k.upper(), v)
|
|
return merged
|
|
|
|
|
|
def _safe_lookup_variable(db: Session, identifier: str) -> Any:
|
|
# 1) FormVariable
|
|
fv = db.query(FormVariable).filter(FormVariable.identifier == identifier, FormVariable.active == 1).first()
|
|
if fv:
|
|
# MVP: use static response if present; otherwise treat as unresolved
|
|
if fv.response is not None:
|
|
return fv.response
|
|
return None
|
|
# 2) ReportVariable
|
|
rv = db.query(ReportVariable).filter(ReportVariable.identifier == identifier, ReportVariable.active == 1).first()
|
|
if rv:
|
|
# MVP: no evaluation yet; unresolved
|
|
return None
|
|
return None
|
|
|
|
|
|
def resolve_tokens(db: Session, tokens: List[str], context: Dict[str, Any]) -> Tuple[Dict[str, Any], List[str]]:
|
|
resolved: Dict[str, Any] = {}
|
|
unresolved: List[str] = []
|
|
for tok in tokens:
|
|
# Order: payload context (case-insensitive via upper) -> FormVariable -> ReportVariable
|
|
value = context.get(tok)
|
|
if value is None:
|
|
value = context.get(tok.upper())
|
|
if value is None:
|
|
value = _safe_lookup_variable(db, tok)
|
|
if value is None:
|
|
unresolved.append(tok)
|
|
else:
|
|
resolved[tok] = value
|
|
return resolved, unresolved
|
|
|
|
|
|
def render_docx(docx_bytes: bytes, context: Dict[str, Any]) -> bytes:
|
|
if not DOCXTPL_AVAILABLE:
|
|
# Return original bytes if docxtpl is not installed
|
|
return docx_bytes
|
|
# Write to BytesIO for docxtpl
|
|
in_buffer = io.BytesIO(docx_bytes)
|
|
tpl = DocxTemplate(in_buffer)
|
|
tpl.render(context)
|
|
out_buffer = io.BytesIO()
|
|
tpl.save(out_buffer)
|
|
return out_buffer.getvalue()
|
|
|
|
|