<commit_before># coding: utf-8
#
# This file is part of Supysonic.
# Supysonic is a Python implementation of the Subsonic server API.
#
# Copyright (C) 2013-2018 Alban 'spl0k' Féron
#
# Distributed under terms of the GNU AGPLv3 license.

import os, os.path
import mimetypes
import mutagen
import time

from pony.orm import db_session

from .covers import find_cover_in_folder, CoverFile
from .db import Folder, Artist, Album, Track, User
from .db import StarredFolder, StarredArtist, StarredAlbum, StarredTrack
from .db import RatingFolder, RatingTrack
from .py23 import strtype

class StatsDetails(object):
    def __init__(self):
        self.artists = 0
        self.albums = 0
        self.tracks = 0

class Stats(object):
    def __init__(self):
        self.added = StatsDetails()
        self.deleted = StatsDetails()
        self.errors = []

class Scanner:
    def __init__(self, force = False, extensions = None):
        if extensions is not None and not isinstance(extensions, list):
            raise TypeError('Invalid extensions type')

        self.__force = force

        self.__stats = Stats()
        self.__extensions = extensions

    def scan(self, folder, progress_callback = None):
        if not isinstance(folder, Folder):
            raise TypeError('Expecting Folder instance, got ' + str(type(folder)))

        # Scan new/updated files
        to_scan = [ folder.path ]
        scanned = 0
        while to_scan:
            path = to_scan.pop()

            try:
                entries = os.listdir(path)
            except OSError:
                continue

            for f in entries:
                try: # test for badly encoded filenames
                    f.encode('utf-8')
                except UnicodeError:
                    self.__stats.errors.append(path)
                    continue

                full_path = os.path.join(path, f)
                if os.path.islink(full_path):
                    continue
                elif os.path.isdir(full_path):
                    to_scan.append(full_path)
                elif os.path.isfile(full_path) and self.__is_valid_path(full_path):
                    self.scan_file(full_path)
                    scanned += 1

                    if progress_callback:
                        progress_callback(scanned)

        # Remove files that have been deleted
        for track in Track.select(lambda t: t.root_folder == folder):
            if not self.__is_valid_path(track.path):
                self.remove_file(track.path)

        # Remove deleted/moved folders and update cover art info
        folders = [ folder ]
        while folders:
            f = folders.pop()

            if not f.root and not os.path.isdir(f.path):
                f.delete() # Pony will cascade
                continue

            self.find_cover(f.path)
            folders += f.children

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

    @db_session
    def finish(self):
        self.__stats.deleted.albums = Album.prune()
        self.__stats.deleted.artists = Artist.prune()
        Folder.prune()

    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

    @db_session
    def scan_file(self, path):
        if not isinstance(path, strtype):
            raise TypeError('Expecting string, got ' + str(type(path)))

        tr = Track.get(path = path)
        if tr is not None:
            if not self.__force and not int(os.path.getmtime(path)) > tr.last_modification:
                return

            tag = self.__try_load_tag(path)
            if tag is None:
                self.remove_file(path)
                return
            trdict = {}
        else:
            tag = self.__try_load_tag(path)
            if tag is None:
                return

            trdict = { 'path': path }

        artist      = self.__try_read_tag(tag, 'artist', '[unknown]')
        album       = self.__try_read_tag(tag, 'album', '[non-album tracks]')
        albumartist = self.__try_read_tag(tag, 'albumartist', artist)

        trdict['disc']     = self.__try_read_tag(tag, 'discnumber',  1, lambda x: int(x.split('/')[0]))
        trdict['number']   = self.__try_read_tag(tag, 'tracknumber', 1, lambda x: int(x.split('/')[0]))
        trdict['title']    = self.__try_read_tag(tag, 'title', os.path.basename(path))
        trdict['year']     = self.__try_read_tag(tag, 'date', None, lambda x: int(x.split('-')[0]))
        trdict['genre']    = self.__try_read_tag(tag, 'genre')
        trdict['duration'] = int(tag.info.length)
        trdict['has_art'] = bool(Track._extract_cover_art(path))

        trdict['bitrate']  = (tag.info.bitrate if hasattr(tag.info, 'bitrate') else int(os.path.getsize(path) * 8 / tag.info.length)) // 1000
        trdict['content_type'] = mimetypes.guess_type(path, False)[0] or 'application/octet-stream'
        trdict['last_modification'] = int(os.path.getmtime(path))

        tralbum = self.__find_album(albumartist, album)
        trartist = self.__find_artist(artist)

        if tr is None:
            trdict['root_folder'] = self.__find_root_folder(path)
            trdict['folder'] = self.__find_folder(path)
            trdict['album'] = tralbum
            trdict['artist'] = trartist

            Track(**trdict)
            self.__stats.added.tracks += 1
        else:
            if tr.album.id != tralbum.id:
                trdict['album'] = tralbum

            if tr.artist.id != trartist.id:<commit_msg>Truncate artist, album and title when scanning<commit_after># coding: utf-8
