Private labeling Lob
Best practices on building atop of Lob's APIs
Billing
If you have less than 10 cost centers
If you have over 10 cost centers
'''
USAGE: python3 pdf_page_count.py
This file will go and fetch mail from the LIST endpoints and basically fetch the letters, load
them into memory and count the # of pages, Spitting out a CSV with the summary.
'''
from pdfminer.pdfparser import PDFParser
from pdfminer.pdfdocument import PDFDocument
from pdfminer.pdfpage import PDFPage
from pdfminer.pdfinterp import resolve1
from io import BytesIO
import pandas as pd
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
import multiprocessing
from functools import partial
import numpy as np
import requests
import json
import sys
def retry_session(retries, session=None, backoff_factor=1):
session = session or requests.Session()
retry = Retry(
total=retries,
read=retries,
connect=retries,
backoff_factor=backoff_factor,
status_forcelist=[500, 502, 503, 504, 429],
allowed_methods=['GET']
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
return session
def parse_pdf(item_list):
result_csv = pd.DataFrame(columns=['id','page_count','double_sided'])
session = retry_session(retries=5)
for item in item_list:
id = item['id']
print(id)
r = session.get(f"{item['url']}")
if r.status_code in (200,204,202):
f = BytesIO(r.content)
f = PDFParser(f)
document = PDFDocument(f)
# This will give you the count of pages
result_csv = result_csv.append({'id' : id, 'page_count' : resolve1(document.catalog['Pages'])['Count'], 'double_sided' : item['double_sided']}, ignore_index=True)
return result_csv
def parse_data(api_key):
url = f"https://api.lob.com/v1/letters"
creative_url = url
#TODO: add in any filtering you see fit (metadata, extra services, etc)
params = {'send_date[gte]' : '2021-10-01', 'send_date[lte]' : '2021-10-12' }
params['limit'] = '100'
session = retry_session(retries=5)
r = session.get(url, auth=requests.auth.HTTPBasicAuth(api_key, ''), params=params)
j = r.json()
# print(j)
results = pd.DataFrame(columns=['id','page_count','double_sided'])
results = parallelize_data(results, j['data'])
url = j['next_url']
#Run through the pagination
if url is not None:
while url is not None:
r = session.get(url, auth=requests.auth.HTTPBasicAuth(api_key, ''))
j = r.json()
results = pd.concat([results,parallelize_data(results, j['data'])], ignore_index=True)
url = j['next_url']
results.to_csv('results_pdf_parse.csv', index=False)
def parallelize_data(results, data_list):
num_cores = int(multiprocessing.cpu_count() - 1) #leave one free to not freeze machine
num_partitions = num_cores * 2 #number of partitions to split dataframe
df_split = np.array_split(data_list, num_partitions)
pool = multiprocessing.Pool(4)
func = partial(parse_pdf)
results = pd.concat(pool.map(func, df_split))
pool.close()
pool.join()
return results
if __name__ == '__main__':
##Get API Key
api_key = input('Enter in your API key (use your live API key to delete live resources): ')
##Delete
parse_data(api_key)yTry before you buy
Use the test environment
Preview mailpieces
Give insight into the delivery process

Last updated
Was this helpful?

