Auto-Generate Compliance Reports and Audit Certificates with an API

pip install requests python-dotenv jinja2

import os
import requests
from datetime import datetime
from dotenv import load_dotenv
 
load_dotenv()
 
DOCFORGE_API_KEY = os.getenv("DOCFORGE_API_KEY")
BASE_URL = "https://apivult.com/docforge/v1"
 
HEADERS = {
    "X-RapidAPI-Key": DOCFORGE_API_KEY,
    "Content-Type": "application/json",
}

def generate_aml_report(
    period_start: str,
    period_end: str,
    screening_results: list[dict],
    institution_name: str,
) -> bytes:
    """
    Generate a formatted AML transaction monitoring report as PDF.
    Returns the raw PDF bytes.
    """
    # Aggregate stats
    total = len(screening_results)
    cleared = sum(1 for r in screening_results if r["status"] == "clear")
    reviewed = sum(1 for r in screening_results if r["status"] == "review_required")
    blocked = sum(1 for r in screening_results if r["status"] == "confirmed_match")
 
    payload = {
        "template": "aml_monitoring_report_v2",
        "format": "pdf",
        "data": {
            "institution_name": institution_name,
            "report_period": {
                "start": period_start,
                "end": period_end,
            },
            "generated_at": datetime.utcnow().isoformat() + "Z",
            "summary": {
                "total_transactions_screened": total,
                "cleared": cleared,
                "flagged_for_review": reviewed,
                "blocked": blocked,
                "alert_rate": f"{((reviewed + blocked) / total * 100):.2f}%",
            },
            "alerts": [
                r for r in screening_results if r["status"] != "clear"
            ],
            "certifying_officer": institution_name + " Compliance Team",
        },
        "options": {
            "watermark": None,
            "page_numbers": True,
            "header_logo": True,
            "confidentiality_notice": "CONFIDENTIAL — For Regulatory Use Only",
        }
    }
 
    response = requests.post(
        f"{BASE_URL}/generate",
        headers=HEADERS,
        json=payload,
        timeout=30,
    )
    response.raise_for_status()
    return response.content
 
# Save the report
pdf_bytes = generate_aml_report(
    period_start="2026-03-01",
    period_end="2026-03-31",
    screening_results=get_monthly_screening_results(),
    institution_name="Acme Financial Services",
)
 
with open("aml_report_march_2026.pdf", "wb") as f:
    f.write(pdf_bytes)
 
print("AML report generated: aml_report_march_2026.pdf")

