- Create a submission
- Poll for job completion
- Download the final output documents
1. List User Teams
The first API call to make is List User Teams. This endpoint returns the teams associated with your user account, including theteam_uuid, division_uuid, and available workflow statuses for each team.
Example code:
send_and_track_submission.py
Copy
Ask AI
import os
import time
from pathlib import Path
import requests
from pingintel_api.pingvision import types as t
POLL_INTERVAL = 10 # seconds
def color_text(text: str, color: str) -> str:
"""Wrap text in ANSI color codes for terminal output."""
colors = {"red": "\033[31m", "yellow": "\033[33m", "green": "\033[32m"}
if color not in colors:
return text
reset = "\033[0m"
return f"{colors[color]}{text}{reset}"
# Configuration
API_KEY = os.environ.get("PINGVISION_AUTH_TOKEN")
BASE_URL = "https://vision.pingintel.com/api/v1"
headers = {"Authorization": f"Token {API_KEY}"}
# File to submit
file_path = Path("demo_email.eml")
files = [("files", (file_path.name, open(file_path, "rb")))]
## ...
# 1. List User Teams
# https://docs.pingintel.com/ping-vision/user-memberships/list-user-teams
list_teams_url = f"{BASE_URL}/user/teams/"
list_teams_response = requests.get(list_teams_url, headers=headers)
if list_teams_response.status_code not in (200, 201):
raise RuntimeError(f"Failed to list teams: {list_teams_response.status_code}")
teams = list_teams_response.json()
# select team by team_uuid or division_uuid
team = teams[0]
team_uuid = team["team_uuid"] # Unique identifier for the team
division_uuid = team["division_uuid"] # Division within the team
# Build a mapping of workflow status names to UUIDs for later use
status_uuids = {s["name"]: s["uuid"] for s in team.get("statuses", [])}
## ...
JSON output
Copy
Ask AI
[
{
"company_name": "Ping Intel",
"company_short_name": "PING",
"company_uuid": "12345678-abc1-cde2-efg3-123456789abc",
"division_name": "Ping Intel",
"division_uuid": "12345678-abc1-cde2-efg3-123456789abcd",
"id": 1,
"membership_type": "admin",
"statuses": [
{
"category": "Received",
"color": "#06A77C",
"division_uuid": "12345678-abc1-cde2-efg3-123456789abcd",
"iconName": "PlaceholderPing",
"id": 1001,
"is_valid_for_user_transition": false,
"name": "Received",
"uuid": "019c6d26-4468-7b89-8328-d840dc9b5e3b"
},
{
"category": "Initializing",
"color": "#8239FC",
"division_uuid": "12345678-abc1-cde2-efg3-123456789abcd",
"iconName": "PlaceholderPing",
"id": 1002,
"is_valid_for_user_transition": false,
"name": "Initializing",
"uuid": "019c6d26-4471-7e1f-a9b3-12acce3b0ba5"
},
"..."
],
"team_name": "Ping Intel",
"team_uuid": "9876543-abc1-cde2-efg3-123456789abcd"
},
"..."
]
2. Create Submission
The second API call to make is Initiate New Submission. This endpoint allows the user to upload files to create a new submission for processing. Example code:send_and_track_submission.py
Copy
Ask AI
## ...
# Step 2. Create Submission
# https://docs.pingintel.com/ping-vision/create-submission/initiate-new-submission
payload = {
"client_ref": "my_salesforce_id", # Optional: your external reference ID
"insured_name": "Acme Corp", # Optional: name of the insured party
"team_uuid": team_uuid, # Required: which team should receive this submission
}
create_submission_url = f"{BASE_URL}/submission"
create_submission_response = requests.post(create_submission_url, data=payload, files=files, headers=headers)
if create_submission_response.status_code not in (200, 201):
raise RuntimeError(f"Failed to create submission: {create_submission_response.status_code}")
submission_id = create_submission_response.json()["id"]
print(f"Created submission: {color_text(submission_id, 'green')}")
## ...
JSON output
Copy
Ask AI
{
"id": "p-mk-ourke-esw5ab",
"message": "OK",
"url": "http://vision.pingintel.com/submission/i/p-mk-ourke-esw5ab"
}
3. Poll for Job Completion
After submission, poll List Recent Submission Activity to monitor job progress. Ping Vision processes the document through multiple jobs:- SOVFIXER: Parses and validates the SOV data, geocodes addresses, enriches with third-party data
- RUN_OUTPUTTERS: Generates final output files in configured formats
send_and_track_submission.py
Copy
Ask AI
## ...
# Step 3. Poll for Job Completion
submission_activity_url = f"{BASE_URL}/submission"
print(color_text("Polling for jobs to complete...", "yellow"))
# track completed jobs
completed_jobs = set()
while True:
# Query for the specific submission by ID
activity_params = {
"id": submission_id,
"page_size": 1,
"team_uuid": team_uuid,
"division_uuid": division_uuid,
}
activity_response = requests.get(submission_activity_url, params=activity_params, headers=headers)
if activity_response.status_code not in (200, 201):
print(color_text(f"Error checking activity: {activity_response.status_code}", "red"))
time.sleep(POLL_INTERVAL)
continue
results = activity_response.json().get("results", [])
if not results:
print(color_text("Waiting for submission data...", "yellow"))
time.sleep(POLL_INTERVAL)
continue
submission = results[0]
jobs = submission.get("jobs", [])
# Display progress for each job
for j in jobs:
job_id = j.get("job_id")
job_type = j.get("job_type", "UNKNOWN") # e.g., "SOVFIXER", "RUN_OUTPUTTERS"
pct = j.get("processing_pct_complete", 0)
status = j.get("processing_status", "?")
message = j.get("processing_last_message", "") # Human-readable status message
if status == "C" and job_id not in completed_jobs:
# job completed
print(color_text(f"✓ {job_type}: {pct:.0f}% - {message}", "green"))
completed_jobs.add(job_id)
elif status != "C" and job_id not in completed_jobs:
# job still in progress
print(color_text(f"⋯ {job_type}: {pct:.0f}% - {message}", "yellow"))
# Check if the final outputters job is complete
outputters_job = next((j for j in jobs if j.get("job_type") == "RUN_OUTPUTTERS"), None)
if outputters_job and outputters_job.get("processing_pct_complete") == 100:
print(color_text("Final outputs ready for download.", "green"))
break
time.sleep(POLL_INTERVAL)
## ...
JSON output
Copy
Ask AI
{
"cursor_id": "p-mk-ourke-esw5ab",
"has_remaining": false,
"results": [
{
"actions": {
"claim": true,
"download_documents": true,
"view_map": true
},
"automated_processing_failed": false,
"automated_processing_failed_reason": null,
"...": "...",
"documents": [
{
"actions": ["download", "archive"],
"archived_on": null,
"archived_reason": null,
"created_time": "2026-02-20T22:21:22.216034Z",
"document_type": "MSG",
"filename": "demo_email.eml",
"id": 625,
"is_archived": false,
"label": "MSG",
"preview_url": null,
"url": "http://vision.pingintel.com/api/v1/submission/p-mk-ourke-esw5ab/document/demo_email.eml"
},
{
"actions": ["download", "archive"],
"document_type": "SOVFIXER_OUTPUT",
"filename": "Ping Marketing SOV - Acme Corp.xlsx",
"is_archived": false,
"url": "http://vision.pingintel.com/api/v1/submission/p-mk-ourke-esw5ab/document/Ping%20Marketing%20SOV%20-%20Acme%20Corp.xlsx"
},
{
"actions": ["download", "archive"],
"document_type": "SOVFIXER_JSON",
"filename": "parse_sov_testfile-20260220222216.json",
"is_archived": false,
"url": "http://vision.pingintel.com/api/v1/submission/p-mk-ourke-esw5ab/document/parse_sov_testfile-20260220222216.json"
},
"..."
],
"jobs": [
{
"created_time": "2026-02-20T22:21:26.655911Z",
"filenames": ["parse_sov_testfile.xlsx", "Email - 02-20-2026.content.html"],
"job_id": "7f3444de-0eaa-11f1-ba93-befce6b35b80",
"job_type": "SOVFIXER",
"job_type_details": {
"job_type_label": "Ping.Extraction",
"sovfixer_request_type": null,
"sovfixer_result_message": null,
"sovfixer_result_status": null,
"sovfixer_sovid": "s-mk-ourke-g1bnkd"
},
"processing_last_message": "Processing complete.",
"processing_pct_complete": 100.0,
"processing_status": "C",
"sovid": "s-mk-ourke-g1bnkd",
"updated_time": "2026-02-20T22:22:16.102475Z",
"user_id": 28
},
{
"job_id": "8a4555ef-1fbb-22g2-cb94-cgadf7c46c91",
"job_type": "RUN_OUTPUTTERS",
"processing_last_message": "Processing complete.",
"processing_pct_complete": 100.0,
"processing_status": "C"
}
],
"...": "..."
}
]
}
4. Download Final Output Documents
The final API call to make is Download Submission Document. This endpoint allows the user to download the processed output documents by providing the submission ID and the document filename. Once processing is complete, the submission contains output documents:- SOVFIXER_OUTPUT: The scrubbed/normalized SOV file (Excel format)
- SOVFIXER_JSON: Machine-readable JSON with all extracted and enriched data
https://vision.pingintel.com/api/v1/submission/p-mk-ourke-esw5ab/document/parse_sov_testfile-20260220222216.json
Example code:
send_and_track_submission.py
Copy
Ask AI
## ...
# Step 4. Download Final Output Documents
# https://docs.pingintel.com/ping-vision/get-submission-data/download-submission-document
os.makedirs("workflow_example_results", exist_ok=True)
for doc in submission.get("documents", []):
doc_type = doc.get("document_type", "")
filename = doc.get("filename", "")
# Download only the final output files
if doc_type in ["SOVFIXER_OUTPUT", "SOVFIXER_JSON"] and not doc.get("is_archived", False):
download_url = f"{BASE_URL}/submission/{submission_id}/document/{filename}"
download_response = requests.get(download_url, headers=headers)
if download_response.status_code not in (200, 201):
raise RuntimeError(f"Failed to download {filename}: {download_response.status_code}")
# Save with submission ID prefix to avoid filename collisions
output_path = f"workflow_example_results/{submission_id}_{filename}"
with open(output_path, "wb") as outfile:
outfile.write(download_response.content)
print(color_text(f"Saved {doc_type}: {output_path}", "green"))
print(color_text(f"\nWorkflow complete for submission {submission_id}", "green"))
##
Python output
Copy
Ask AI
Saved SOVFIXER_OUTPUT: workflow_example_results/p-mk-ourke-esw5ab_Ping Marketing SOV - Acme Corp.xlsx
Saved SOVFIXER_JSON: workflow_example_results/p-mk-ourke-esw5ab_parse_sov_testfile-20260220222216.json
Workflow complete for submission p-mk-ourke-esw5ab
Python Demo
Download Python Script here ||| Download Example Email heresend_and_track_submission.py
Copy
Ask AI
import os
import time
from pathlib import Path
import requests
from pingintel_api.pingvision import types as t
POLL_INTERVAL = 10 # seconds
def color_text(text: str, color: str) -> str:
"""Wrap text in ANSI color codes for terminal output."""
colors = {"red": "\033[31m", "yellow": "\033[33m", "green": "\033[32m"}
if color not in colors:
return text
reset = "\033[0m"
return f"{colors[color]}{text}{reset}"
# Configuration
API_KEY = os.environ.get("PINGVISION_AUTH_TOKEN")
BASE_URL = "https://vision.pingintel.com/api/v1"
headers = {"Authorization": f"Token {API_KEY}"}
# File to submit
file_path = Path("demo_email.eml")
files = [("files", (file_path.name, open(file_path, "rb")))]
## ...
# 1. List User Teams
# https://docs.pingintel.com/ping-vision/user-memberships/list-user-teams
list_teams_url = f"{BASE_URL}/user/teams/"
list_teams_response = requests.get(list_teams_url, headers=headers)
if list_teams_response.status_code not in (200, 201):
raise RuntimeError(f"Failed to list teams: {list_teams_response.status_code}")
teams = list_teams_response.json()
# select team by team_uuid or division_uuid
team = teams[0]
team_uuid = team["team_uuid"] # Unique identifier for the team
division_uuid = team["division_uuid"] # Division within the team
# Build a mapping of workflow status names to UUIDs for later use
status_uuids = {s["name"]: s["uuid"] for s in team.get("statuses", [])}
## ...
# Step 2. Create Submission
# https://docs.pingintel.com/ping-vision/create-submission/initiate-new-submission
payload = {
"client_ref": "my_salesforce_id", # Optional: your external reference ID
"insured_name": "Acme Corp", # Optional: name of the insured party
"team_uuid": team_uuid, # Required: which team should receive this submission
}
create_submission_url = f"{BASE_URL}/submission"
create_submission_response = requests.post(create_submission_url, data=payload, files=files, headers=headers)
if create_submission_response.status_code not in (200, 201):
raise RuntimeError(f"Failed to create submission: {create_submission_response.status_code}")
submission_id = create_submission_response.json()["id"]
print(f"Created submission: {color_text(submission_id, 'green')}")
## ...
# Step 3. Poll for Job Completion
submission_activity_url = f"{BASE_URL}/submission"
print(color_text("Polling for jobs to complete...", "yellow"))
# track completed jobs
completed_jobs = set()
while True:
# Query for the specific submission by ID
activity_params = {
"id": submission_id,
"page_size": 1,
"team_uuid": team_uuid,
"division_uuid": division_uuid,
}
activity_response = requests.get(submission_activity_url, params=activity_params, headers=headers)
if activity_response.status_code not in (200, 201):
print(color_text(f"Error checking activity: {activity_response.status_code}", "red"))
time.sleep(POLL_INTERVAL)
continue
results = activity_response.json().get("results", [])
if not results:
print(color_text("Waiting for submission data...", "yellow"))
time.sleep(POLL_INTERVAL)
continue
submission = results[0]
jobs = submission.get("jobs", [])
# Display progress for each job
for j in jobs:
job_id = j.get("job_id")
job_type = j.get("job_type", "UNKNOWN") # e.g., "SOVFIXER", "RUN_OUTPUTTERS"
pct = j.get("processing_pct_complete", 0)
status = j.get("processing_status", "?")
message = j.get("processing_last_message", "") # Human-readable status message
if status == "C" and job_id not in completed_jobs:
# job completed
print(color_text(f"✓ {job_type}: {pct:.0f}% - {message}", "green"))
completed_jobs.add(job_id)
elif status != "C" and job_id not in completed_jobs:
# job still in progress
print(color_text(f"⋯ {job_type}: {pct:.0f}% - {message}", "yellow"))
# Check if the final outputters job is complete
outputters_job = next((j for j in jobs if j.get("job_type") == "RUN_OUTPUTTERS"), None)
if outputters_job and outputters_job.get("processing_pct_complete") == 100:
print(color_text("Final outputs ready for download.", "green"))
break
time.sleep(POLL_INTERVAL)
## ...
# Step 4. Download Final Output Documents
# https://docs.pingintel.com/ping-vision/get-submission-data/download-submission-document
os.makedirs("workflow_example_results", exist_ok=True)
for doc in submission.get("documents", []):
doc_type = doc.get("document_type", "")
filename = doc.get("filename", "")
# Download only the final output files
if doc_type in ["SOVFIXER_OUTPUT", "SOVFIXER_JSON"] and not doc.get("is_archived", False):
download_url = f"{BASE_URL}/submission/{submission_id}/document/{filename}"
download_response = requests.get(download_url, headers=headers)
if download_response.status_code not in (200, 201):
raise RuntimeError(f"Failed to download {filename}: {download_response.status_code}")
# Save with submission ID prefix to avoid filename collisions
output_path = f"workflow_example_results/{submission_id}_{filename}"
with open(output_path, "wb") as outfile:
outfile.write(download_response.content)
print(color_text(f"Saved {doc_type}: {output_path}", "green"))
print(color_text(f"\nWorkflow complete for submission {submission_id}", "green"))
##