Cum urmăresc un fișier pentru modificări? (Programare, Python, Fișier, Pywin32, Ceas)

Jon Cage a intrebat.

Am un fișier de jurnal scris de un alt proces pe care vreau să-l urmăresc pentru modificări. De fiecare dată când are loc o schimbare aș dori să citesc noile date pentru a face o anumită prelucrare asupra lor.

Care este cea mai bună modalitate de a face acest lucru? Speram să existe un fel de cârlig din biblioteca PyWin32. Am găsit win32file.FindNextChangeNotification dar nu am nicio idee despre cum să îi cer să urmărească un anumit fișier.

Dacă cineva a făcut ceva de genul acesta, aș fi foarte recunoscător să aud cum…

[Editează] Ar fi trebuit să menționez că am fost după o soluție care nu necesită polling.

[Editare] Blesteme! Se pare că acest lucru nu funcționează pe o unitate de rețea mapată. Bănuiesc că Windows nu „aude” nicio actualizare a fișierului așa cum o face pe un disc local.

Comentarii

  • Pe Linux se poate folosi strace monitorizare write apeluri pentru acest lucru –  > Por test30.
  • Răspunsul lui @simao folosește python-watchdog. Python-Watchdog are o documentație grozavă –> aici este un link către documentația [„QuickStart”] care oferă un exemplu de cod minim care urmărește directorul de lucru curent. –  > Por Trevor Boyd Smith.
25 răspunsuri
Horst Gutmann

V-ați uitat deja la documentația disponibilă pe http://timgolden.me.uk/python/win32_how_do_i/watch_directory_for_changes.html? Dacă aveți nevoie să funcționeze doar sub Windows, al 2-lea exemplu pare să fie exact ceea ce doriți (dacă schimbați calea directorului cu unul dintre fișierele pe care doriți să le urmăriți).

Altfel, probabil că pollingul va fi singura opțiune cu adevărat independentă de platformă.

Notă: Nu am încercat niciuna dintre aceste soluții.

Comentarii

  • Acest răspuns este specific Windows, dar se pare că au fost postate aici și unele soluții cross-platform pentru această problemă. –  > Por Anderson Green.
  • Există un punct de referință, dacă acest proces este mai lent decât implementarea într-un limbaj nativ precum c++ ? –  > Por user1767754.
  • Este mai bine să inserați conținutul relevant din sursele citate, deoarece acestea pot deveni depășite. –  > Por Trilarion.
  • (1.) la finalul acestui răspuns este o dezmințire puternică… „Nu am încercat niciuna dintre aceste soluții”. (2.) acest răspuns este mai mult sau mai puțin un răspuns de tip „doar link” (3.) răspunsul menționează „sondare”, dar nu adaugă nimic util după aceea … în timp ce răspunsul lui @Deestan oferă informații bune despre sondare –  > Por Trevor Boyd Smith.
simao

Ați încercat să utilizați Watchdog?

Biblioteca API Python și utilitățile shell pentru a monitoriza evenimentele din sistemul de fișiere.

Monitorizarea directoarelor a devenit ușoară cu

  • un API pentru mai multe platforme.
  • Un instrument de shell pentru a rula comenzi ca răspuns la modificările de directoare.

Începeți rapid cu un exemplu simplu în Quickstart

Comentarii

    57

  • Instalabil cu easy_install? Verificați. Licență gratuită? Verificați. Rezolvă problema pe platformele mari? Verificat. Susțin acest răspuns. Singura observație: se poate observa că exemplu de pe pagina lor de proiect nu funcționează din start. Utilizați cel de pe github-ul lor în schimb. –  > Por Inaimathi.
  • Noi folosim watchdog. Este posibil să trecem la QFileSystemWatcher. Doar un avertisment corect – watchdog este bun, dar departe de a fi perfect pe toate platformele (în acest moment). Fiecare sistem de operare are particularitățile sale. Așadar, dacă nu vă dedicați să îl faceți perfect, vă veți trage de păr. Dacă vreți să urmăriți doar 10 fișiere sau cam așa ceva, eu aș face un sondaj. Memoria cache pe disc a sistemului de operare este foarte matură, iar Watchdog implică oricum API-uri de interogare. Este în principal pentru a urmări structuri de dosare uriașe, IMHO. –  > Por SilentSteel.
  • Singura mea nemulțumire cu watchdog este că are multe dependențe. Mai puține decât PyQt, bineînțeles, dar nu funcționează și nu se simte ca o soluție minimă, de bună practică, care face o singură treabă și o face bine. –  > Por AndreasT.
  • Are @denfromufa dreptate aici? Chiar blochează watchdog fișierele, astfel încât acestea să nu poată fi editate concomitent cu watchdog care le supraveghează? Mi-e greu să cred asta, ar fi complet inutil. –  > Por Michel Müller.
  • @MichelMüller Tocmai am verificat acest exemplu (vezi linkul de mai jos) și funcționează! nu sunt sigur ce a fost greșit înainte, dar acest răspuns nu oferă niciun exemplu. stackoverflow.com/a/18599427/2230844 –  > Por denfromufa.
