7ed06780bb
Capture-to-form signature placement so the customer's drawn or typed signature lands right on the signature rule of the actual form, not in a sidecar page. - migration 085: esign_records.signature_anchors (JSONB exact PDF coords, lower-left origin, points) + signed_document_minio_key - signature_stamper.py: signature_box() anchors; anchors_from_acroform() pulls the signature field /Rect from a real AcroForm (e.g. MCS-150 certifySignature); stamp_signature() overlays PNG (auto-trimmed so ink rests on the rule) or typed name, scaled to actual page size - state_trucking_authorization.py: renders the Limited Authorization to File PDF and returns (pdf_bytes, anchors) - esign_stamp.py: stamp_esign_document() downloads unsigned PDF, stamps, uploads _signed.pdf, sets signed_document_minio_key (idempotent) - dot_esign.py: extract certifySignature anchor for MCS-150/closeout forms so the federal perjury cert is signed on the line - state_trucking.py: authorization gate — first run emails signing link and PAUSES; resumes with client_approved after signing - job_server handle_esign_completed: stamp then re-dispatch - tests: test_signature_placement.py (custom form), and test_mcs150_signature_placement.py (official AcroForm) both assert the signature lands inside the recorded signature box (verified visually)
122 lines
4.6 KiB
Python
122 lines
4.6 KiB
Python
"""Stamp the captured signature onto the form PDF stored in MinIO.
|
|
|
|
Glue between the e-sign record (signature image + recorded anchor coordinates)
|
|
and the actual form document. After a client signs, call
|
|
:func:`stamp_esign_document` to:
|
|
|
|
1. read the signature, type, and signature_anchors from esign_records;
|
|
2. download the unsigned form PDF (document_minio_key) from MinIO;
|
|
3. stamp the signature onto the recorded signature line(s);
|
|
4. upload the flattened signed PDF and store signed_document_minio_key.
|
|
|
|
Safe to call more than once — it is idempotent on signed_document_minio_key.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import tempfile
|
|
|
|
LOG = logging.getLogger("workers.services.esign_stamp")
|
|
|
|
|
|
def stamp_esign_document(esign_record_id: int) -> str | None:
|
|
"""Stamp the signature onto the form for one esign_records row.
|
|
|
|
Returns the signed document's MinIO key, or None if nothing was stamped
|
|
(no document, no anchors, or already stamped).
|
|
"""
|
|
try:
|
|
import psycopg2
|
|
except ImportError: # pragma: no cover
|
|
LOG.error("psycopg2 unavailable — cannot stamp signature")
|
|
return None
|
|
|
|
conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""SELECT order_number, document_type, signature_type,
|
|
signature_data, document_minio_key, signature_anchors,
|
|
signed_document_minio_key
|
|
FROM esign_records WHERE id = %s""",
|
|
(esign_record_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
LOG.warning("[esign_stamp] No esign record %s", esign_record_id)
|
|
return None
|
|
|
|
(order_number, document_type, signature_type, signature_data,
|
|
document_key, anchors_raw, already_signed_key) = row
|
|
|
|
if already_signed_key:
|
|
LOG.info("[esign_stamp] %s already stamped (%s)", order_number, already_signed_key)
|
|
return already_signed_key
|
|
|
|
if not document_key:
|
|
LOG.info("[esign_stamp] %s has no form PDF to stamp", order_number)
|
|
return None
|
|
|
|
anchors = anchors_raw if isinstance(anchors_raw, list) else (
|
|
json.loads(anchors_raw) if anchors_raw else []
|
|
)
|
|
if not anchors:
|
|
LOG.info("[esign_stamp] %s has no signature anchors — skipping stamp", order_number)
|
|
return None
|
|
|
|
if not signature_type or not signature_data:
|
|
LOG.warning("[esign_stamp] %s missing signature data", order_number)
|
|
return None
|
|
|
|
# Download the unsigned form PDF.
|
|
try:
|
|
from scripts.document_gen.minio_client import MinioStorage
|
|
except ImportError: # pragma: no cover - alt sys.path layouts
|
|
from document_gen.minio_client import MinioStorage # type: ignore
|
|
try:
|
|
from .signature_stamper import stamp_signature
|
|
except ImportError: # pragma: no cover
|
|
from signature_stamper import stamp_signature # type: ignore
|
|
|
|
storage = MinioStorage()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
local_in = os.path.join(tmp, "form.pdf")
|
|
storage.download(document_key, local_in)
|
|
with open(local_in, "rb") as f:
|
|
pdf_bytes = f.read()
|
|
|
|
signed_bytes = stamp_signature(
|
|
pdf_bytes,
|
|
anchors,
|
|
signature_type=signature_type,
|
|
signature_data=signature_data,
|
|
signer_field="signer",
|
|
)
|
|
|
|
local_out = os.path.join(tmp, "form_signed.pdf")
|
|
with open(local_out, "wb") as f:
|
|
f.write(signed_bytes)
|
|
|
|
# Signed key sits next to the original with a _signed suffix.
|
|
if document_key.endswith(".pdf"):
|
|
signed_key = document_key[:-4] + "_signed.pdf"
|
|
else:
|
|
signed_key = document_key + "_signed.pdf"
|
|
storage.upload(local_out, signed_key, content_type="application/pdf")
|
|
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"UPDATE esign_records SET signed_document_minio_key = %s, updated_at = NOW() WHERE id = %s",
|
|
(signed_key, esign_record_id),
|
|
)
|
|
conn.commit()
|
|
LOG.info("[esign_stamp] %s signature stamped onto form -> %s", order_number, signed_key)
|
|
return signed_key
|
|
except Exception as exc:
|
|
conn.rollback()
|
|
LOG.error("[esign_stamp] Failed for record %s: %s", esign_record_id, exc)
|
|
return None
|
|
finally:
|
|
conn.close()
|