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

Passed

Pull Request — master (#559)

by Konstantin

created 2021-01-21 18:16 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_CACHE_HOME, XDG_CONFIG_HOME, XDG_DATA_HOME, VIRTUAL_ENV
from ocrd_utils.os import list_all_resources, pushd_popd

from .constants import RESOURCE_LIST_FILENAME, RESOURCE_USER_LIST_COMMENT
from .config import load_config_file

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()
            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_CACHE_HOME, XDG_CONFIG_HOME, XDG_DATA_HOME]]
            if VIRTUAL_ENV:
                parent_dirs += [join(VIRTUAL_ENV, 'share', 'ocrd-resources')]
            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

    def location_to_resource_dir(self, location):
        return join(VIRTUAL_ENV, 'share', 'ocrd-resources') if location == 'virtualenv' and VIRTUAL_ENV else \
                join(XDG_CACHE_HOME, 'ocrd-resources') if location == 'cache' else \
                join(XDG_DATA_HOME, 'ocrd-resources') if location == 'data' else \
                join(XDG_CONFIG_HOME, 'ocrd-resources') if location == 'config' else \
                getcwd()

    def resource_dir_to_location(self, resource_path):
        resource_path = str(resource_path)
        return 'virtualenv' if VIRTUAL_ENV and resource_path.startswith(join(VIRTUAL_ENV, 'share', 'ocrd-resources')) else \
               'cache' if resource_path.startswith(join(XDG_CACHE_HOME, 'ocrd-resources')) else \
               'data' if resource_path.startswith(join(XDG_DATA_HOME, 'ocrd-resources')) else \
               'config' if resource_path.startswith(join(XDG_CONFIG_HOME, 'ocrd-resources')) else \
               resource_path

    @property
    def default_resource_dir(self):
        config = load_config_file()
        return self.location_to_resource_dir(config.resource_location)

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

OCR-D / core

Pull Request — master (#559)

OcrdResourceManager.find_resources() D

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like