Deestan

Dacă pollingul este suficient de bun pentru tine, aș urmări doar dacă se schimbă statusul fișierului „modified time”. Pentru a o citi:

os.stat(filename).st_mtime

(Rețineți, de asemenea, că soluția nativă a evenimentului de modificare din Windows nu funcționează în toate circumstanțele, de exemplu, pe unitățile de rețea).

import os

class Monkey(object):
    def __init__(self):
        self._cached_stamp = 0
        self.filename = '/path/to/file'

    def ook(self):
        stamp = os.stat(self.filename).st_mtime
        if stamp != self._cached_stamp:
            self._cached_stamp = stamp
            # File has changed, so do something...

Comentarii

  • Cum puteți face acest lucru pe un interval? –  > Por dopatraman.
  • @dopatraman Iată cum se poate face acest lucru pe un interval ` import sys import time pub = Monkey() while True: try: time.sleep(1) pub.watch() except KeyboardInterrupt: print(‘
    Done’) break except: print(f’Unhandled error: {sys.exc_info()[0]}’) ` –  > Por Vlad Bezden.
  • Grozavă soluție simplă! Am adăugat o verificare pentru a nu raporta fișierul modificat la prima rulare: if self._cached_stamp is not None. –  > Por Noumenon.
  • nu există watch metoda așa cum a scris @VladBezden. lipsește codul? –  > Por Lei Yang.
hipersayan_x

Dacă doriți o soluție multiplatformă, atunci verificați QFileSystemWatcher.Iată un exemplu de cod (neigienizat):

from PyQt4 import QtCore

@QtCore.pyqtSlot(str)
def directory_changed(path):
    print('Directory Changed!!!')

@QtCore.pyqtSlot(str)
def file_changed(path):
    print('File Changed!!!')

fs_watcher = QtCore.QFileSystemWatcher(['/path/to/files_1', '/path/to/files_2', '/path/to/files_3'])

fs_watcher.connect(fs_watcher, QtCore.SIGNAL('directoryChanged(QString)'), directory_changed)
fs_watcher.connect(fs_watcher, QtCore.SIGNAL('fileChanged(QString)'), file_changed)

Comentarii

  • Cred că acesta este, probabil, cel mai bun răspuns dintre toate, având în vedere că acestea fie a) se bazează pe obiectul FileSystemwatcher de la Win32 și nu pot fi portate, fie b) solicită fișierul (ceea ce este rău pentru performanță și nu va fi scalabil). Este păcat că Python nu are această facilitate încorporată, deoarece PyQt este o dependență uriașă dacă tot ce folosiți este clasa QFileSystemWatcher. –  > Por CadentOrange.
  • Îmi place această soluție. Am vrut să subliniez faptul că veți avea nevoie de o instanță QApplication pentru ca aceasta să funcționeze, am adăugat „app = QtGui.QApplication(sys.argv)” chiar sub importuri și apoi „app.exec_()” după conexiunile de semnal. –  > Por spencewah.
  • Tocmai am testat acest lucru pe o cutie Linux, văd că metoda directory_changed este apelată, dar nu și file_changed. –  > Por Ken Kinder.
  • @CadentOrange, dacă nu vă place dependența de pyQt, se poate folosi metoda watchdog este răspunsul corect –  > Por Mike Pennington.
  • de ce să nu folosiți PySide pentru asta în loc de PyQt pentru o utilizare atât de mică. –  > Por Ciasto piekarz.
