Automate Compliance Certificate Generation at Scale with DocForge API

import requests
import json
import base64
from pathlib import Path
from typing import Optional
from dataclasses import dataclass
from datetime import date
 
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://apivult.com/api/docforge"
 
@dataclass
class CustomerComplianceData:
    """Data required to generate a GDPR DPA."""
    customer_name: str
    customer_address: str
    customer_contact_email: str
    effective_date: date
    data_categories: list[str]  # e.g. ["personal_data", "sensitive_data"]
    processing_purposes: list[str]
    retention_period_months: int
    sub_processors: list[dict]   # list of {name, country, purpose}
    jurisdiction: str            # e.g. "EU", "UK", "US"
    contract_reference: str

DPA_TEMPLATE = """
# Data Processing Agreement
 
**Effective Date:** {{effective_date}}  
**Contract Reference:** {{contract_reference}}
 
## Parties
 
**Data Controller:** {{customer_name}}  
**Address:** {{customer_address}}  
**Contact:** {{customer_contact_email}}
 
**Data Processor:** APIVult Ltd  
**Registered Office:** [Your Company Address]
 
---
 
## 1. Subject Matter and Duration
 
This Agreement governs the processing of personal data by the Processor on behalf of the Controller pursuant to the services described in Contract Reference {{contract_reference}}.
 
The processing activities covered by this Agreement are:
{{#each processing_purposes}}
- {{this}}
{{/each}}
 
The Agreement is effective from {{effective_date}} and continues for the duration of the underlying services agreement.
 
## 2. Categories of Data Subjects and Personal Data
 
The Processor processes the following categories of personal data on behalf of the Controller:
{{#each data_categories}}
- {{this}}
{{/each}}
 
## 3. Retention and Deletion
 
Personal data processed under this Agreement will be retained for no longer than {{retention_period_months}} months from the date of collection, unless longer retention is required by applicable law.
 
## 4. Sub-Processors
 
The Controller authorizes the Processor to engage the following sub-processors:
 
| Sub-Processor | Country | Purpose |
|---|---|---|
{{#each sub_processors}}
| {{name}} | {{country}} | {{purpose}} |
{{/each}}
 
## 5. Governing Law
 
This Agreement is governed by the laws of {{jurisdiction}}.
 
---
 
*Generated automatically on {{generation_date}}*
"""

def register_template(template_name: str, template_content: str) -> str:
    """
    Upload a document template to DocForge.
    Returns the template_id for subsequent document generation calls.
    """
    response = requests.post(
        f"{BASE_URL}/templates",
        json={
            "name": template_name,
            "content": template_content,
            "format": "markdown",
            "output_format": "pdf",
            "pdf_options": {
                "page_size": "A4",
                "margin_mm": 20,
                "include_page_numbers": True,
                "watermark": None  # Set to "DRAFT" for review copies
            }
        },
        headers={"X-RapidAPI-Key": API_KEY}
    )
    result = response.json()
    return result["template_id"]
 
# Register once; reuse for all DPAs
DPA_TEMPLATE_ID = register_template("gdpr_dpa_v1", DPA_TEMPLATE)
print(f"Template registered: {DPA_TEMPLATE_ID}")

from datetime import datetime
 