#
# This file is part of Supysonic.
# Supysonic is a Python implementation of the Subsonic server API.
#
# Copyright (C) 2013-2018 Alban 'spl0k' Féron
#
# Distributed under terms of the GNU AGPLv3 license.

import os, os.path
import mimetypes
import mutagen
import time

from pony.orm import db_session

from .covers import find_cover_in_folder, CoverFile
from .db import Folder, Artist, Album, Track, User
from .db import StarredFolder, StarredArtist, StarredAlbum, StarredTrack
from .db import RatingFolder, RatingTrack
from .py23 import strtype

class StatsDetails(object):
    def __init__(self):
        self.artists = 0
        self.albums = 0
        self.tracks = 0

class Stats(object):
    def __init__(self):
        self.added = StatsDetails()
        self.deleted = StatsDetails()
        self.errors = []

class Scanner:
    def __init__(self, force = False, extensions = None):
        if extensions is not None and not isinstance(extensions, list):
            raise TypeError('Invalid extensions type')

        self.__force = force

        self.__stats = Stats()
        self.__extensions = extensions

    def scan(self, folder, progress_callback = None):
        if not isinstance(folder, Folder):
            raise TypeError('Expecting Folder instance, got ' + str(type(folder)))

        # Scan new/updated files
        to_scan = [ folder.path ]
        scanned = 0
        while to_scan:
            path = to_scan.pop()

            try:
                entries = os.listdir(path)
            except OSError:
                continue

            for f in entries:
                try: # test for badly encoded filenames
                    f.encode('utf-8')
                except UnicodeError:
                    self.__stats.errors.append(path)
                    continue

                full_path = os.path.join(path, f)
                if os.path.islink(full_path):
                    continue
                elif os.path.isdir(full_path):
                    to_scan.append(full_path)
                elif os.path.isfile(full_path) and self.__is_valid_path(full_path):
                    self.scan_file(full_path)
                    scanned += 1

                    if progress_callback:
                        progress_callback(scanned)

        # Remove files that have been deleted
        for track in Track.select(lambda t: t.root_folder == folder):
            if not self.__is_valid_path(track.path):
                self.remove_file(track.path)

        # Remove deleted/moved folders and update cover art info
        folders = [ folder ]
        while folders:
            f = folders.pop()

            if not f.root and not os.path.isdir(f.path):
                f.delete() # Pony will cascade
                continue

            self.find_cover(f.path)
            folders += f.children

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

    @db_session
    def finish(self):
        self.__stats.deleted.albums = Album.prune()
        self.__stats.deleted.artists = Artist.prune()
        Folder.prune()

    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

    @db_session
    def scan_file(self, path):
        if not isinstance(path, strtype):
            raise TypeError('Expecting string, got ' + str(type(path)))

        tr = Track.get(path = path)
        if tr is not None:
            if not self.__force and not int(os.path.getmtime(path)) > tr.last_modification:
                return

            tag = self.__try_load_tag(path)
            if tag is None:
                self.remove_file(path)
                return
            trdict = {}
        else:
            tag = self.__try_load_tag(path)
            if tag is None:
                return

            trdict = { 'path': path }

        artist      = self.__try_read_tag(tag, 'artist', '[unknown]')[:255]
        album       = self.__try_read_tag(tag, 'album', '[non-album tracks]')[:255]
        albumartist = self.__try_read_tag(tag, 'albumartist', artist)[:255]

        trdict['disc']     = self.__try_read_tag(tag, 'discnumber',  1, lambda x: int(x.split('/')[0]))
        trdict['number']   = self.__try_read_tag(tag, 'tracknumber', 1, lambda x: int(x.split('/')[0]))
        trdict['title']    = self.__try_read_tag(tag, 'title', os.path.basename(path))[:255]
        trdict['year']     = self.__try_read_tag(tag, 'date', None, lambda x: int(x.split('-')[0]))
        trdict['genre']    = self.__try_read_tag(tag, 'genre')
        trdict['duration'] = int(tag.info.length)
        trdict['has_art'] = bool(Track._extract_cover_art(path))

        trdict['bitrate']  = (tag.info.bitrate if hasattr(tag.info, 'bitrate') else int(os.path.getsize(path) * 8 / tag.info.length)) // 1000
        trdict['content_type'] = mimetypes.guess_type(path, False)[0] or 'application/octet-stream'
        trdict['last_modification'] = int(os.path.getmtime(path))

        tralbum = self.__find_album(albumartist, album)
        trartist = self.__find_artist(artist)

        if tr is None:
            trdict['root_folder'] = self.__find_root_folder(path)
            trdict['folder'] = self.__find_folder(path)
            trdict['album'] = tralbum
            trdict['artist'] = trartist

            Track(**trdict)
            self.__stats.added.tracks += 1
        else:
            if tr.album.id != tralbum.id:
                trdict['album'] = tralbum

            if tr.artist.id != trartist.id:
{'fraction_non_alphanumeric': 0.10784747412191453, 'fraction_numerical': 0.006900822790409626, 'mean_word_length': 2.0007963897000267, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '1638869', 'n_tokens_mistral': 3451, 'n_tokens_neox': 3261, 'n_words': 803}