""" app/modules/editor/pdf_router.py FastAPI router for native PDF block editing and extraction. Integrated with the project's Drive storage and Auth system. """ from typing import Any, Optional, List, Dict from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel from pypdf import PdfReader import io import json import logging from sqlalchemy.orm import Session from app.db.database import get_db from app.middleware.auth import get_current_user from app.modules.auth.models.user_model import User from app.modules.drive.services.file_service import FileService from app.modules.drive.services.activity_service import ActivityService from app.modules.drive.constants import DriveRole, ActivityType from app.modules.signing.models.signature_imprint import SignatureImprint from app.modules.signing.models.signing_request import SigningRequest from app.modules.signing.services.signing_service import SigningService from app.modules.editor.pdf_geometry import Rect as PdfRect, visual_to_physical_rect, page_display_size from .pdf_controller import extract_blocks, apply_edits logger = logging.getLogger(__name__) def _rect_area(bbox: List[float]) -> float: return max(0.0, bbox[2] - bbox[0]) * max(0.0, bbox[3] - bbox[1]) def _bbox_overlap_ratio(a: List[float], b: List[float]) -> float: if not a or not b or len(a) < 4 or len(b) < 4: return 0.0 ix0 = max(a[0], b[0]) iy0 = max(a[1], b[1]) ix1 = min(a[2], b[2]) iy1 = min(a[3], b[3]) if ix1 <= ix0 or iy1 <= iy0: return 0.0 inter = (ix1 - ix0) * (iy1 - iy0) base = min(_rect_area(a), _rect_area(b)) or 1.0 return inter / base def _load_locked_regions_for_file( db: Session, file_id: int, pdf_bytes: Optional[bytes] = None, ) -> Dict[int, List[List[float]]]: """Load locked signature regions for a file keyed by page number. Returns a dict {page_number: [[x0, y0, x1, y1], ...]}. """ regions_by_page: Dict[int, List[List[float]]] = {} imprints = ( db.query(SignatureImprint) .filter(SignatureImprint.drive_file_id == file_id) .all() ) for imp in imprints: regions_by_page.setdefault(imp.page_number, []).append(list(imp.bbox_points)) completed = ( db.query(SigningRequest) .filter( SigningRequest.drive_file_id == file_id, SigningRequest.status == "completed", ) .all() ) have_imprints = {imp.signing_request_id for imp in imprints} missing = [ sr for sr in completed if sr.id not in have_imprints and (sr.placements_json or []) ] if missing and pdf_bytes: try: reader = PdfReader(io.BytesIO(pdf_bytes)) for sr in missing: sr_sig_id = sr.signature_id or "" for p in (sr.placements_json or []): if (p or {}).get("element_type") != "signature": continue page_idx = int(p.get("page", 0) or 0) - 1 if page_idx < 0 or page_idx >= len(reader.pages): continue pg = reader.pages[page_idx] pw = float(pg.mediabox.width) ph = float(pg.mediabox.height) rotation = int(pg.rotation or 0) % 360 pw_v, ph_v = page_display_size(pw, ph, rotation) vx = float(p.get("x", 0) or 0) vy = float(p.get("y", 0) or 0) vw = float(p.get("width", 0) or 0) vh = float(p.get("height", 0) or 0) vr = PdfRect(vx * pw_v, vy * ph_v, (vx + vw) * pw_v, (vy + vh) * ph_v) pr = visual_to_physical_rect(vr, pw, ph, rotation) screen_w = p.get("page_width_px") or 750.0 scale = pw_v / (screen_w or 750.0) base_font = float(p.get("font_size") or 20) s_name = (p.get("signer_name") or "").strip() s_desig = (p.get("signer_designation") or "").strip() fr = SigningService._compute_signature_full_rect( pr, base_font, scale, sr_sig_id, s_name, s_desig ) if (fr.x1 - fr.x0) <= 0 or (fr.y1 - fr.y0) <= 0: continue regions_by_page.setdefault(page_idx, []).append([ round(fr.x0, 2), round(fr.y0, 2), round(fr.x1, 2), round(fr.y1, 2), ]) except Exception as e: logger.warning(f"Lock-region fallback failed for file {file_id}: {e}") return regions_by_page router = APIRouter(prefix="/pdf", tags=["PDF Editor"]) class BlockEdit(BaseModel): block_id: str new_text: str font: Optional[str] = None size: Optional[float] = None color: Optional[str] = None bold: Optional[bool] = None italic: Optional[bool] = None underline: Optional[bool] = None bbox: Optional[List[float]] = None align: Optional[int] = 0 class PdfBlockAddition(BaseModel): type: str page: int x: float y: float width: float height: float content: str font_size: Optional[float] = 14 color: Optional[str] = "#000000" bg_color: Optional[str] = None align: Optional[int] = 0 bold: Optional[bool] = False italic: Optional[bool] = False font: Optional[str] = None class PageOperation(BaseModel): type: str page_index: Optional[int] = None target_index: Optional[int] = None width: Optional[float] = None height: Optional[float] = None class SaveEditsRequest(BaseModel): edits: List[BlockEdit] = [] additions: List[PdfBlockAddition] = [] page_ops: List[PageOperation] = [] @router.get("/extract-blocks/{file_id}") async def extract_blocks_route( file_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user), ): """Extract text/image blocks using PyMuPDF (extraction only — no Stirling call). Also attaches `locked_regions` per page so the frontend knows which areas are part of a signature imprint and must not be edited. """ file_svc = FileService(db) try: stream_data = file_svc.stream_file(file_id, user) pdf_bytes = stream_data["body"].read() except HTTPException: raise except Exception as e: logger.error(f"Failed to fetch PDF for extraction: {e}") raise HTTPException(status_code=500, detail="Could not read file from storage") try: result = extract_blocks(pdf_bytes) except HTTPException: raise except Exception as e: logger.exception("Block extraction failed") raise HTTPException(status_code=500, detail="Extraction failed") try: regions_by_page = _load_locked_regions_for_file(db, file_id, pdf_bytes) for page_entry in result.get("pages", []): pnum = page_entry.get("page_number") rects = regions_by_page.get(pnum, []) page_entry["locked_regions"] = [{"bbox": r} for r in rects] total = sum(len(v) for v in regions_by_page.values()) logger.info( "extract-blocks file_id=%s returning %d locked_region(s) across %d page(s)", file_id, total, len(regions_by_page), ) except Exception as e: logger.warning(f"Failed to attach locked_regions for file {file_id}: {e}") for page_entry in result.get("pages", []): page_entry.setdefault("locked_regions", []) return JSONResponse(result) @router.post("/save-blocks/{file_id}") async def save_blocks_route( file_id: int, body: SaveEditsRequest, db: Session = Depends(get_db), user: User = Depends(get_current_user), ): """ Apply edits via Stirling-PDF REST API and persist the result as a new version. Pipeline (inside apply_edits): 1. Stirling /redact — wipe original content 2. PyMuPDF overlay — draw replacement text 3. Stirling /overlay — composite onto base 4. PyMuPDF — insert replacement images """ file_svc = FileService(db) try: stream_data = file_svc.stream_file(file_id, user) original_bytes = stream_data["body"].read() except HTTPException: raise except Exception as e: logger.error(f"Failed to fetch PDF for saving: {e}") raise HTTPException(status_code=500, detail="Could not read file from storage") print(f"[DEBUG] Processing {len(body.edits)} edits and {len(body.additions)} additions for file {file_id}") for i, e in enumerate(body.edits): print(f" Edit {i}: block={e.block_id} text='{e.new_text[:30]}' color={e.color} bbox={e.bbox}") for i, a in enumerate(body.additions): print(f" Add {i}: type={a.type} text='{a.content[:30]}' x={a.x} y={a.y}") regions_by_page = _load_locked_regions_for_file(db, file_id, original_bytes) if regions_by_page: blocked: List[Dict[str, Any]] = [] original_blocks_by_page: Dict[int, Dict[str, List[float]]] = {} original_types_by_page: Dict[int, Dict[str, str]] = {} try: full_block_data = extract_blocks(original_bytes) for page_entry in full_block_data.get("pages", []): pnum = page_entry.get("page_number") original_blocks_by_page[pnum] = { b["block_id"]: b["bbox"] for b in page_entry.get("blocks", []) } original_types_by_page[pnum] = { b["block_id"]: b.get("type", "text") for b in page_entry.get("blocks", []) } except Exception as ex: logger.warning(f"Failed to pre-extract blocks for lock check: {ex}") def _page_num_from_block_id(block_id: str) -> int: try: parts = block_id.split("_") return int(parts[0][1:]) except Exception: return -1 def _intersects(pnum: int, bbox: List[float]) -> bool: for region in regions_by_page.get(pnum, []): if _bbox_overlap_ratio(bbox, region) > 0: return True return False def _center_in(pnum: int, bbox: List[float]) -> bool: if not bbox or len(bbox) < 4: return False cx = (bbox[0] + bbox[2]) / 2.0 cy = (bbox[1] + bbox[3]) / 2.0 for region in regions_by_page.get(pnum, []): if region[0] <= cx <= region[2] and region[1] <= cy <= region[3]: return True return False for edit in body.edits: bid = edit.block_id or "" base_bid = bid.split("_l")[0] pnum = _page_num_from_block_id(bid) if pnum not in regions_by_page: continue orig_bbox = original_blocks_by_page.get(pnum, {}).get(base_bid) btype = original_types_by_page.get(pnum, {}).get(base_bid, "text") if orig_bbox: if _intersects(pnum, orig_bbox): blocked.append({"kind": "edit", "block_id": bid, "reason": "original_in_signed_region"}) continue if btype == "image" and _center_in(pnum, orig_bbox): blocked.append({"kind": "edit", "block_id": bid, "reason": "image_center_in_signed_region"}) continue if edit.bbox and _intersects(pnum, list(edit.bbox)): blocked.append({"kind": "edit", "block_id": bid, "reason": "new_in_signed_region"}) try: _reader = PdfReader(io.BytesIO(original_bytes)) except Exception: _reader = None if _reader is not None: for add in body.additions: pnum = (add.page or 1) - 1 if pnum not in regions_by_page or pnum >= len(_reader.pages): continue pg = _reader.pages[pnum] mb_w, mb_h = float(pg.mediabox.width), float(pg.mediabox.height) rotation = int(pg.rotation or 0) % 360 pw, ph = page_display_size(mb_w, mb_h, rotation) add_bbox = [add.x * pw, add.y * ph, (add.x + add.width) * pw, (add.y + add.height) * ph] if _intersects(pnum, add_bbox): blocked.append({"kind": "addition", "page": pnum, "reason": "in_signed_region"}) for op in body.page_ops: if op.type == "delete" and op.page_index is not None and op.page_index in regions_by_page: blocked.append({"kind": "page_delete", "page_index": op.page_index, "reason": "page_has_signed_region"}) if blocked: try: ActivityService(db).log( resource_type="file", resource_id=file_id, actor_id=user.id, activity_type=ActivityType.EDIT_BLOCKED_ON_SIGNED_REGION, metadata=json.dumps({"blocked": blocked}), tenant_id=user.tenant_id, ) with db.begin_nested(): db.flush() except Exception as ex: logger.warning(f"Failed to log EDIT_BLOCKED_ON_SIGNED_REGION: {ex}") raise HTTPException( status_code=409, detail={ "error": "edit_blocked_on_signed_region", "message": "This area is part of a signature and cannot be modified.", "blocked": blocked, }, ) try: edited_bytes = await apply_edits( pdf_bytes = original_bytes, edits = [e.model_dump() for e in body.edits], additions = [a.model_dump() for a in body.additions], page_ops = [op.model_dump() for op in body.page_ops], ) except HTTPException: raise except Exception as e: logger.exception("Stirling save pipeline failed") raise HTTPException(status_code=500, detail="Edit application failed") try: await file_svc.update_pdf_content(file_id, edited_bytes, user) new_blocks = extract_blocks(edited_bytes) try: fresh_regions = _load_locked_regions_for_file(db, file_id, edited_bytes) for page_entry in new_blocks.get("pages", []): pnum = page_entry.get("page_number") rects = fresh_regions.get(pnum, []) page_entry["locked_regions"] = [{"bbox": r} for r in rects] except Exception as ex: logger.warning(f"Failed to attach locked_regions post-save: {ex}") for page_entry in new_blocks.get("pages", []): page_entry.setdefault("locked_regions", []) return JSONResponse({"status": "success", "new_blocks": new_blocks}) except HTTPException: raise except Exception as e: logger.error(f"Failed to persist modified PDF: {e}") raise HTTPException(status_code=500, detail="Failed to persist changes to storage")