ocrd.resource_manager.OcrdResourceManager._copy_impl() - 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-02-22 18:27 UTC

OcrdResourceManager._copy_impl() C

↳ Parent: ocrd.resource_manager

Complexity

Conditions

Size

Total Lines	27
Code Lines	24

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	24
dl	0
loc	27
rs	5.4
c	0
b	0
f	0
cc	11
nop	4

How to fix Complexity

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

OCR-D / core

Pull Request — master (#800)

OcrdResourceManager._copy_impl() C

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like