Как просмотреть файл на предмет изменений?

Я не знаю никаких функций, специфичных для Windows. Вы можете попробовать получать хэш файла MD5 каждую секунду / минуту / час (зависит от того, насколько быстро он вам нужен) и сравнивать его с последним хешем. Если он отличается, вы знаете, что файл был изменен, и вы читаете самые новые строки.

Я бы попробовал что-нибудь подобное.

    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()

Цикл проверяет, есть ли новые строки с момента последнего чтения файла - если есть, она считывается и передается в функцию functionThatAnalisesTheLine. В противном случае сценарий ждет 1 секунду и повторяет процесс.

Проверьте мой ответ на аналогичный вопрос. Вы можете попробовать тот же цикл в Python. Эта страница предлагает:

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

Также см. Вопрос tail () файл с Python.

Вы уже смотрели документацию, доступную на http://timgolden.me.uk/python/win32_how_do_i/watch_directory_for_changes.html? Если вам нужно, чтобы он работал только под Windows, второй пример кажется именно тем, что вам нужно (если вы поменяете путь к каталогу с одним из файлов, которые хотите просмотреть).

В противном случае опрос, вероятно, будет единственным действительно независимым от платформы вариантом.

Примечание: Я не пробовал ни одного из этих решений.

Если вам достаточно опроса, я бы просто посмотрел, изменится ли статистика файла «Время изменения». Чтобы прочитать это:

os.stat(filename).st_mtime

(Также обратите внимание, что собственное решение для событий изменения Windows работает не во всех обстоятельствах, например, на сетевых дисках.)

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...

Как вы можете видеть в Статья Тима Голдена, указанном Хорст Гутманн, WIN32 относительно сложен и отслеживает каталоги, а не отдельный файл.

Я хотел бы предложить вам изучить IronPython, который является реализацией Python .СЕТЬ. С IronPython вы можете использовать все функции .СЕТЬ, включая

System.IO.FileSystemWatcher

Которая обрабатывает отдельные файлы с помощью простого интерфейса Мероприятие.

Что ж, поскольку вы используете Python, вы можете просто открыть файл и продолжать читать строки из него.

f = open('file.log')

Если прочитанная строка - не пусто, вы ее обрабатываете.

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

Возможно, вам не хватает того, что продолжать звонить readline в EOF - это нормально. В этом случае он просто будет возвращать пустую строку. И когда что-то добавляется в файл журнала, чтение будет продолжаться с того места, где оно было остановлено, как вам нужно.

Если вы ищете решение, использующее события или конкретную библиотеку, укажите это в своем вопросе. В остальном, я думаю, это решение вполне подойдет.

Что ж, после небольшого взлома сценария Тима Голдена у меня есть следующее, которое, похоже, работает довольно хорошо:

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 )

Вероятно, это можно было бы сделать с помощью дополнительной проверки ошибок, но для простого просмотра файла журнала и выполнения некоторой обработки, прежде чем выплевывать его на экран, это работает хорошо.

Спасибо всем за ваш вклад - отличный материал!

Он не должен работать в Windows (может быть, с cygwin?), Но для пользователя unix вы должны использовать системный вызов "fcntl". Вот пример на Python. В основном это тот же код, если вам нужно написать его на C (те же имена функций)

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)

Вот упрощенная версия кода Кендера, которая, похоже, выполняет тот же трюк и не импортирует весь файл:

# Check file for new data.

import time

f = open(r'c:\temp\test.txt', 'r')

while True:

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

Проверьте пиинофикация.

inotify заменяет dnotify (из более раннего ответа) в новых Linux и позволяет отслеживать на уровне файлов, а не на уровне каталогов.

Вы пробовали использовать Сторожевая собака?

Python API library and shell utilities to monitor file system events.

Directory monitoring made easy with

• A cross-platform API.
• A shell tool to run commands in response to directory changes.

Get started quickly with a simple example in Quickstart...

Если вам нужно многоплатформенное решение, отметьте QFileSystemWatcher. Вот пример кода (не дезинфицированный):

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)

Это еще одна модификация сценария Тима Голдана, который работает с типами unix и добавляет простой наблюдатель для модификации файлов с помощью dict (file => time).

использование: somethingName.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

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
                ...

Это пример проверки файла на наличие изменений. Возможно, это не лучший способ сделать это, но это точно короткий путь.

Удобный инструмент для перезапуска приложения после внесения изменений в исходный код. Я сделал это во время игры с pygame, поэтому я могу видеть, что эффекты срабатывают сразу после сохранения файла.

При использовании в pygame убедитесь, что материал из цикла while помещается в ваш игровой цикл, известный как update или что-то еще. В противном случае ваше приложение застрянет в бесконечном цикле, и вы не увидите обновления своей игры.

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

Если вам нужен код перезапуска, который я нашел в Интернете. Вот. (Не имеет отношения к вопросу, хотя может пригодиться)

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

Получайте удовольствие, заставляя электроны делать то, что вы от них хотите.

Самым простым решением для меня является использование сторожевого пса watchmedo.

Из https://pypi.python.org/pypi/watchdog у меня теперь есть процесс, который ищет файлы sql в каталоге и при необходимости выполняет их.

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

Вот пример, предназначенный для просмотра входных файлов, которые пишут не более одной строки в секунду, но обычно намного меньше. Цель состоит в том, чтобы добавить последнюю строку (самую последнюю запись) в указанный выходной файл. Я скопировал это из одного из своих проектов и просто удалил все ненужные строки. Вам нужно будет заполнить или изменить недостающие символы.

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])

Конечно, охватывающий класс QMainWindow строго не требуется, т.е. вы можете использовать только QFileSystemWatcher.

Лучшее и простое решение - использовать 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)

Для просмотра одного файла с опросом и минимальными зависимостями вот полностью проработанный пример, основанный на ответе Deestan (см. Выше):

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('\nDone') 
                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

Вы также можете использовать простую библиотеку под названием повторять, вот пример:

repyt ./app.py

Кажется, что никто не отправил fswatch. Это кроссплатформенный наблюдатель файловой системы. Просто установите его, запустите и следуйте подсказкам.

Я использовал его с программами python и golang, и он просто работает.

связанное решение @ 4Oh4 плавное изменение списка файлов для просмотра;

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('\nDone')
                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

Если вы используете Windows, создайте этот файл POLL.CMD

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

затем вы можете ввести «poll dir1 dir2», и он скопирует все файлы из dir1 в dir2 и будет проверять наличие обновлений один раз в секунду.

«Найти» не обязательно, чтобы консоль была менее шумной.

Это не рекурсивно. Возможно, вы могли бы сделать его рекурсивным, используя / e в xcopy.

Поскольку он установлен глобально, мой любимый подход - использовать nodemon. Если ваш исходный код находится на src, а ваша точка входа - src/app.py, то это так же просто, как:

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

... где -e py,html позволяет вам контролировать, какие типы файлов отслеживать изменения.