Sometimes I was given 10k+ ONIX files, but I needed to correct a couple of records (by ISBN13). This script helps extract only those records I need so I can correct them and re-ingest only those records.

extract-onix-records.py

#!/usr/local/bin/python3
from lxml import etree
import sys
import os
import re
import logging
import shutil
import argparse
from itertools import islice
import chardet
import subprocess
from chardet.universaldetector import UniversalDetector

def main(onixfile, isbns):
	print("Extracting",str(len(isbns)),"ISBNs:"," ".join(isbns))
	print(isbns)
	onixfile = onixfile[0]
	print(onixfile)
	logging.basicConfig(filename=sys.argv[0]+".log", level=logging.DEBUG, format="%(asctime)s %(message)s", datefmt="%Y/%m/%d %H:%M:%S")
	logging.debug("======================================")
	logging.debug("Extracting " + str(len(isbns)) + " ISBNs: " + " ".join(isbns))
	outputfile = os.path.join(os.path.split(os.path.abspath(onixfile))[0], os.path.split(os.path.abspath(onixfile))[1] + "_Extracted.xml")
	movedfile = os.path.join(os.path.split(os.path.abspath(onixfile))[0], os.path.split(os.path.abspath(onixfile))[1])
	notfoundfile = os.path.join(os.path.split(os.path.abspath(onixfile))[0], os.path.split(os.path.abspath(onixfile))[1] + "_NotFound.txt")

	
	# test that we can open the file
	try:
		with open(onixfile, 'r') as f: logging.debug("Tested file existence %s", onixfile)
	except IOError as e:
		logging.critical("Can't open the file: %s" % os.path.abspath(onixfile))
		logging.critical(e)
		sys.exit(1)
	try:
		logging.debug("Parsing ONIX file.")
		root = etree.parse(onixfile, etree.XMLParser(resolve_entities=False, dtd_validation=False, strip_cdata=False)).getroot()
	except etree.XMLSyntaxError as e:
		logging.critical("Could not parse %s", onixfile)
		logging.critical(e.error_log.filter_from_level(etree.ErrorLevels.FATAL))
		sys.exit(1)
	
	version = root.get("release")
	# grab header information and prepare new file
	with open(onixfile, 'r') as f:
		headlist = list(islice(f, 10))
		head = ''.join(headlist)

		if '<Header>' in head:
			print("Long tags")
			newRoot = etree.Element("ONIXMessage", Release=version)
			productxpath = '//ONIXMessage/Product[ProductIdentifier/IDValue[../ProductIDType[text()="15"]]/text()="'
			headerxpath = "//ONIXMessage/Header"
		elif '<header>' in head:
			print("Short tags")
			newRoot = etree.Element("ONIXmessage", release=version)
			productxpath = '//ONIXmessage/product[productidentifier/b244[../b221[text()="15"]]/text()="'
			headerxpath = "//ONIXmessage/header"

		# copy header over
		try:
			header = root.xpath(headerxpath)
			newRoot.append(header[0])
		except:
			logging.critical("Couldn't find the ONIX header")
			sys.exit(1)
	
		# iterate over isbns and copy each over
		notfound = []
		found = []
		for isbn in isbns:
			try:
				products = root.xpath(productxpath+isbn+'"]')
				for product in products:
					newRoot.append(product)
					found.append(isbn)
				if len(products) == 0:
					notfound.append(isbn)
					print(notfound)
			except:
				logging.warning(isbn + "Invalid XPATH Expression when searching for ISBN")
		
		
		print(etree.tostring(newRoot, pretty_print=False))
	
	# determine encoding, so we're writing the extraction correctly.
	if len(found) > 0:
		detector = UniversalDetector()
		for line in open(onixfile, 'rb'):
			detector.feed(line)
			if detector.done:	break
		detector.close()
		whatEncoding = detector.result['encoding']
		print("File is found to be",whatEncoding,"encoding")
		logging.debug("File is found to be " + whatEncoding + " encoding")
	
		try:
			with open(outputfile, 'wb') as f:
				f.write(etree.tostring(newRoot, xml_declaration=True, encoding=whatEncoding))
				print("Saved file at " + outputfile)
		except IOError as e:
			logging.critical("Could not write extracted ONIX file: %s" % os.path.abspath(outputfile))
			logging.critical(e)
			print("Failed saving the file at " + outputfile)
			sys.exit(1)
	
	if len(notfound) > 0:
		print("There were some unfound ISBNs: " + ' '.join(notfound))
		try:
			with open(notfoundfile, 'w') as f:
				f.write("ISBNs not found in this file: " + os.path.split(os.path.abspath(onixfile))[1] + "\r\n")
				f.write('\r\n'.join(notfound))
				print("Saved NotFound file at " + notfoundfile)
		except IOError as e:
			logging.critical("Could not write NotFound file: %s" % os.path.abspath(notfoundfile))
			logging.critical(e)
			print("Failed saving the file at " + notfoundfile)
			sys.exit(1)
		

parser = argparse.ArgumentParser(description="Create a new ONIX file containing the ISBNs specified with -i")
parser.add_argument('-f', '--file', nargs=1, required=True, help="The ONIX file that you'd like to retrieve from")
parser.add_argument('-i', '--isbns', nargs='*', required=True, help="A list of ISBNs that you'd like to retrieve. No dashes.")
args = parser.parse_args()
main(args.file, args.isbns)