--- updatedAt: 2025-12-15T10:07:05.000Z agentTools: projectIndex: https://developer.onetrust.com/onetrust/llms.txt --- # Stream Consent Receipt Import Automation ```python Python import requests import json import os import urllib import time from tqdm.auto import tqdm import pandas as pd from datetime import datetime, timezone max_checks = 10 # may be changed sleep_time = 60 # may be changed use_sequential_processing = False # may be changed file_dir = "your directory here" api_host = "your host name here" client_id = "your client ID here" client_secret = "your client secret here" job_desc = "desired description for ALL imports" post_jobs_endpoint = api_host + "/api/schedule/v1/jobs?job=" get_jobs_endpoint = api_host + "/api/schedule/v1/jobs/" post_oauth_endpoint = api_host + "/api/access/v1/oauth/token" file_job_ids = [] file_job_names = [] completed_files = [] failed_files = [] unfinished_files = [] def get_token(): payload = f"client_id={client_id}&client_secret={client_secret}&grant_type=client_credentials" headers = {"Content-Type": "application/x-www-form-urlencoded"} response = requests.request( "POST", post_oauth_endpoint, headers=headers, data=payload ) token = json.loads(response.text)["access_token"] return token def locate_files(dir=file_dir): file_names = [] file_paths = [] valid_extensions = (".jsonl", ".gz") for file_name in os.listdir(dir): if file_name.endswith(valid_extensions): file_names.append(file_name) file_paths.append(dir + "/" + file_name) return file_names, file_paths def import_files(api_token, file_paths, file_names, sequential=False): # check for array validity if ( min([len(file_paths), len(file_names)]) - min([len(file_paths), len(file_names)]) != 0 ): print("The arrays for file data (paths and names) are invalid!") else: print("YOUR IMPORT FILES ARE BEING UPLOADED TO ONETRUST!") for index in range(0, len(file_paths)): if sequential: print(f"\nNow importing and validating {file_names[index]}") upload_file(api_token, file_paths[index], file_names[index], sequential) return def upload_file(api_token, path, name, sequential): job_name = os.path.splitext(name)[0] file_job_names.append(job_name) job = create_job(job_name, job_desc) body = {} body["file"] = path body["filename"] = name url = post_jobs_endpoint + job payload = {"filename": name} files = [("file", (name, open(path, "rb"), "application/octet-stream"))] headers = {"Authorization": f"Bearer {api_token}"} response = requests.request("POST", url, headers=headers, data=payload, files=files) if response.ok: # returns true if status code is less than 400 job_id = json.loads(response.text)["Id"] file_job_ids.append(job_id) if sequential: job_success, status = validate_upload(job_id, api_token) if job_success: completed_files.append(job_id) elif status == "Error": failed_files.append(job_name) else: unfinished_files.append(job_name) else: failed_files.append(job_name) return def create_job(job_name, job_desc): job_data = {} job_data["Name"] = job_name job_data["Description"] = job_desc job_data["Type"] = "stream_consent_receipts" job_json = json.dumps(job_data) return urllib.parse.quote(job_json) def validate_upload(job_id, api_token, max_checks=max_checks, sleep_time=sleep_time): checkCount = 0 while checkCount < max_checks: time.sleep(sleep_time) job_complete, status = get_job_status(api_token, job_id) checkCount += 1 if job_complete: break elif status == "Error": break return job_complete, status def get_job_status(api_token, job_id): url = get_jobs_endpoint + job_id payload = {} headers = {"Authorization": f"Bearer {api_token}"} response = requests.request("GET", url, headers=headers, data=payload) dictResponse = dict(response.json()) status = dictResponse["Status"] job_passed = False if status == "Completed": job_passed = True else: job_passed = False return job_passed, status def validate_all_imports(api_token, job_ids=file_job_ids, job_names=file_job_names): file_count = len(job_names) for index in range(0, len(job_names)): print(f"Now validating {job_names[index]}") job_success, status = validate_upload(job_ids[index], api_token) if job_success: completed_files.append(job_ids[index]) elif status == "Error": failed_files.append(job_ids[index]) else: unfinished_files.append(job_ids[index]) return def generate_import_summary( all_jobs=file_job_ids, failed_jobs=failed_files, completed_jobs=completed_files, names=file_job_names, ): job_status = {} for job in range(0, len(all_jobs)): name = names[job] job_id = all_jobs[job] if job_id in failed_jobs: job_status[name] = [job_id, "Failure"] elif job_id in completed_jobs: job_status[name] = [job_id, "Success"] else: job_status[name] = [job_id, "Unfinished/Pending"] report_name = datetime.utcnow().strftime( "Bulk Import Summary %Y-%m-%d--%I-%M-%S.xlsx" ) report = pd.DataFrame.from_dict( data=job_status, orient="index", columns=["Job ID", "Job Status"] ) report.index.name = "Job Name" print(job_status) print("\n\n") print(report) report.to_excel(report_name) def __main__(sequential=False): api_token = get_token() file_names, file_paths = locate_files() all_jobs = import_files(api_token, file_paths, file_names, sequential) if sequential == False: validate_all_imports(api_token) generate_import_summary() return __main__(use_sequential_processing ``` # Introduction This script is beneficial for users who need to import multiple consent receipt files into OneTrust without manually uploading each file through the OneTrust application (GUI). Some of the benefits include: - Faster import times - Automated validation of imports - A summary report of the import process > 📘 Important Notes > > - **Credentials**: You'll need to create a OneTrust client ID and client secret. This recipe provides a simple method for obtaining an OAuth Access Token; for a more robust solution, refer to the [Generate OAuth Access Token](https://developer.onetrust.com/onetrust/recipes/generate-oauth-access-token) recipe. > - **Configurations**: Adjust `max_checks` and `sleep_time` for optimal performance based on your file sizes. Unless the configuration of the `max_checks` and `sleep_time` is appropriate for the import file sizes, all imports will likely result in either 'Error' or 'In Progress' statuses due to inadequate time to fully ingest the data. # Global Variables The script uses several global variables, both static and dynamic, which you need to configure before running it. The following steps will introduce each of these variables when used; therefore, making the updates is unnecessary now. **Static Global Variables** - `max_checks`: The maximum attempts to check job statuses. - `sleep_time`: The number of seconds to wait between checks. - `client_id`: Your OneTrust client ID. For more information, refer to the [Generate OAuth Access Token](https://developer.onetrust.com/onetrust/recipes/generate-oauth-access-token) recipe. - `client_secret`: Your OneTrust client secret. For more information, refer to the [Generate OAuth Access Token](https://developer.onetrust.com/onetrust/recipes/generate-oauth-access-token) recipe. - `post_jobs_endpoint`: The API endpoint for uploading files. - `get_jobs_endpoint`: The API endpoint for checking job status. - `file_dir`: The directory where your import files are located. - `job_desc`: A description for all import jobs. - `use_sequential_processing`: Whether to process files sequentially (True) or in bulk (False). - `api_host`: Your OneTrust application hostname (e.g., https://app.onetrust.com). **Dynamic Global Variables** - `file_job_ids`: List of job IDs returned by the GET Jobs API. - `file_job_names`: List of job names assigned to file imports. - `completed_files`: List of job IDs with a "COMPLETED" status. - `failed_files`: List of job IDs with an "ERROR" status. - `unfinished_files`: List of job IDs that are not yet completed. # Script Configuration Update the following variables in the script to meet your needs. - `max_checks`: The maximum attempts to check job statuses. - `sleep_time`: The number of seconds to wait between checks. - `use_sequential_processing`: Whether to process files sequentially (True) or in bulk (False). - `file_dir`: The directory where your import files are located. - `api_host`: Your OneTrust application hostname (e.g., https://app.onetrust.com). - `client_id`: Your OneTrust client ID. For more information, refer to the [Generate OAuth Access Token](https://developer.onetrust.com/onetrust/recipes/generate-oauth-access-token) recipe. - `client_secret`: Your OneTrust client secret. For more information, refer to the [Generate OAuth Access Token](https://developer.onetrust.com/onetrust/recipes/generate-oauth-access-token) recipe. - `job_desc`: A description for all import jobs. > 📘 Important Notes > > - **Credentials**: You'll need to create a OneTrust client ID and client secret. This recipe provides a simple method for obtaining an OAuth Access Token; for a more robust solution, refer to the [Generate OAuth Access Token](https://developer.onetrust.com/onetrust/recipes/generate-oauth-access-token) recipe. > - **Configurations**: Adjust `max_checks` and `sleep_time` for optimal performance based on your file sizes. Unless the configuration of the `max_checks` and `sleep_time` is appropriate for the import file sizes, all imports will likely result in either 'Error' or 'In Progress' statuses due to inadequate time to fully ingest the data. # Authentication This function retrieves an OAuth 2.0 Bearer token using your OneTrust client credentials. This script generates an OAuth access token using minimal implementation. For a robust implementation, refer to the [Generate OAuth Access Token](https://developer.onetrust.com/onetrust/recipes/generate-oauth-access-token) recipe. # Locate Files This function searches the specified directory, `file_dir`, for `.jsonl` or `.gz` files and returns their names and paths. # Import Files This function uploads each file to OneTrust using the POST Job API. If `use_sequential_processing` is True, it validates each file before proceeding to the next. # Upload File This function uploads a single file to OneTrust using the POST Job API and retrieves the job ID, `Id`, assigned to the upload. # Create Job This function uses the job name and description from the `upload_file` function to generate the necessary JSON-encoded string for the Jobs API query parameter. # Validate Upload This function repeatedly checks the status of an import job until it completes or exceeds the maximum check attempts. This is only leveraged for SEQUENTIAL imports. E.g., `use_sequential_processing` is True. # Get Job Status This function retrieves the current status of a specified import job using the GET Jobs API. # Validate All Imports This function validates all imports if `use_sequential_processing` is False. # Generate Import Summary This function generates an Excel report summarizing the status of each import. The summary records the import's Job Name, ID, and Status as found by the validation functions. The output is an Excel file, `Bulk Import Summary %Y-%m-%d--%I-%M-%S.xlsx`, created via Pandas. # Main Function This function orchestrates the entire process: getting the token, locating files, importing files, validating them, and generating the summary report. # Running the Script 1. **Install Dependencies**: Ensure you have the required Python packages installed: ``` pip install requests pandas tqdm ``` 1. **Configure Variables**: Update the static global variables in the script with your credentials, API endpoints, and file directory. 1. **Execute the Script**: Run the script using Python: ``` python your_script_name.py ``` This will automate the import process, validate the uploads, and generate a summary report. Adjust the configuration as needed for your specific use case.