Maxime

Nu ar trebui să funcționeze pe windows (poate cu cygwin ?), dar pentru utilizatorul unix, ar trebui să folosiți apelul de sistem „fcntl”. Iată un exemplu în Python. Este în mare parte același cod dacă trebuie să-l scrieți în C (aceleași nume de funcții)

import time
import fcntl
import os
import signal

FNAME = "/HOME/TOTO/FILETOWATCH"

def handler(signum, frame):
    print "File %s modified" % (FNAME,)

signal.signal(signal.SIGIO, handler)
fd = os.open(FNAME,  os.O_RDONLY)
fcntl.fcntl(fd, fcntl.F_SETSIG, 0)
fcntl.fcntl(fd, fcntl.F_NOTIFY,
            fcntl.DN_MODIFY | fcntl.DN_CREATE | fcntl.DN_MULTISHOT)

while True:
    time.sleep(10000)

Comentarii

  • Funcționează de minune cu kernelul Linux 2.6.31 pe un sistem de fișiere ext4 (pe Ubuntu 10.04), deși numai pentru directoare – ridică un IOError „not a directory” dacă îl folosesc cu un fișier. –  > Por David Underhill.
  • GREAT! Același lucru pentru mine, funcționează doar pentru director și urmărește fișierele din acest director. Dar nu va funcționa pentru fișierele modificate în subdirectoare, așa că se pare că trebuie să parcurgeți subdirectoarele și să le supravegheați pe toate. (sau există o modalitate mai bună de a face acest lucru?) -.  > Por lfagundes.
Michael Palmer

Verifică pyinotify.

inotify înlocuiește dnotify (dintr-un răspuns anterior) în linuxurile mai noi și permite monitorizarea la nivel de fișier mai degrabă decât la nivel de director.

Comentarii

  • Nu vreau să pun o piedică pe acest răspuns, dar după ce am citit acest articol, aș spune că s-ar putea să nu fie o soluție atât de fascinantă pe cât se credea. serpentine.com/blog/2008/01/04/04/why-you-should-not-use-pyinotify –  > Por NuclearPeon.
  • pyinotify are o mulțime de dezavantaje, începând de la baza de cod foarte nepionică până la consumul de memorie. Mai bine să căutați alte opțiuni… –  > Por NightOwl19.

Ei bine, după un pic de hacking al scriptului lui Tim Golden, am următorul care pare să funcționeze destul de bine:

import os

import win32file
import win32con

path_to_watch = "." # look at the current directory
file_to_watch = "test.txt" # look for changes to a file called test.txt

def ProcessNewData( newData ):
    print "Text added: %s"%newData

# Set up the bits we'll need for output
ACTIONS = {
  1 : "Created",
  2 : "Deleted",
  3 : "Updated",
  4 : "Renamed from something",
  5 : "Renamed to something"
}
FILE_LIST_DIRECTORY = 0x0001
hDir = win32file.CreateFile (
  path_to_watch,
  FILE_LIST_DIRECTORY,
  win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
  None,
  win32con.OPEN_EXISTING,
  win32con.FILE_FLAG_BACKUP_SEMANTICS,
  None
)

# Open the file we're interested in
a = open(file_to_watch, "r")

# Throw away any exising log data
a.read()

# Wait for new data and call ProcessNewData for each new chunk that's written
while 1:
  # Wait for a change to occur
  results = win32file.ReadDirectoryChangesW (
    hDir,
    1024,
    False,
    win32con.FILE_NOTIFY_CHANGE_LAST_WRITE,
    None,
    None
  )

  # For each change, check to see if it's updating the file we're interested in
  for action, file in results:
    full_filename = os.path.join (path_to_watch, file)
    #print file, ACTIONS.get (action, "Unknown")
    if file == file_to_watch:
        newText = a.read()
        if newText != "":
            ProcessNewData( newText )

Probabil că ar putea face cu o sarcină mai mult de verificare a erorilor, dar pentru a urmări pur și simplu un fișier jurnal și a face o anumită procesare pe el înainte de a-l scuipa pe ecran, acest lucru funcționează bine.

Mulțumesc tuturor pentru contribuțiile voastre – foarte bine!

