/usr/share/pyshared/livereload/task.py is in python-livereload 1.0.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | # -*- coding: utf-8 -*-
"""livereload.task
Task management for LiveReload Server.
A basic syntax overview::
from livereload.task import Task
Task.add('file.css')
def do_some_thing():
pass
Task.add('file.css', do_some_thing)
"""
import os
import glob
import logging
try:
import pyinotify
from tornado import ioloop
class TaskEventHandler(pyinotify.ProcessEvent):
def my_init(self, **kwargs):
self.func = kwargs['func']
def process_default(self, event):
if Task.watch():
self.func()
HAS_PYINOTIFY = True
except ImportError:
HAS_PYINOTIFY = False
IGNORE = [
'.pyc', '.pyo', '.o', '.swp'
]
class Task(object):
tasks = {}
_modified_times = {}
last_modified = None
if HAS_PYINOTIFY:
wm = pyinotify.WatchManager()
notifier = None
@classmethod
def add(cls, path, func=None):
logging.info('Add task: %s' % path)
if HAS_PYINOTIFY:
cls.wm.add_watch(path, pyinotify.IN_CREATE | pyinotify.IN_DELETE | pyinotify.IN_MODIFY, rec=True, do_glob=True, auto_add=True)
cls.tasks[path] = func
@classmethod
def start(cls, func):
if HAS_PYINOTIFY:
if not cls.notifier:
cls.notifier = pyinotify.TornadoAsyncNotifier(cls.wm, ioloop.IOLoop.instance(), default_proc_fun=TaskEventHandler(func=func))
Task.watch() # initial run so we don't miss the first change
return HAS_PYINOTIFY
@classmethod
def watch(cls):
_changed = False
for path in cls.tasks:
if cls.is_changed(path):
_changed = True
func = cls.tasks[path]
func and func()
return _changed
@classmethod
def is_changed(cls, path):
def is_file_changed(path):
if not os.path.isfile(path):
return False
_, ext = os.path.splitext(path)
if ext in IGNORE:
return False
modified = int(os.stat(path).st_mtime)
if path not in cls._modified_times:
cls._modified_times[path] = modified
return True
if path in cls._modified_times and \
cls._modified_times[path] != modified:
logging.info('file changed: %s' % path)
cls._modified_times[path] = modified
cls.last_modified = path
return True
cls._modified_times[path] = modified
return False
def is_folder_changed(path):
_changed = False
for root, dirs, files in os.walk(path, followlinks=True):
if '.git' in dirs:
dirs.remove('.git')
if '.hg' in dirs:
dirs.remove('.hg')
if '.svn' in dirs:
dirs.remove('.svn')
if '.cvs' in dirs:
dirs.remove('.cvs')
for f in files:
if is_file_changed(os.path.join(root, f)):
_changed = True
return _changed
def is_glob_changed(path):
_changed = False
for f in glob.glob(path):
if is_file_changed(f):
_changed = True
return _changed
if os.path.isfile(path):
return is_file_changed(path)
elif os.path.isdir(path):
return is_folder_changed(path)
else:
return is_glob_changed(path)
return False
|