<commit_before># coding: utf-8

# This file is part of Supysonic.
#
# Supysonic is a Python implementation of the Subsonic server API.
# Copyright (C) 2013, 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 os, os.path
import time, mimetypes
import mutagen

from storm.expr import ComparableExpr, compile, Like
from storm.exceptions import NotSupportedError

from supysonic import config
from supysonic.db import Folder, Artist, Album, Track

def get_mime(ext):
	return mimetypes.guess_type('dummy.' + ext, False)[0] or config.get('mimetypes', ext) or 'application/octet-stream'

# Hacking in support for a concatenation expression
class Concat(ComparableExpr):
	__slots__ = ("left", "right", "db")

	def __init__(self, left, right, db):
		self.left = left
		self.right = right
		self.db = db

@compile.when(Concat)
def compile_concat(compile, concat, state):
	left = compile(concat.left, state)
	right = compile(concat.right, state)
	if concat.db in ('sqlite', 'postgres'):
		statement = "%s||%s"
	elif concat.db == 'mysql':
		statement = "CONCAT(%s, %s)"
	else:
		raise NotSupportedError("Unspported database (%s)" % concat.db)
	return statement % (left, right)

class Scanner:
	def __init__(self, store):
		self.__store = store

		self.__added_artists = 0
		self.__added_albums  = 0
		self.__added_tracks  = 0
		self.__deleted_artists = 0
		self.__deleted_albums  = 0
		self.__deleted_tracks  = 0

		extensions = config.get('base', 'scanner_extensions')
		self.__extensions = map(str.lower, extensions.split()) if extensions else None

		self.__folders_to_check = set()
		self.__artists_to_check = set()
		self.__albums_to_check = set()

	def __del__(self):
		if self.__folders_to_check or self.__artists_to_check or self.__albums_to_check:
			raise Exception("There's still something to check. Did you run Scanner.finish()?")

	def scan(self, folder, progress_callback = None):
		# Scan new/updated files
		files = [ os.path.join(root, f) for root, _, fs in os.walk(folder.path) for f in fs if self.__is_valid_path(os.path.join(root, f)) ]
		total = len(files)
		current = 0

		for path in files:
			self.scan_file(path)
			current += 1
			if progress_callback:
				progress_callback(current, total)

		# Remove files that have been deleted
		for track in [ t for t in self.__store.find(Track, Track.root_folder_id == folder.id) if not self.__is_valid_path(t.path) ]:
			self.remove_file(track.path)

		# Update cover art info
		folders = [ folder ]
		while folders:
			f = folders.pop()
			f.has_cover_art = os.path.isfile(os.path.join(f.path, 'cover.jpg'))
			folders += f.children

		folder.last_scan = int(time.time())

	def finish(self):
		for album in [ a for a in self.__albums_to_check if not a.tracks.count() ]:
			self.__artists_to_check.add(album.artist)
			self.__store.remove(album)
			self.__deleted_albums += 1
		self.__albums_to_check.clear()

		for artist in [ a for a in self.__artists_to_check if not a.albums.count() ]:
			self.__store.remove(artist)
			self.__deleted_artists += 1
		self.__artists_to_check.clear()

		while self.__folders_to_check:
			folder = self.__folders_to_check.pop()
			if folder.root:
				continue

			if not folder.tracks.count() and not folder.children.count():
				self.__folders_to_check.add(folder.parent)
				self.__store.remove(folder)

	def __is_valid_path(self, path):
		if not os.path.exists(path):
			return False
		if not self.__extensions:
			return True
		return os.path.splitext(path)[1][1:].lower() in self.__extensions

	def scan_file(self, path):
		tr = self.__store.find(Track, Track.path == path).one()
		add = False
		if tr:
			if not int(os.path.getmtime(path)) > tr.last_modification:
				return

			tag = self.__try_load_tag(path)
			if not tag:
				self.remove_file(path)
				return
		else:
			tag = self.__try_load_tag(path)
			if not tag:
				return

			tr = Track()
			tr.path = path
			add = True

		tr.disc     = self.__try_read_tag(tag, 'discnumber',  1, lambda x: int(x[0].split('/')[0]))
		tr.number   = self.__try_read_tag(tag, 'tracknumber', 1, lambda x: int(x[0].split('/')[0]))
		tr.title    = self.__try_read_tag(tag, 'title', '')
		tr.year     = self.__try_read_tag(tag, 'date', None, lambda x: int(x[0].split('-')[0]))
		tr.genre    = self.__try_read_tag(tag, 'genre')
		tr.duration = int(tag.info.length)
		if not add:
			tr.album = self.__find_album(self.__try_read_tag(tag, 'artist', ''), self.__try_read_tag(tag, 'album', ''))<commit_msg>B"H check old albums when track metadata changed<commit_after># coding: utf-8

# This file is part of Supysonic.
#
# Supysonic is a Python implementation of the Subsonic server API.
# Copyright (C) 2013, 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 os, os.path
import time, mimetypes
import mutagen

