Python: encuentra expresiones regulares en un archivo

Tener:

f = open(...) r = re.compile(...) 

Necesitar:
¿Encuentra la posición (inicio y final) de una primera expresión regular coincidente en un archivo grande?
(a partir de current_pos=... )

¿Cómo puedo hacer esto?


Quiero tener esta función:

 def find_first_regex_in_file(f, regexp, start_pos=0): f.seek(start_pos) .... (searching f for regexp starting from start_pos) HOW? return [match_start, match_end] 

Se espera que el archivo ‘f’ sea grande.

Una forma de buscar en archivos grandes es usar la biblioteca mmap para asignar el archivo a una gran porción de memoria. Entonces puedes buscar a través de él sin tener que leerlo explícitamente.

Por ejemplo, algo como:

 size = os.stat(fn).st_size f = open(fn) data = mmap.mmap(f.fileno(), size, access=mmap.ACCESS_READ) m = re.search(r"867-?5309", data) 

Esto funciona bien para archivos muy grandes (lo he hecho para un archivo de más de 30 GB, pero necesitará un sistema operativo de 64 bits si su archivo es más de uno o dos GB).

El siguiente código funciona razonablemente bien con archivos de prueba de alrededor de 2 GB de tamaño.

 def search_file(pattern, filename, offset=0): with open(filename) as f: f.seek(offset) for line in f: m = pattern.search(line) if m: search_offset = f.tell() - len(line) - 1 return search_offset + m.start(), search_offset + m.end() 

Tenga en cuenta que la expresión regular no debe abarcar varias líneas.

NOTA: esto ha sido probado en python2.7. Es posible que tengas que modificar las cosas en Python 3 para manejar cadenas vs bytes, pero no debería ser demasiado doloroso.

Es posible que los archivos asignados en la memoria no sean ideales para su situación (el modo de 32 bits aumenta la posibilidad de que no haya suficiente memoria virtual contigua, no se pueda leer desde tuberías u otros archivos que no sean archivos, etc.).

Aquí hay una solución que lee 128k bloques a la vez y siempre que su expresión regular coincida con una cadena más pequeña que ese tamaño, esto funcionará. También tenga en cuenta que no está restringido utilizando expresiones regulares de una sola línea. Esta solución funciona bastante rápido, aunque sospecho que será ligeramente más lento que usar mmap. Probablemente, depende más de lo que esté haciendo con las coincidencias, así como del tamaño / complejidad de la expresión regular que está buscando.

El método se asegurará de mantener solo un máximo de 2 bloques en la memoria. Es posible que desee imponer al menos 1 coincidencia por bloque como comprobación de validez en algunos casos de uso, pero este método se truncará para mantener el máximo de 2 bloques en la memoria. También se asegura de que NO se produzca ninguna coincidencia de expresiones regulares que coman hasta el final del bloque actual y, en cambio, se guarde la última posición para cuando se agote la verdadera entrada o tengamos otro bloque con el que coincida la expresión regular antes del final de, en para que coincida mejor con patrones como “[^ \ n] +” o “xxx $”. Es posible que aún puedas romper cosas si tienes un lookahead al final de la expresión regular como xx (?! Xyz), donde yz está en el siguiente bloque, pero en la mayoría de los casos puedes evitar el uso de tales patrones.

 import re def regex_stream(regex,stream,block_size=128*1024): stream_read=stream.read finditer=regex.finditer block=stream_read(block_size) if not block: return lastpos = 0 for mo in finditer(block): if mo.end()!=len(block): yield mo lastpos = mo.end() else: break while True: new_buffer = stream_read(block_size) if not new_buffer: break if lastpos: size_to_append=len(block)-lastpos if size_to_append > block_size: block='%s%s'%(block[-block_size:],new_buffer) else: block='%s%s'%(block[lastpos:],new_buffer) else: size_to_append=len(block) if size_to_append > block_size: block='%s%s'%(block[-block_size:],new_buffer) else: block='%s%s'%(block,new_buffer) lastpos = 0 for mo in finditer(block): if mo.end()!=len(block): yield mo lastpos = mo.end() else: break if lastpos: block=block[lastpos:] for mo in finditer(block): yield mo 

Para probar / explorar, puede ejecutar esto:

 # NOTE: you can substitute a real file stream here for t_in but using this as a test t_in=cStringIO.StringIO('testing this is a 1regexxx\nanother 2regexx\nmore 3regexes') block_size=len('testing this is a regex') re_pattern=re.compile(r'\dregex+',re.DOTALL) for match_obj in regex_stream(re_pattern,t_in,block_size=block_size): print 'found regex in block of len %s/%s: "%s[[[%s]]]%s"'%( len(match_obj.string), block_size,match_obj.string[:match_obj.start()].encode('string_escape'), match_obj.group(), match_obj.string[match_obj.end():].encode('string_escape')) 

Aquí está la salida:

 found regex in block of len 46/23: "testing this is a [[[1regexxx]]]\nanother 2regexx\nmor" found regex in block of len 46/23: "testing this is a 1regexxx\nanother [[[2regexx]]]\nmor" found regex in block of len 14/23: "\nmore [[[3regex]]]es" 

Esto puede ser útil junto con un análisis rápido de un XML grande donde se puede dividir en mini-DOMs basados ​​en un subelemento como raíz, en lugar de tener que sumergirse en el manejo de devoluciones de llamada y estados cuando se usa un analizador SAX. También te permite filtrar a través de XML más rápido también. Pero también lo he usado para muchos otros propósitos. ¡Me sorprenden las recetas de este tipo que no están disponibles en la red!

Una cosa más: el análisis en Unicode debería funcionar siempre que la secuencia transmitida esté generando cadenas Unicode, y si está utilizando las clases de caracteres como \ w, deberá agregar la marca re.U a re.compile construcción de patrones. En este caso, block_size en realidad significa conteo de caracteres en lugar de conteo de bytes.