def generate_dsar_response(
    subject: dict,
    data_inventory: dict,
    request_received_date: str,
) -> bytes:
    """
    Generate a GDPR DSAR response document.
    subject: { name, email, id, request_reference }
    data_inventory: { categories, retention_periods, processing_purposes, third_party_sharing }
    """
    response = requests.post(
        f"{BASE_URL}/generate",
        headers=HEADERS,
        json={
            "template": "gdpr_dsar_response",
            "format": "pdf",
            "data": {
                "data_controller": "Your Company Name",
                "dpo_contact": "[email protected]",
                "subject": subject,
                "request_received": request_received_date,
                "response_date": datetime.utcnow().strftime("%Y-%m-%d"),
                "data_inventory": data_inventory,
                "rights_information": {
                    "right_to_erasure": True,
                    "right_to_restriction": True,
                    "right_to_portability": True,
                    "complaint_authority": "ICO (ico.org.uk) for UK residents",
                }
            },
            "options": {
                "page_numbers": True,
                "header_logo": True,
                "include_appendix": True,
            }
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.content
 
# Trigger from your DSAR request queue
dsar_pdf = generate_dsar_response(
    subject={
        "name": "Jane Doe",
        "email": "[email protected]",
        "id": "USER-88821",
        "request_reference": "DSAR-2026-0341",
    },
    data_inventory={
        "categories": ["profile_data", "transaction_history", "communications"],
        "retention_periods": {
            "profile_data": "Active account + 2 years",
            "transaction_history": "7 years (financial regulation)",
            "communications": "3 years",
        },
        "processing_purposes": [
            "Account management",
            "Fraud prevention",
            "Legal compliance",
        ],
        "third_party_sharing": [
            {"party": "Payment Processor", "purpose": "Transaction processing"},
        ]
    },
    request_received_date="2026-03-10",
)

def generate_vendor_certificate(
    vendor: dict,
    assessment: dict,
    risk_level: str,
    approved_by: str,
) -> bytes:
    """
    Generate a vendor risk assessment certificate.
    risk_level: 'low', 'medium', 'high', 'critical'
    """
    expiry_year = datetime.utcnow().year + 1
 
    response = requests.post(
        f"{BASE_URL}/generate",
        headers=HEADERS,
        json={
            "template": "vendor_risk_certificate",
            "format": "pdf",
            "data": {
                "vendor": vendor,
                "assessment_date": datetime.utcnow().strftime("%Y-%m-%d"),
                "valid_until": f"{expiry_year}-{datetime.utcnow().strftime('%m-%d')}",
                "risk_level": risk_level,
                "assessment_summary": assessment,
                "approved_by": approved_by,
                "certificate_number": f"VRC-{datetime.utcnow().strftime('%Y%m%d')}-{vendor['id']}",
                "next_review": f"{expiry_year}-{datetime.utcnow().strftime('%m-%d')}",
            },
            "options": {
                "include_qr_verification": True,  # QR code links to verification page
                "digital_seal": True,
                "watermark": risk_level.upper() if risk_level in ("high", "critical") else None,
            }
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.content

def generate_audit_evidence_package(
    audit_id: str,
    evidence_items: list[dict],
    output_path: str,
) -> str:
    """
    Generate multiple compliance documents and zip them into one package.
    Returns the path to the downloaded zip file.
    """
    response = requests.post(
        f"{BASE_URL}/batch",
        headers=HEADERS,
        json={
            "documents": evidence_items,
            "package": {
                "format": "zip",
                "filename": f"audit_evidence_{audit_id}",
                "include_index": True,  # auto-generate an index.pdf listing all docs
            }
        },
        timeout=120,
    )
    response.raise_for_status()
 
    zip_path = f"{output_path}/audit_evidence_{audit_id}.zip"
    with open(zip_path, "wb") as f:
        f.write(response.content)
 
    print(f"Audit evidence package saved: {zip_path}")
    return zip_path
 
# Example: Generate a SOC 2 audit evidence package
evidence_package = generate_audit_evidence_package(
    audit_id="SOC2-2026-Q1",
    evidence_items=[
        {
            "template": "access_control_log",
            "data": get_access_logs(period="2026-Q1"),
        },
        {
            "template": "incident_response_log",
            "data": get_incidents(period="2026-Q1"),
        },
        {
            "template": "change_management_log",
            "data": get_change_log(period="2026-Q1"),
        },
        {
            "template": "vendor_risk_summary",
            "data": get_vendor_risk_summary(),
        },
    ],
    output_path="./audit_packages",
)

import schedule
import time
 
def monthly_aml_report_job():
    from calendar import monthrange
    now = datetime.utcnow()
    last_month = now.month - 1 if now.month > 1 else 12
    year = now.year if now.month > 1 else now.year - 1
    last_day = monthrange(year, last_month)[1]
 
    pdf_bytes = generate_aml_report(
        period_start=f"{year}-{last_month:02d}-01",
        period_end=f"{year}-{last_month:02d}-{last_day}",
        screening_results=get_monthly_screening_results(year, last_month),
        institution_name="Acme Financial Services",
    )
 
    report_path = f"./reports/aml_report_{year}_{last_month:02d}.pdf"
    with open(report_path, "wb") as f:
        f.write(pdf_bytes)
 
    notify_compliance_team(report_path)
    print(f"[✓] Monthly AML report delivered: {report_path}")
 
# Run on the 1st of each month at 8 AM
schedule.every().month.at("08:00").do(monthly_aml_report_job)
 
while True:
    schedule.run_pending()
    time.sleep(3600)