Seguidor de archivos en Python
20200608102855
Seguidor de archivos en Python
¿Qué es el seguidor de archivos?
Un seguidor de archivos file_follower = FileFollower(my_file, pattern) busca dentro de un archivo my_file las líneas de texto que activen una expresión regular pattern. Por defecto, file_follower comienza su búsqueda al final de my_file, por lo que sólo buscará dentro de las líneas nuevas que otra aplicación escriba en my_file.
Por ejemplo, si al moment en que ejecutemos file_follower.start() el archivo my_file tuviera el contenido presentado en el listado 1, y pattern=[a-z]{3}[0-9]{3}: entonces file_follower sólo buscará líneas que contengan al menos una palabra formada por tres letras minúsculas y tres números después de la línea que dice "start of search" (cuarta línea).
start of the file
abc123
xyz987
start of search
Listado 1. Contenido inicial de my_file antes de crear activar el seguidor de archivos file_follower.
Cada vez que file_follower encuentra una línea que activa la expresión regular pattern, añade el objeto resultante de dicha activación (de tipo re.Match) a su cola de resultados file_follower.result_queue. Y otros hilos de ejecución pueden tomar esos resultados utilizando el método file_follower.get_result().
Cuando la aplicación principal ya no necesite de file_follower, puede detenerlo utilizando su método file_follower.quit().
¿Cómo leemos un archivo línea por línea?
En Python, podemos leer un archivo my_file línea por línea utilizando el método readline desde el objeto handler = open(my_file, "r"). En cada invocación de readline, el cursor de handler avanza hasta la posición posterior al siguiente salto de línea ("\n") o hasta el final del archivo si ha leído la última línea. Y una vez que el cursor haya alcanzado el final del archivo, el método readline regresará cadenas vacías ("") hasta que haya un nuevo salto de línea después de la posición del cursor.
¿Y qué pasa si el archivo reduce su tamaño?
El objeto handler no actualiza su cursor automáticamente cuando el contenido del archivo my_file reduce su tamaño. Por lo tanto, tenemos que detectar si el contenido del archivo se contrae, y mover el cursor a la última posición del archivo manualmente.
Supongamos que el archivo my_file se contrae: su contenido pasa de tener n caracteres, a m caracteres (por lo que m < n). Además, supongamos que para entonces, el cursor de handler ha llegado hasta el final del archivo. Entonces, podemos detectar la reducción del tamaño:
- Solicitando el tamaño
mdel archivo al sistema operativo. - Solicitando la posición del cursor
nal objetohandler. - Verificando si
m < n. En caso contrario, el archivo: o sigue teniendo su tamaño original o su contenido ha crecido en tamaño.
Finalmente, si el archivo ha reducido su tamaño, para leer las nuevas líneas que se agreguen al contenido tenemos que mover el cursor a la posición m. Esto último lo podemos hacer con handler.seek(0, os.SEEK_END).
El código
El listado 2 contiene la implementación de la clase FileFollower. Además tiene una aplicación de demostración demo. Suponiendo que el archivo de implementación se llama file_follower.py, el listado 3 contiene las instrucciones de uso de la aplicación de demostración (que también podemos obtener al ejecutar python3 ./file_follower.py --help).
#! /usr/bin/env python3
import sys
import time
import re
import threading
import queue
import os
import argparse
class FileFollower(threading.Thread):
COMMAND_QUEUE_SIZE = 1
def __init__(
self,
input_file_path,
regular_expression,
flags=0,
check_period=0.5,
result_queue_size=3,
start_from_eof=True
):
# Setup the thread interface with the Thread object constructor
threading.Thread.__init__(self, name="file_follower")
self.check_period = check_period
self.command_queue = queue.Queue(FileFollower.COMMAND_QUEUE_SIZE)
self.file_path = input_file_path
self.flags = flags
self.regular_expression = regular_expression
self.result_queue = queue.Queue(result_queue_size)
self.start_from_eof = start_from_eof
def _received_quit_command(self):
try:
command = self.command_queue.get_nowait()
return command == "quit"
except queue.Empty as queue_is_empty:
del queue_is_empty
return False
def _add_result(self, result):
self.result_queue.put(
item=result,
block=True,
timeout=None
)
def run(self):
with open(self.file_path, "r") as file_handler:
if self.start_from_eof:
file_handler.seek(0, os.SEEK_END)
while True:
if self._received_quit_command():
return
new_line = file_handler.readline().strip()
if new_line:
match = re.search(
pattern=self.regular_expression,
string=new_line,
flags=self.flags
)
if match:
self._add_result(match)
else:
file_size = os.path.getsize(self.file_path)
cursor_position = file_handler.tell()
the_file_got_smaller = file_size < cursor_position
if the_file_got_smaller:
# Move cursor to the end of the file
file_handler.seek(0, os.SEEK_END)
time.sleep(self.check_period)
def get_result(self, block=True, timeout=None):
if not self.is_alive():
return None
try:
return self.result_queue.get(block=block, timeout=timeout)
except queue.Empty as queue_is_empty:
del queue_is_empty
return None
def quit(self):
if not self.is_alive():
return False
return self.command_queue.put_nowait("quit")
def demo():
parser = argparse.ArgumentParser()
parser.add_argument(
"--pattern",
type=str,
required=True,
help="A regular expression."
)
parser.add_argument(
"--file",
type=str,
required=True,
help="The file to follow."
)
parser.add_argument(
"--number_of_matches",
type=int,
required=True,
help="Number of searches for the pattern in the input file."
)
arguments = parser.parse_args()
file_follower = FileFollower(
input_file_path=arguments.file,
regular_expression=arguments.pattern,
check_period=0.25)
file_follower.start()
for n in range(arguments.number_of_matches):
result = file_follower.get_result()
if result:
print(
"({}) '{}' matches '{}'.".format(
n+1,
result.string,
sys.argv[2]
)
)
print("Done! Bye bye!")
file_follower.quit()
if __name__ == "__main__":
demo()
Listado 2. file_follower.py: Implementación de la clase FileFollower y una aplicación de demostración demo.
$ python3 ./file_follower.py --help
usage: file_follower.py [-h] --pattern PATTERN --file FILE --number_of_matches NUMBER_OF_MATCHES
optional arguments:
-h, --help show this help message and exit
--pattern PATTERN A regular expression.
--file FILE The file to follow.
--number_of_matches NUMBER_OF_MATCHES
Number of searches for the pattern in the input file.
Listado 3. Instrucciones para ejecutar la aplicación de demostración contenida en file_follower.py.
Comentarios
Publicar un comentario