148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
"""
|
|
Template service helpers extracted from API layer for document template and version operations.
|
|
|
|
These functions centralize database lookups, validation, storage interactions, and
|
|
preview/download resolution so that API endpoints remain thin.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional, List, Tuple, Dict, Any
|
|
import os
|
|
import hashlib
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.templates import DocumentTemplate, DocumentTemplateVersion, TemplateKeyword
|
|
from app.services.storage import get_default_storage
|
|
from app.services.template_merge import extract_tokens_from_bytes, build_context, resolve_tokens, render_docx
|
|
|
|
|
|
def get_template_or_404(db: Session, template_id: int) -> DocumentTemplate:
|
|
tpl = db.query(DocumentTemplate).filter(DocumentTemplate.id == template_id).first()
|
|
if not tpl:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template not found")
|
|
return tpl
|
|
|
|
|
|
def list_template_versions(db: Session, template_id: int) -> List[DocumentTemplateVersion]:
|
|
_ = get_template_or_404(db, template_id)
|
|
return (
|
|
db.query(DocumentTemplateVersion)
|
|
.filter(DocumentTemplateVersion.template_id == template_id)
|
|
.order_by(DocumentTemplateVersion.created_at.desc())
|
|
.all()
|
|
)
|
|
|
|
|
|
def add_template_version(
|
|
db: Session,
|
|
*,
|
|
template_id: int,
|
|
semantic_version: str,
|
|
changelog: Optional[str],
|
|
approve: bool,
|
|
content: bytes,
|
|
filename_hint: str,
|
|
content_type: Optional[str],
|
|
created_by: Optional[str],
|
|
) -> DocumentTemplateVersion:
|
|
tpl = get_template_or_404(db, template_id)
|
|
if not content:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No file uploaded")
|
|
|
|
sha256 = hashlib.sha256(content).hexdigest()
|
|
storage = get_default_storage()
|
|
storage_path = storage.save_bytes(content=content, filename_hint=filename_hint or "template.bin", subdir="templates")
|
|
|
|
version = DocumentTemplateVersion(
|
|
template_id=template_id,
|
|
semantic_version=semantic_version,
|
|
storage_path=storage_path,
|
|
mime_type=content_type,
|
|
size=len(content),
|
|
checksum=sha256,
|
|
changelog=changelog,
|
|
created_by=created_by,
|
|
is_approved=bool(approve),
|
|
)
|
|
db.add(version)
|
|
db.flush()
|
|
if approve:
|
|
tpl.current_version_id = version.id
|
|
db.commit()
|
|
return version
|
|
|
|
|
|
def resolve_template_preview(
|
|
db: Session,
|
|
*,
|
|
template_id: int,
|
|
version_id: Optional[int],
|
|
context: Dict[str, Any],
|
|
) -> Tuple[Dict[str, Any], List[str], bytes, str]:
|
|
tpl = get_template_or_404(db, template_id)
|
|
resolved_version_id = version_id or tpl.current_version_id
|
|
if not resolved_version_id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Template has no versions")
|
|
|
|
ver = (
|
|
db.query(DocumentTemplateVersion)
|
|
.filter(DocumentTemplateVersion.id == resolved_version_id)
|
|
.first()
|
|
)
|
|
if not ver:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Version not found")
|
|
|
|
storage = get_default_storage()
|
|
content = storage.open_bytes(ver.storage_path)
|
|
tokens = extract_tokens_from_bytes(content)
|
|
built_context = build_context(context or {}, "template", str(template_id))
|
|
resolved, unresolved = resolve_tokens(db, tokens, built_context)
|
|
|
|
output_bytes = content
|
|
output_mime = ver.mime_type
|
|
if ver.mime_type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
|
|
output_bytes = render_docx(content, resolved)
|
|
output_mime = ver.mime_type
|
|
|
|
return resolved, unresolved, output_bytes, output_mime
|
|
|
|
|
|
def get_download_payload(
|
|
db: Session,
|
|
*,
|
|
template_id: int,
|
|
version_id: Optional[int],
|
|
) -> Tuple[bytes, str, str]:
|
|
tpl = get_template_or_404(db, template_id)
|
|
resolved_version_id = version_id or tpl.current_version_id
|
|
if not resolved_version_id:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Template has no approved version")
|
|
|
|
ver = (
|
|
db.query(DocumentTemplateVersion)
|
|
.filter(
|
|
DocumentTemplateVersion.id == resolved_version_id,
|
|
DocumentTemplateVersion.template_id == tpl.id,
|
|
)
|
|
.first()
|
|
)
|
|
if not ver:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Version not found")
|
|
|
|
storage = get_default_storage()
|
|
try:
|
|
content = storage.open_bytes(ver.storage_path)
|
|
except Exception:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Stored file not found")
|
|
|
|
base = os.path.basename(ver.storage_path)
|
|
if "_" in base:
|
|
original_name = base.split("_", 1)[1]
|
|
else:
|
|
original_name = base
|
|
return content, ver.mime_type, original_name
|
|
|
|
|