ocrd.resource_manager.OcrdResourceManager.find_resources() - Code Metrics - Inspection of "Resmgr tess config" - OCR-D/core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#680)

by Konstantin

created 2021-03-05 18:11 UTC

OcrdResourceManager.find_resources() D

↳ Parent: ocrd.resource_manager

Complexity

Conditions

Size

Total Lines	18
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	15
dl	0
loc	18
rs	4.2
c	0
b	0
f	0
cc	13
nop	5

How to fix Complexity

from pathlib import Path
from os.path import join
from os import environ, listdir, getcwd
import re
from shutil import copytree
from datetime import datetime
from tarfile import open as open_tarfile
from urllib.parse import urlparse, unquote

import requests
from yaml import safe_load, safe_dump

from ocrd_validators import OcrdResourceListValidator
from ocrd_utils import getLogger
from ocrd_utils.constants import HOME, XDG_DATA_HOME, XDG_CONFIG_HOME
from ocrd_utils.os import list_all_resources, pushd_popd

from .constants import RESOURCE_LIST_FILENAME, RESOURCE_USER_LIST_COMMENT

class OcrdResourceManager():

    """
    Managing processor resources
    """
    def __init__(self):
        self.log = getLogger('ocrd.resource_manager')
        self.database = {}
        self.load_resource_list(Path(RESOURCE_LIST_FILENAME))
        self.user_list = Path(XDG_CONFIG_HOME, 'ocrd', 'resources.yml')
        if not self.user_list.exists():
            if not self.user_list.parent.exists():
                self.user_list.parent.mkdir(parents=True)
            with open(str(self.user_list), 'w', encoding='utf-8') as f:
                f.write(RESOURCE_USER_LIST_COMMENT)
        self.load_resource_list(self.user_list)

    def load_resource_list(self, list_filename, database=None):
        if not database:
            database = self.database
        if list_filename.is_file():
            with open(list_filename, 'r', encoding='utf-8') as f:
                list_loaded = safe_load(f) or {}
            report = OcrdResourceListValidator.validate(list_loaded)
            if not report.is_valid:
                self.log.error('\n'.join(report.errors))
                raise ValueError("Resource list %s is invalid!" % (list_filename))
            for executable, resource_list in list_loaded.items():
                if executable not in database:
                    database[executable] = []
                # Prepend, so user provided is sorted before builtin
                database[executable] = list_loaded[executable] + database[executable]
        return database

    def list_available(self, executable=None):
        """
        List models available for download by processor
        """
        if executable:
            return [(executable, self.database[executable])]
        return [(x, y) for x, y in self.database.items()]

    def list_installed(self, executable=None):
        """
        List installed resources, matching with registry by ``name``
        """
        ret = []
        if executable:
            all_executables = [executable]
        else:
            # resources we know about
            all_executables = list(self.database.keys())
            # resources in the file system
            parent_dirs = [join(x, 'ocrd-resources') for x in [XDG_DATA_HOME, '/usr/local/share']]
            for parent_dir in parent_dirs:
                if Path(parent_dir).exists():
                    all_executables += [x for x in listdir(parent_dir) if x.startswith('ocrd-')]
        for this_executable in set(all_executables):
            reslist = []
            for res_filename in list_all_resources(this_executable):
                res_name = Path(res_filename).name
                resdict = [x for x in self.database.get(this_executable, []) if x['name'] == res_name]
                if not resdict:
                    self.log.info("%s resource '%s' (%s) not a known resource, creating stub in %s'" % (this_executable, res_name, res_filename, self.user_list))
                    resdict = [self.add_to_user_database(this_executable, res_filename)]
                resdict[0]['path'] = res_filename
                reslist.append(resdict[0])
            ret.append((this_executable, reslist))
        return ret

    def add_to_user_database(self, executable, res_filename, url=None):
        """
        Add a stub entry to the user resource.yml
        """
        res_name = Path(res_filename).name
        res_size = Path(res_filename).stat().st_size
        with open(self.user_list, 'r', encoding='utf-8') as f:
            user_database = safe_load(f) or {}
        if executable not in user_database:
            user_database[executable] = []
        if not self.find_resources(executable=executable, name=res_name, database=user_database):
            resdict = {
                'name': res_name,
                'url': url if url else '???',
                'description': 'Found at %s on %s' % (self.resource_dir_to_location(res_filename), datetime.now()),
                'version_range': '???',
                'size': res_size
            }
            user_database[executable].append(resdict)
        with open(self.user_list, 'w', encoding='utf-8') as f:
            f.write(RESOURCE_USER_LIST_COMMENT)
            f.write('\n')
            f.write(safe_dump(user_database))
        return resdict


    def find_resources(self, executable=None, name=None, url=None, database=None):
        """
        Find resources in the registry
        """
        if not database:
            database = self.database
        ret = []
        if executable and executable not in database.keys():
            return ret
        for executable in [executable] if executable else database.keys():
            for resdict in database[executable]:
                if not name and not url:
                    ret.append((executable, resdict))
                elif url and url == resdict['url']:
                    ret.append((executable, resdict))
                elif name and name == resdict['name']:
                    ret.append((executable, resdict))
        return ret

    @property
    def default_resource_dir(self):
        return self.location_to_resource_dir('data')

    def location_to_resource_dir(self, location):
        return '/usr/local/share/ocrd-resources' if location == 'system' else \
                join(XDG_DATA_HOME, 'ocrd-resources') if location == 'data' else \
                getcwd()

    def resource_dir_to_location(self, resource_path):
        resource_path = str(resource_path)
        return 'system' if resource_path.startswith('/usr/local/share/ocrd-resources') else \
               'data' if resource_path.startswith(join(XDG_DATA_HOME, 'ocrd-resources')) else \
               'cwd' if resource_path.startswith(getcwd()) else \
               resource_path

    def parameter_usage(self, name, usage='as-is'):
        if usage == 'as-is':
            return name
        if usage == 'without-extension':
            return Path(name).stem

    def _download_impl(self, url, filename, progress_cb=None, size=None):
        log = getLogger('ocrd.resource_manager._download_impl')
        log.info("Downloading %s to %s" % (url, filename))
        with open(filename, 'wb') as f:
            with requests.get(url, stream=True) as r:
                total = size if size else int(r.headers.get('content-length'))
                for data in r.iter_content(chunk_size=4096):
                    if progress_cb:
                        progress_cb(len(data))
                    f.write(data)

    def _copy_impl(self, src_filename, filename, progress_cb=None):
        log = getLogger('ocrd.resource_manager._copy_impl')
        log.info("Copying %s" % src_filename)
        with open(filename, 'wb') as f_out, open(src_filename, 'rb') as f_in:
            while True:
                chunk = f_in.read(4096)
                if chunk:
                    f_out.write(chunk)
                    if progress_cb:
                        progress_cb(len(chunk))
                else:
                    break

    # TODO Proper caching (make head request for size, If-Modified etc)
    def download(
        self,
        executable,
        url,
        basedir,
        overwrite=False,
        no_subdir=False,
        name=None,
        resource_type='file',
        path_in_archive='.',
        progress_cb=None,
        size=None,
    ):
        """
        Download a resource by URL
        """
        log = getLogger('ocrd.resource_manager.download')
        destdir = Path(basedir) if no_subdir else Path(basedir, executable)
        if not name:
            url_parsed = urlparse(url)
            name = Path(unquote(url_parsed.path)).name
        fpath = Path(destdir, name)
        is_url = url.startswith('https://') or url.startswith('http://')
        if fpath.exists() and not overwrite:
            log.info("%s to be %s to %s which already exists and overwrite is False" % (url, 'downloaded' if is_url else 'copied', fpath))
            return fpath
        destdir.mkdir(parents=True, exist_ok=True)
        if resource_type == 'file':
            if is_url:
                self._download_impl(url, fpath, progress_cb)
            else:
                self._copy_impl(url, fpath, progress_cb)
        elif resource_type == 'tarball':
            with pushd_popd(tempdir=True):
                if is_url:
                    self._download_impl(url, 'download.tar.xx', progress_cb, size)
                else:
                    self._copy_impl(url, 'download.tar.xx', progress_cb)
                Path('out').mkdir()
                with pushd_popd('out'):
                    log.info("Extracting tarball")
                    with open_tarfile('../download.tar.xx', 'r:*') as tar:
                        tar.extractall()
                    log.info("Copying '%s' from tarball to %s" % (path_in_archive, fpath))
                    copytree(path_in_archive, str(fpath))
        # TODO
        # elif resource_type == 'github-dir':
        return fpath


