285 lines
11 KiB
Python
285 lines
11 KiB
Python
import logging
|
|
from fastapi import APIRouter, HTTPException, Response, status
|
|
|
|
from app.schemas.annotation import AnnotationResponse
|
|
from app.schemas.page_model import PageModelResponse
|
|
from app.schemas.search import SearchMatch
|
|
from app.services import engine
|
|
from app.services.page_model import build_page_model_response
|
|
from app.services.search import compute_search_matches
|
|
from app.services.store import document_store
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(tags=["documents"])
|
|
|
|
|
|
@router.get("/{document_id}/raw")
|
|
def get_document_raw(document_id: str) -> Response:
|
|
"""Raw PDF bytes of the current version — loaded into the in-browser WASM engine for the
|
|
pixel-identical live-edit preview. Gated on copy permission (same as export)."""
|
|
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 {}
|
|
if perms.get("canCopy", True) is False:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not permitted (canCopy).")
|
|
data = doc_info.get("bytes_data")
|
|
if not data:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Document bytes unavailable")
|
|
return Response(content=bytes(data), media_type="application/pdf")
|
|
|
|
|
|
@router.get("/{document_id}/search", response_model=list[SearchMatch])
|
|
def search_document(
|
|
document_id: str,
|
|
q: str,
|
|
case_sensitive: bool = False,
|
|
whole_words: bool = False,
|
|
) -> list[SearchMatch]:
|
|
if not engine.is_available():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
|
detail="Engine bridge (bindings/python) not yet available.",
|
|
)
|
|
|
|
if not q:
|
|
return []
|
|
|
|
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")
|
|
|
|
try:
|
|
doc = doc_info["doc_instance"]
|
|
return compute_search_matches(doc, q, case_sensitive, whole_words)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
|
|
|
|
|
|
@router.get("/{document_id}/pages/{page_index}/model", response_model=PageModelResponse)
|
|
def get_page_model(document_id: str, page_index: int) -> PageModelResponse:
|
|
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")
|
|
|
|
try:
|
|
doc = doc_info["doc_instance"]
|
|
page = doc.get_page(page_index)
|
|
model = page.extract_document_model()
|
|
return build_page_model_response(model)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
|
except IndexError as e:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
|
except Exception as e:
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
|
|
|
|
|
|
@router.get("/{document_id}/annotations", response_model=list[AnnotationResponse])
|
|
def get_document_annotations(document_id: str) -> list[AnnotationResponse]:
|
|
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")
|
|
|
|
try:
|
|
doc = doc_info["doc_instance"]
|
|
all_annots = []
|
|
for i in range(doc.page_count):
|
|
try:
|
|
page = doc.get_page(i)
|
|
annots = page.extract_annotations()
|
|
for a in annots:
|
|
all_annots.append(AnnotationResponse(
|
|
id=a.id,
|
|
type=a.type,
|
|
x=a.x,
|
|
y=a.y,
|
|
width=a.width,
|
|
height=a.height,
|
|
color=a.color,
|
|
author=a.author,
|
|
content=a.content,
|
|
timestamp=getattr(a, "timestamp", None),
|
|
pageIndex=a.page_index,
|
|
thickness=getattr(a, "thickness", None),
|
|
paths=[[{"x": p.x, "y": p.y} for p in stroke] for stroke in getattr(a, "paths", [])],
|
|
quadPoints=getattr(a, "quad_points", []),
|
|
fieldName=getattr(a, "field_name", None),
|
|
fieldValue=getattr(a, "field_value", None),
|
|
fieldType=getattr(a, "field_type", None),
|
|
fieldFlags=getattr(a, "field_flags", None),
|
|
fieldOptions=getattr(a, "field_options", None),
|
|
))
|
|
except Exception:
|
|
pass
|
|
return all_annots
|
|
except Exception as e:
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
|
|
|
|
|
|
@router.get("/{document_id}/pages/{page_index}/display_list")
|
|
def get_display_list(document_id: str, page_index: int):
|
|
if not engine.is_available():
|
|
raise HTTPException(status_code=501, detail="Engine not available")
|
|
|
|
doc_info = document_store.get_document(document_id)
|
|
if not doc_info:
|
|
raise HTTPException(status_code=404, detail="Document not found")
|
|
|
|
try:
|
|
doc = doc_info["doc_instance"]
|
|
page = doc.get_page(page_index)
|
|
json_str = page.extract_display_list()
|
|
return Response(content=json_str, media_type="application/json")
|
|
except Exception as e:
|
|
logger.warning(f"Display list extraction failed for doc {document_id} page {page_index}: {e}")
|
|
return Response(content="[]", media_type="application/json")
|
|
|
|
|
|
@router.get("/{document_id}/pages/{page_index}/xobjects/{name}")
|
|
def get_image_xobject(document_id: str, page_index: int, name: str):
|
|
"""
|
|
Serve a PDF XObject (image or form) as PNG.
|
|
Primary: use C++ extractImageXObject.
|
|
Fallback: parse display list to find the CTM for this XObject name,
|
|
then render that region of the page via render_tile.
|
|
"""
|
|
if not engine.is_available():
|
|
raise HTTPException(status_code=501, detail="Engine not available")
|
|
|
|
doc_info = document_store.get_document(document_id)
|
|
if not doc_info:
|
|
raise HTTPException(status_code=404, detail="Document not found")
|
|
|
|
doc = doc_info["doc_instance"]
|
|
page = doc.get_page(page_index)
|
|
|
|
# --- Primary: C++ extraction (may be unimplemented) ---
|
|
try:
|
|
img_bytes = page.extract_image_xobject(name)
|
|
if img_bytes:
|
|
return Response(content=img_bytes, media_type="image/png")
|
|
except Exception:
|
|
pass # fall through to display-list render fallback
|
|
|
|
# --- Fallback: parse display list, find CTM for this XObject, render tile ---
|
|
try:
|
|
import json
|
|
import struct
|
|
|
|
dl_str = page.extract_display_list()
|
|
if not dl_str:
|
|
raise HTTPException(status_code=404, detail=f"XObject '{name}' not found and no display list available")
|
|
|
|
dl_ops = json.loads(dl_str)
|
|
|
|
# Walk ops tracking the current transformation matrix stack (cm ops)
|
|
# The last cm before a Do with matching name gives us the image region.
|
|
current_matrix = [1.0, 0.0, 0.0, 1.0, 0.0, 0.0]
|
|
matrix_stack: list[list[float]] = []
|
|
target_matrix: list[float] | None = None
|
|
|
|
for op in dl_ops:
|
|
op_name = op.get("op")
|
|
args = op.get("args", [])
|
|
|
|
if op_name == "q":
|
|
matrix_stack.append(list(current_matrix))
|
|
elif op_name == "Q" and matrix_stack:
|
|
current_matrix = matrix_stack.pop()
|
|
elif op_name == "cm" and len(args) >= 6:
|
|
# Concatenate matrix: new = args * current
|
|
a, b, c, d, e, f = [float(x) for x in args[:6]]
|
|
# Simple concatenation (row-major)
|
|
ca, cb, cc, cd, ce, cf = current_matrix
|
|
current_matrix = [
|
|
a * ca + b * cc,
|
|
a * cb + b * cd,
|
|
c * ca + d * cc,
|
|
c * cb + d * cd,
|
|
e * ca + f * cc + ce,
|
|
e * cb + f * cd + cf,
|
|
]
|
|
elif op_name == "Do" and len(args) >= 1:
|
|
xname = str(args[0]).lstrip("/")
|
|
if xname == name.lstrip("/"):
|
|
target_matrix = list(current_matrix)
|
|
break # found it
|
|
|
|
if target_matrix is None:
|
|
raise HTTPException(status_code=404, detail=f"XObject '{name}' not referenced on page {page_index}")
|
|
|
|
# Extract image region from matrix
|
|
# a=width scale, d=height scale, e=x, f=y (PDF bottom-left origin)
|
|
a, b, c, d, e, f_val = target_matrix
|
|
page_height = float(page.height)
|
|
|
|
img_w = abs(a) if abs(a) > 1 else abs(d)
|
|
img_h = abs(d) if abs(d) > 1 else abs(a)
|
|
x_pt = e
|
|
# Convert PDF Y (bottom-left) -> top-left
|
|
y_pt = page_height - f_val - img_h
|
|
|
|
# Clamp to page bounds
|
|
x_pt = max(0.0, x_pt)
|
|
y_pt = max(0.0, y_pt)
|
|
if img_w < 4:
|
|
img_w = float(page.width)
|
|
if img_h < 4:
|
|
img_h = float(page.height)
|
|
|
|
DPI = 144 # 2x for crisp rendering
|
|
_w, _h, raw_bgra = page.render_tile(DPI, x_pt, y_pt, img_w, img_h)
|
|
|
|
# Convert raw BGRA bytes to PNG using only stdlib
|
|
width_px: int = int(_w)
|
|
height_px: int = int(_h)
|
|
raw_bytes = bytes(raw_bgra)
|
|
|
|
import zlib
|
|
|
|
def _png_chunk(tag: bytes, data: bytes) -> bytes:
|
|
chunk = tag + data
|
|
return struct.pack(">I", len(data)) + chunk + struct.pack(">I", zlib.crc32(chunk) & 0xFFFFFFFF)
|
|
|
|
# IHDR
|
|
ihdr_data = struct.pack(">IIBBBBB", width_px, height_px, 8, 2, 0, 0, 0)
|
|
# Convert RGBA rows with filter byte 0 (render_tile already outputs RGBA)
|
|
rows = bytearray()
|
|
for row_i in range(height_px):
|
|
rows.append(0) # PNG filter byte: None
|
|
for col_i in range(width_px):
|
|
idx = (row_i * width_px + col_i) * 4
|
|
rows.append(raw_bytes[idx]) # R
|
|
rows.append(raw_bytes[idx + 1]) # G
|
|
rows.append(raw_bytes[idx + 2]) # B
|
|
|
|
idat_data = zlib.compress(bytes(rows), 6)
|
|
|
|
png = (
|
|
b"\x89PNG\r\n\x1a\n"
|
|
+ _png_chunk(b"IHDR", ihdr_data)
|
|
+ _png_chunk(b"IDAT", idat_data)
|
|
+ _png_chunk(b"IEND", b"")
|
|
)
|
|
return Response(content=png, media_type="image/png",
|
|
headers={"Cache-Control": "public, max-age=3600"})
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as ex:
|
|
raise HTTPException(status_code=500, detail=f"Failed to render XObject '{name}': {ex}")
|