ocrd.resource_manager.OcrdResourceManager.download() - Code Metrics - Inspection of "resmgr: support zip archives and non-folder archiv..." - OCR-D/core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#967)

by Konstantin

created 2022-12-09 17:38 UTC

OcrdResourceManager.download() F

↳ Parent: ocrd.resource_manager

Complexity

Conditions

Size

Total Lines	54
Code Lines	46

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	46
dl	0
loc	54
rs	2.4
c	0
b	0
f	0
cc	16
nop	10

How to fix Long Method Complexity Many Parameters

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

If many parameters/temporary variables are present:
- Replace temporary variables with Query
- Introduce parameter object; often combined with preserve whole object
- If the above two are insufficient: Replace method with method object
If you have long conditionals: Decompose Conditional
Otherwise: Extract method

Complexity

Complex classes like ocrd.resource_manager.OcrdResourceManager.download() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

Many Parameters

Methods with many parameters are not only hard to understand, but their parameters also often become inconsistent when you need more, or different data.

There are several approaches to avoid long parameter lists:

If you can get the data from another object that the method knows about already: Replace the parameter with method call
If several parameters are part of another object: Preserve Whole Object
If parameters are not yet connected: Introduce Parameter Object

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

import requests
from yaml import safe_load, safe_dump
import magic

# https://github.com/OCR-D/core/issues/867
# https://stackoverflow.com/questions/50900727/skip-converting-entities-while-loading-a-yaml-string-using-pyyaml
import yaml.constructor
yaml.constructor.SafeConstructor.yaml_constructors[u'tag:yaml.org,2002:timestamp'] = \
    yaml.constructor.SafeConstructor.yaml_constructors[u'tag:yaml.org,2002:str']

from ocrd_validators import OcrdResourceListValidator
from ocrd_utils import getLogger, directory_size, get_moduledir
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, skip_init=False):
        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')

        if not skip_init:
            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, name=None, database=None, url=None):
        """
        List models available for download by processor
        """
        if not database:
            database = self.database
        if not executable:
            return database.items()
        if dynamic:
            for exec_dir in environ['PATH'].split(':'):
                for exec_path in Path(exec_dir).glob(f'{executable}'):
                    self.log.debug(f"Inspecting '{exec_path} --dump-json' for resources")
                    ocrd_tool = get_ocrd_tool_json(exec_path)
                    for resdict in ocrd_tool.get('resources', ()):
                        if exec_path.name not in database:
                            database[exec_path.name] = []
                        database[exec_path.name].append(resdict)
            database = self._dedup_database(database)
        found = False
        ret = []
        for k in database:
            if apply_glob([k], executable):
                found = True
                restuple = (k, [])
                ret.append(restuple)
                for resdict in database[k]:
                    if name and resdict['name'] != name:
                        continue
                    if url and resdict['url'] != url:
                        continue
                    restuple[1].append(resdict)
        if not found:
            ret = [(executable, [])]
        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)
            moduledir = get_moduledir(this_executable)
            for res_filename in list_all_resources(this_executable, moduled=moduledir, 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
                res_type = 'file' if res_filename.is_file() else 'directory'
                res_size = res_filename.stat().st_size if res_filename.is_file() else directory_size(res_filename)
                resdict_list = [x for x in self.database.get(this_executable, []) if x['name'] == res_name]
                if resdict_list:
                    resdict = resdict_list[0]
                elif str(res_filename.parent) == moduledir:
                    resdict = {
                        'name': res_name, 
                        'url': str(res_filename), 
                        'description': 'Found at module', 
                        'type': res_type,
                        'size': res_size
                    }
                else:
                    resdict = self.add_to_user_database(this_executable, res_filename, resource_type=res_type)
                resdict['path'] = str(res_filename)
                reslist.append(resdict)
            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
        self.log.info("%s resource '%s' (%s) not a known resource, creating stub in %s'", executable, res_name, str(res_filename), self.user_list)
        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.list_available(executable=executable, name=res_name, database=user_database)[0][1]
        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]
        self.save_user_list(user_database)
        self.load_resource_list(self.user_list)
        return resdict

    @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):
        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:
                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,
    ):
        """
        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':
            archive_fname = 'download.tar.xx'
            with pushd_popd(tempdir=True) as tempdir:
                if is_url:
                    self._download_impl(url, archive_fname, progress_cb)
                else:
                    self._copy_impl(url, archive_fname, progress_cb)
                Path('out').mkdir()
                with pushd_popd('out'):
                    mimetype = magic.from_file(f'../{archive_fname}', mime=True)
                    log.info("Extracting %s archive to %s/out" % (mimetype, tempdir))
                    if mimetype == 'application/zip':
                        with ZipFile(f'../{archive_fname}', 'r') as zipf:
                            zipf.extractall()
                    else:
                        with open_tarfile(f'../{archive_fname}', 'r:*') as tar:
                            tar.extractall()
                    log.info("Copying '%s' from archive to %s" % (path_in_archive, fpath))
                    if Path(path_in_archive).is_dir():
                        copytree(path_in_archive, str(fpath))
                    else:
                        copy(path_in_archive, str(fpath))
        return fpath

    def _dedup_database(self, database=None, dedup_key='name'):
        """
        Deduplicate resources by name
        """
        if not database:
            database = self.database
        for executable, reslist in database.items():
            reslist_dedup = []
            for resdict in reslist:
                if not any(r[dedup_key] == resdict[dedup_key] for r in reslist_dedup):
                    reslist_dedup.append(resdict)
            database[executable] = reslist_dedup
        return database


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

OCR-D / core

Pull Request — master (#967)

OcrdResourceManager.download() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity Many Parameters

Long Method

Complexity

Many Parameters

Duplication Side-by-Side

Filter issues like