1			from pathlib import Path
2			from os.path import join
3			from os import environ, listdir, getcwd
4			import re
5			from shutil import copytree
6			from datetime import datetime
7			from tarfile import open as open_tarfile
8			from urllib.parse import urlparse, unquote
9
10			import requests
11			from yaml import safe_load, safe_dump
12
13			from ocrd_validators import OcrdResourceListValidator
14			from ocrd_utils import getLogger
15			from ocrd_utils.constants import HOME, XDG_DATA_HOME, XDG_CONFIG_HOME
16			from ocrd_utils.os import list_all_resources, pushd_popd
17
18			from .constants import RESOURCE_LIST_FILENAME, RESOURCE_USER_LIST_COMMENT
19
20			class OcrdResourceManager():
21
22			"""
23			Managing processor resources
24			"""
25			def __init__(self):
26			self.log = getLogger('ocrd.resource_manager')
27			self.database = {}
28			self.load_resource_list(Path(RESOURCE_LIST_FILENAME))
29			self.user_list = Path(XDG_CONFIG_HOME, 'ocrd', 'resources.yml')
30			if not self.user_list.exists():
31			if not self.user_list.parent.exists():
32			self.user_list.parent.mkdir(parents=True)
33			with open(str(self.user_list), 'w', encoding='utf-8') as f:
34			f.write(RESOURCE_USER_LIST_COMMENT)
35			self.load_resource_list(self.user_list)
36
37			def load_resource_list(self, list_filename, database=None):
38			if not database:
39			database = self.database
40			if list_filename.is_file():
41			with open(list_filename, 'r', encoding='utf-8') as f:
42			list_loaded = safe_load(f) or {}
43			report = OcrdResourceListValidator.validate(list_loaded)
44			if not report.is_valid:
45			self.log.error('\n'.join(report.errors))
46			raise ValueError("Resource list %s is invalid!" % (list_filename))
47			for executable, resource_list in list_loaded.items():
48			if executable not in database:
49			database[executable] = []
50			# Prepend, so user provided is sorted before builtin
51			database[executable] = list_loaded[executable] + database[executable]
52			return database
53
54			def list_available(self, executable=None):
55			"""
56			List models available for download by processor
57			"""
58			if executable:
59			return [(executable, self.database[executable])]
60			return [(x, y) for x, y in self.database.items()]
61
62			def list_installed(self, executable=None):
63			"""
64			List installed resources, matching with registry by ``name``
65			"""
66			ret = []
67			if executable:
68			all_executables = [executable]
69			else:
70			# resources we know about
71			all_executables = list(self.database.keys())
72			# resources in the file system
73			parent_dirs = [join(x, 'ocrd-resources') for x in [XDG_DATA_HOME, '/usr/local/share']]
74			for parent_dir in parent_dirs:
75			if Path(parent_dir).exists():
76			all_executables += [x for x in listdir(parent_dir) if x.startswith('ocrd-')]
77			for this_executable in set(all_executables):
78			reslist = []
79			for res_filename in list_all_resources(this_executable):
80			res_name = Path(res_filename).name
81			resdict = [x for x in self.database.get(this_executable, []) if x['name'] == res_name]
82			if not resdict:
83			self.log.info("%s resource '%s' (%s) not a known resource, creating stub in %s'" % (this_executable, res_name, res_filename, self.user_list))
84			resdict = [self.add_to_user_database(this_executable, res_filename)]
85			resdict[0]['path'] = res_filename
86			reslist.append(resdict[0])
87			ret.append((this_executable, reslist))
88			return ret
89
90			def add_to_user_database(self, executable, res_filename, url=None):
91			"""
92			Add a stub entry to the user resource.yml
93			"""
94			res_name = Path(res_filename).name
95			res_size = Path(res_filename).stat().st_size
96			with open(self.user_list, 'r', encoding='utf-8') as f:
97			user_database = safe_load(f) or {}
98			if executable not in user_database:
99			user_database[executable] = []
100			if not self.find_resources(executable=executable, name=res_name, database=user_database):
101			resdict = {
102			'name': res_name,
103			'url': url if url else '???',
104			'description': 'Found at %s on %s' % (self.resource_dir_to_location(res_filename), datetime.now()),
105			'version_range': '???',
106			'size': res_size
107			}
108			user_database[executable].append(resdict)
109			with open(self.user_list, 'w', encoding='utf-8') as f:
110			f.write(RESOURCE_USER_LIST_COMMENT)
111			f.write('\n')
112			f.write(safe_dump(user_database))
113			return resdict
			0 ignored issues – show introduced 2021-01-05 13:16 UTC by Report Bug Copy Issue Report The variable `resdict` does not seem to be defined in case `BooleanNotNode` on line `100` is `False`. Are you sure this can never be the case? Loading history...
