"""Backfill signature_imprints rows for documents signed before the signature-flattening feature was introduced. For each completed SigningRequest that does not yet have any SignatureImprint rows, this script: 1. Loads the signed PDF bytes from storage. 2. Iterates the stored placements_json. 3. Uses the same bbox math as signing_service to compute the full enclosing rectangle (frame + label + ID + name + designation) per signature. 4. Inserts one SignatureImprint row per signature placement. The script does NOT modify PDF bytes — pre-existing signed documents keep their live-text signatures as they were. It only records the locked regions so the PDF block editor will refuse to move, delete, or overlay the signature. Idempotent: safe to run multiple times. Use --dry-run to preview without writing. Usage (from the docqube_backend directory): python -m scripts.backfill_signature_imprints [--dry-run] """ import argparse import io import logging import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from pypdf import PdfReader from app.db.database import SessionLocal from app.modules.drive.models.drive_model import DriveFile, DriveFileVersion from app.modules.drive.storage.drive_storage_service import DriveStorageService from app.modules.editor.pdf_geometry import Rect as PdfRect, visual_to_physical_rect, page_display_size from app.modules.signing.models.signing_request import SigningRequest from app.modules.signing.models.signature_imprint import SignatureImprint from app.modules.signing.services.signing_service import SigningService logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger("backfill_signature_imprints") def backfill(dry_run: bool = False) -> None: db = SessionLocal() storage = DriveStorageService(db) total_scanned = 0 total_skipped_existing = 0 total_skipped_failed = 0 total_inserted_rows = 0 total_requests_backfilled = 0 try: completed_requests = ( db.query(SigningRequest) .filter(SigningRequest.status == "completed") .order_by(SigningRequest.created_at.asc()) .all() ) logger.info("Found %d completed signing requests", len(completed_requests)) for sr in completed_requests: total_scanned += 1 existing_count = ( db.query(SignatureImprint) .filter(SignatureImprint.signing_request_id == sr.id) .count() ) if existing_count > 0: total_skipped_existing += 1 continue placements = sr.placements_json or [] if not placements: continue drive_file: DriveFile = ( db.query(DriveFile).filter(DriveFile.id == sr.drive_file_id).first() ) if not drive_file: logger.warning( "SigningRequest %s references missing drive_file %s; skipping", sr.id, sr.drive_file_id, ) total_skipped_failed += 1 continue # Identify which version number was produced by this signing request signed_version_number = None meta = sr.signing_metadata or {} if isinstance(meta, dict): signed_version_number = meta.get("version_number") version: DriveFileVersion if signed_version_number is not None: version = ( db.query(DriveFileVersion) .filter( DriveFileVersion.file_id == drive_file.id, DriveFileVersion.version_number == signed_version_number, ) .first() ) else: version = ( db.query(DriveFileVersion) .filter(DriveFileVersion.file_id == drive_file.id) .order_by(DriveFileVersion.version_number.desc()) .first() ) if not version: logger.warning( "No version found for drive_file %s (signing_request %s); skipping", drive_file.id, sr.id, ) total_skipped_failed += 1 continue try: obj = storage.get_object(version.s3_key, tenant_id=drive_file.tenant_id) pdf_bytes = obj["Body"].read() except Exception as e: logger.warning( "Failed to load PDF for drive_file %s (signing_request %s): %s", drive_file.id, sr.id, e, ) total_skipped_failed += 1 continue try: doc = PdfReader(io.BytesIO(pdf_bytes)) except Exception as e: logger.warning( "Failed to open PDF for drive_file %s: %s", drive_file.id, e ) total_skipped_failed += 1 continue try: rows_for_this_request = 0 for p in placements: 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(doc.pages): continue page = doc.pages[page_idx] mb_w, mb_h = float(page.mediabox.width), float(page.mediabox.height) rotation = int(page.rotation or 0) % 360 pw_v, ph_v = page_display_size(mb_w, mb_h, rotation) x = float(p.get("x", 0) or 0) y = float(p.get("y", 0) or 0) w = float(p.get("width", 0) or 0) h = float(p.get("height", 0) or 0) visual_rect = PdfRect( x * pw_v, y * ph_v, (x + w) * pw_v, (y + h) * ph_v, ) physical_rect = visual_to_physical_rect(visual_rect, mb_w, mb_h, 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) sig_id = sr.signature_id or "" signer_name = (p.get("signer_name") or "").strip() signer_designation = (p.get("signer_designation") or "").strip() full_rect = SigningService._compute_signature_full_rect( physical_rect, base_font, scale, sig_id, signer_name, signer_designation, ) if (full_rect.x1 - full_rect.x0) <= 0 or (full_rect.y1 - full_rect.y0) <= 0: continue bbox_points = [ round(full_rect.x0, 2), round(full_rect.y0, 2), round(full_rect.x1, 2), round(full_rect.y1, 2), ] if dry_run: logger.info( "[dry-run] would insert imprint: file=%s page=%s bbox=%s sr=%s", drive_file.id, page_idx, bbox_points, sr.id, ) else: db.add( SignatureImprint( signing_request_id=sr.id, drive_file_id=drive_file.id, page_number=page_idx, bbox_points=bbox_points, ) ) rows_for_this_request += 1 if rows_for_this_request > 0: total_requests_backfilled += 1 total_inserted_rows += rows_for_this_request logger.info( "Backfilled %d imprint row(s) for signing_request %s (file %s)", rows_for_this_request, sr.id, drive_file.id, ) if not dry_run: db.commit() except Exception as e: logger.warning( "Failed while processing placements for drive_file %s: %s", drive_file.id, e ) total_skipped_failed += 1 finally: db.close() logger.info("─── Summary ───") logger.info("Scanned signing requests: %d", total_scanned) logger.info("Already had imprints: %d", total_skipped_existing) logger.info("Backfilled signing requests: %d", total_requests_backfilled) logger.info("Total imprint rows inserted: %d", total_inserted_rows) logger.info("Skipped (errors / no version): %d", total_skipped_failed) if dry_run: logger.info("(dry-run — no rows were actually inserted)") def main(): parser = argparse.ArgumentParser( description="Backfill signature_imprints rows for pre-existing signed files." ) parser.add_argument( "--dry-run", action="store_true", help="Compute imprint rects and log them but do NOT insert rows.", ) args = parser.parse_args() backfill(dry_run=args.dry_run) if __name__ == "__main__": main()