def generate_dpa(
    template_id: str,
    customer_data: CustomerComplianceData,
    output_path: Optional[str] = None
) -> bytes:
    """
    Generate a GDPR DPA PDF for a specific customer.
    
    Returns the PDF as bytes; optionally saves to file.
    """
    # Build template variables
    template_data = {
        "customer_name": customer_data.customer_name,
        "customer_address": customer_data.customer_address,
        "customer_contact_email": customer_data.customer_contact_email,
        "effective_date": customer_data.effective_date.strftime("%d %B %Y"),
        "contract_reference": customer_data.contract_reference,
        "processing_purposes": customer_data.processing_purposes,
        "data_categories": customer_data.data_categories,
        "retention_period_months": customer_data.retention_period_months,
        "sub_processors": customer_data.sub_processors,
        "jurisdiction": customer_data.jurisdiction,
        "generation_date": datetime.utcnow().strftime("%d %B %Y at %H:%M UTC")
    }
 
    response = requests.post(
        f"{BASE_URL}/generate",
        json={
            "template_id": template_id,
            "data": template_data,
            "metadata": {
                "document_type": "dpa",
                "customer_id": customer_data.contract_reference,
                "generated_at": datetime.utcnow().isoformat()
            }
        },
        headers={"X-RapidAPI-Key": API_KEY}
    )
    response.raise_for_status()
    result = response.json()
 
    # Download the generated PDF
    pdf_response = requests.get(
        result["download_url"],
        headers={"X-RapidAPI-Key": API_KEY}
    )
    pdf_bytes = pdf_response.content
 
    if output_path:
        Path(output_path).write_bytes(pdf_bytes)
        print(f"DPA saved: {output_path}")
 
    return pdf_bytes
 
# Generate a single DPA
customer = CustomerComplianceData(
    customer_name="Acme Corp Ltd",
    customer_address="1 Business Park, London EC1A 1BB",
    customer_contact_email="[email protected]",
    effective_date=date(2026, 4, 9),
    data_categories=["Employee personal data", "Customer contact data"],
    processing_purposes=[
        "HR management and payroll processing",
        "Customer relationship management"
    ],
    retention_period_months=84,  # 7 years
    sub_processors=[
        {"name": "AWS EMEA", "country": "Ireland", "purpose": "Cloud infrastructure"},
        {"name": "Stripe Inc", "country": "USA", "purpose": "Payment processing"}
    ],
    jurisdiction="EU (GDPR)",
    contract_reference="ACME-2026-0412"
)
 
pdf = generate_dpa(DPA_TEMPLATE_ID, customer, "output/acme_corp_dpa_2026.pdf")

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
 
def generate_bulk_dpa_batch(
    template_id: str,
    customers: list[CustomerComplianceData],
    output_dir: str,
    max_workers: int = 10
) -> dict:
    """
    Generate DPAs for multiple customers in parallel.
    
    Returns a summary with success/failure counts.
    """
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
 
    results = {"success": 0, "failed": 0, "files": []}
 
    def generate_for_customer(customer: CustomerComplianceData) -> tuple:
        filename = f"dpa_{customer.contract_reference.lower().replace('-', '_')}.pdf"
        filepath = str(output_path / filename)
        try:
            generate_dpa(template_id, customer, filepath)
            return (True, customer.contract_reference, filepath)
        except Exception as e:
            return (False, customer.contract_reference, str(e))
 
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(generate_for_customer, c): c
            for c in customers
        }
        for future in as_completed(futures):
            success, ref, detail = future.result()
            if success:
                results["success"] += 1
                results["files"].append(detail)
                print(f"✓ Generated DPA for {ref}")
            else:
                results["failed"] += 1
                print(f"✗ Failed for {ref}: {detail}")
 
    return results
 
# Generate 3 sample DPAs in parallel
customers = [
    CustomerComplianceData(
        customer_name="TechStart GmbH",
        customer_address="Unter den Linden 10, 10117 Berlin",
        customer_contact_email="[email protected]",
        effective_date=date(2026, 4, 9),
        data_categories=["Customer personal data"],
        processing_purposes=["CRM and customer analytics"],
        retention_period_months=36,
        sub_processors=[
            {"name": "AWS EMEA", "country": "Ireland", "purpose": "Cloud infrastructure"}
        ],
        jurisdiction="EU (GDPR)",
        contract_reference="TECH-2026-0089"
    ),
    # ... add more customers
]
 
summary = generate_bulk_dpa_batch(DPA_TEMPLATE_ID, customers, "output/dpa_batch")
print(f"\nCompleted: {summary['success']} generated, {summary['failed']} failed")