load project
This commit is contained in:
0
backend/app/services/__init__.py
Normal file
0
backend/app/services/__init__.py
Normal file
66
backend/app/services/access.py
Normal file
66
backend/app/services/access.py
Normal file
@@ -0,0 +1,66 @@
|
||||
from sqlalchemy import or_
|
||||
from sqlalchemy.orm import Query, Session
|
||||
|
||||
from app.models.document import FilledDocument
|
||||
from app.models.template import DocumentTemplate
|
||||
from app.models.user import User, UserRole
|
||||
|
||||
|
||||
def is_admin(user: User) -> bool:
|
||||
return user.role == UserRole.admin
|
||||
|
||||
|
||||
def can_access_template(user: User, template: DocumentTemplate) -> bool:
|
||||
return (
|
||||
is_admin(user)
|
||||
or template.owner_id == user.id
|
||||
or template.is_public
|
||||
)
|
||||
|
||||
|
||||
def can_manage_template(user: User, template: DocumentTemplate) -> bool:
|
||||
return is_admin(user) or template.owner_id == user.id
|
||||
|
||||
|
||||
def can_access_document(user: User, document: FilledDocument) -> bool:
|
||||
return is_admin(user) or document.owner_id == user.id
|
||||
|
||||
|
||||
def accessible_templates_query(db: Session, user: User) -> Query:
|
||||
return db.query(DocumentTemplate).filter(
|
||||
or_(
|
||||
DocumentTemplate.owner_id == user.id,
|
||||
DocumentTemplate.is_public == True,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def accessible_template(
|
||||
db: Session, user: User, template_id: int
|
||||
) -> DocumentTemplate | None:
|
||||
template = db.query(DocumentTemplate).filter(
|
||||
DocumentTemplate.id == template_id
|
||||
).first()
|
||||
if template and can_access_template(user, template):
|
||||
return template
|
||||
return None
|
||||
|
||||
|
||||
def manageable_template(
|
||||
db: Session, user: User, template_id: int
|
||||
) -> DocumentTemplate | None:
|
||||
template = db.query(DocumentTemplate).filter(
|
||||
DocumentTemplate.id == template_id
|
||||
).first()
|
||||
if template and can_manage_template(user, template):
|
||||
return template
|
||||
return None
|
||||
|
||||
|
||||
def accessible_document(
|
||||
db: Session, user: User, document_id: int
|
||||
) -> FilledDocument | None:
|
||||
doc = db.query(FilledDocument).filter(FilledDocument.id == document_id).first()
|
||||
if doc and can_access_document(user, doc):
|
||||
return doc
|
||||
return None
|
||||
377
backend/app/services/docx_parser.py
Normal file
377
backend/app/services/docx_parser.py
Normal file
@@ -0,0 +1,377 @@
|
||||
import json
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
from docx import Document
|
||||
from docx.enum.text import WD_ALIGN_PARAGRAPH
|
||||
from docx.oxml.ns import qn
|
||||
from docx.shared import Pt
|
||||
from docx.table import Table
|
||||
from docx.text.paragraph import Paragraph
|
||||
from jinja2 import Environment, meta
|
||||
|
||||
|
||||
JINJA_VAR_PATTERN = re.compile(r"\{\{[^{}]*?\}\}")
|
||||
JINJA_BLOCK_PATTERN = re.compile(r"\{%[^{}]*?%\}")
|
||||
TABLE_ROW_LOOP_PATTERN = re.compile(
|
||||
r"\{%\s*tr\s+for\s+(\w+)\s+in\s+(\w+)\s*%\}", re.IGNORECASE
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParsedVariable:
|
||||
name: str
|
||||
label: str
|
||||
field_type: str
|
||||
order: int
|
||||
parent_variable: str | None = None
|
||||
style_params: dict[str, Any] = field(default_factory=dict)
|
||||
default_value: str | None = None
|
||||
is_required: bool = True
|
||||
|
||||
|
||||
def _extract_run_style(run) -> dict[str, Any]:
|
||||
style: dict[str, Any] = {}
|
||||
if run.font.name:
|
||||
style["font_name"] = run.font.name
|
||||
if run.font.size:
|
||||
style["font_size"] = run.font.size.pt
|
||||
if run.bold:
|
||||
style["bold"] = True
|
||||
if run.italic:
|
||||
style["italic"] = True
|
||||
if run.underline:
|
||||
style["underline"] = True
|
||||
if run.font.color and run.font.color.rgb:
|
||||
style["color"] = str(run.font.color.rgb)
|
||||
return style
|
||||
|
||||
|
||||
def _alignment_to_str(alignment) -> str | None:
|
||||
if alignment is None:
|
||||
return None
|
||||
mapping = {
|
||||
WD_ALIGN_PARAGRAPH.LEFT: "left",
|
||||
WD_ALIGN_PARAGRAPH.CENTER: "center",
|
||||
WD_ALIGN_PARAGRAPH.RIGHT: "right",
|
||||
WD_ALIGN_PARAGRAPH.JUSTIFY: "justify",
|
||||
}
|
||||
return mapping.get(alignment)
|
||||
|
||||
|
||||
def _get_paragraph_alignment_str(paragraph: Paragraph) -> str | None:
|
||||
"""Read paragraph alignment, including Word values like 'start'/'end'."""
|
||||
try:
|
||||
alignment = paragraph.alignment
|
||||
if alignment is not None:
|
||||
return _alignment_to_str(alignment)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
p_pr = paragraph._p.pPr
|
||||
if p_pr is not None:
|
||||
jc = p_pr.find(qn("w:jc"))
|
||||
if jc is not None:
|
||||
val = jc.get(qn("w:val"))
|
||||
xml_map = {
|
||||
"left": "left",
|
||||
"right": "right",
|
||||
"center": "center",
|
||||
"both": "justify",
|
||||
"justify": "justify",
|
||||
"start": "left",
|
||||
"end": "right",
|
||||
"distribute": "justify",
|
||||
}
|
||||
return xml_map.get(val)
|
||||
return None
|
||||
|
||||
|
||||
def _extract_paragraph_style(paragraph: Paragraph) -> dict[str, Any]:
|
||||
style: dict[str, Any] = {}
|
||||
alignment = _get_paragraph_alignment_str(paragraph)
|
||||
if alignment is not None:
|
||||
style["alignment"] = alignment
|
||||
if paragraph.runs:
|
||||
run_style = _extract_run_style(paragraph.runs[0])
|
||||
style.update(run_style)
|
||||
return style
|
||||
|
||||
|
||||
def _extract_cell_style(cell) -> dict[str, Any]:
|
||||
style: dict[str, Any] = {}
|
||||
if cell.paragraphs:
|
||||
style.update(_extract_paragraph_style(cell.paragraphs[0]))
|
||||
if cell.width:
|
||||
style["width"] = cell.width.pt if hasattr(cell.width, "pt") else None
|
||||
|
||||
tc = cell._tc
|
||||
tc_pr = tc.tcPr
|
||||
if tc_pr is not None:
|
||||
shd = tc_pr.find(qn("w:shd"))
|
||||
if shd is not None and shd.get(qn("w:fill")):
|
||||
style["background_color"] = shd.get(qn("w:fill"))
|
||||
return style
|
||||
|
||||
|
||||
def _extract_table_style(table: Table) -> dict[str, Any]:
|
||||
style: dict[str, Any] = {"rows": len(table.rows), "cols": len(table.columns)}
|
||||
if table.rows:
|
||||
style["row_styles"] = [
|
||||
[_extract_cell_style(cell) for cell in row.cells]
|
||||
for row in table.rows
|
||||
]
|
||||
return style
|
||||
|
||||
|
||||
def _get_text_from_element(element) -> str:
|
||||
if isinstance(element, Paragraph):
|
||||
return element.text
|
||||
if isinstance(element, Table):
|
||||
parts = []
|
||||
for row in element.rows:
|
||||
for cell in row.cells:
|
||||
for p in cell.paragraphs:
|
||||
parts.append(p.text)
|
||||
return "\n".join(parts)
|
||||
return ""
|
||||
|
||||
|
||||
DOTTED_VAR_PATTERN = re.compile(r"\{\{\s*(\w+)\.(\w+)\s*\}\}")
|
||||
|
||||
|
||||
def _find_dotted_variables_in_text(
|
||||
text: str, table_loops: dict[str, str]
|
||||
) -> list[tuple[str, str, str]]:
|
||||
"""Return list of (field_name, parent_list_var, loop_item_var)."""
|
||||
results: list[tuple[str, str, str]] = []
|
||||
for match in DOTTED_VAR_PATTERN.finditer(text):
|
||||
item_var, field_name = match.groups()
|
||||
parent = table_loops.get(item_var)
|
||||
if parent:
|
||||
results.append((field_name, parent, item_var))
|
||||
return results
|
||||
|
||||
|
||||
MALFORMED_VAR_PATTERN = re.compile(r"\{\{\s*(\w+)\s*\}\}?")
|
||||
|
||||
|
||||
def _find_jinja_variables_in_text(text: str) -> list[str]:
|
||||
names: set[str] = set()
|
||||
env = Environment()
|
||||
try:
|
||||
ast = env.parse(text)
|
||||
names.update(meta.find_undeclared_variables(ast))
|
||||
except Exception:
|
||||
pass
|
||||
names.update(re.findall(r"\{\{\s*(\w+)\s*\}\}", text))
|
||||
names.update(MALFORMED_VAR_PATTERN.findall(text))
|
||||
return sorted(names)
|
||||
|
||||
|
||||
def _detect_field_type(var_name: str, text: str, in_table: bool, is_loop_item: bool) -> str:
|
||||
if is_loop_item:
|
||||
return "table_cell"
|
||||
if in_table:
|
||||
return "table_cell"
|
||||
lower_text = text.lower()
|
||||
if any(kw in lower_text for kw in ["description", "comment", "notes", "textarea"]):
|
||||
return "textarea"
|
||||
if any(kw in var_name.lower() for kw in ["date", "дата"]):
|
||||
return "date"
|
||||
if any(kw in var_name.lower() for kw in ["amount", "price", "sum", "qty", "count", "number"]):
|
||||
return "number"
|
||||
return "text"
|
||||
|
||||
|
||||
def _humanize(name: str) -> str:
|
||||
return name.replace("_", " ").replace("-", " ").title()
|
||||
|
||||
|
||||
def parse_docx_template(file_path: str) -> list[ParsedVariable]:
|
||||
doc = Document(file_path)
|
||||
variables: list[ParsedVariable] = []
|
||||
seen: set[str] = set()
|
||||
order = 0
|
||||
|
||||
table_loops: dict[str, str] = {}
|
||||
full_text_parts: list[str] = []
|
||||
|
||||
for element in doc.element.body:
|
||||
tag = element.tag.split("}")[-1]
|
||||
if tag == "p":
|
||||
para = Paragraph(element, doc)
|
||||
full_text_parts.append(para.text)
|
||||
text = para.text
|
||||
for match in TABLE_ROW_LOOP_PATTERN.finditer(text):
|
||||
item_var, list_var = match.groups()
|
||||
table_loops[item_var] = list_var
|
||||
elif tag == "tbl":
|
||||
table = Table(element, doc)
|
||||
for row in table.rows:
|
||||
for cell in row.cells:
|
||||
for p in cell.paragraphs:
|
||||
full_text_parts.append(p.text)
|
||||
text = p.text
|
||||
for match in TABLE_ROW_LOOP_PATTERN.finditer(text):
|
||||
item_var, list_var = match.groups()
|
||||
table_loops[item_var] = list_var
|
||||
|
||||
full_text = "\n".join(full_text_parts)
|
||||
|
||||
dotted_fields: dict[str, tuple[str, str]] = {}
|
||||
for text_part in full_text_parts:
|
||||
for field_name, parent, item_var in _find_dotted_variables_in_text(
|
||||
text_part, table_loops
|
||||
):
|
||||
key = f"{parent}.{field_name}"
|
||||
if key not in dotted_fields:
|
||||
dotted_fields[key] = (field_name, parent)
|
||||
|
||||
for list_var in sorted(set(table_loops.values())):
|
||||
if list_var not in seen:
|
||||
seen.add(list_var)
|
||||
variables.append(
|
||||
ParsedVariable(
|
||||
name=list_var,
|
||||
label=_humanize(list_var),
|
||||
field_type="table_row",
|
||||
order=order,
|
||||
style_params={"is_repeating_table": True},
|
||||
)
|
||||
)
|
||||
order += 1
|
||||
|
||||
for key, (field_name, parent) in sorted(dotted_fields.items()):
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
table_style = _find_table_style_for_variable(doc, field_name)
|
||||
field_type = _detect_field_type(field_name, field_name, True, True)
|
||||
variables.append(
|
||||
ParsedVariable(
|
||||
name=field_name,
|
||||
label=_humanize(field_name),
|
||||
field_type=field_type,
|
||||
order=order,
|
||||
parent_variable=parent,
|
||||
style_params=table_style,
|
||||
)
|
||||
)
|
||||
order += 1
|
||||
|
||||
for item_var in table_loops:
|
||||
seen.add(item_var)
|
||||
|
||||
for element in doc.element.body:
|
||||
tag = element.tag.split("}")[-1]
|
||||
in_table = tag == "tbl"
|
||||
|
||||
if tag == "p":
|
||||
para = Paragraph(element, doc)
|
||||
text = para.text
|
||||
var_names = _find_jinja_variables_in_text(text)
|
||||
para_style = _extract_paragraph_style(para)
|
||||
for var_name in var_names:
|
||||
if var_name in seen or var_name in table_loops:
|
||||
continue
|
||||
seen.add(var_name)
|
||||
field_type = _detect_field_type(var_name, text, False, False)
|
||||
variables.append(
|
||||
ParsedVariable(
|
||||
name=var_name,
|
||||
label=_humanize(var_name),
|
||||
field_type=field_type,
|
||||
order=order,
|
||||
style_params=para_style,
|
||||
)
|
||||
)
|
||||
order += 1
|
||||
|
||||
elif tag == "tbl":
|
||||
table = Table(element, doc)
|
||||
table_style = _extract_table_style(table)
|
||||
for row in table.rows:
|
||||
for cell in row.cells:
|
||||
for p in cell.paragraphs:
|
||||
text = p.text
|
||||
var_names = _find_jinja_variables_in_text(text)
|
||||
cell_style = _extract_cell_style(cell)
|
||||
cell_style["table_style"] = table_style
|
||||
for var_name in var_names:
|
||||
if var_name in seen or var_name in table_loops:
|
||||
continue
|
||||
if any(
|
||||
f"{table_loops[iv]}.{fn}" in seen
|
||||
for iv, fn in [
|
||||
(m.group(1), m.group(2))
|
||||
for m in DOTTED_VAR_PATTERN.finditer(text)
|
||||
]
|
||||
if iv in table_loops
|
||||
):
|
||||
continue
|
||||
seen.add(var_name)
|
||||
field_type = _detect_field_type(var_name, text, True, False)
|
||||
variables.append(
|
||||
ParsedVariable(
|
||||
name=var_name,
|
||||
label=_humanize(var_name),
|
||||
field_type=field_type,
|
||||
order=order,
|
||||
style_params=cell_style,
|
||||
)
|
||||
)
|
||||
order += 1
|
||||
|
||||
env = Environment()
|
||||
try:
|
||||
ast = env.parse(full_text)
|
||||
all_vars = meta.find_undeclared_variables(ast)
|
||||
for var_name in sorted(all_vars):
|
||||
if var_name not in seen:
|
||||
seen.add(var_name)
|
||||
variables.append(
|
||||
ParsedVariable(
|
||||
name=var_name,
|
||||
label=_humanize(var_name),
|
||||
field_type="text",
|
||||
order=order,
|
||||
)
|
||||
)
|
||||
order += 1
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return variables
|
||||
|
||||
|
||||
def _find_table_style_for_variable(doc: Document, var_name: str) -> dict[str, Any]:
|
||||
for element in doc.element.body:
|
||||
tag = element.tag.split("}")[-1]
|
||||
if tag != "tbl":
|
||||
continue
|
||||
table = Table(element, doc)
|
||||
for row in table.rows:
|
||||
for cell in row.cells:
|
||||
for p in cell.paragraphs:
|
||||
if var_name in p.text or f".{var_name}" in p.text:
|
||||
cell_style = _extract_cell_style(cell)
|
||||
cell_style["table_style"] = _extract_table_style(table)
|
||||
return cell_style
|
||||
return {}
|
||||
|
||||
|
||||
def variables_to_json(variables: list[ParsedVariable]) -> list[dict[str, Any]]:
|
||||
return [
|
||||
{
|
||||
"name": v.name,
|
||||
"label": v.label,
|
||||
"field_type": v.field_type,
|
||||
"order": v.order,
|
||||
"parent_variable": v.parent_variable,
|
||||
"style_params": v.style_params,
|
||||
"default_value": v.default_value,
|
||||
"is_required": v.is_required,
|
||||
}
|
||||
for v in variables
|
||||
]
|
||||
353
backend/app/services/docx_renderer.py
Normal file
353
backend/app/services/docx_renderer.py
Normal file
@@ -0,0 +1,353 @@
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
from html import escape
|
||||
from typing import Any
|
||||
|
||||
from docx import Document
|
||||
from docx.oxml.ns import qn
|
||||
from docx.shared import Pt, RGBColor
|
||||
from docx.table import Table, _Cell
|
||||
from docxtpl import DocxTemplate
|
||||
|
||||
from app.services.docx_parser import _get_paragraph_alignment_str
|
||||
|
||||
TABLE_ROW_LOOP_PATTERN = re.compile(
|
||||
r"\{%\s*tr\s+for\s+(\w+)\s+in\s+(\w+)\s*%\}", re.IGNORECASE
|
||||
)
|
||||
TABLE_ROW_END_PATTERN = re.compile(r"\{%\s*tr\s+endfor\s*%\}", re.IGNORECASE)
|
||||
DOTTED_VAR_PATTERN = re.compile(r"\{\{\s*(\w+)\.(\w+)\s*\}\}")
|
||||
SIMPLE_VAR_PATTERN = re.compile(r"\{\{\s*(\w+)\s*\}\}")
|
||||
TOLERANT_VAR_PATTERN = re.compile(r"\{\{\s*(\w+)\s*\}\}?")
|
||||
|
||||
|
||||
def _prepare_context(field_data: dict[str, Any]) -> dict[str, Any]:
|
||||
return dict(field_data)
|
||||
|
||||
|
||||
def _clean_jinja_markers(text: str) -> str:
|
||||
text = TABLE_ROW_LOOP_PATTERN.sub("", text)
|
||||
text = TABLE_ROW_END_PATTERN.sub("", text)
|
||||
return text.strip()
|
||||
|
||||
|
||||
def _replace_variables_in_text(text: str, context: dict[str, Any]) -> str:
|
||||
def replacer(match: re.Match) -> str:
|
||||
name = match.group(1)
|
||||
value = context.get(name, "")
|
||||
return str(value) if value is not None else ""
|
||||
|
||||
text = SIMPLE_VAR_PATTERN.sub(replacer, text)
|
||||
if "{{" in text:
|
||||
text = TOLERANT_VAR_PATTERN.sub(replacer, text)
|
||||
return text
|
||||
|
||||
|
||||
def _replace_simple_vars(text: str, context: dict[str, Any]) -> str:
|
||||
return _replace_variables_in_text(text, context)
|
||||
|
||||
|
||||
def _replace_vars_in_paragraph(paragraph, context: dict[str, Any]) -> None:
|
||||
"""Replace Jinja variables while preserving run formatting."""
|
||||
if "{{" not in paragraph.text:
|
||||
return
|
||||
|
||||
full_text = "".join(run.text for run in paragraph.runs)
|
||||
new_text = _replace_variables_in_text(full_text, context)
|
||||
if new_text == full_text:
|
||||
return
|
||||
|
||||
if paragraph.runs:
|
||||
paragraph.runs[0].text = new_text
|
||||
for run in paragraph.runs[1:]:
|
||||
run.text = ""
|
||||
else:
|
||||
paragraph.add_run(new_text)
|
||||
|
||||
|
||||
def _expand_table_rows(doc: Document, context: dict[str, Any]) -> None:
|
||||
for table in doc.tables:
|
||||
rows_to_process: list[tuple[int, str, str, list[str]]] = []
|
||||
|
||||
for row_idx, row in enumerate(table.rows):
|
||||
row_text = " ".join(
|
||||
p.text for cell in row.cells for p in cell.paragraphs
|
||||
)
|
||||
loop_match = TABLE_ROW_LOOP_PATTERN.search(row_text)
|
||||
if not loop_match:
|
||||
continue
|
||||
|
||||
item_var, list_var = loop_match.groups()
|
||||
field_names: list[str] = []
|
||||
for cell in row.cells:
|
||||
for p in cell.paragraphs:
|
||||
for m in DOTTED_VAR_PATTERN.finditer(p.text):
|
||||
if m.group(1) == item_var:
|
||||
field = m.group(2)
|
||||
if field not in field_names:
|
||||
field_names.append(field)
|
||||
|
||||
rows_to_process.append((row_idx, item_var, list_var, field_names))
|
||||
|
||||
for row_idx, item_var, list_var, field_names in reversed(rows_to_process):
|
||||
template_row = table.rows[row_idx]
|
||||
items = context.get(list_var, [])
|
||||
if not isinstance(items, list):
|
||||
items = []
|
||||
|
||||
if not items:
|
||||
for cell in template_row.cells:
|
||||
for p in cell.paragraphs:
|
||||
_replace_vars_in_paragraph(p, context)
|
||||
cleaned = _clean_jinja_markers(p.text)
|
||||
if cleaned != p.text and p.runs:
|
||||
p.runs[0].text = cleaned
|
||||
for run in p.runs[1:]:
|
||||
run.text = ""
|
||||
continue
|
||||
|
||||
template_cells = [
|
||||
[_clean_jinja_markers(p.text) for p in cell.paragraphs]
|
||||
for cell in template_row.cells
|
||||
]
|
||||
|
||||
for item in items:
|
||||
new_row = table.add_row()
|
||||
for cell_idx, cell in enumerate(new_row.cells):
|
||||
if cell_idx < len(template_cells):
|
||||
template_text = (
|
||||
template_cells[cell_idx][0]
|
||||
if template_cells[cell_idx]
|
||||
else ""
|
||||
)
|
||||
result = template_text
|
||||
for field in field_names:
|
||||
placeholder = f"{{{{ {item_var}.{field} }}}}"
|
||||
value = ""
|
||||
if isinstance(item, dict):
|
||||
value = item.get(field, "")
|
||||
result = result.replace(placeholder, str(value))
|
||||
result = DOTTED_VAR_PATTERN.sub(
|
||||
lambda m, it=item: str(it.get(m.group(2), ""))
|
||||
if isinstance(it, dict) and m.group(1) == item_var
|
||||
else m.group(0),
|
||||
result,
|
||||
)
|
||||
if cell.paragraphs:
|
||||
_replace_vars_in_paragraph(cell.paragraphs[0], context)
|
||||
if cell.paragraphs[0].text != result and cell.paragraphs[0].runs:
|
||||
cell.paragraphs[0].runs[0].text = result
|
||||
for run in cell.paragraphs[0].runs[1:]:
|
||||
run.text = ""
|
||||
else:
|
||||
cell.text = result
|
||||
|
||||
tbl = table._tbl
|
||||
tr = template_row._tr
|
||||
tbl.remove(tr)
|
||||
|
||||
|
||||
def _replace_paragraph_vars(doc: Document, context: dict[str, Any]) -> None:
|
||||
for para in doc.paragraphs:
|
||||
_replace_vars_in_paragraph(para, context)
|
||||
|
||||
for table in doc.tables:
|
||||
for row in table.rows:
|
||||
for cell in row.cells:
|
||||
for para in cell.paragraphs:
|
||||
if "{{" in para.text and not DOTTED_VAR_PATTERN.search(para.text):
|
||||
_replace_vars_in_paragraph(para, context)
|
||||
|
||||
|
||||
def render_docx(template_path: str, field_data: dict[str, Any], output_path: str) -> str:
|
||||
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
|
||||
context = _prepare_context(field_data)
|
||||
|
||||
try:
|
||||
tpl = DocxTemplate(template_path)
|
||||
tpl.render(context)
|
||||
tpl.save(output_path)
|
||||
return output_path
|
||||
except Exception:
|
||||
doc = Document(template_path)
|
||||
_expand_table_rows(doc, context)
|
||||
_replace_paragraph_vars(doc, context)
|
||||
doc.save(output_path)
|
||||
return output_path
|
||||
|
||||
|
||||
def _apply_style_to_run(run, style: dict[str, Any]) -> None:
|
||||
if style.get("font_name"):
|
||||
run.font.name = style["font_name"]
|
||||
if style.get("font_size"):
|
||||
run.font.size = Pt(style["font_size"])
|
||||
if style.get("bold"):
|
||||
run.bold = True
|
||||
if style.get("italic"):
|
||||
run.italic = True
|
||||
if style.get("underline"):
|
||||
run.underline = True
|
||||
if style.get("color"):
|
||||
try:
|
||||
run.font.color.rgb = RGBColor.from_string(style["color"])
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _run_to_html(run) -> str:
|
||||
text = escape(run.text)
|
||||
if not text:
|
||||
return ""
|
||||
styles: list[str] = []
|
||||
if run.bold:
|
||||
styles.append("font-weight:bold")
|
||||
if run.italic:
|
||||
styles.append("font-style:italic")
|
||||
if run.underline:
|
||||
styles.append("text-decoration:underline")
|
||||
if run.font.size:
|
||||
styles.append(f"font-size:{run.font.size.pt}pt")
|
||||
if run.font.name:
|
||||
styles.append(f"font-family:'{run.font.name}'")
|
||||
if styles:
|
||||
return f'<span style="{";".join(styles)}">{text}</span>'
|
||||
return text
|
||||
|
||||
|
||||
def _paragraph_to_html(paragraph) -> str:
|
||||
parts = [_run_to_html(run) for run in paragraph.runs]
|
||||
return "".join(parts) if any(parts) else escape(paragraph.text)
|
||||
|
||||
|
||||
def _get_cell_colspan(tc) -> int:
|
||||
tc_pr = tc.tcPr
|
||||
if tc_pr is None:
|
||||
return 1
|
||||
grid_span = tc_pr.find(qn("w:gridSpan"))
|
||||
if grid_span is not None:
|
||||
return int(grid_span.get(qn("w:val"), 1))
|
||||
return 1
|
||||
|
||||
|
||||
def _get_vmerge_state(tc) -> str | None:
|
||||
tc_pr = tc.tcPr
|
||||
if tc_pr is None:
|
||||
return None
|
||||
v_merge = tc_pr.find(qn("w:vMerge"))
|
||||
if v_merge is None:
|
||||
return None
|
||||
return v_merge.get(qn("w:val")) or "continue"
|
||||
|
||||
|
||||
def _cell_to_html(cell: _Cell) -> str:
|
||||
return "<br>".join(_paragraph_to_html(p) for p in cell.paragraphs)
|
||||
|
||||
|
||||
def _table_to_html(table: Table) -> str:
|
||||
"""Render table using actual w:tc elements (handles merged cells)."""
|
||||
rows_html: list[str] = []
|
||||
|
||||
for row in table.rows:
|
||||
tr = row._tr
|
||||
cells_html: list[str] = []
|
||||
|
||||
for tc in tr.tc_lst:
|
||||
if _get_vmerge_state(tc) == "continue":
|
||||
continue
|
||||
|
||||
cell = _Cell(tc, table)
|
||||
colspan = _get_cell_colspan(tc)
|
||||
attrs = ""
|
||||
if colspan > 1:
|
||||
attrs += f' colspan="{colspan}"'
|
||||
|
||||
cells_html.append(f"<td{attrs}>{_cell_to_html(cell)}</td>")
|
||||
|
||||
rows_html.append(f"<tr>{''.join(cells_html)}</tr>")
|
||||
|
||||
return (
|
||||
'<table class="doc-table" border="1" cellpadding="4" cellspacing="0">'
|
||||
+ "".join(rows_html)
|
||||
+ "</table>"
|
||||
)
|
||||
|
||||
|
||||
def docx_to_html(file_path: str) -> str:
|
||||
doc = Document(file_path)
|
||||
html_parts = ['<div class="doc-preview">']
|
||||
|
||||
for element in doc.element.body:
|
||||
tag = element.tag.split("}")[-1]
|
||||
if tag == "p":
|
||||
from docx.text.paragraph import Paragraph
|
||||
|
||||
para = Paragraph(element, doc)
|
||||
align = ""
|
||||
alignment = _get_paragraph_alignment_str(para)
|
||||
if alignment is not None:
|
||||
align = f' style="text-align:{alignment}"'
|
||||
text = _paragraph_to_html(para)
|
||||
html_parts.append(f"<p{align}>{text}</p>")
|
||||
elif tag == "tbl":
|
||||
table = Table(element, doc)
|
||||
html_parts.append(_table_to_html(table))
|
||||
|
||||
html_parts.append("</div>")
|
||||
return "\n".join(html_parts)
|
||||
|
||||
|
||||
def export_to_pdf(docx_path: str, output_path: str) -> str:
|
||||
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[
|
||||
"libreoffice",
|
||||
"--headless",
|
||||
"--convert-to",
|
||||
"pdf",
|
||||
"--outdir",
|
||||
os.path.dirname(output_path),
|
||||
docx_path,
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60,
|
||||
)
|
||||
if result.returncode == 0:
|
||||
base = os.path.splitext(os.path.basename(docx_path))[0]
|
||||
generated = os.path.join(os.path.dirname(output_path), f"{base}.pdf")
|
||||
if os.path.exists(generated) and generated != output_path:
|
||||
os.rename(generated, output_path)
|
||||
return output_path
|
||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||
pass
|
||||
|
||||
return _fallback_pdf_export(docx_path, output_path)
|
||||
|
||||
|
||||
def _fallback_pdf_export(docx_path: str, output_path: str) -> str:
|
||||
html = docx_to_html(docx_path)
|
||||
full_html = f"""<!DOCTYPE html>
|
||||
<html><head><meta charset="utf-8">
|
||||
<style>
|
||||
body {{ font-family: 'Times New Roman', serif; margin: 40px; }}
|
||||
.doc-table {{ border-collapse: collapse; width: 100%; margin: 10px 0; }}
|
||||
.doc-table td, .doc-table th {{ border: 1px solid #333; padding: 6px 10px; }}
|
||||
p {{ margin: 6px 0; }}
|
||||
</style></head><body>{html}</body></html>"""
|
||||
|
||||
try:
|
||||
import weasyprint
|
||||
|
||||
weasyprint.HTML(string=full_html).write_pdf(output_path)
|
||||
return output_path
|
||||
except ImportError:
|
||||
html_path = output_path.replace(".pdf", ".html")
|
||||
with open(html_path, "w", encoding="utf-8") as f:
|
||||
f.write(full_html)
|
||||
raise RuntimeError(
|
||||
"PDF export requires LibreOffice (libreoffice) or weasyprint. "
|
||||
f"HTML preview saved to {html_path}"
|
||||
)
|
||||
57
backend/app/services/email_service.py
Normal file
57
backend/app/services/email_service.py
Normal file
@@ -0,0 +1,57 @@
|
||||
import logging
|
||||
import smtplib
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.mime.text import MIMEText
|
||||
|
||||
from app.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _send_smtp(to_email: str, subject: str, body: str) -> None:
|
||||
if not settings.smtp_host:
|
||||
logger.warning(
|
||||
"SMTP not configured. Email to %s — subject: %s\n%s",
|
||||
to_email,
|
||||
subject,
|
||||
body,
|
||||
)
|
||||
return
|
||||
|
||||
msg = MIMEMultipart("alternative")
|
||||
msg["Subject"] = subject
|
||||
msg["From"] = settings.smtp_from or settings.smtp_user
|
||||
msg["To"] = to_email
|
||||
msg.attach(MIMEText(body, "plain", "utf-8"))
|
||||
|
||||
with smtplib.SMTP(settings.smtp_host, settings.smtp_port) as server:
|
||||
if settings.smtp_use_tls:
|
||||
server.starttls()
|
||||
if settings.smtp_user and settings.smtp_password:
|
||||
server.login(settings.smtp_user, settings.smtp_password)
|
||||
server.sendmail(msg["From"], [to_email], msg.as_string())
|
||||
|
||||
|
||||
def send_activation_email(to_email: str, username: str, code: str) -> None:
|
||||
activate_url = f"{settings.frontend_url.rstrip('/')}/activate"
|
||||
body = (
|
||||
f"Hello {username},\n\n"
|
||||
f"Thank you for registering at Document Template Editor.\n\n"
|
||||
f"Your activation code: {code}\n\n"
|
||||
f"Enter this code at: {activate_url}\n\n"
|
||||
f"The code expires in {settings.activation_code_expire_hours} hours.\n"
|
||||
)
|
||||
_send_smtp(to_email, "Activate your account", body)
|
||||
|
||||
|
||||
def send_password_reset_email(to_email: str, username: str, code: str) -> None:
|
||||
reset_url = f"{settings.frontend_url.rstrip('/')}/reset-password"
|
||||
body = (
|
||||
f"Hello {username},\n\n"
|
||||
f"You requested a password reset.\n\n"
|
||||
f"Your reset code: {code}\n\n"
|
||||
f"Enter this code at: {reset_url}\n\n"
|
||||
f"The code expires in {settings.password_reset_code_expire_hours} hours.\n\n"
|
||||
f"If you did not request this, ignore this email.\n"
|
||||
)
|
||||
_send_smtp(to_email, "Password reset code", body)
|
||||
36
backend/app/services/user_cleanup.py
Normal file
36
backend/app/services/user_cleanup.py
Normal file
@@ -0,0 +1,36 @@
|
||||
import os
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.document import FilledDocument
|
||||
from app.models.template import DocumentTemplate
|
||||
from app.models.user import User
|
||||
|
||||
|
||||
def cleanup_user_files(db: Session, user: User) -> None:
|
||||
"""Remove uploaded template and rendered document files for a user."""
|
||||
templates = (
|
||||
db.query(DocumentTemplate).filter(DocumentTemplate.owner_id == user.id).all()
|
||||
)
|
||||
template_ids = [t.id for t in templates]
|
||||
|
||||
for template in templates:
|
||||
if template.file_path and os.path.exists(template.file_path):
|
||||
os.remove(template.file_path)
|
||||
|
||||
documents = db.query(FilledDocument).filter(
|
||||
FilledDocument.owner_id == user.id
|
||||
).all()
|
||||
if template_ids:
|
||||
documents += (
|
||||
db.query(FilledDocument)
|
||||
.filter(FilledDocument.template_id.in_(template_ids))
|
||||
.all()
|
||||
)
|
||||
|
||||
seen_paths: set[str] = set()
|
||||
for doc in documents:
|
||||
if doc.rendered_docx_path and doc.rendered_docx_path not in seen_paths:
|
||||
seen_paths.add(doc.rendered_docx_path)
|
||||
if os.path.exists(doc.rendered_docx_path):
|
||||
os.remove(doc.rendered_docx_path)
|
||||
67
backend/app/services/verification.py
Normal file
67
backend/app/services/verification.py
Normal file
@@ -0,0 +1,67 @@
|
||||
import random
|
||||
import string
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.config import settings
|
||||
from app.models.user import User
|
||||
from app.models.verification_token import TokenType, VerificationToken
|
||||
|
||||
|
||||
def _generate_code(length: int = 6) -> str:
|
||||
return "".join(random.choices(string.digits, k=length))
|
||||
|
||||
|
||||
def _invalidate_tokens(db: Session, user_id: int, token_type: TokenType) -> None:
|
||||
db.query(VerificationToken).filter(
|
||||
VerificationToken.user_id == user_id,
|
||||
VerificationToken.token_type == token_type,
|
||||
).delete()
|
||||
|
||||
|
||||
def create_verification_token(
|
||||
db: Session, user: User, token_type: TokenType
|
||||
) -> VerificationToken:
|
||||
_invalidate_tokens(db, user.id, token_type)
|
||||
|
||||
if token_type == TokenType.activation:
|
||||
hours = settings.activation_code_expire_hours
|
||||
else:
|
||||
hours = settings.password_reset_code_expire_hours
|
||||
|
||||
token = VerificationToken(
|
||||
user_id=user.id,
|
||||
code=_generate_code(),
|
||||
token_type=token_type,
|
||||
expires_at=datetime.utcnow() + timedelta(hours=hours),
|
||||
)
|
||||
db.add(token)
|
||||
db.flush()
|
||||
return token
|
||||
|
||||
|
||||
def verify_code(
|
||||
db: Session, email: str, code: str, token_type: TokenType
|
||||
) -> User | None:
|
||||
from app.auth import get_user_by_email
|
||||
|
||||
user = get_user_by_email(db, email)
|
||||
if not user:
|
||||
return None
|
||||
|
||||
token = (
|
||||
db.query(VerificationToken)
|
||||
.filter(
|
||||
VerificationToken.user_id == user.id,
|
||||
VerificationToken.token_type == token_type,
|
||||
VerificationToken.code == code.strip(),
|
||||
VerificationToken.expires_at > datetime.utcnow(),
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not token:
|
||||
return None
|
||||
|
||||
db.delete(token)
|
||||
return user
|
||||
Reference in New Issue
Block a user