385 lines
15 KiB
Python
385 lines
15 KiB
Python
"""
|
|
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") |