file_sync_tool.transfer.utility.get_authorization() - Code Metrics - jackd248/file-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

get_authorization() B
last analyzed 2025-04-10 06:33 UTC

↳ Parent: file_sync_tool.transfer.utility

Complexity

Conditions

Size

Total Lines	25
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	12
nop	1
dl	0
loc	25
rs	8
c	0
b	0
f	0

#!/usr/bin/env python3

# -*- coding: future_fstrings -*-

import re
import datetime
import os
import shutil
from db_sync_tool.utility import system, mode, output
from file_sync_tool.utility import helper


# Default options for rsync command
# https://wiki.ubuntuusers.de/rsync/
default_options = [
    '--delete',
    '-a',
    '-z',
    '--stats',
    '--human-readable',
    '--iconv=UTF-8',
    '--chmod=D2770,F660'
]

# Temporary data directory (for PROXY mode)
temp_data_dir = None



def get_password_environment(client):
    """
    Optionally create a password environment variable for sshpass password authentication
    https://www.redhat.com/sysadmin/ssh-automation-sshpass
    :param client: String
    :return:
    """
    if not client:
        return ''

    if system.config['use_sshpass'] and not 'ssh_key' in system.config[client] and 'password' in system.config[client]:

        return f'SSHPASS=\'{system.config[client]["password"]}\' '
    return ''


def get_authorization(client):
    """
    Define authorization arguments for rsync command
    :param client: String
    :return: String
    """
    _ssh_key = None
    if not client:
        return ''

    if 'ssh_key' in system.config[client]:
        _ssh_key = system.config[mode.Client.ORIGIN]['ssh_key']

    _ssh_port = system.config[client]['port'] if 'port' in system.config[client] else 22

    if _ssh_key is None:
        if system.config['use_sshpass'] and get_password_environment(client):

            # In combination with SSHPASS environment variable
            # https://www.redhat.com/sysadmin/ssh-automation-sshpass
            return f'--rsh="sshpass -e ssh -p{_ssh_port} -o StrictHostKeyChecking=no -l {system.config[client]["user"]}"'

        else:
            return f'-e "ssh -p{_ssh_port} -o StrictHostKeyChecking=no"'
    else:
        # Provide ssh key file path for ssh authentication
        return f'-e "ssh -i {_ssh_key} -p{_ssh_port}"'


def get_host(client):
    """
    Return user@host if client is not local
    :param client: String
    :return: String
    """
    if mode.is_remote(client):
        return f'{system.config[client]["user"]}@{system.config[client]["host"]}:'
    return ''


def get_options():
    """
    Prepare rsync options with stored default options and provided addtional options
    :return: String
    """
    _options = f'{" ".join(default_options)}'
    if 'option' in system.config['files']:
        _options += f'{" ".join(system.config["files"]["option"])}'
    return _options


def get_excludes(excludes):
    """
    Prepare rsync excludes as arguments
    :param excludes:
    :return:
    """
    _excludes = ''
    for exclude in excludes:
        _excludes += f'--exclude {exclude} '
    return _excludes


def read_stats(stats):
    """
    Read rsync stats and print a summary
    :param stats: String
    :return:
    """
    if system.config['verbose']:
        print(f'{output.Subject.DEBUG}{output.CliFormat.BLACK}{stats}{output.CliFormat.ENDC}')

    _file_number = parse_string(stats, r'Number of regular files transferred:\s*([\d.]+)')
    _file_size = parse_string(stats, r'Total transferred file size:\s*([\d.]+[MKG]?)')

    if _file_number and _file_size:
        output.message(
            output.Subject.INFO,
            f'Status: {_file_number[0]} file(s) transferred {output.CliFormat.BLACK}({_file_size[0]}Bytes){output.CliFormat.ENDC}'

        )


def parse_string(string, regex):
    """
    Parse string by given regex
    :param string: String
    :param regex: String
    :return:
    """
    _file_size_pattern = regex
    _regex_matcher = re.compile(_file_size_pattern)
    return _regex_matcher.findall(string)


def generate_temp_dir_name():
    """
    Generate a temporary directory name, e.g. _tmp_28-12-2020_15-59
    :return:
    """
    global temp_data_dir

    _now = datetime.datetime.now()
    temp_data_dir = f'/tmp/_tmp_{_now.strftime("%d-%m-%Y_%H-%M")}'


def remove_temporary_dir():
    """
    Remove temporary directory
    :return:
    """
    global temp_data_dir

    if os.path.exists(temp_data_dir):
        shutil.rmtree(temp_data_dir)
        output.message(
            output.Subject.LOCAL,
            'Cleaning up',
            True
        )


1			#!/usr/bin/env python3
			0 ignored issues – show introduced 2020-12-28 17:01 UTC by Report Bug Copy Issue Report Missing module docstring Loading history...
2			# -- coding: future_fstrings --
3
4			import re
5			import datetime
6			import os
7			import shutil
8			from db_sync_tool.utility import system, mode, output
9			from file_sync_tool.utility import helper
			0 ignored issues – show Unused Code introduced 2020-12-28 17:01 UTC by Report Bug Copy Issue Report Unused helper imported from file_sync_tool.utility Loading history...
10
11			# Default options for rsync command
12			# https://wiki.ubuntuusers.de/rsync/
13			default_options = [
14			'--delete',
15			'-a',
16			'-z',
17			'--stats',
18			'--human-readable',
19			'--iconv=UTF-8',
20			'--chmod=D2770,F660'
21			]
22
23			# Temporary data directory (for PROXY mode)
24			temp_data_dir = None
			0 ignored issues – show Coding Style Naming introduced 2020-12-28 17:01 UTC by Report Bug Copy Issue Report Constant name "temp_data_dir" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