114
115			def find_resources(self, executable=None, name=None, url=None, database=None):
116			"""
117			Find resources in the registry
118			"""
119			if not database:
120			database = self.database
121			ret = []
122			if executable and executable not in database.keys():
123			return ret
124			for executable in [executable] if executable else database.keys():
125			for resdict in database[executable]:
126			if not name and not url:
127			ret.append((executable, resdict))
128			elif url and url == resdict['url']:
129			ret.append((executable, resdict))
130			elif name and name == resdict['name']:
131			ret.append((executable, resdict))
132			return ret
133
134			@property
135			def default_resource_dir(self):
136			return self.location_to_resource_dir('data')
137
138			def location_to_resource_dir(self, location):
139			return '/usr/local/share/ocrd-resources' if location == 'system' else \
140			join(XDG_DATA_HOME, 'ocrd-resources') if location == 'data' else \
141			getcwd()
142
143			def resource_dir_to_location(self, resource_path):
144			resource_path = str(resource_path)
145			return 'system' if resource_path.startswith('/usr/local/share/ocrd-resources') else \
146			'data' if resource_path.startswith(join(XDG_DATA_HOME, 'ocrd-resources')) else \
147			'cwd' if resource_path.startswith(getcwd()) else \
148			resource_path
149
150			def parameter_usage(self, name, usage='as-is'):
151			if usage == 'as-is':
152			return name
153			if usage == 'without-extension':
154			return Path(name).stem
155
156			def _download_impl(self, url, filename, progress_cb=None, size=None):
157			log = getLogger('ocrd.resource_manager._download_impl')
158			log.info("Downloading %s to %s" % (url, filename))
159			with open(filename, 'wb') as f:
160			with requests.get(url, stream=True) as r:
161			total = size if size else int(r.headers.get('content-length'))
162			for data in r.iter_content(chunk_size=4096):
163			if progress_cb:
164			progress_cb(len(data))
165			f.write(data)
166
167			def _copy_impl(self, src_filename, filename, progress_cb=None):
168			log = getLogger('ocrd.resource_manager._copy_impl')
169			log.info("Copying %s" % src_filename)
170			with open(filename, 'wb') as f_out, open(src_filename, 'rb') as f_in:
171			while True:
172			chunk = f_in.read(4096)
173			if chunk:
174			f_out.write(chunk)
175			if progress_cb:
176			progress_cb(len(chunk))
177			else:
178			break
179
180			# TODO Proper caching (make head request for size, If-Modified etc)
181			def download(
182			self,
183			executable,
184			url,
185			basedir,
186			overwrite=False,
187			no_subdir=False,
188			name=None,
189			resource_type='file',
190			path_in_archive='.',
191			progress_cb=None,
192			size=None,
193			):
194			"""
195			Download a resource by URL
196			"""
197			log = getLogger('ocrd.resource_manager.download')
198			destdir = Path(basedir) if no_subdir else Path(basedir, executable)
199			if not name:
200			url_parsed = urlparse(url)
201			name = Path(unquote(url_parsed.path)).name
202			fpath = Path(destdir, name)
203			is_url = url.startswith('https://') or url.startswith('http://')
204			if fpath.exists() and not overwrite:
205			log.info("%s to be %s to %s which already exists and overwrite is False" % (url, 'downloaded' if is_url else 'copied', fpath))
206			return fpath
207			destdir.mkdir(parents=True, exist_ok=True)
208			if resource_type == 'file':
209			if is_url:
210			self._download_impl(url, fpath, progress_cb)
211			else:
212			self._copy_impl(url, fpath, progress_cb)
213			elif resource_type == 'tarball':
214			with pushd_popd(tempdir=True):
215			if is_url:
216			self._download_impl(url, 'download.tar.xx', progress_cb, size)
217			else:
218			self._copy_impl(url, 'download.tar.xx', progress_cb)
219			Path('out').mkdir()
220			with pushd_popd('out'):
221			log.info("Extracting tarball")
222			with open_tarfile('../download.tar.xx', 'r:*') as tar:
223			tar.extractall()
224			log.info("Copying '%s' from tarball to %s" % (path_in_archive, fpath))
225			copytree(path_in_archive, str(fpath))
226			# TODO
227			# elif resource_type == 'github-dir':
228			return fpath
229

OCR-D / core

Pull Request — master (#680)

OcrdResourceManager.find_resources() D

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like