Python XML parser .

DB COLUMN

+-------+--------------+-----------------+------+-----+---------+-------+---------------------------------+---------+
| Field | Type         | Collation       | Null | Key | Default | Extra | Privileges                      | Comment |
+-------+--------------+-----------------+------+-----+---------+-------+---------------------------------+---------+
| url   | varchar(800) | utf8_general_ci | YES  |     | NULL    |       | select,insert,update,references |         |
| title | varchar(800) | utf8_general_ci | YES  |     | NULL    |       | select,insert,update,references |         |
| descr | varchar(800) | utf8_general_ci | YES  |     | NULL    |       | select,insert,update,references |         |
+-------+--------------+-----------------+------+-----+---------+-------+---------------------------------+---------+

gistfile1.py

#!/usr/bin/env python

import lxml.etree as tr
import oursql
import itertools
import time
from multiprocessing import Process, Queue, Pool
from collections import deque

errors = 0

def mainParser(queue):
	context = tr.iterparse('content.rdf.u8', events=('end',), tag='{http://dmoz.org/rdf/}ExternalPage')
	XMLNS_D = '{http://purl.org/dc/elements/1.0/}'

	xmlChunk = [] # Array with xml nodes to be Queued
	chunkNumber = 10000 # Number of nodes of xml

	for event, elem in context:
		if(len(xmlChunk) < chunkNumber):
			xmlChunk.append(tr.tostring(elem))			
		else:
			next(queue).put(xmlChunk)
			xmlChunk = []

	    	elem.clear()
		while elem.getprevious() is not None:
		        del elem.getparent()[0]


def dataExtractor(node):
	rows = []
	XMLNS_D = '{http://purl.org/dc/elements/1.0/}'
	for i in node:
		elem = tr.fromstring(i)
		#print elem
		title = elem.find(XMLNS_D + 'Title').text
		if title is not None:
			title = title.encode('utf8', 'replace')
		else:
			title = "nnn"
		descr = elem.find(XMLNS_D + 'Description').text
		if descr is not None:
			descr = descr.encode('utf8', 'replace')
		else:
			descr = "nnn"
		url =  elem.attrib['about'].encode('utf8', 'replace')
		row = {'url': 'pp', 'title': 'pp', 'descr': 'pp'}
		print type(url)
		print type(title)
		print type(descr)
		rows.append(row)
	return rows


def xmlParser(queue):	    
	    conn = oursql.connect("hidden")
	    curs = conn.cursor(try_plain_query=False)
            curs.execute("SET NAMES utf8")

	    while(queue.qsize() >= 0):
		theChunk = queue.get(block=True, timeout=100)
		curs.executemany('INSERT INTO dmoz (url, title, descr) VALUES(?, ?, ?)', [(i['url'], i['title'], i['descr']) for i in dataExtractor(theChunk)])
		print "inserted 10000"


		
def xmlSecondParser(queue):
	    conn = oursql.connect("hidden")
	    curs = conn.cursor(try_plain_query=False)
            curs.execute("SET NAMES utf8")

	    while(queue.qsize() >= 0):
		theChunk = queue.get(block=True, timeout=100)		
		curs.executemany('INSERT INTO dmoz (url, title, descr) VALUES(?, ?, ?)', [(i['url'], i['title'], i['descr']) for i in dataExtractor(theChunk)])
		print "inserted 10000"



if __name__ == '__main__':
	q = [Queue(12), Queue(12)]
	roundRobin = itertools.cycle(q)
	
	mainParser = Process(target=mainParser, args=(roundRobin,)) #MAIN XML PARSE SINGLE NODES

	xmlParser = Process(target=xmlParser, args=(q[0],)) #XML PARSER PARSES DATA FROM NODES
	xmlSecondParser = Process(target=xmlSecondParser, args=(q[1],)) #XML PARSER PARSES DATA FROM NODES

	mainParser.start()

	xmlParser.start()	
	xmlSecondParser.start()

	#print "main "+str(mainParser.pid)
        #print "one "+str(xmlParser.pid)
        #print "sec "+str(xmlSecondParser.pid)
	

last traceback

Traceback (most recent call last):
  File "/usr/lib/python2.6/multiprocessing/process.py", line 232, in _bootstrap
    self.run()
  File "/usr/lib/python2.6/multiprocessing/process.py", line 88, in run
    self._target(*self._args, **self._kwargs)
  File "xmlFastParser.py", line 89, in xmlParser
    curs.executemany('INSERT INTO dmoz (url, title, descr) VALUES(?, ?, ?)', [(i['url'], i['title'], i['descr']) for i in dataExtractor(theChunk)])
  File "cursor.pyx", line 138, in oursql.Cursor.executemany (oursqlx/oursql.c:16181)
  File "statement.pyx", line 407, in oursql._Statement.execute (oursqlx/oursql.c:10345)
  File "util.pyx", line 91, in oursql._do_warnings_query (oursqlx/oursql.c:3459)
CollatedWarningsError: (None, 'query caused warnings', [(<class 'oursql.Warning'>, (u"Data truncated for column 'descr' at row 1", 1265L))])

New Traceback

Traceback (most recent call last):
  File "/usr/lib/python2.6/multiprocessing/process.py", line 232, in _bootstrap
    self.run()
  File "/usr/lib/python2.6/multiprocessing/process.py", line 88, in run
    self._target(*self._args, **self._kwargs)
  File "xmlFastParser.py", line 103, in xmlSecondParser
    curs.executemany('INSERT INTO dmoz (url, title, descr) VALUES(?, ?, ?)', [(i['url'], i['title'], i['descr']) for i in dataExtractor(theChunk)])
  File "cursor.pyx", line 138, in oursql.Cursor.executemany (oursqlx/oursql.c:16181)
  File "statement.pyx", line 407, in oursql._Statement.execute (oursqlx/oursql.c:10345)
  File "util.pyx", line 91, in oursql._do_warnings_query (oursqlx/oursql.c:3459)
CollatedWarningsError: (None, 'query caused warnings', [(<class 'oursql.Warning'>, (u"Incorrect string value: '\\xCFbrahi...' for column 'descr' at row 1", 1366L))])

Traceback

Process Process-3:
Traceback (most recent call last):
  File "/usr/lib/python2.6/multiprocessing/process.py", line 232, in _bootstrap
    self.run()
  File "/usr/lib/python2.6/multiprocessing/process.py", line 88, in run
    self._target(*self._args, **self._kwargs)
  File "xmlFastParser.py", line 80, in xmlSecondParser
    curs.executemany('INSERT INTO dmoz (url, title, descr) VALUES(?, ?, ?)', [(i['url'], i['title'], i['descr']) for i in dataExtractor(theChunk)])
  File "cursor.pyx", line 138, in oursql.Cursor.executemany (oursqlx/oursql.c:16181)
  File "statement.pyx", line 316, in oursql._Statement.execute (oursqlx/oursql.c:9313)
UnicodeDecodeError: 'ascii' codec can't decode byte 0xc3 in position 0: ordinal not in range(128)