25
26
27			def get_password_environment(client):
28			"""
29			Optionally create a password environment variable for sshpass password authentication
30			https://www.redhat.com/sysadmin/ssh-automation-sshpass
31			:param client: String
32			:return:
33			"""
34			if not client:
35			return ''
36
37			if system.config['use_sshpass'] and not 'ssh_key' in system.config[client] and 'password' in system.config[client]:
			0 ignored issues – show Coding Style introduced 2020-12-28 17:01 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (119/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
38			return f'SSHPASS=\'{system.config[client]["password"]}\' '
39			return ''
40
41
42			def get_authorization(client):
43			"""
44			Define authorization arguments for rsync command
45			:param client: String
46			:return: String
47			"""
48			_ssh_key = None
49			if not client:
50			return ''
51
52			if 'ssh_key' in system.config[client]:
53			_ssh_key = system.config[mode.Client.ORIGIN]['ssh_key']
54
55			_ssh_port = system.config[client]['port'] if 'port' in system.config[client] else 22
56
57			if _ssh_key is None:
58			if system.config['use_sshpass'] and get_password_environment(client):
			0 ignored issues – show unused-code introduced 2020-12-28 17:01 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
59			# In combination with SSHPASS environment variable
60			# https://www.redhat.com/sysadmin/ssh-automation-sshpass
61			return f'--rsh="sshpass -e ssh -p{_ssh_port} -o StrictHostKeyChecking=no -l {system.config[client]["user"]}"'
			0 ignored issues – show Coding Style introduced 2020-12-28 17:01 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (121/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
62			else:
63			return f'-e "ssh -p{_ssh_port} -o StrictHostKeyChecking=no"'
64			else:
65			# Provide ssh key file path for ssh authentication
66			return f'-e "ssh -i {_ssh_key} -p{_ssh_port}"'
67
68
69			def get_host(client):
70			"""
71			Return user@host if client is not local
72			:param client: String
73			:return: String
74			"""
75			if mode.is_remote(client):
76			return f'{system.config[client]["user"]}@{system.config[client]["host"]}:'
77			return ''
78
79
80			def get_options():
81			"""
82			Prepare rsync options with stored default options and provided addtional options
83			:return: String
84			"""
85			_options = f'{" ".join(default_options)}'
86			if 'option' in system.config['files']:
87			_options += f'{" ".join(system.config["files"]["option"])}'
88			return _options
89
90
91			def get_excludes(excludes):
92			"""
93			Prepare rsync excludes as arguments
94			:param excludes:
95			:return:
96			"""
97			_excludes = ''
98			for exclude in excludes:
99			_excludes += f'--exclude {exclude} '
100			return _excludes
101
102
103			def read_stats(stats):
104			"""
105			Read rsync stats and print a summary
106			:param stats: String
107			:return:
108			"""
109			if system.config['verbose']:
110			print(f'{output.Subject.DEBUG}{output.CliFormat.BLACK}{stats}{output.CliFormat.ENDC}')
111
112			_file_number = parse_string(stats, r'Number of regular files transferred:\s*([\d.]+)')
113			_file_size = parse_string(stats, r'Total transferred file size:\s*([\d.]+[MKG]?)')
114
115			if _file_number and _file_size:
116			output.message(
117			output.Subject.INFO,
118			f'Status: {_file_number[0]} file(s) transferred {output.CliFormat.BLACK}({_file_size[0]}Bytes){output.CliFormat.ENDC}'
			0 ignored issues – show Coding Style introduced 2020-12-28 17:01 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (130/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
119			)
120
121
122			def parse_string(string, regex):
123			"""
124			Parse string by given regex
125			:param string: String
126			:param regex: String
127			:return:
128			"""
129			_file_size_pattern = regex
130			_regex_matcher = re.compile(_file_size_pattern)
131			return _regex_matcher.findall(string)
132
133
134			def generate_temp_dir_name():
135			"""
136			Generate a temporary directory name, e.g. _tmp_28-12-2020_15-59
137			:return:
138			"""
139			global temp_data_dir
			0 ignored issues – show Coding Style Naming introduced 2020-12-28 17:01 UTC by Report Bug Copy Issue Report Constant name "temp_data_dir" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history... Coding Style introduced 2020-12-28 17:01 UTC by Report Bug Copy Issue Report Usage of the `global` statement should be avoided. Usage of `global` can make code hard to read and test, its usage is generally not recommended unless you are dealing with legacy code. Loading history...
140			_now = datetime.datetime.now()
141			temp_data_dir = f'/tmp/_tmp_{_now.strftime("%d-%m-%Y_%H-%M")}'
142
143
144			def remove_temporary_dir():
145			"""
146			Remove temporary directory
147			:return:
148			"""
149			global temp_data_dir
			0 ignored issues – show Coding Style Naming introduced 2020-12-28 17:01 UTC by Report Bug Copy Issue Report Constant name "temp_data_dir" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history... Coding Style introduced 2020-12-28 17:01 UTC by Report Bug Copy Issue Report Usage of the `global` statement should be avoided. Usage of `global` can make code hard to read and test, its usage is generally not recommended unless you are dealing with legacy code. Loading history...
150			if os.path.exists(temp_data_dir):
151			shutil.rmtree(temp_data_dir)
152			output.message(
153			output.Subject.LOCAL,
154			'Cleaning up',
155			True
156			)
157

jackd248 / file-sync-tool

get_authorization() B last analyzed 2025-04-10 06:33 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

get_authorization() B
last analyzed 2025-04-10 06:33 UTC