4Oh4

Pentru urmărirea unui singur fișier cu interogare și dependențe minime, iată un exemplu complet, bazat pe răspunsul lui Deestan (mai sus):

import os
import sys 
import time

class Watcher(object):
    running = True
    refresh_delay_secs = 1

    # Constructor
    def __init__(self, watch_file, call_func_on_change=None, *args, **kwargs):
        self._cached_stamp = 0
        self.filename = watch_file
        self.call_func_on_change = call_func_on_change
        self.args = args
        self.kwargs = kwargs

    # Look for changes
    def look(self):
        stamp = os.stat(self.filename).st_mtime
        if stamp != self._cached_stamp:
            self._cached_stamp = stamp
            # File has changed, so do something...
            print('File changed')
            if self.call_func_on_change is not None:
                self.call_func_on_change(*self.args, **self.kwargs)

    # Keep watching in a loop        
    def watch(self):
        while self.running: 
            try: 
                # Look for changes
                time.sleep(self.refresh_delay_secs) 
                self.look() 
            except KeyboardInterrupt: 
                print('
Done') 
                break 
            except FileNotFoundError:
                # Action on file not found
                pass
            except: 
                print('Unhandled error: %s' % sys.exc_info()[0])

# Call this function each time a change happens
def custom_action(text):
    print(text)

watch_file = 'my_file.txt'

# watcher = Watcher(watch_file)  # simple
watcher = Watcher(watch_file, custom_action, text='yes, changed')  # also call custom action function
watcher.watch()  # start the watch going

Comentarii

  • Ați putea face watch_file și _cached_stamp în liste și să le parcurgeți într-o buclă for. Totuși, nu se adaptează prea bine la un număr mare de fișiere…  > Por 4Oh4.
  • Asta nu declanșează acțiunea de fiecare dată când este rulată? _cached_stamp este setat la 0 și apoi comparat cu os.stat(self.filename).st_mtime. _cached_stamp ar trebui să fie setat la os.stat(self.filename).st_mtime în constructor, nu? –  > Por Seanonymous.
  • call_func_on_change() va fi declanșată la prima execuție a look(), , dar apoi _cached_stamp este actualizat, deci nu va fi declanșat din nou până când valoarea lui os.stat(self.filename).st_mtime. _cached_stamp se modifică. –  > Por 4Oh4.
  • Ați putea seta valoarea lui _cached_stamp în constructor, dacă nu doriți ca call_func_on_change() să fie apelată la prima execuție –  > Por 4Oh4.
  • Am folosit scriptul tău pentru a apela o funcție la schimbarea fișierului. Funcția mea nu primește niciun argument, spre deosebire de a ta. M-am gândit că pentru a o face să funcționeze trebuie să înlătur *args, **kwargs Se pare că (am pus doar liniile cu modificări): self.call_func_on_change(self) def custom_action(): watcher = Watcher(watch_file, custom_action()) Dar acest lucru nu a funcționat. Acțiunea a fost apelată doar în timpul primei iterații: Fișier schimbat da, schimbat Fișier schimbat Fișier schimbat Fișier schimbat Fișier schimbat A început să funcționeze când am păstrat *args și am apelat-o: watcher = Watcher(watch_file, custom_action) Mă chinui să mă întreb de ce? –  > Por zwornik.
Bruno De Fraine

Verificați răspunsul meu la o întrebare similară. Ați putea încerca aceeași buclă în Python. Această pagină sugerează:

import time

while 1:
    where = file.tell()
    line = file.readline()
    if not line:
        time.sleep(1)
        file.seek(where)
    else:
        print line, # already has newline

Vedeți și întrebarea tail() un fișier cu Python.

Comentarii

  • Puteți sys.stdout.write(line). Codul dvs. nu funcționează dacă fișierul este trunchiat. Python are încorporată funcția file(). –  > Por jfs.
  • Am postat o versiune modificată a codului tău. Puteți să o încorporați în răspunsul dumneavoastră dacă funcționează pentru dumneavoastră. –  > Por jfs.
redestructa

Cea mai simplă soluție pentru mine este să folosesc instrumentul watchmedo al lui watchdog

De la https://pypi.python.org/pypi/watchdog Acum am un proces care caută fișierele sql într-un director și le execută dacă este necesar.

watchmedo shell-command 
--patterns="*.sql" 
--recursive 
--command='~/Desktop/load_files_into_mysql_database.sh' 
.

ronedg

Aceasta este o altă modificare a scriptului lui Tim Goldan care rulează pe tipuri unix și adaugă un observator simplu pentru modificarea fișierelor prin utilizarea unui dict (file=>time).

utilizare: whateverName.py path_to_dir_to_watch

#!/usr/bin/env python

import os, sys, time

def files_to_timestamp(path):
    files = [os.path.join(path, f) for f in os.listdir(path)]
    return dict ([(f, os.path.getmtime(f)) for f in files])

if __name__ == "__main__":

    path_to_watch = sys.argv[1]
    print('Watching {}..'.format(path_to_watch))

    before = files_to_timestamp(path_to_watch)

    while 1:
        time.sleep (2)
        after = files_to_timestamp(path_to_watch)

        added = [f for f in after.keys() if not f in before.keys()]
        removed = [f for f in before.keys() if not f in after.keys()]
        modified = []

        for f in before.keys():
            if not f in removed:
                if os.path.getmtime(f) != before.get(f):
                    modified.append(f)

        if added: print('Added: {}'.format(', '.join(added)))
        if removed: print('Removed: {}'.format(', '.join(removed)))
        if modified: print('Modified: {}'.format(', '.join(modified)))

        before = after

Comentarii

  • Actualizat pentru a suporta python3 –  > Por ronedg.
seuvitor

Ei bine, din moment ce folosiți Python, puteți pur și simplu să deschideți un fișier și să continuați să citiți linii din el.

f = open('file.log')

Dacă linia citită este nu este goală, , o procesați.

line = f.readline()
if line:
    // Do what you want with the line

S-ar putea să vă lipsească faptul că este în regulă să continuați să apelați readline la EOF. În acest caz, va continua să returneze un șir gol. Iar atunci când se adaugă ceva la fișierul jurnal, citirea va continua de unde s-a oprit, după cum aveți nevoie.

Dacă sunteți în căutarea unei soluții care utilizează evenimente sau o anumită bibliotecă, vă rugăm să specificați acest lucru în întrebare. În caz contrar, cred că această soluție este foarte bună.

AlaXul

Iată o versiune simplificată a codului lui Kender care pare să facă același truc și nu importă întregul fișier:

# Check file for new data.

import time

f = open(r'c:temptest.txt', 'r')

while True:

    line = f.readline()
    if not line:
        time.sleep(1)
        print 'Nothing New'
    else:
        print 'Call Function: ', line

gimel

După cum puteți vedea în articolul lui Tim Golden, , indicat de Horst Gutmann, WIN32 este relativ complex și urmărește directoare, nu un singur fișier.

Aș dori să vă sugerez să vă uitați la IronPython, , care este un .NET python.Cu IronPython puteți folosi toate funcțiile .NET inclusiv

System.IO.FileSystemWatcher

care gestionează fișiere unice cu o simplă Eveniment interfață simplă.

Comentarii

  • @Ciasto, deoarece atunci trebuie să aveți la dispoziție Iron Python, mai degrabă decât o instalare Python de bază. –  > Por Jon Cage.
Bassim Huis

Acesta este un exemplu de verificare a modificării unui fișier. Unul care poate nu este cel mai bun mod de a face acest lucru, dar cu siguranță este un mod scurt.

Un instrument util pentru repornirea aplicației atunci când au fost făcute modificări în sursă. Am făcut acest lucru atunci când mă jucam cu pygame, astfel încât să pot vedea efectele care au loc imediat după salvarea fișierului.

Atunci când este folosit în pygame, asigurați-vă că lucrurile din bucla „while” sunt plasate în bucla de joc aka update sau orice altceva. În caz contrar, aplicația dvs. va rămâne blocată într-o buclă infinită și nu veți vedea actualizarea jocului.

file_size_stored = os.stat('neuron.py').st_size

  while True:
    try:
      file_size_current = os.stat('neuron.py').st_size
      if file_size_stored != file_size_current:
        restart_program()
    except: 
      pass

În cazul în care ați dorit codul de repornire pe care l-am găsit pe web. Iată-l aici. (Nu este relevant pentru întrebare, deși ar putea fi util)

def restart_program(): #restart application
    python = sys.executable
    os.execl(python, python, * sys.argv)

Distracție plăcută în a face electronii să facă ceea ce vrei tu să facă.

Comentarii

  • Se pare că folosind .st_mtime în loc de .st_size ar fi mai fiabil și o modalitate la fel de scurtă de a face acest lucru, deși OP a indicat că nu a vrut să o facă prin sondaj. –  > Por martineau.
imp
ACTIONS = {
  1 : "Created",
  2 : "Deleted",
  3 : "Updated",
  4 : "Renamed from something",
  5 : "Renamed to something"
}
FILE_LIST_DIRECTORY = 0x0001

class myThread (threading.Thread):
    def __init__(self, threadID, fileName, directory, origin):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.fileName = fileName
        self.daemon = True
        self.dir = directory
        self.originalFile = origin
    def run(self):
        startMonitor(self.fileName, self.dir, self.originalFile)

def startMonitor(fileMonitoring,dirPath,originalFile):
    hDir = win32file.CreateFile (
        dirPath,
        FILE_LIST_DIRECTORY,
        win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
        None,
        win32con.OPEN_EXISTING,
        win32con.FILE_FLAG_BACKUP_SEMANTICS,
        None
    )
    # Wait for new data and call ProcessNewData for each new chunk that's
    # written
    while 1:
        # Wait for a change to occur
        results = win32file.ReadDirectoryChangesW (
            hDir,
            1024,
            False,
            win32con.FILE_NOTIFY_CHANGE_LAST_WRITE,
            None,
            None
        )
        # For each change, check to see if it's updating the file we're
        # interested in
        for action, file_M in results:
            full_filename = os.path.join (dirPath, file_M)
            #print file, ACTIONS.get (action, "Unknown")
            if len(full_filename) == len(fileMonitoring) and action == 3:
                #copy to main file
                ...

utilizator1277936

Iată un exemplu orientat spre urmărirea fișierelor de intrare care nu scriu mai mult de o linie pe secundă, dar de obicei mult mai puțin. Scopul este de a adăuga ultima linie (cea mai recentă scriere) la fișierul de ieșire specificat. Am copiat acest lucru de la unul dintre proiectele mele și am șters pur și simplu toate liniile irelevante. Va trebui să completați sau să modificați simbolurile lipsă.

from PyQt5.QtCore import QFileSystemWatcher, QSettings, QThread
from ui_main_window import Ui_MainWindow   # Qt Creator gen'd 

class MainWindow(QMainWindow, Ui_MainWindow):
    def __init__(self, parent=None):
        QMainWindow.__init__(self, parent)
        Ui_MainWindow.__init__(self)
        self._fileWatcher = QFileSystemWatcher()
        self._fileWatcher.fileChanged.connect(self.fileChanged)

    def fileChanged(self, filepath):
        QThread.msleep(300)    # Reqd on some machines, give chance for write to complete
        # ^^ About to test this, may need more sophisticated solution
        with open(filepath) as file:
            lastLine = list(file)[-1]
        destPath = self._filemap[filepath]['dest file']
        with open(destPath, 'a') as out_file:               # a= append
            out_file.writelines([lastLine])

Desigur, clasa cuprinzătoare QMainWindow nu este strict necesară, adică puteți folosi QFileSystemWatcher singur.

Gus

Se pare că nimeni nu a postat fswatch. Este un observator al sistemului de fișiere pentru mai multe platforme. Doar instalați-l, rulați-l și urmați instrucțiunile.

L-am folosit cu programe python și golang și funcționează pur și simplu.

Magnus

Din moment ce îl am instalat la nivel global, abordarea mea preferată este să folosesc nodemon. Dacă codul sursă se află în src, , iar punctul dvs. de intrare este src/app.py, , atunci este la fel de simplu ca:

nodemon -w 'src/**' -e py,html --exec python src/app.py

… unde -e py,html vă permite să controlați ce tipuri de fișiere să urmărească modificările.

Rafal Enden

De asemenea, puteți utiliza o bibliotecă simplă numită repyt, , iată un exemplu:

repyt ./app.py

mexekanez

legat de soluția @4Oh4 o schimbare lină pentru o listă de fișiere de urmărit;

import os
import sys
import time

class Watcher(object):
    running = True
    refresh_delay_secs = 1

    # Constructor
    def __init__(self, watch_files, call_func_on_change=None, *args, **kwargs):
        self._cached_stamp = 0
        self._cached_stamp_files = {}
        self.filenames = watch_files
        self.call_func_on_change = call_func_on_change
        self.args = args
        self.kwargs = kwargs

    # Look for changes
    def look(self):
        for file in self.filenames:
            stamp = os.stat(file).st_mtime
            if not file in self._cached_stamp_files:
                self._cached_stamp_files[file] = 0
            if stamp != self._cached_stamp_files[file]:
                self._cached_stamp_files[file] = stamp
                # File has changed, so do something...
                file_to_read = open(file, 'r')
                value = file_to_read.read()
                print("value from file", value)
                file_to_read.seek(0)
                if self.call_func_on_change is not None:
                    self.call_func_on_change(*self.args, **self.kwargs)

    # Keep watching in a loop
    def watch(self):
        while self.running:
            try:
                # Look for changes
                time.sleep(self.refresh_delay_secs)
                self.look()
            except KeyboardInterrupt:
                print('
Done')
                break
            except FileNotFoundError:
                # Action on file not found
                pass
            except Exception as e:
                print(e)
                print('Unhandled error: %s' % sys.exc_info()[0])

# Call this function each time a change happens
def custom_action(text):
    print(text)
    # pass

watch_files = ['/Users/mexekanez/my_file.txt', '/Users/mexekanez/my_file1.txt']

# watcher = Watcher(watch_file)  # simple



if __name__ == "__main__":
    watcher = Watcher(watch_files, custom_action, text='yes, changed')  # also call custom action function
    watcher.watch()  # start the watch going

george

Cea mai bună și mai simplă soluție este să folosiți pygtail: https://pypi.python.org/pypi/pygtail

from pygtail import Pygtail
import sys

while True:
    for line in Pygtail("some.log"):
        sys.stdout.write(line)

John Henckel

Dacă utilizați Windows, creați acest fișier POLL.CMD

@echo off
:top
xcopy /m /y %1 %2 | find /v "File(s) copied"
timeout /T 1 > nul
goto :top

apoi puteți tasta „poll dir1 dir2” și acesta va copia toate fișierele din dir1 în dir2 și va verifica dacă există actualizări o dată pe secundă.

„find” este opțional, doar pentru a face consola mai puțin zgomotoasă.

Acest lucru nu este recursiv. Poate că ați putea să o faceți recursivă folosind /e la xcopy.

scable

Nu cunosc nicio funcție specifică pentru Windows. Ați putea încerca să obțineți hash-ul MD5 al fișierului la fiecare secundă/minut/oră (depinde de cât de repede aveți nevoie de el) și să-l comparați cu ultimul hash. Când diferă, știi că fișierul a fost modificat și citești cele mai noi linii.

kender

Eu aș încerca ceva de genul acesta.

    try:
            f = open(filePath)
    except IOError:
            print "No such file: %s" % filePath
            raw_input("Press Enter to close window")
    try:
            lines = f.readlines()
            while True:
                    line = f.readline()
                    try:
                            if not line:
                                    time.sleep(1)
                            else:
                                    functionThatAnalisesTheLine(line)
                    except Exception, e:
                            # handle the exception somehow (for example, log the trace) and raise the same exception again
                            raw_input("Press Enter to close window")
                            raise e
    finally:
            f.close()

Bucla verifică dacă există linii noi de la ultima citire a fișierului – dacă există, acestea sunt citite și trecute la functionThatAnalisesTheLine funcție. Dacă nu, scriptul așteaptă 1 secundă și reîncearcă procesul.

Comentarii

  • -1: Deschiderea fișierului și citirea liniilor nu este o idee prea bună atunci când fișierele pot avea 100 de MB. De asemenea, ar trebui să îl executați pentru fiecare fișier în parte, ceea ce ar fi rău atunci când doriți să urmăriți 1000 de fișiere. –  > Por Jon Cage.
  • Chiar așa? Deschideți fișierul pentru modificări? –  > Por Farshid Ashouri.