from storm.expr import ComparableExpr, compile, Like
from storm.exceptions import NotSupportedError

from supysonic import config
from supysonic.db import Folder, Artist, Album, Track

def get_mime(ext):
	return mimetypes.guess_type('dummy.' + ext, False)[0] or config.get('mimetypes', ext) or 'application/octet-stream'

# Hacking in support for a concatenation expression
class Concat(ComparableExpr):
	__slots__ = ("left", "right", "db")

	def __init__(self, left, right, db):
		self.left = left
		self.right = right
		self.db = db

@compile.when(Concat)
def compile_concat(compile, concat, state):
	left = compile(concat.left, state)
	right = compile(concat.right, state)
	if concat.db in ('sqlite', 'postgres'):
		statement = "%s||%s"
	elif concat.db == 'mysql':
		statement = "CONCAT(%s, %s)"
	else:
		raise NotSupportedError("Unspported database (%s)" % concat.db)
	return statement % (left, right)

class Scanner:
	def __init__(self, store):
		self.__store = store

		self.__added_artists = 0
		self.__added_albums  = 0
		self.__added_tracks  = 0
		self.__deleted_artists = 0
		self.__deleted_albums  = 0
		self.__deleted_tracks  = 0

		extensions = config.get('base', 'scanner_extensions')
		self.__extensions = map(str.lower, extensions.split()) if extensions else None

		self.__folders_to_check = set()
		self.__artists_to_check = set()
		self.__albums_to_check = set()

	def __del__(self):
		if self.__folders_to_check or self.__artists_to_check or self.__albums_to_check:
			raise Exception("There's still something to check. Did you run Scanner.finish()?")

	def scan(self, folder, progress_callback = None):
		# Scan new/updated files
		files = [ os.path.join(root, f) for root, _, fs in os.walk(folder.path) for f in fs if self.__is_valid_path(os.path.join(root, f)) ]
		total = len(files)
		current = 0

		for path in files:
			self.scan_file(path)
			current += 1
			if progress_callback:
				progress_callback(current, total)

		# Remove files that have been deleted
		for track in [ t for t in self.__store.find(Track, Track.root_folder_id == folder.id) if not self.__is_valid_path(t.path) ]:
			self.remove_file(track.path)

		# Update cover art info
		folders = [ folder ]
		while folders:
			f = folders.pop()
			f.has_cover_art = os.path.isfile(os.path.join(f.path, 'cover.jpg'))
			folders += f.children

		folder.last_scan = int(time.time())

	def finish(self):
		for album in [ a for a in self.__albums_to_check if not a.tracks.count() ]:
			self.__artists_to_check.add(album.artist)
			self.__store.remove(album)
			self.__deleted_albums += 1
		self.__albums_to_check.clear()

		for artist in [ a for a in self.__artists_to_check if not a.albums.count() ]:
			self.__store.remove(artist)
			self.__deleted_artists += 1
		self.__artists_to_check.clear()

		while self.__folders_to_check:
			folder = self.__folders_to_check.pop()
			if folder.root:
				continue

			if not folder.tracks.count() and not folder.children.count():
				self.__folders_to_check.add(folder.parent)
				self.__store.remove(folder)

	def __is_valid_path(self, path):
		if not os.path.exists(path):
			return False
		if not self.__extensions:
			return True
		return os.path.splitext(path)[1][1:].lower() in self.__extensions

	def scan_file(self, path):
		tr = self.__store.find(Track, Track.path == path).one()
		add = False
		if tr:
			if not int(os.path.getmtime(path)) > tr.last_modification:
				return

			tag = self.__try_load_tag(path)
			if not tag:
				self.remove_file(path)
				return
		else:
			tag = self.__try_load_tag(path)
			if not tag:
				return

			tr = Track()
			tr.path = path
			add = True

		tr.disc     = self.__try_read_tag(tag, 'discnumber',  1, lambda x: int(x[0].split('/')[0]))
		tr.number   = self.__try_read_tag(tag, 'tracknumber', 1, lambda x: int(x[0].split('/')[0]))
		tr.title    = self.__try_read_tag(tag, 'title', '')
		tr.year     = self.__try_read_tag(tag, 'date', None, lambda x: int(x[0].split('-')[0]))
		tr.genre    = self.__try_read_tag(tag, 'genre')
		tr.duration = int(tag.info.length)
		if not add:
			old_album = tr.album
			new_album = self.__find_album(self.__try_read_tag(tag, 'artist', ''), self.__try_read_tag(tag, 'album', ''))
			if old_album.id != new_album.id:
				tr.album = new_album
				self.__albums_to_check.add(old_album)
{'fraction_non_alphanumeric': 0.11863420312348433, 'fraction_numerical': 0.006208167620525755, 'mean_word_length': 4.009718172983479, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7351234', 'n_tokens_mistral': 3836, 'n_tokens_neox': 3475, 'n_words': 999}