425 lines
11 KiB
Python
425 lines
11 KiB
Python
import json
|
|
from typing import Annotated, Literal
|
|
|
|
from fastapi import APIRouter, HTTPException, status
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.services import engine
|
|
from app.services.store import document_store
|
|
|
|
router = APIRouter(prefix="/documents/{document_id}/edits", tags=["edits"])
|
|
compat_router = APIRouter(tags=["edits"])
|
|
|
|
|
|
class TextOverlayData(BaseModel):
|
|
text: str
|
|
x: float
|
|
y: float
|
|
width: float
|
|
height: float
|
|
fontSize: float = Field(..., gt=0)
|
|
fontFamily: str
|
|
color: str
|
|
|
|
|
|
class RedactionData(BaseModel):
|
|
x: float
|
|
y: float
|
|
width: float
|
|
height: float
|
|
fillColor: str = "#000000"
|
|
|
|
|
|
class ImageOverlayData(BaseModel):
|
|
x: float
|
|
y: float
|
|
width: float
|
|
height: float
|
|
imageData: str
|
|
|
|
|
|
class HighlightQuadPoint(BaseModel):
|
|
x1: float
|
|
y1: float
|
|
x2: float
|
|
y2: float
|
|
x3: float
|
|
y3: float
|
|
x4: float
|
|
y4: float
|
|
|
|
|
|
class HighlightData(BaseModel):
|
|
quadPoints: list[HighlightQuadPoint]
|
|
color: str
|
|
opacity: float = 1.0
|
|
author: str
|
|
content: str | None = None
|
|
|
|
|
|
class FreeTextData(BaseModel):
|
|
x: float
|
|
y: float
|
|
width: float
|
|
height: float
|
|
text: str
|
|
fontSize: float = Field(12.0, gt=0)
|
|
color: str = "#000000"
|
|
|
|
|
|
class StickyNoteData(BaseModel):
|
|
x: float
|
|
y: float
|
|
author: str
|
|
content: str
|
|
timestamp: str | None = None
|
|
|
|
|
|
class FreehandPoint(BaseModel):
|
|
x: float
|
|
y: float
|
|
|
|
|
|
class FreehandData(BaseModel):
|
|
paths: list[list[FreehandPoint]]
|
|
color: str
|
|
thickness: float
|
|
|
|
|
|
class PageRotationData(BaseModel):
|
|
rotation: Literal[0, 90, 180, 270]
|
|
|
|
|
|
class TextOverlayOperation(BaseModel):
|
|
id: str
|
|
type: Literal["text_overlay"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: TextOverlayData
|
|
|
|
|
|
class RedactionOperation(BaseModel):
|
|
id: str
|
|
type: Literal["redaction"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: RedactionData
|
|
|
|
|
|
class ImageOverlayOperation(BaseModel):
|
|
id: str
|
|
type: Literal["image_overlay"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: ImageOverlayData
|
|
|
|
|
|
class HighlightOperation(BaseModel):
|
|
id: str
|
|
type: Literal["highlight"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: HighlightData
|
|
|
|
|
|
class FreeTextOperation(BaseModel):
|
|
id: str
|
|
type: Literal["free_text"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: FreeTextData
|
|
|
|
|
|
class CommentOperation(BaseModel):
|
|
id: str
|
|
type: Literal["comment"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: StickyNoteData
|
|
|
|
|
|
class FreehandOperation(BaseModel):
|
|
id: str
|
|
type: Literal["freehand"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: FreehandData
|
|
|
|
|
|
class PageRotationOperation(BaseModel):
|
|
id: str
|
|
type: Literal["page_rotation"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: PageRotationData
|
|
|
|
|
|
class PageDeletionData(BaseModel):
|
|
pass
|
|
|
|
|
|
class PageDeletionOperation(BaseModel):
|
|
id: str
|
|
type: Literal["page_deletion"]
|
|
pageIndex: int
|
|
data: PageDeletionData
|
|
|
|
|
|
class PageReorderData(BaseModel):
|
|
destPageIndex: int
|
|
|
|
|
|
class PageReorderOperation(BaseModel):
|
|
id: str
|
|
type: Literal["page_reorder"]
|
|
pageIndex: int
|
|
data: PageReorderData
|
|
|
|
|
|
class UpdateFieldData(BaseModel):
|
|
value: str | bool
|
|
annotationId: str | None = None
|
|
|
|
|
|
class UpdateFieldOperation(BaseModel):
|
|
id: str
|
|
type: Literal["update_field"]
|
|
pageIndex: int
|
|
data: UpdateFieldData
|
|
|
|
|
|
class DeleteAnnotationData(BaseModel):
|
|
annotationId: str
|
|
|
|
|
|
class DeleteAnnotationOperation(BaseModel):
|
|
id: str
|
|
type: Literal["delete_annotation"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: DeleteAnnotationData
|
|
|
|
|
|
class UpdateAnnotationData(BaseModel):
|
|
annotationId: str
|
|
x: float | None = None
|
|
y: float | None = None
|
|
width: float | None = None
|
|
height: float | None = None
|
|
color: str | None = None
|
|
thickness: float | None = None
|
|
text: str | None = None
|
|
|
|
|
|
class UpdateAnnotationOperation(BaseModel):
|
|
id: str
|
|
type: Literal["update_annotation"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: UpdateAnnotationData
|
|
|
|
|
|
class ReplaceTextData(BaseModel):
|
|
objectIndices: list[int]
|
|
text: str
|
|
internalFontId: str
|
|
fontSize: float
|
|
disableJustify: bool = False
|
|
|
|
|
|
class ReplaceTextOperation(BaseModel):
|
|
id: str
|
|
type: Literal["replace_text"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: ReplaceTextData
|
|
|
|
|
|
class ReflowRun(BaseModel):
|
|
text: str
|
|
internalFontId: str
|
|
fontSize: float
|
|
color: str = "#000000"
|
|
advances: list[float] | None = None
|
|
|
|
|
|
class ReflowParagraphData(BaseModel):
|
|
objectIndices: list[int]
|
|
runs: list[ReflowRun]
|
|
columnLeft: float
|
|
columnRight: float
|
|
firstBaselineY: float
|
|
leading: float
|
|
oldLineCount: int = 1
|
|
align: Literal["left", "justify", "center", "right"] = "left"
|
|
pushColumnLeft: float | None = None
|
|
paraId: str | None = None
|
|
lines: list[list[ReflowRun]] | None = None
|
|
lineBaselineY: list[float] | None = None
|
|
lineX: list[float] | None = None
|
|
hangingIndent: float = 0.0
|
|
listMarker: str | None = None
|
|
listKind: Literal["bullet", "ordered"] | None = None
|
|
listLevel: int | None = None
|
|
|
|
|
|
class ReflowParagraphOperation(BaseModel):
|
|
id: str
|
|
type: Literal["reflow_paragraph"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: ReflowParagraphData
|
|
|
|
|
|
class DecorationData(BaseModel):
|
|
quadPoints: list[HighlightQuadPoint]
|
|
color: str = "#000000"
|
|
author: str = "User"
|
|
content: str | None = None
|
|
|
|
class UnderlineOperation(BaseModel):
|
|
id: str
|
|
type: Literal["underline"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: DecorationData
|
|
|
|
class StrikeoutOperation(BaseModel):
|
|
id: str
|
|
type: Literal["strikeout"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: DecorationData
|
|
|
|
class SquigglyOperation(BaseModel):
|
|
id: str
|
|
type: Literal["squiggly"]
|
|
pageIndex: int = Field(..., ge=0)
|
|
data: DecorationData
|
|
|
|
EditOperation = Annotated[
|
|
TextOverlayOperation
|
|
| RedactionOperation
|
|
| ImageOverlayOperation
|
|
| HighlightOperation
|
|
| FreeTextOperation
|
|
| CommentOperation
|
|
| FreehandOperation
|
|
| PageRotationOperation
|
|
| PageDeletionOperation
|
|
| PageReorderOperation
|
|
| UpdateFieldOperation
|
|
| DeleteAnnotationOperation
|
|
| UpdateAnnotationOperation
|
|
| ReplaceTextOperation
|
|
| ReflowParagraphOperation
|
|
| UnderlineOperation
|
|
| StrikeoutOperation
|
|
| SquigglyOperation,
|
|
Field(discriminator="type"),
|
|
]
|
|
|
|
|
|
class EditsRequest(BaseModel):
|
|
version: Literal["1.0"]
|
|
operations: list[EditOperation]
|
|
|
|
_OP_PERMISSION = {
|
|
"highlight": "canAnnotate", "underline": "canAnnotate", "strikeout": "canAnnotate",
|
|
"squiggly": "canAnnotate", "comment": "canAnnotate", "freehand": "canAnnotate",
|
|
"free_text": "canAnnotate", "text_overlay": "canAnnotate", "image_overlay": "canAnnotate",
|
|
"delete_annotation": "canAnnotate", "update_annotation": "canAnnotate",
|
|
"replace_text": "canModify", "reflow_paragraph": "canModify", "redaction": "canModify",
|
|
"update_field": "canFillForms",
|
|
"page_rotation": "canAssemble", "page_deletion": "canAssemble", "page_reorder": "canAssemble",
|
|
}
|
|
|
|
|
|
def apply_edits_impl(document_id: str, request: EditsRequest):
|
|
if not engine.is_available():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
|
detail="Engine bridge (bindings/python) not yet available.",
|
|
)
|
|
|
|
doc_info = document_store.get_document(document_id)
|
|
if not doc_info:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Document not found")
|
|
|
|
perms = doc_info.get("permissions") or {}
|
|
for op in request.operations:
|
|
required = _OP_PERMISSION.get(op.type)
|
|
if required and perms.get(required, True) is False:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Operation '{op.type}' is not permitted by this document's restrictions ({required}).",
|
|
)
|
|
|
|
created_temp_files = []
|
|
try:
|
|
pdfengine = engine.require()
|
|
doc = doc_info["doc_instance"]
|
|
|
|
req_dict = request.model_dump(exclude_none=True)
|
|
for op in req_dict.get("operations", []):
|
|
if op.get("type") == "image_overlay":
|
|
img_data_str = op["data"].get("imageData", "")
|
|
if img_data_str:
|
|
import base64
|
|
import io
|
|
import os
|
|
import tempfile
|
|
|
|
from PIL import Image
|
|
|
|
if "," in img_data_str:
|
|
img_data_str = img_data_str.split(",", 1)[1]
|
|
|
|
raw_bytes = base64.b64decode(img_data_str)
|
|
img = Image.open(io.BytesIO(raw_bytes))
|
|
img_rgba = img.convert("RGBA")
|
|
|
|
r, g, b, a = img_rgba.split()
|
|
img_bgra = Image.merge("RGBA", (b, g, r, a))
|
|
|
|
bgra_bytes = img_bgra.tobytes()
|
|
|
|
fd, temp_path = tempfile.mkstemp(suffix=".bin", prefix="pdf_pixel_")
|
|
created_temp_files.append(temp_path)
|
|
try:
|
|
with os.fdopen(fd, "wb") as tmp:
|
|
tmp.write(bgra_bytes)
|
|
except Exception:
|
|
os.close(fd)
|
|
raise
|
|
|
|
op["data"]["pixelDataPath"] = temp_path
|
|
op["data"]["pixelWidth"] = img.width
|
|
op["data"]["pixelHeight"] = img.height
|
|
|
|
if "imageData" in op["data"]:
|
|
del op["data"]["imageData"]
|
|
|
|
edits_json = json.dumps(req_dict)
|
|
doc_copy = pdfengine.PdfDocument.load_from_memory(doc_info["bytes_data"])
|
|
invalidated_regions = doc_copy.apply_edits(edits_json)
|
|
|
|
full_save_types = {"redaction", "replace_text", "reflow_paragraph"}
|
|
needs_full = any(op.get("type") in full_save_types for op in req_dict.get("operations", []))
|
|
new_bytes = doc_copy.save_full() if needs_full else doc_copy.save_incremental()
|
|
|
|
new_doc = pdfengine.PdfDocument.load_from_memory(new_bytes)
|
|
|
|
new_info = document_store.add_document(
|
|
filename=doc_info["filename"], bytes_data=new_bytes, doc_instance=new_doc,
|
|
permissions=doc_info.get("permissions"),
|
|
)
|
|
|
|
return {"success": True, "newDocumentId": new_info["id"], "invalidatedRegions": invalidated_regions}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
|
|
finally:
|
|
import contextlib
|
|
import os
|
|
|
|
for temp_path in created_temp_files:
|
|
if os.path.exists(temp_path):
|
|
with contextlib.suppress(Exception):
|
|
os.remove(temp_path)
|
|
|
|
|
|
@router.post("")
|
|
def apply_edits(document_id: str, request: EditsRequest):
|
|
return apply_edits_impl(document_id, request)
|
|
|
|
|
|
@compat_router.post("/edits/{document_id}")
|
|
def apply_edits_compat(document_id: str, request: EditsRequest):
|
|
return apply_edits_impl(document_id, request)
|