from fastapi import APIRouter, File, HTTPException, Response, UploadFile, status from pydantic import BaseModel from app.services import engine from app.services.store import document_store router = APIRouter(prefix="/documents", tags=["documents"]) class PageInfoResponse(BaseModel): index: int width: float height: float class PermissionsResponse(BaseModel): isEncrypted: bool = False encryption: str = "None" securityRevision: int = -1 ownerUnlocked: bool = False canPrint: bool = True canPrintHighRes: bool = True canModify: bool = True canCopy: bool = True canAnnotate: bool = True canFillForms: bool = True canExtractForAccessibility: bool = True canAssemble: bool = True class DocumentInfoResponse(BaseModel): id: str filename: str sizeBytes: int totalPages: int pageWidth: float pageHeight: float uploadedAt: str status: str pages: list[PageInfoResponse] = [] permissions: PermissionsResponse = PermissionsResponse() def make_document_response(d: dict) -> DocumentInfoResponse: pages_list = [] if "doc_instance" in d: doc = d["doc_instance"] for i in range(doc.page_count): try: page = doc.get_page(i) pages_list.append(PageInfoResponse(index=i, width=page.width, height=page.height)) except Exception: pass perms = d.get("permissions") 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"], pages=pages_list, permissions=PermissionsResponse(**perms) if perms else PermissionsResponse(), ) @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 make_document_response(info) 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: {e!s}" ) @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 [make_document_response(d) 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 make_document_response(d) @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, 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") def _is_word_char(ch: str) -> bool: return ch.isalnum() or ch == "_" def _find_all(haystack: str, needle: str) -> list[int]: """Return start indices of all non-overlapping occurrences of needle in haystack.""" results: list[int] = [] start = 0 needle_len = len(needle) while True: pos = haystack.find(needle, start) if pos == -1: break if whole_words: before_ok = pos == 0 or not _is_word_char(haystack[pos - 1]) after_ok = (pos + needle_len) >= len(haystack) or not _is_word_char( haystack[pos + needle_len] ) if before_ok and after_ok: results.append(pos) else: results.append(pos) start = pos + 1 return results try: doc = doc_info["doc_instance"] matches = [] search_needle = q if case_sensitive else q.lower() query_len = len(search_needle) 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: list[int] = [] 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) search_text = text_str if case_sensitive else text_str.lower() for idx in _find_all(search_text, search_needle): if idx + query_len - 1 >= len(char_to_glyph): continue 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] ) ) return matches except Exception as e: raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) class GlyphModel(BaseModel): text: str unicode: int font_name: str flags: int font_size: float origin_x: float origin_y: float bbox_x: float bbox_y: float bbox_w: float bbox_h: float angle: float page_object_index: int = -1 class TextRunModel(BaseModel): text: str font_name: str flags: int font_size: float internal_font_id: str is_embedded: bool type: str glyphs: list[GlyphModel] x: float y: float w: float h: float object_indices: list[int] = [] class TextLineModel(BaseModel): runs: list[TextRunModel] baseline_y: float x: float y: float w: float h: float class ParagraphModel(BaseModel): lines: list[TextLineModel] x: float y: float w: float h: float class PageModelResponse(BaseModel): paragraphs: list[ParagraphModel] width: float height: float page_index: int @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() paragraphs = [] for p in model.paragraphs: lines = [] for line in p.lines: runs = [] for r in line.runs: glyphs = [] for g in r.glyphs: glyphs.append( GlyphModel( text=g.text, unicode=g.unicode, font_name=g.font_name, flags=g.flags, font_size=g.font_size, origin_x=g.origin_x, origin_y=g.origin_y, bbox_x=g.bbox_x, bbox_y=g.bbox_y, bbox_w=g.bbox_w, bbox_h=g.bbox_h, angle=g.angle, page_object_index=g.page_object_index, ) ) runs.append( TextRunModel( text=r.text, font_name=r.font_name, flags=r.flags, font_size=r.font_size, internal_font_id=r.internal_font_id, is_embedded=r.is_embedded, type=r.type, glyphs=glyphs, x=r.x, y=r.y, w=r.w, h=r.h, object_indices=r.object_indices, ) ) lines.append( TextLineModel( runs=runs, baseline_y=line.baseline_y, x=line.x, y=line.y, w=line.w, h=line.h, ) ) paragraphs.append(ParagraphModel(lines=lines, x=p.x, y=p.y, w=p.w, h=p.h)) return PageModelResponse( paragraphs=paragraphs, width=model.width, height=model.height, page_index=model.page_index, ) 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)) class AnnotationResponse(BaseModel): id: str type: str x: float y: float width: float height: float color: str author: str content: str timestamp: str | None = None pageIndex: int # Stroke geometry for ink annotations (top-left page-point space), so the # frontend can redraw them as an interactive overlay rather than a flat image. paths: list[list[dict[str, float]]] = [] # Form field properties fieldName: str | None = None fieldValue: str | None = None fieldType: str | None = None fieldFlags: int | None = None fieldOptions: list[str] | None = None @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, paths=[[{"x": p.x, "y": p.y} for p in stroke] for stroke in getattr(a, "paths", [])], 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)) class OutlineItemResponse(BaseModel): title: str pageIndex: int level: int @router.get("/{document_id}/outline", response_model=list[OutlineItemResponse]) def get_document_outline(document_id: str) -> list[OutlineItemResponse]: 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"] items = doc.extract_outline() return [ OutlineItemResponse(title=it["title"], pageIndex=it["pageIndex"], level=it["level"]) for it in items ] except Exception as e: raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e)) @router.get("/{document_id}/export") def export_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.", ) d = document_store.get_document(document_id) if not d: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Document not found") # Exporting a decrypted copy is a form of content extraction — gate on copy permission. perms = d.get("permissions") or {} if perms.get("canCopy", True) is False: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Exporting is not permitted by this document's restrictions (canCopy).", ) try: doc = d["doc_instance"] bytes_data = doc.save_full() filename = d["filename"] if not filename.endswith(".pdf"): filename += ".pdf" # pyrefly: ignore [missing-import] import sys import os roaming_path = os.path.join(os.environ.get("APPDATA", "C:\\Users\\azeem\\AppData\\Roaming"), "Python", "Python312", "site-packages") if roaming_path not in sys.path: sys.path.append(roaming_path) import pypdf import io # Parse the raw bytes and force NeedAppearances so the viewer # actually renders the filled values. reader = pypdf.PdfReader(io.BytesIO(bytes_data)) writer = pypdf.PdfWriter() writer.append(reader) acro_form = writer.root_object.get("/AcroForm") if acro_form is not None: acro_form_dict = acro_form.get_object() acro_form_dict[pypdf.generic.NameObject("/NeedAppearances")] = pypdf.generic.BooleanObject(True) out_stream = io.BytesIO() writer.write(out_stream) bytes_data = out_stream.getvalue() return Response( content=bytes_data, media_type="application/pdf", headers={ "Content-Disposition": f'attachment; filename="{filename}"', "Content-Length": str(len(bytes_data)), }, ) except Exception as e: raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))