305 lines
11 KiB
Python
305 lines
11 KiB
Python
from typing import List
|
|
from fastapi import APIRouter, HTTPException, status, File, UploadFile
|
|
from pydantic import BaseModel
|
|
|
|
from app.services import engine
|
|
from app.services.store import document_store
|
|
|
|
router = APIRouter(prefix="/documents", tags=["documents"])
|
|
|
|
class DocumentInfoResponse(BaseModel):
|
|
id: str
|
|
filename: str
|
|
sizeBytes: int
|
|
totalPages: int
|
|
pageWidth: float
|
|
pageHeight: float
|
|
uploadedAt: str
|
|
status: str
|
|
|
|
@router.post("", response_model=DocumentInfoResponse, status_code=status.HTTP_201_CREATED)
|
|
async def upload_document(file: UploadFile = File(...), password: str = "") -> DocumentInfoResponse:
|
|
if not engine.is_available():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
|
detail="Engine bridge (bindings/python) not yet available."
|
|
)
|
|
|
|
bytes_data = await file.read()
|
|
try:
|
|
pdfengine = engine.require()
|
|
doc = pdfengine.PdfDocument.load_from_memory(bytes_data, password)
|
|
info = document_store.add_document(file.filename, bytes_data, doc)
|
|
return DocumentInfoResponse(
|
|
id=info["id"],
|
|
filename=info["filename"],
|
|
sizeBytes=info["sizeBytes"],
|
|
totalPages=info["totalPages"],
|
|
pageWidth=info["pageWidth"],
|
|
pageHeight=info["pageHeight"],
|
|
uploadedAt=info["uploadedAt"],
|
|
status=info["status"]
|
|
)
|
|
except ValueError as e:
|
|
detail = str(e)
|
|
if "Password required" in detail:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Password required")
|
|
elif "Invalid password" in detail:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid password")
|
|
else:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"Failed to load PDF: {str(e)}")
|
|
|
|
@router.get("", response_model=List[DocumentInfoResponse])
|
|
def list_documents() -> List[DocumentInfoResponse]:
|
|
if not engine.is_available():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
|
detail="Engine bridge (bindings/python) not yet available."
|
|
)
|
|
|
|
docs = document_store.list_documents()
|
|
return [
|
|
DocumentInfoResponse(
|
|
id=d["id"],
|
|
filename=d["filename"],
|
|
sizeBytes=d["sizeBytes"],
|
|
totalPages=d["totalPages"],
|
|
pageWidth=d.get("pageWidth", 612.0),
|
|
pageHeight=d.get("pageHeight", 792.0),
|
|
uploadedAt=d["uploadedAt"],
|
|
status=d["status"]
|
|
)
|
|
for d in docs
|
|
]
|
|
|
|
@router.get("/{document_id}", response_model=DocumentInfoResponse)
|
|
def get_document(document_id: str) -> DocumentInfoResponse:
|
|
if not engine.is_available():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
|
detail="Engine bridge (bindings/python) not yet available."
|
|
)
|
|
|
|
d = document_store.get_document(document_id)
|
|
if not d:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Document not found")
|
|
|
|
return DocumentInfoResponse(
|
|
id=d["id"],
|
|
filename=d["filename"],
|
|
sizeBytes=d["sizeBytes"],
|
|
totalPages=d["totalPages"],
|
|
pageWidth=d.get("pageWidth", 612.0),
|
|
pageHeight=d.get("pageHeight", 792.0),
|
|
uploadedAt=d["uploadedAt"],
|
|
status=d["status"]
|
|
)
|
|
|
|
@router.delete("/{document_id}")
|
|
def delete_document(document_id: str):
|
|
if not engine.is_available():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
|
detail="Engine bridge (bindings/python) not yet available."
|
|
)
|
|
|
|
deleted = document_store.delete_document(document_id)
|
|
if not deleted:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Document not found")
|
|
|
|
return {"success": True}
|
|
|
|
class DocumentMetadataResponse(BaseModel):
|
|
title: str
|
|
author: str
|
|
creator: str
|
|
producer: str
|
|
creation_date: str
|
|
modification_date: str
|
|
|
|
@router.get("/{document_id}/metadata", response_model=DocumentMetadataResponse)
|
|
def get_document_metadata(document_id: str) -> DocumentMetadataResponse:
|
|
if not engine.is_available():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
|
detail="Engine bridge (bindings/python) not yet available."
|
|
)
|
|
|
|
d = document_store.get_document(document_id)
|
|
if not d:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Document not found")
|
|
|
|
try:
|
|
doc = d["doc_instance"]
|
|
meta = doc.metadata
|
|
return DocumentMetadataResponse(
|
|
title=meta.title,
|
|
author=meta.author,
|
|
creator=meta.creator,
|
|
producer=meta.producer,
|
|
creation_date=meta.creation_date,
|
|
modification_date=meta.modification_date
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
|
|
|
|
class FontInfoResponse(BaseModel):
|
|
fontName: str
|
|
type: str
|
|
isEmbedded: bool
|
|
isSubset: bool
|
|
isVertical: bool
|
|
encoding: str
|
|
hasToUnicode: bool
|
|
cmapName: str
|
|
cidSystemInfo: str
|
|
subsetTag: str
|
|
sourceType: str
|
|
substitutedFrom: str
|
|
substitutedTo: str
|
|
normalizedFamily: str
|
|
internalFontId: str
|
|
flags: int
|
|
ascent: float
|
|
descent: float
|
|
capHeight: float
|
|
|
|
@router.get("/{document_id}/fonts", response_model=List[FontInfoResponse])
|
|
def get_document_fonts(document_id: str, start_page: int = 0, end_page: int = -1) -> List[FontInfoResponse]:
|
|
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"]
|
|
fonts = doc.get_fonts(start_page, end_page)
|
|
return [
|
|
FontInfoResponse(
|
|
fontName=f.font_name,
|
|
type=f.type,
|
|
isEmbedded=f.is_embedded,
|
|
isSubset=f.is_subset,
|
|
isVertical=f.is_vertical,
|
|
encoding=f.encoding,
|
|
hasToUnicode=f.has_to_unicode,
|
|
cmapName=f.cmap_name,
|
|
cidSystemInfo=f.cid_system_info,
|
|
subsetTag=f.subset_tag,
|
|
sourceType=f.source_type,
|
|
substitutedFrom=f.substituted_from,
|
|
substitutedTo=f.substituted_to,
|
|
normalizedFamily=f.normalized_family,
|
|
internalFontId=f.internal_font_id,
|
|
flags=f.flags,
|
|
ascent=f.ascent,
|
|
descent=f.descent,
|
|
capHeight=f.cap_height
|
|
)
|
|
for f in fonts
|
|
]
|
|
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_400_BAD_REQUEST, detail=str(e))
|
|
|
|
class SearchRect(BaseModel):
|
|
x: float
|
|
y: float
|
|
w: float
|
|
h: float
|
|
|
|
class SearchMatch(BaseModel):
|
|
pageIndex: int
|
|
rects: List[SearchRect]
|
|
text: str
|
|
|
|
@router.get("/{document_id}/search", response_model=List[SearchMatch])
|
|
def search_document(document_id: str, q: str) -> 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"]
|
|
matches = []
|
|
lower_query = q.lower()
|
|
query_len = len(lower_query)
|
|
|
|
for page_idx in range(doc.page_count):
|
|
page = doc.get_page(page_idx)
|
|
glyphs = page.extract_text_with_bounds()
|
|
if not glyphs:
|
|
continue
|
|
|
|
text_str = ""
|
|
char_to_glyph = []
|
|
for i, g in enumerate(glyphs):
|
|
s = g.get("text", "")
|
|
start_len = len(text_str)
|
|
text_str += s
|
|
for _ in range(len(text_str) - start_len):
|
|
char_to_glyph.append(i)
|
|
|
|
lower_text = text_str.lower()
|
|
idx = 0
|
|
while True:
|
|
idx = lower_text.find(lower_query, idx)
|
|
if idx == -1:
|
|
break
|
|
|
|
start_glyph_idx = char_to_glyph[idx]
|
|
end_glyph_idx = char_to_glyph[idx + query_len - 1]
|
|
|
|
rects = []
|
|
current_rect = None
|
|
|
|
for g_idx in range(start_glyph_idx, end_glyph_idx + 1):
|
|
g = glyphs[g_idx]
|
|
dom_y = page.height - (g["y"] + g["h"])
|
|
|
|
if current_rect is None:
|
|
current_rect = {"x": g["x"], "y": dom_y, "w": g["w"], "h": g["h"]}
|
|
else:
|
|
if abs(dom_y - current_rect["y"]) < g.get("fontSize", 12) * 0.5:
|
|
max_x = max(current_rect["x"] + current_rect["w"], g["x"] + g["w"])
|
|
current_rect["w"] = max_x - current_rect["x"]
|
|
current_rect["y"] = min(current_rect["y"], dom_y)
|
|
current_rect["h"] = max(current_rect["h"], g["h"])
|
|
else:
|
|
rects.append(SearchRect(**current_rect))
|
|
current_rect = {"x": g["x"], "y": dom_y, "w": g["w"], "h": g["h"]}
|
|
|
|
if current_rect:
|
|
rects.append(SearchRect(**current_rect))
|
|
|
|
matches.append(SearchMatch(
|
|
pageIndex=page_idx,
|
|
rects=rects,
|
|
text=text_str[idx:idx + query_len]
|
|
))
|
|
|
|
idx += 1
|
|
|
|
return matches
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) |