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

Passed

Pull Request — master (#800)

by Konstantin

created 2022-05-20 13:50 UTC

OcrdResourceManager.download() D

↳ Parent: ocrd.resource_manager

Complexity

Conditions

Size

Total Lines	46
Code Lines	40

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	40
dl	0
loc	46
rs	4.2
c	0
b	0
f	0
cc	13
nop	11

How to fix Complexity Many Parameters

from pathlib import Path
from os.path import join
from json import loads
from json.decoder import JSONDecodeError
from os import environ, listdir, getcwd, path
from fnmatch import filter as apply_glob
from shutil import copytree
from datetime import datetime
from tarfile import open as open_tarfile
from urllib.parse import urlparse, unquote
from subprocess import run, PIPE

import requests
from yaml import safe_load, safe_dump

from ocrd_validators import OcrdResourceListValidator
from ocrd_utils import getLogger, directory_size
from ocrd_utils.os import get_processor_resource_types, list_all_resources, pushd_popd, get_ocrd_tool_json
from .constants import RESOURCE_LIST_FILENAME, RESOURCE_USER_LIST_COMMENT

class OcrdResourceManager():

    """
    Managing processor resources
    """
    def __init__(self, userdir=None, xdg_config_home=None, xdg_data_home=None):
        self.log = getLogger('ocrd.resource_manager')
        self.database = {}

        self._xdg_data_home = xdg_data_home
        self._xdg_config_home = xdg_config_home
        self._userdir = userdir
        self.user_list = Path(self.xdg_config_home, 'ocrd', 'resources.yml')

        self.load_resource_list(Path(RESOURCE_LIST_FILENAME))
        if not self.user_list.exists():
            if not self.user_list.parent.exists():
                self.user_list.parent.mkdir(parents=True)
            self.save_user_list()
        self.load_resource_list(self.user_list)

    @property
    def userdir(self):
        if not self._userdir:
            self._userdir = path.expanduser('~')
            if 'HOME' in environ and environ['HOME'] != path.expanduser('~'):
                self._userdir = environ['HOME']
        return self._userdir

    @property
    def xdg_data_home(self):
        if not self._xdg_data_home:
            if 'XDG_DATA_HOME' in environ:
                self._xdg_data_home = environ['XDG_DATA_HOME']
            else:
                self._xdg_data_home = join(self.userdir, '.local', 'share')
        return self._xdg_data_home

    @property
    def xdg_config_home(self):
        if not self._xdg_config_home:
            if 'XDG_CONFIG_HOME' in environ:
                self._xdg_config_home = environ['XDG_CONFIG_HOME']
            else:
                self._xdg_config_home = join(self.userdir, '.config')
        return self._xdg_config_home

    def save_user_list(self, database=None):
        if not database:
            database = self.database
        with open(self.user_list, 'w', encoding='utf-8') as f:
            f.write(RESOURCE_USER_LIST_COMMENT)
            f.write('\n')
            f.write(safe_dump(database))

    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, dynamic=True):
        """
        List models available for download by processor
        """
        if not executable:
            return self.database.items()
        if dynamic:
            for exec_dir in environ['PATH'].split(':'):
                for exec_path in Path(exec_dir).glob(f'{executable}'):
                    self.log.info(f"Inspecting '{exec_path} --dump-json' for resources")
                    try:
                        ocrd_tool = get_ocrd_tool_json(exec_path)
                    except JSONDecodeError:
                        self.log.info(f"Failed to parse {exec_path} --dump-json output - not an OCR-D processor?")
                    for resdict in ocrd_tool.get('resources', ()):
                        for res_remove in (res for res in self.database.get(executable, []) if res['name'] == resdict['name']):
                            self.database.get(executable).remove(res_remove)
                        self.database[exec_path.name].append(resdict)
        ret = []
        for k in self.database:
            if apply_glob([k], executable):
                ret.append((k, self.database[k]))
        return ret

    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 [self.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 = []
            mimetypes = get_processor_resource_types(this_executable)
            for res_filename in list_all_resources(this_executable, xdg_data_home=self.xdg_data_home):
                res_filename = Path(res_filename)
                if not '*/*' in mimetypes:
                    if res_filename.is_dir() and not 'text/directory' in mimetypes:
                        continue
                    if res_filename.is_file() and ['text/directory'] == mimetypes:
                        continue
                res_name = 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, str(res_filename), self.user_list)
                    resdict = [self.add_to_user_database(this_executable, res_filename)]
                resdict[0]['path'] = str(res_filename)
                reslist.append(resdict[0])
            ret.append((this_executable, reslist))
        return ret

    def add_to_user_database(self, executable, res_filename, url=None, resource_type='file'):
        """
        Add a stub entry to the user resource.yml
        """
        res_name = Path(res_filename).name
        if Path(res_filename).is_dir():
            res_size = directory_size(res_filename)
        else:
            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] = []
        resources_found = self.find_resources(executable=executable, name=res_name, database=user_database)
        if not resources_found:
            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': '???',
                'type': resource_type,
                'size': res_size
            }
            user_database[executable].append(resdict)
        else:
            resdict = resources_found[0][1]
        self.save_user_list(user_database)
        self.load_resource_list(self.user_list)
        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(self.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(self.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
        elif usage == 'without-extension':
            return Path(name).stem
        raise ValueError("No such usage '%s'" % usage)

    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 to %s", src_filename, filename)
        if Path(src_filename).is_dir():
            log.info(f"Copying recursively from {src_filename} to {filename}")
            for child in Path(src_filename).rglob('*'):
                child_dst = Path(filename) / child.relative_to(src_filename)
                child_dst.parent.mkdir(parents=True, exist_ok=True)
                with open(child_dst, 'wb') as f_out, open(child, '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
        else:
            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 in ('file', 'directory'):
            if is_url:
                self._download_impl(url, fpath, progress_cb)
            else:
                self._copy_impl(url, fpath, progress_cb)
        elif resource_type == 'archive':
            with pushd_popd(tempdir=True) as tempdir:
                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 archive to %s/out" % tempdir)
                    with open_tarfile('../download.tar.xx', 'r:*') as tar:
                        tar.extractall()
                    log.info("Copying '%s' from archive to %s" % (path_in_archive, fpath))
                    copytree(path_in_archive, str(fpath))
        return fpath


1			from pathlib import Path
2			from os.path import join
3			from json import loads
4			from json.decoder import JSONDecodeError
5			from os import environ, listdir, getcwd, path
6			from fnmatch import filter as apply_glob
7			from shutil import copytree
8			from datetime import datetime
9			from tarfile import open as open_tarfile
10			from urllib.parse import urlparse, unquote
11			from subprocess import run, PIPE
12
13			import requests
14			from yaml import safe_load, safe_dump
15
16			from ocrd_validators import OcrdResourceListValidator
17			from ocrd_utils import getLogger, directory_size
18			from ocrd_utils.os import get_processor_resource_types, list_all_resources, pushd_popd, get_ocrd_tool_json
19			from .constants import RESOURCE_LIST_FILENAME, RESOURCE_USER_LIST_COMMENT
20
21			class OcrdResourceManager():
22
23			"""
24			Managing processor resources
25			"""
26			def __init__(self, userdir=None, xdg_config_home=None, xdg_data_home=None):
27			self.log = getLogger('ocrd.resource_manager')
28			self.database = {}
29
30			self._xdg_data_home = xdg_data_home
31			self._xdg_config_home = xdg_config_home
32			self._userdir = userdir
33			self.user_list = Path(self.xdg_config_home, 'ocrd', 'resources.yml')
34
35			self.load_resource_list(Path(RESOURCE_LIST_FILENAME))
36			if not self.user_list.exists():
37			if not self.user_list.parent.exists():
38			self.user_list.parent.mkdir(parents=True)
39			self.save_user_list()
40			self.load_resource_list(self.user_list)
41
42			@property
43			def userdir(self):
44			if not self._userdir:
45			self._userdir = path.expanduser('~')
46			if 'HOME' in environ and environ['HOME'] != path.expanduser('~'):
47			self._userdir = environ['HOME']
48			return self._userdir
49
50			@property
51			def xdg_data_home(self):
52			if not self._xdg_data_home:
53			if 'XDG_DATA_HOME' in environ:
54			self._xdg_data_home = environ['XDG_DATA_HOME']
55			else:
56			self._xdg_data_home = join(self.userdir, '.local', 'share')
57			return self._xdg_data_home
58
59			@property
60			def xdg_config_home(self):
61			if not self._xdg_config_home:
62			if 'XDG_CONFIG_HOME' in environ:
63			self._xdg_config_home = environ['XDG_CONFIG_HOME']
64			else:
65			self._xdg_config_home = join(self.userdir, '.config')
66			return self._xdg_config_home
67
68			def save_user_list(self, database=None):
69			if not database:
70			database = self.database
71			with open(self.user_list, 'w', encoding='utf-8') as f:
72			f.write(RESOURCE_USER_LIST_COMMENT)
73			f.write('\n')
74			f.write(safe_dump(database))
75
76			def load_resource_list(self, list_filename, database=None):
77			if not database:
78			database = self.database
79			if list_filename.is_file():
80			with open(list_filename, 'r', encoding='utf-8') as f:
81			list_loaded = safe_load(f) or {}
82			report = OcrdResourceListValidator.validate(list_loaded)
83			if not report.is_valid:
84			self.log.error('\n'.join(report.errors))
85			raise ValueError("Resource list %s is invalid!" % (list_filename))
86			for executable, resource_list in list_loaded.items():
87			if executable not in database:
88			database[executable] = []
89			# Prepend, so user provided is sorted before builtin
90			database[executable] = list_loaded[executable] + database[executable]
91			return database
92
93			def list_available(self, executable=None, dynamic=True):
94			"""
95			List models available for download by processor
96			"""
97			if not executable:
98			return self.database.items()
99			if dynamic:
100			for exec_dir in environ['PATH'].split(':'):
101			for exec_path in Path(exec_dir).glob(f'{executable}'):
102			self.log.info(f"Inspecting '{exec_path} --dump-json' for resources")
103			try:
104			ocrd_tool = get_ocrd_tool_json(exec_path)
105			except JSONDecodeError:
106			self.log.info(f"Failed to parse {exec_path} --dump-json output - not an OCR-D processor?")
107			for resdict in ocrd_tool.get('resources', ()):
108			for res_remove in (res for res in self.database.get(executable, []) if res['name'] == resdict['name']):
109			self.database.get(executable).remove(res_remove)
110			self.database[exec_path.name].append(resdict)
111			ret = []
112			for k in self.database:
113			if apply_glob([k], executable):
114			ret.append((k, self.database[k]))
115			return ret
116
117			def list_installed(self, executable=None):
118			"""
119			List installed resources, matching with registry by ``name``
120			"""
121			ret = []
122			if executable:
123			all_executables = [executable]
124			else:
125			# resources we know about
126			all_executables = list(self.database.keys())
127			# resources in the file system
128			parent_dirs = [join(x, 'ocrd-resources') for x in [self.xdg_data_home, '/usr/local/share']]
129			for parent_dir in parent_dirs:
130			if Path(parent_dir).exists():
131			all_executables += [x for x in listdir(parent_dir) if x.startswith('ocrd-')]
132			for this_executable in set(all_executables):
133			reslist = []
134			mimetypes = get_processor_resource_types(this_executable)
135			for res_filename in list_all_resources(this_executable, xdg_data_home=self.xdg_data_home):
136			res_filename = Path(res_filename)
137			if not '/' in mimetypes:
138			if res_filename.is_dir() and not 'text/directory' in mimetypes:
139			continue
140			if res_filename.is_file() and ['text/directory'] == mimetypes:
141			continue
142			res_name = res_filename.name
143			resdict = [x for x in self.database.get(this_executable, []) if x['name'] == res_name]
144			if not resdict:
145			self.log.info("%s resource '%s' (%s) not a known resource, creating stub in %s'", this_executable, res_name, str(res_filename), self.user_list)
146			resdict = [self.add_to_user_database(this_executable, res_filename)]
147			resdict[0]['path'] = str(res_filename)
148			reslist.append(resdict[0])
149			ret.append((this_executable, reslist))
150			return ret
151
152			def add_to_user_database(self, executable, res_filename, url=None, resource_type='file'):
153			"""
154			Add a stub entry to the user resource.yml
155			"""
156			res_name = Path(res_filename).name
157			if Path(res_filename).is_dir():
158			res_size = directory_size(res_filename)
159			else:
160			res_size = Path(res_filename).stat().st_size
161			with open(self.user_list, 'r', encoding='utf-8') as f:
162			user_database = safe_load(f) or {}
163			if executable not in user_database:
164			user_database[executable] = []
165			resources_found = self.find_resources(executable=executable, name=res_name, database=user_database)
166			if not resources_found:
167			resdict = {
168			'name': res_name,
169			'url': url if url else '???',
170			'description': 'Found at %s on %s' % (self.resource_dir_to_location(res_filename), datetime.now()),
171			'version_range': '???',
172			'type': resource_type,
173			'size': res_size
174			}
175			user_database[executable].append(resdict)
176			else:
177			resdict = resources_found[0][1]
178			self.save_user_list(user_database)
179			self.load_resource_list(self.user_list)
180			return resdict
181
182			def find_resources(self, executable=None, name=None, url=None, database=None):
183			"""
184			Find resources in the registry
185			"""
186			if not database:
187			database = self.database
188			ret = []
189			if executable and executable not in database.keys():
190			return ret
191			for executable in [executable] if executable else database.keys():
192			for resdict in database[executable]:
193			if not name and not url:
194			ret.append((executable, resdict))
195			elif url and url == resdict['url']:
196			ret.append((executable, resdict))
197			elif name and name == resdict['name']:
198			ret.append((executable, resdict))
199			return ret
200
201			@property
202			def default_resource_dir(self):
203			return self.location_to_resource_dir('data')
204
205			def location_to_resource_dir(self, location):
206			return '/usr/local/share/ocrd-resources' if location == 'system' else \
207			join(self.xdg_data_home, 'ocrd-resources') if location == 'data' else \
208			getcwd()
209
210			def resource_dir_to_location(self, resource_path):
211			resource_path = str(resource_path)
212			return 'system' if resource_path.startswith('/usr/local/share/ocrd-resources') else \
213			'data' if resource_path.startswith(join(self.xdg_data_home, 'ocrd-resources')) else \
214			'cwd' if resource_path.startswith(getcwd()) else \
215			resource_path
216
217			def parameter_usage(self, name, usage='as-is'):
218			if usage == 'as-is':
219			return name
220			elif usage == 'without-extension':
221			return Path(name).stem
222			raise ValueError("No such usage '%s'" % usage)
223
224			def _download_impl(self, url, filename, progress_cb=None, size=None):
225			log = getLogger('ocrd.resource_manager._download_impl')
226			log.info("Downloading %s to %s" % (url, filename))
227			with open(filename, 'wb') as f:
228			with requests.get(url, stream=True) as r:
229			total = size if size else int(r.headers.get('content-length'))
230			for data in r.iter_content(chunk_size=4096):
231			if progress_cb:
232			progress_cb(len(data))
233			f.write(data)
234
235			def _copy_impl(self, src_filename, filename, progress_cb=None):
236			log = getLogger('ocrd.resource_manager._copy_impl')
237			log.info("Copying %s to %s", src_filename, filename)
238			if Path(src_filename).is_dir():
239			log.info(f"Copying recursively from {src_filename} to {filename}")
240			for child in Path(src_filename).rglob('*'):
241			child_dst = Path(filename) / child.relative_to(src_filename)
242			child_dst.parent.mkdir(parents=True, exist_ok=True)
243			with open(child_dst, 'wb') as f_out, open(child, 'rb') as f_in:
244			while True:
245			chunk = f_in.read(4096)
246			if chunk:
247			f_out.write(chunk)
248			if progress_cb:
249			progress_cb(len(chunk))
250			else:
251			break
252			else:
253			with open(filename, 'wb') as f_out, open(src_filename, 'rb') as f_in:
254			while True:
255			chunk = f_in.read(4096)
256			if chunk:
257			f_out.write(chunk)
258			if progress_cb:
259			progress_cb(len(chunk))
260			else:
261			break
262
263			# TODO Proper caching (make head request for size, If-Modified etc)
264			def download(
265			self,
266			executable,
267			url,
268			basedir,
269			overwrite=False,
270			no_subdir=False,
271			name=None,
272			resource_type='file',
273			path_in_archive='.',
274			progress_cb=None,
275			size=None,
276			):
277			"""
278			Download a resource by URL
279			"""
280			log = getLogger('ocrd.resource_manager.download')
281			destdir = Path(basedir) if no_subdir else Path(basedir, executable)
282			if not name:
283			url_parsed = urlparse(url)
284			name = Path(unquote(url_parsed.path)).name
285			fpath = Path(destdir, name)
286			is_url = url.startswith('https://') or url.startswith('http://')
287			if fpath.exists() and not overwrite:
288			log.info("%s to be %s to %s which already exists and overwrite is False" % (url, 'downloaded' if is_url else 'copied', fpath))
289			return fpath
290			destdir.mkdir(parents=True, exist_ok=True)
291			if resource_type in ('file', 'directory'):
292			if is_url:
293			self._download_impl(url, fpath, progress_cb)
294			else:
295			self._copy_impl(url, fpath, progress_cb)
296			elif resource_type == 'archive':
297			with pushd_popd(tempdir=True) as tempdir:
298			if is_url:
299			self._download_impl(url, 'download.tar.xx', progress_cb, size)
300			else:
301			self._copy_impl(url, 'download.tar.xx', progress_cb)
302			Path('out').mkdir()
303			with pushd_popd('out'):
304			log.info("Extracting archive to %s/out" % tempdir)
305			with open_tarfile('../download.tar.xx', 'r:*') as tar:
306			tar.extractall()
307			log.info("Copying '%s' from archive to %s" % (path_in_archive, fpath))
308			copytree(path_in_archive, str(fpath))
309			return fpath
310

OCR-D / core

Pull Request — master (#800)

OcrdResourceManager.download() D

Complexity

Size

Duplication

Importance

How to fix Complexity Many Parameters

Complexity

Many Parameters

Duplication Side-by-Side

Filter issues like