262 lines
9.7 KiB
Python
262 lines
9.7 KiB
Python
"""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()
|