supysonic/bin/supysonic-watcher

#!/usr/bin/python
# coding: utf-8

# This file is part of Supysonic.
#
# Supysonic is a Python implementation of the Subsonic server API.
# Copyright (C) 2014  Alban 'spl0k' Féron
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import time, sys
import logging
from threading import Thread, Condition, Timer
from logging.handlers import TimedRotatingFileHandler
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

from supysonic import config, db
from supysonic.scanner import Scanner

class SupysonicWatcherEventHandler(PatternMatchingEventHandler):
	def __init__(self, queue, logger):
		extensions = config.get('base', 'scanner_extensions')
		patterns = map(lambda e: "*." + e.lower(), extensions.split()) if extensions else None
		super(SupysonicWatcherEventHandler, self).__init__(patterns = patterns, ignore_directories = True)

		self.__queue = queue
		self.__logger = logger

	def on_created(self, event):
		self.__logger.debug("File created: '%s'", event.src_path)
		self.__queue.put(event.src_path)

	def on_deleted(self, event):
		self.__logger.debug("File deleted: '%s'", event.src_path)
		store = db.get_store(config.get('base', 'database_uri'))
		track = store.find(db.Track, db.Track.path == event.src_path).one()
		if track:
			folder = track.root_folder
			Scanner(store).prune(folder)
			store.commit()
		else:
			self.__logger.debug("Deleted file %s not in the database", event.src_path)
		store.close()

	def on_modified(self, event):
		self.__logger.debug("File modified: '%s'", event.src_path)
		self.__queue.put(event.src_path)

	def on_moved(self, event):
		pass

class ScannerProcessingQueue(Thread):
	def __init__(self, logger):
		super(ScannerProcessingQueue, self).__init__()

		self.__logger = logger
		self.__cond = Condition()
		self.__timer = None
		self.__queue = {}
		self.__running = True

	def run(self):
		while self.__running:
			time.sleep(0.1)

			with self.__cond:
				self.__cond.wait()

				if not self.__queue:
					continue

			self.__logger.debug("Instantiating scanner")
			store = db.get_store(config.get('base', 'database_uri'))
			scanner = Scanner(store)

			path = self.__next_item()
			while path:
				self.__logger.info("Scanning: '%s'", path)
				scanner.scan_file(path)
				path = self.__next_item()

			store.commit()
			store.close()
			self.__logger.debug("Freeing scanner")
			del scanner

	def stop(self):
		self.__running = False
		with self.__cond:
			self.__cond.notify()

	def put(self, path):
		if not self.__running:
			raise RuntimeError("Trying to put an item in a stopped queue")

		with self.__cond:
			self.__queue[path] = time.time()
			if self.__timer:
				self.__timer.cancel()
			self.__timer = Timer(5, self.__wakeup)
			self.__timer.start()

	def __wakeup(self):
		with self.__cond:
			self.__cond.notify()
			self.__timer = None

	def __next_item(self):
		with self.__cond:
			if not self.__queue:
				return None

			next = sorted(self.__queue.iteritems(), key = lambda i: i[1])[0]
			if not self.__running or next[1] + 5 < time.time():
				del self.__queue[next[0]]
				return next[0]

			return None

if __name__ == "__main__":
	if not config.check():
		sys.exit(1)

	logger = logging.getLogger(__name__)
	if config.get('daemon', 'log_file'):
		log_handler = TimedRotatingFileHandler(config.get('daemon', 'log_file'), when = 'midnight')
	else:
		log_handler = logging.NullHandler()
	log_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
	logger.addHandler(log_handler)
	if config.get('daemon', 'log_level'):
		mapping = {
			'DEBUG':   logging.DEBUG,
			'INFO':    logging.INFO,
			'WARNING': logging.WARNING,
			'ERROR':   logging.ERROR,
			'CRTICAL': logging.CRITICAL
		}
		logger.setLevel(mapping.get(config.get('daemon', 'log_level').upper(), logging.NOTSET))

	store = db.get_store(config.get('base', 'database_uri'))
	folders = store.find(db.Folder, db.Folder.root == True)

	if not folders.count():
		logger.info("No folder set. Exiting.")
		store.close()
		sys.exit(0)

	queue = ScannerProcessingQueue(logger)
	handler = SupysonicWatcherEventHandler(queue, logger)
	observer = Observer()

	for folder in folders:
		logger.info("Starting watcher for %s", folder.path)
		observer.schedule(handler, folder.path, recursive = True)

	store.close()
	queue.start()
	observer.start()
	try:
		while True:
			time.sleep(1)
	except KeyboardInterrupt:
		logger.info("Stopping watcher")
		observer.stop()
	observer.join()
	queue.stop()
	queue.join()