Rescue Paperjet v1 implementation
This commit is contained in:
parent
7ff5c8130d
commit
db9e2ca51b
83 changed files with 6186 additions and 3894 deletions
12
backend/app/services/assets.py
Normal file
12
backend/app/services/assets.py
Normal file
|
|
@ -0,0 +1,12 @@
|
|||
"""Storage helpers for binary annotation assets."""
|
||||
|
||||
import shutil
|
||||
|
||||
from app.config import settings
|
||||
|
||||
|
||||
def delete_asset_directory(document_id: str) -> None:
|
||||
"""Delete all uploaded annotation assets belonging to a document."""
|
||||
directory = settings.PDF_STORAGE_PATH / "assets" / document_id
|
||||
if directory.exists():
|
||||
shutil.rmtree(directory)
|
||||
1
backend/app/services/export/__init__.py
Normal file
1
backend/app/services/export/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
"""PDF export services."""
|
||||
465
backend/app/services/export/renderer.py
Normal file
465
backend/app/services/export/renderer.py
Normal file
|
|
@ -0,0 +1,465 @@
|
|||
"""Flatten canonical annotation data into a PDF with PyMuPDF."""
|
||||
|
||||
import logging
|
||||
import math
|
||||
import re
|
||||
import uuid
|
||||
from collections.abc import Callable, Mapping, Sequence
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pymupdf
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
Annotation = Mapping[str, Any]
|
||||
Handler = Callable[[pymupdf.Page, Annotation, Path], None]
|
||||
|
||||
|
||||
def _rect(annotation: Annotation) -> pymupdf.Rect | None:
|
||||
value = annotation.get("rect")
|
||||
if not isinstance(value, Mapping):
|
||||
return None
|
||||
try:
|
||||
x = float(value["x"])
|
||||
y = float(value["y"])
|
||||
width = float(value["width"])
|
||||
height = float(value["height"])
|
||||
except (KeyError, TypeError, ValueError):
|
||||
return None
|
||||
if width < 0 or height < 0:
|
||||
return None
|
||||
return pymupdf.Rect(x, y, x + width, y + height)
|
||||
|
||||
|
||||
def _props(annotation: Annotation) -> Mapping[str, Any]:
|
||||
value = annotation.get("props")
|
||||
return value if isinstance(value, Mapping) else {}
|
||||
|
||||
|
||||
def _color(
|
||||
value: Any,
|
||||
fallback: tuple[float, float, float] = (0.0, 0.0, 0.0),
|
||||
) -> tuple[float, float, float]:
|
||||
if not isinstance(value, str):
|
||||
return fallback
|
||||
text = value.strip().lstrip("#")
|
||||
if len(text) == 3:
|
||||
text = "".join(char * 2 for char in text)
|
||||
if len(text) != 6:
|
||||
return fallback
|
||||
try:
|
||||
return tuple(int(text[index : index + 2], 16) / 255 for index in (0, 2, 4)) # type: ignore[return-value]
|
||||
except ValueError:
|
||||
return fallback
|
||||
|
||||
|
||||
def _opacity(value: Any, fallback: float = 1.0) -> float:
|
||||
try:
|
||||
return max(0.0, min(1.0, float(value)))
|
||||
except (TypeError, ValueError):
|
||||
return fallback
|
||||
|
||||
|
||||
def _font_name(family: Any, bold: Any = False, italic: Any = False) -> str:
|
||||
name = str(family or "Liberation Sans").lower()
|
||||
is_serif = "times" in name or "serif" in name
|
||||
is_mono = "courier" in name or "mono" in name
|
||||
if is_serif:
|
||||
normal, bold_name, italic_name, bold_italic_name = "tiro", "tibo", "tiit", "tibi"
|
||||
elif is_mono:
|
||||
normal, bold_name, italic_name, bold_italic_name = "cour", "cobo", "coit", "cobi"
|
||||
else:
|
||||
normal, bold_name, italic_name, bold_italic_name = "helv", "hebo", "heit", "hebi"
|
||||
if bold and italic:
|
||||
return bold_italic_name
|
||||
if bold:
|
||||
return bold_name
|
||||
if italic:
|
||||
return italic_name
|
||||
return normal
|
||||
|
||||
|
||||
_SIGNATURE_FONT_FILES = {
|
||||
"great vibes": "GreatVibes-Regular.ttf",
|
||||
"allura": "allura.ttf",
|
||||
"sacramento": "sacramento.ttf",
|
||||
"dancing script": "dancing-script.ttf",
|
||||
"caveat": "caveat.ttf",
|
||||
}
|
||||
|
||||
|
||||
def _signature_font_file(family: Any) -> Path | None:
|
||||
filename = _SIGNATURE_FONT_FILES.get(str(family or "").strip().lower())
|
||||
if not filename:
|
||||
return None
|
||||
path = Path(__file__).parents[2] / "assets" / "fonts" / filename
|
||||
return path if path.is_file() else None
|
||||
|
||||
|
||||
def _signature_font_size(
|
||||
rect: pymupdf.Rect, text: str, font_file: Path | None, font_name: str
|
||||
) -> float:
|
||||
size = max(1.0, min(48.0, rect.height * 0.55))
|
||||
try:
|
||||
font = (
|
||||
pymupdf.Font(fontfile=str(font_file)) if font_file else pymupdf.Font(fontname=font_name)
|
||||
)
|
||||
text_width = font.text_length(text, fontsize=size)
|
||||
if text_width > rect.width and text_width > 0:
|
||||
size *= max(0.1, rect.width / text_width) * 0.95
|
||||
except Exception:
|
||||
# A missing optional font must not prevent a PDF export.
|
||||
pass
|
||||
return max(1.0, size)
|
||||
|
||||
|
||||
def _points_from_props(annotation: Annotation, rect: pymupdf.Rect) -> list[pymupdf.Point]:
|
||||
props = _props(annotation)
|
||||
raw_points = props.get("paths")
|
||||
points: list[pymupdf.Point] = []
|
||||
if isinstance(raw_points, Sequence) and not isinstance(raw_points, (str, bytes)):
|
||||
for raw_point in raw_points:
|
||||
if isinstance(raw_point, Sequence) and len(raw_point) == 2:
|
||||
try:
|
||||
points.append(
|
||||
pymupdf.Point(
|
||||
rect.x0 + float(raw_point[0]),
|
||||
rect.y0 + float(raw_point[1]),
|
||||
)
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if len(points) >= 2:
|
||||
return points
|
||||
|
||||
# Older clients stored a Fabric SVG path. Approximate its command end
|
||||
# points and normalize them into the canonical annotation rectangle.
|
||||
svg_path = props.get("svgPath")
|
||||
if not isinstance(svg_path, str):
|
||||
return []
|
||||
raw = [float(value) for value in re.findall(r"[-+]?\d*\.?\d+(?:[eE][-+]?\d+)?", svg_path)]
|
||||
if len(raw) < 4:
|
||||
return []
|
||||
pairs = list(zip(raw[::2], raw[1::2], strict=False))
|
||||
min_x = min(point[0] for point in pairs)
|
||||
max_x = max(point[0] for point in pairs)
|
||||
min_y = min(point[1] for point in pairs)
|
||||
max_y = max(point[1] for point in pairs)
|
||||
source_width = max(max_x - min_x, 1e-6)
|
||||
source_height = max(max_y - min_y, 1e-6)
|
||||
return [
|
||||
pymupdf.Point(
|
||||
rect.x0 + (x - min_x) / source_width * rect.width,
|
||||
rect.y0 + (y - min_y) / source_height * rect.height,
|
||||
)
|
||||
for x, y in pairs
|
||||
]
|
||||
|
||||
|
||||
def _draw_polyline(
|
||||
page: pymupdf.Page,
|
||||
points: list[pymupdf.Point],
|
||||
*,
|
||||
color: tuple[float, float, float],
|
||||
width: float,
|
||||
opacity: float,
|
||||
) -> None:
|
||||
if len(points) < 2:
|
||||
return
|
||||
shape = page.new_shape()
|
||||
shape.draw_polyline(points)
|
||||
shape.finish(color=color, width=max(0.1, width), stroke_opacity=opacity)
|
||||
shape.commit(overlay=True)
|
||||
|
||||
|
||||
def _style_at(styles: Any, line_number: int, character_number: int) -> Mapping[str, Any]:
|
||||
if not isinstance(styles, Mapping):
|
||||
return {}
|
||||
line_styles = styles.get(str(line_number), styles.get(line_number))
|
||||
if not isinstance(line_styles, Mapping):
|
||||
return {}
|
||||
character_style = line_styles.get(str(character_number), line_styles.get(character_number))
|
||||
return character_style if isinstance(character_style, Mapping) else {}
|
||||
|
||||
|
||||
def _render_styled_text(
|
||||
page: pymupdf.Page,
|
||||
rect: pymupdf.Rect,
|
||||
text: str,
|
||||
props: Mapping[str, Any],
|
||||
styles: Any,
|
||||
) -> bool:
|
||||
if not isinstance(styles, Mapping) or not styles:
|
||||
return False
|
||||
|
||||
base_size = max(1.0, float(props.get("fontSize", 14)))
|
||||
line_height = max(0.5, float(props.get("lineHeight", 1.2)))
|
||||
lines = text.split("\n")
|
||||
y = rect.y0
|
||||
for line_number, line in enumerate(lines):
|
||||
measured: list[tuple[str, float, str, tuple[float, float, float], float, str | None]] = []
|
||||
for character_number, character in enumerate(line):
|
||||
style = _style_at(styles, line_number, character_number)
|
||||
font_size = max(1.0, float(style.get("fontSize", base_size)))
|
||||
font_family = style.get("fontFamily", props.get("fontFamily"))
|
||||
font_name = _font_name(
|
||||
font_family,
|
||||
style.get("fontWeight") == "bold",
|
||||
style.get("fontStyle") == "italic",
|
||||
)
|
||||
color = _color(style.get("fill", props.get("color")))
|
||||
font = pymupdf.Font(fontname=font_name)
|
||||
width = float(font.text_length(character, fontsize=font_size))
|
||||
background = style.get("textBackgroundColor", props.get("highlightColor"))
|
||||
measured.append((character, font_size, font_name, color, width, background))
|
||||
|
||||
total_width = sum(item[4] for item in measured)
|
||||
align = str(props.get("align", "left"))
|
||||
x = rect.x0
|
||||
if align == "center":
|
||||
x += max(0.0, (rect.width - total_width) / 2)
|
||||
elif align == "right":
|
||||
x += max(0.0, rect.width - total_width)
|
||||
|
||||
max_size = max((item[1] for item in measured), default=base_size)
|
||||
for character, font_size, font_name, color, width, background in measured:
|
||||
if (
|
||||
isinstance(background, str)
|
||||
and background.lower() not in {"", "transparent", "none"}
|
||||
):
|
||||
page.draw_rect(
|
||||
pymupdf.Rect(x, y, x + width, y + max_size * 1.15),
|
||||
color=None,
|
||||
fill=_color(background),
|
||||
fill_opacity=0.3,
|
||||
overlay=True,
|
||||
)
|
||||
if character != " ":
|
||||
page.insert_text(
|
||||
pymupdf.Point(x, y + font_size),
|
||||
character,
|
||||
fontsize=font_size,
|
||||
fontname=font_name,
|
||||
color=color,
|
||||
overlay=True,
|
||||
)
|
||||
x += width
|
||||
y += max_size * line_height
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def render_text(page: pymupdf.Page, annotation: Annotation, _assets_dir: Path) -> None:
|
||||
rect = _rect(annotation)
|
||||
if rect is None:
|
||||
return
|
||||
props = _props(annotation)
|
||||
text = str(props.get("text", ""))
|
||||
if not text:
|
||||
return
|
||||
highlight = props.get("highlightColor")
|
||||
if isinstance(highlight, str) and highlight and highlight.lower() != "transparent":
|
||||
page.draw_rect(rect, color=None, fill=_color(highlight), fill_opacity=0.3, overlay=True)
|
||||
if _render_styled_text(page, rect, text, props, props.get("styles")):
|
||||
return
|
||||
align = {"left": 0, "center": 1, "right": 2}.get(str(props.get("align", "left")), 0)
|
||||
page.insert_textbox(
|
||||
rect,
|
||||
text,
|
||||
fontsize=max(1.0, float(props.get("fontSize", 14))),
|
||||
fontname=_font_name(props.get("fontFamily"), props.get("bold"), props.get("italic")),
|
||||
color=_color(props.get("color")),
|
||||
align=align,
|
||||
overlay=True,
|
||||
)
|
||||
|
||||
|
||||
def render_draw(page: pymupdf.Page, annotation: Annotation, _assets_dir: Path) -> None:
|
||||
rect = _rect(annotation)
|
||||
if rect is None:
|
||||
return
|
||||
props = _props(annotation)
|
||||
_draw_polyline(
|
||||
page,
|
||||
_points_from_props(annotation, rect),
|
||||
color=_color(props.get("strokeColor")),
|
||||
width=float(props.get("strokeWidth", 2)),
|
||||
opacity=_opacity(props.get("opacity")),
|
||||
)
|
||||
|
||||
|
||||
def _asset_path(assets_dir: Path, ref: Any) -> Path | None:
|
||||
if not isinstance(ref, str):
|
||||
return None
|
||||
try:
|
||||
safe_ref = str(uuid.UUID(ref))
|
||||
except ValueError:
|
||||
return None
|
||||
path = assets_dir / safe_ref
|
||||
return path if path.is_file() else None
|
||||
|
||||
|
||||
def render_image(page: pymupdf.Page, annotation: Annotation, assets_dir: Path) -> None:
|
||||
rect = _rect(annotation)
|
||||
asset = _asset_path(assets_dir, _props(annotation).get("ref"))
|
||||
if rect is None or asset is None:
|
||||
return
|
||||
page.insert_image(rect, filename=str(asset), keep_proportion=False, overlay=True)
|
||||
|
||||
|
||||
def render_signature(page: pymupdf.Page, annotation: Annotation, assets_dir: Path) -> None:
|
||||
rect = _rect(annotation)
|
||||
props = _props(annotation)
|
||||
if rect is None:
|
||||
return
|
||||
if props.get("mode") == "draw":
|
||||
asset = _asset_path(assets_dir, props.get("ref"))
|
||||
if asset is not None:
|
||||
page.insert_image(rect, filename=str(asset), keep_proportion=False, overlay=True)
|
||||
return
|
||||
text = str(props.get("text", ""))
|
||||
if text:
|
||||
font_file = _signature_font_file(props.get("fontFamily"))
|
||||
font_name = _font_name(props.get("fontFamily"), False, False)
|
||||
options: dict[str, Any] = {
|
||||
# Script fonts have a taller ascender than the built-in PDF fonts;
|
||||
# leave room inside the editor's saved bounding rectangle so a
|
||||
# valid signature is never silently dropped for not fitting.
|
||||
"fontsize": _signature_font_size(rect, text, font_file, font_name),
|
||||
"color": _color(props.get("color")),
|
||||
"overlay": True,
|
||||
}
|
||||
if font_file:
|
||||
options["fontname"] = f"paperjet-{font_file.stem.lower()}"
|
||||
options["fontfile"] = str(font_file)
|
||||
else:
|
||||
options["fontname"] = font_name
|
||||
page.insert_textbox(rect, text, **options)
|
||||
|
||||
|
||||
def render_highlight(page: pymupdf.Page, annotation: Annotation, _assets_dir: Path) -> None:
|
||||
rect = _rect(annotation)
|
||||
if rect is None:
|
||||
return
|
||||
props = _props(annotation)
|
||||
page.draw_rect(
|
||||
rect,
|
||||
color=None,
|
||||
fill=_color(props.get("color"), (1.0, 0.92, 0.1)),
|
||||
fill_opacity=_opacity(props.get("opacity"), 0.3),
|
||||
overlay=True,
|
||||
)
|
||||
|
||||
|
||||
def render_shape(page: pymupdf.Page, annotation: Annotation, _assets_dir: Path) -> None:
|
||||
rect = _rect(annotation)
|
||||
if rect is None:
|
||||
return
|
||||
props = _props(annotation)
|
||||
kind = str(props.get("kind", "rect"))
|
||||
stroke = _color(props.get("strokeColor"))
|
||||
fill_value = props.get("fillColor")
|
||||
fill = None if fill_value in (None, "", "transparent", "none") else _color(fill_value)
|
||||
width = max(0.1, float(props.get("strokeWidth", 2)))
|
||||
if kind == "ellipse":
|
||||
page.draw_oval(rect, color=stroke, fill=fill, width=width, overlay=True)
|
||||
elif kind in {"line", "arrow"}:
|
||||
start = props.get("start")
|
||||
end = props.get("end")
|
||||
if (
|
||||
isinstance(start, Sequence)
|
||||
and not isinstance(start, (str, bytes))
|
||||
and len(start) == 2
|
||||
and isinstance(end, Sequence)
|
||||
and not isinstance(end, (str, bytes))
|
||||
and len(end) == 2
|
||||
):
|
||||
try:
|
||||
start_point = pymupdf.Point(
|
||||
rect.x0 + float(start[0]) * rect.width,
|
||||
rect.y0 + float(start[1]) * rect.height,
|
||||
)
|
||||
end_point = pymupdf.Point(
|
||||
rect.x0 + float(end[0]) * rect.width,
|
||||
rect.y0 + float(end[1]) * rect.height,
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
start_point, end_point = rect.tl, rect.br
|
||||
else:
|
||||
start_point, end_point = rect.tl, rect.br
|
||||
page.draw_line(start_point, end_point, color=stroke, width=width, overlay=True)
|
||||
if kind == "arrow":
|
||||
angle = math.atan2(end_point.y - start_point.y, end_point.x - start_point.x)
|
||||
length = min(
|
||||
12.0,
|
||||
max(
|
||||
5.0, math.hypot(end_point.x - start_point.x, end_point.y - start_point.y) * 0.2
|
||||
),
|
||||
)
|
||||
left = pymupdf.Point(
|
||||
end_point.x - length * math.cos(angle - math.pi / 6),
|
||||
end_point.y - length * math.sin(angle - math.pi / 6),
|
||||
)
|
||||
right = pymupdf.Point(
|
||||
end_point.x - length * math.cos(angle + math.pi / 6),
|
||||
end_point.y - length * math.sin(angle + math.pi / 6),
|
||||
)
|
||||
page.draw_polyline([left, end_point, right], color=stroke, width=width, overlay=True)
|
||||
else:
|
||||
page.draw_rect(rect, color=stroke, fill=fill, width=width, overlay=True)
|
||||
|
||||
|
||||
HANDLERS: dict[str, Handler] = {
|
||||
"text": render_text,
|
||||
"draw": render_draw,
|
||||
"signature": render_signature,
|
||||
"image": render_image,
|
||||
"highlight": render_highlight,
|
||||
"shape": render_shape,
|
||||
}
|
||||
|
||||
|
||||
def export_annotations(
|
||||
pdf_path: Path,
|
||||
annotations: Sequence[Annotation],
|
||||
assets_dir: Path,
|
||||
) -> bytes:
|
||||
"""Return a flattened PDF while preserving the source page rotations."""
|
||||
with pymupdf.open(pdf_path) as document:
|
||||
rotations = [page.rotation for page in document]
|
||||
try:
|
||||
for page in document:
|
||||
if page.rotation:
|
||||
page.set_rotation(0)
|
||||
page_annotations: dict[int, list[Annotation]] = {}
|
||||
for annotation in annotations:
|
||||
try:
|
||||
page_number = int(annotation.get("page", -1))
|
||||
except (TypeError, ValueError):
|
||||
logger.warning("Skipping annotation with invalid page: %r", annotation)
|
||||
continue
|
||||
if 0 <= page_number < len(document):
|
||||
page_annotations.setdefault(page_number, []).append(annotation)
|
||||
else:
|
||||
logger.warning("Skipping annotation outside document pages: %r", annotation)
|
||||
|
||||
for page_number, page in enumerate(document):
|
||||
ordered = sorted(
|
||||
page_annotations.get(page_number, []),
|
||||
key=lambda item: int(item.get("z", 0)),
|
||||
)
|
||||
for annotation in ordered:
|
||||
annotation_type = annotation.get("type")
|
||||
handler = HANDLERS.get(str(annotation_type))
|
||||
if handler is None:
|
||||
logger.warning("Skipping unsupported annotation type %r", annotation_type)
|
||||
continue
|
||||
try:
|
||||
handler(page, annotation, assets_dir)
|
||||
except Exception:
|
||||
logger.exception("Skipping malformed %s annotation", annotation_type)
|
||||
finally:
|
||||
for page, rotation in zip(document, rotations, strict=True):
|
||||
if page.rotation != rotation:
|
||||
page.set_rotation(rotation)
|
||||
return document.tobytes(garbage=4, deflate=True)
|
||||
|
|
@ -1,53 +1,71 @@
|
|||
"""Document storage service."""
|
||||
"""Document and binary storage helpers."""
|
||||
|
||||
import shutil
|
||||
import uuid
|
||||
import pymupdf
|
||||
from pathlib import Path
|
||||
from fastapi import UploadFile, HTTPException
|
||||
|
||||
import pymupdf
|
||||
from fastapi import HTTPException, UploadFile
|
||||
|
||||
from app.config import settings
|
||||
|
||||
def validate_pdf(filepath: Path) -> None:
|
||||
"""Validate PDF magic bytes and PyMuPDF openability."""
|
||||
with open(filepath, "rb") as f:
|
||||
header = f.read(5)
|
||||
if header != b"%PDF-":
|
||||
_COPY_CHUNK_SIZE = 1024 * 1024
|
||||
|
||||
|
||||
def validate_pdf(filepath: Path) -> int:
|
||||
"""Validate a PDF's magic bytes and return its page count."""
|
||||
with filepath.open("rb") as file:
|
||||
if file.read(5) != b"%PDF-":
|
||||
raise ValueError("Not a valid PDF file (missing magic bytes)")
|
||||
|
||||
try:
|
||||
doc = pymupdf.open(filepath)
|
||||
doc.close()
|
||||
except Exception as e:
|
||||
raise ValueError(f"Failed to open PDF with PyMuPDF: {e}")
|
||||
with pymupdf.open(filepath) as document:
|
||||
page_count = len(document)
|
||||
except Exception as err:
|
||||
raise ValueError(f"Failed to open PDF with PyMuPDF: {err}") from err
|
||||
|
||||
if page_count == 0:
|
||||
raise ValueError("PDF does not contain any pages")
|
||||
return page_count
|
||||
|
||||
|
||||
def safe_filename(filename: str | None) -> str:
|
||||
"""Return a display-safe basename without allowing path components."""
|
||||
name = Path(filename or "Untitled.pdf").name.strip()
|
||||
return name or "Untitled.pdf"
|
||||
|
||||
|
||||
async def save_upload_file(upload_file: UploadFile) -> tuple[str, int, int]:
|
||||
"""Save an uploaded file to disk and validate it."""
|
||||
"""Stream an uploaded PDF to storage, enforcing the configured size limit."""
|
||||
file_id = str(uuid.uuid4())
|
||||
settings.PDF_STORAGE_PATH.mkdir(parents=True, exist_ok=True)
|
||||
filepath = settings.PDF_STORAGE_PATH / f"{file_id}.pdf"
|
||||
max_bytes = settings.MAX_UPLOAD_MB * 1024 * 1024
|
||||
total_bytes = 0
|
||||
|
||||
try:
|
||||
with open(filepath, "wb") as f:
|
||||
shutil.copyfileobj(upload_file.file, f)
|
||||
except Exception as e:
|
||||
if filepath.exists():
|
||||
filepath.unlink()
|
||||
raise HTTPException(status_code=500, detail="Failed to save file")
|
||||
with filepath.open("wb") as destination:
|
||||
while chunk := upload_file.file.read(_COPY_CHUNK_SIZE):
|
||||
total_bytes += len(chunk)
|
||||
if total_bytes > max_bytes:
|
||||
raise HTTPException(status_code=413, detail="File too large")
|
||||
destination.write(chunk)
|
||||
except HTTPException:
|
||||
filepath.unlink(missing_ok=True)
|
||||
raise
|
||||
except Exception as err:
|
||||
filepath.unlink(missing_ok=True)
|
||||
raise HTTPException(status_code=500, detail="Failed to save file") from err
|
||||
|
||||
page_count = 0
|
||||
try:
|
||||
doc = pymupdf.open(filepath)
|
||||
page_count = len(doc)
|
||||
doc.close()
|
||||
except Exception as e:
|
||||
filepath.unlink()
|
||||
raise HTTPException(status_code=400, detail=f"Failed to open PDF with PyMuPDF: {e}")
|
||||
|
||||
size = filepath.stat().st_size
|
||||
return file_id, size, page_count
|
||||
page_count = validate_pdf(filepath)
|
||||
except ValueError as err:
|
||||
filepath.unlink(missing_ok=True)
|
||||
raise HTTPException(status_code=400, detail=str(err)) from err
|
||||
|
||||
return file_id, total_bytes, page_count
|
||||
|
||||
|
||||
def delete_pdf_file(file_id: str) -> None:
|
||||
"""Delete a PDF file from storage."""
|
||||
filepath = settings.PDF_STORAGE_PATH / f"{file_id}.pdf"
|
||||
if filepath.exists():
|
||||
filepath.unlink()
|
||||
filepath.unlink(missing_ok=True)
|
||||
|
|
|
|||
|
|
@ -1,38 +1,40 @@
|
|||
"""Thumbnail generation service."""
|
||||
|
||||
import pymupdf
|
||||
from pathlib import Path
|
||||
|
||||
from app.config import settings
|
||||
|
||||
def generate_thumbnail(file_id: str) -> None:
|
||||
|
||||
def generate_thumbnail(file_id: str) -> bool:
|
||||
"""Generate a PNG thumbnail for the first page of a PDF."""
|
||||
pdf_path = settings.PDF_STORAGE_PATH / f"{file_id}.pdf"
|
||||
thumb_path = settings.THUMBNAILS_PATH / f"{file_id}.png"
|
||||
|
||||
|
||||
if not pdf_path.exists():
|
||||
return
|
||||
|
||||
return False
|
||||
|
||||
try:
|
||||
doc = pymupdf.open(pdf_path)
|
||||
if len(doc) > 0:
|
||||
with pymupdf.open(pdf_path) as doc:
|
||||
if len(doc) == 0:
|
||||
return False
|
||||
page = doc[0]
|
||||
# Zoom to approximately 600px width
|
||||
rect = page.rect
|
||||
zoom = 600.0 / rect.width if rect.width > 0 else 1.0
|
||||
mat = pymupdf.Matrix(zoom, zoom)
|
||||
|
||||
|
||||
# Render pixmap
|
||||
pix = page.get_pixmap(matrix=mat, alpha=False)
|
||||
|
||||
|
||||
# Save as PNG
|
||||
pix.save(thumb_path, output="png")
|
||||
doc.close()
|
||||
except Exception as e:
|
||||
print(f"Failed to generate thumbnail for {file_id}: {e}")
|
||||
return True
|
||||
except Exception as err:
|
||||
print(f"Failed to generate thumbnail for {file_id}: {err}")
|
||||
return False
|
||||
|
||||
|
||||
def delete_thumbnail(file_id: str) -> None:
|
||||
"""Delete a thumbnail file."""
|
||||
thumb_path = settings.THUMBNAILS_PATH / f"{file_id}.png"
|
||||
if thumb_path.exists():
|
||||
thumb_path.unlink()
|
||||
thumb_path.unlink(missing_ok=True)
|
||||
|
|
|
|||
|
|
@ -1,48 +1,50 @@
|
|||
"""Background tasks for sweeping trash and old versions."""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.db import get_db
|
||||
from app.models.document import Document
|
||||
from app.config import settings
|
||||
from app.models.annotation_state import AnnotationState
|
||||
from app.models.document import Document
|
||||
from app.models.version import Version
|
||||
from app.services.assets import delete_asset_directory
|
||||
from app.services.storage import delete_pdf_file
|
||||
from app.services.thumbnails import delete_thumbnail
|
||||
from app.config import settings
|
||||
|
||||
|
||||
def sweep_trash(db: Session) -> None:
|
||||
"""Permanently delete documents that have been in the trash past the retention period."""
|
||||
cutoff = datetime.now(timezone.utc) - timedelta(days=settings.TRASH_RETENTION_DAYS)
|
||||
cutoff = datetime.now(UTC) - timedelta(days=settings.TRASH_RETENTION_DAYS)
|
||||
cutoff_iso = cutoff.isoformat()
|
||||
|
||||
docs_to_delete = db.query(Document).filter(
|
||||
Document.in_trash == True,
|
||||
Document.deleted_at <= cutoff_iso
|
||||
).all()
|
||||
|
||||
|
||||
docs_to_delete = (
|
||||
db.query(Document)
|
||||
.filter(Document.deleted_at.is_not(None), Document.deleted_at <= cutoff_iso)
|
||||
.all()
|
||||
)
|
||||
|
||||
for doc in docs_to_delete:
|
||||
# Delete files from disk
|
||||
delete_pdf_file(doc.id)
|
||||
delete_thumbnail(doc.id)
|
||||
|
||||
delete_asset_directory(doc.id)
|
||||
|
||||
# Delete DB associations
|
||||
db.query(AnnotationState).filter(AnnotationState.document_id == doc.id).delete()
|
||||
db.query(Version).filter(Version.document_id == doc.id).delete()
|
||||
|
||||
|
||||
# Delete document record
|
||||
db.delete(doc)
|
||||
|
||||
|
||||
db.commit()
|
||||
|
||||
|
||||
def prune_auto_versions(db: Session) -> None:
|
||||
"""Delete automatic versions older than the retention period."""
|
||||
cutoff = datetime.now(timezone.utc) - timedelta(days=settings.AUTO_VERSION_RETENTION_DAYS)
|
||||
cutoff = datetime.now(UTC) - timedelta(days=settings.AUTO_VERSION_RETENTION_DAYS)
|
||||
cutoff_iso = cutoff.isoformat()
|
||||
|
||||
db.query(Version).filter(
|
||||
Version.kind == "auto",
|
||||
Version.created_at <= cutoff_iso
|
||||
).delete()
|
||||
|
||||
|
||||
db.query(Version).filter(Version.kind == "auto", Version.created_at <= cutoff_iso).delete()
|
||||
|
||||
db.commit()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue