adobe-to-docusign-migrator/bulk-send/bulk_send.py

#!/usr/bin/env python3
"""
bulk_send.py — DocuSign Bulk Send Prototype
--------------------------------------------
Sends a template to multiple recipients from a CSV file using
the DocuSign eSignature Bulk Send API (v2.1).

Usage:
    python3 bulk_send.py --template-id <TEMPLATE_ID> --csv recipients.csv

DocuSign API reference:
    https://developers.docusign.com/docs/esign-rest-api/how-to/bulk-send-envelopes/
"""

import os
import csv
import json
import argparse
import requests
from dotenv import load_dotenv
from auth_helper import get_access_token  # reuses existing JWT auth

load_dotenv()

ACCOUNT_ID = os.getenv("DOCUSIGN_ACCOUNT_ID")
BASE_URL = f"https://demo.docusign.net/restapi/v2.1/accounts/{ACCOUNT_ID}"


def load_recipients(csv_path: str) -> list[dict]:
    """Load recipients from CSV file."""
    recipients = []
    with open(csv_path, newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            recipients.append(row)
    print(f"✅ Loaded {len(recipients)} recipients from {csv_path}")
    return recipients


def create_bulk_send_list(token: str, recipients: list[dict], list_name: str = "Bulk Send Demo") -> str:
    """Create a bulk send list in DocuSign and return its ID."""
    url = f"{BASE_URL}/bulk_send_lists"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    bulk_copies = []
    for r in recipients:
        bulk_copies.append({
            "recipients": {
                "signers": [{
                    "name": r["Name"],
                    "email": r["Email"],
                    "roleName": "Signer",   # must match template role name
                    "tabs": {
                        "textTabs": [
                            {"tabLabel": "Company", "value": r.get("Company", "")},
                            {"tabLabel": "Title",   "value": r.get("Title", "")},
                        ]
                    }
                }]
            },
            "customFields": {
                "textCustomFields": [{
                    "name": "CustomField1",
                    "value": r.get("CustomField1", "")
                }]
            }
        })

    payload = {
        "name": list_name,
        "bulkCopies": bulk_copies
    }

    resp = requests.post(url, headers=headers, json=payload)
    resp.raise_for_status()
    list_id = resp.json()["listId"]
    print(f"✅ Bulk send list created: {list_id}")
    return list_id


def create_envelope_from_template(token: str, template_id: str, bulk_list_id: str) -> str:
    """Create a draft envelope from template, linked to the bulk send list."""
    url = f"{BASE_URL}/envelopes"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    payload = {
        "status": "created",   # draft — bulk send will send it
        "templateId": template_id,
        "templateRoles": [{
            "roleName": "Signer",
            "name": "Bulk Recipient",    # placeholder, overridden by bulk list
            "email": "bulk@placeholder.invalid"
        }],
        "customFields": {
            "textCustomFields": [{
                "name": "mailingListId",
                "value": bulk_list_id,
                "required": "false",
                "show": "false"
            }]
        }
    }

    resp = requests.post(url, headers=headers, json=payload)
    resp.raise_for_status()
    envelope_id = resp.json()["envelopeId"]
    print(f"✅ Draft envelope created: {envelope_id}")
    return envelope_id


def send_bulk(token: str, envelope_id: str, bulk_list_id: str) -> dict:
    """Trigger the bulk send."""
    url = f"{BASE_URL}/bulk_send_lists/{bulk_list_id}/send"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {"envelopeOrTemplateId": envelope_id}

    resp = requests.post(url, headers=headers, json=payload)
    resp.raise_for_status()
    result = resp.json()
    print(f"✅ Bulk send triggered!")
    print(f"   Batch ID:  {result.get('batchId')}")
    print(f"   Queued:    {result.get('queued')}")
    print(f"   Envelopes: {result.get('totalEnvelopes')}")
    return result


def main():
    parser = argparse.ArgumentParser(description="DocuSign Bulk Send Prototype")
    parser.add_argument("--template-id", required=True, help="DocuSign template ID to send")
    parser.add_argument("--csv", default="bulk-send/recipients.csv", help="Path to recipients CSV")
    parser.add_argument("--list-name", default="Bulk Send Demo", help="Name for the bulk send list")
    parser.add_argument("--dry-run", action="store_true", help="Create list + envelope but don't send")
    args = parser.parse_args()

    print("\n🚀 DocuSign Bulk Send Prototype")
    print("================================")

    token = get_access_token()
    recipients = load_recipients(args.csv)
    bulk_list_id = create_bulk_send_list(token, recipients, args.list_name)
    envelope_id = create_envelope_from_template(token, args.template_id, bulk_list_id)

    if args.dry_run:
        print("\n⚠️  Dry run — skipping actual send.")
        print(f"   Bulk List ID: {bulk_list_id}")
        print(f"   Envelope ID:  {envelope_id}")
    else:
        result = send_bulk(token, envelope_id, bulk_list_id)
        print(f"\n✅ Done! Batch ID: {result.get('batchId')}")


if __name__ == "__main__":
    main()