kerapu.command.TestsetShredderCommand.TestShredderCommand.__shred_xml_bestand() - Code Metrics - Inspection of "Enable code rating." - SetBased/py-kerapu - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( be813a...3d8811 )

by P.R.

created 2019-07-01 13:49 UTC

TestShredderCommand.__shred_xml_bestand() C

↳ Parent: kerapu.command.TestsetShredderCommand

Complexity

Conditions

Size

Total Lines	93
Code Lines	62

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	110

Importance

Changes

Metric	Value
cc	10
eloc	62
nop	2
dl	0
loc	93
ccs	0
cts	44
cp	0
crap	110
rs	5.4436
c	0
b	0
f	0

How to fix Long Method Complexity

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 kerapu.command.TestsetShredderCommand.TestShredderCommand.__shred_xml_bestand() 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.

"""
Kerapu
"""
import csv
import datetime
import os
import random
import shutil
import string
import zipfile
from typing import Iterable

from cleo import Command
from lxml import etree

from kerapu.style.KerapuStyle import KerapuStyle


class TestShredderCommand(Command):
    """
    Converteert XML-bestand met de testset naar een CSV-bestand

    kerapu:test-shredder
        {testset-zip : ZIP-bestand met de testset}
        {testset-csv : Path waar het CSV-bestand met de tests moeten worden opgeslagen}
    """

    # ------------------------------------------------------------------------------------------------------------------
    def __extract_zip_file(self, zip_filename: str, tmp_dir: str):
        """
        Extracts het ZIP-bestand met de testset in een folder.

        :param str zip_filename: Het path naar het ZIP-bestand met de testset.
        :param str tmp_dir: Path naar de folder.
        """
        self.output.writeln('Uitpakken van <fso>{}</fso> in <fso>{}</fso>'.format(zip_filename, tmp_dir))

        with zipfile.ZipFile(zip_filename, 'r') as zip_ref:
            zip_ref.extractall(tmp_dir)

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def ordinal(path: str) -> int:
        """
        Geeft het volgnummer van een test.

        :param str path: Het path naar het XML-bestand met de test case.
        """
        parts = os.path.basename(path).split('_')

        return int(parts[6])

    # ------------------------------------------------------------------------------------------------------------------
    def __lees_test_cases_lijst(self, folder: str) -> list:
        """
        Geeft een lijst met alle bestanden in een folder.

        :param str folder: Het path naar de folder.
        """
        entries = os.listdir(folder)
        filenames = list()
        for entry in entries:
            path = os.path.join(folder, entry)
            if os.path.isfile(path):
                filenames.append(path)

        self.output.writeln('Aantal gevonden test cases: {}'.format(len(filenames)))

        return sorted(filenames, key=TestShredderCommand.ordinal)

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __maak_xpath(parts: Iterable) -> str:
        """
        Maakt een string met een xpath.

        :param tuple parts: The onderdelen van het xpath.

        :rtype: str
        """
        xpath = ''
        for part in parts:
            if xpath:
                xpath += '/'
            xpath += 'xmlns:' + part

        return xpath

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __convert_date(date: str) -> str:
        """
        Converteert een datum in YYYYMMDD formaat naar YYYY-MM-DD format.

        :param str date: De datum in YYYYMMDD format.

        :rtype: str
        """
        return date[:4] + '-' + date[4:6] + '-' + date[6:8]

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __leeftijd_geboorte_datum(date: str, leeftijd: int) -> str:
        """
        Geeft de geboortedatum gegeven een datum en een leeftijd (en de persoon is niet jarig).

        :param str date: De gegeven datum in YYYY-MM-DD format.
        :param int leeftijd: De leeftijd in jaren.

        :rtype: int
        """
        date = datetime.date(int(date[:4]) - leeftijd, int(date[5:7]), int(date[8:10]))
        date -= datetime.timedelta(days=1)

        return date.isoformat()

    # ------------------------------------------------------------------------------------------------------------------
    def __shred_xml_bestand(self, filename: str) -> dict:
        """
        Leest de relevante data in een XML-bestand met een test case.

        :param str filename: De filenaam van het XML bestand.

        :rtype: dict
        """
        doc = etree.parse(filename)

        xpath = '/soapenv:Envelope/soapenv:Body/xmlns:FICR_IN900101NL04'
        namespaces = {'soapenv': 'http://schemas.xmlsoap.org/soap/envelope/',
                      'xmlns':   'urn:hl7-org:v3'}

        # Lees declaratiecode.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'id')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        declaratie_code = elements[0].get('extension')

        # Lees specialismecode.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'derivedFrom',
                 'zorgtraject', 'responsibleParty', 'assignedPerson', 'code')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        specialisme_code = elements[0].get('code')

        # Lees diagnosecode.
        parts = (
            'ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'pertinentInformation1',
            'typerendeDiagnose', 'value')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        diagnose_code = elements[0].get('code')

        # Lees zorgtypecode.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'code')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        zorg_type_code = elements[0].get('code') if elements else None

        # Lees zorgvraagcode.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'derivedFrom',
                 'zorgtraject', 'reason', 'zorgvraag', 'value')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        zorg_vraag_code = elements[0].get('code') if elements else None

        # Lees begindatum.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'effectiveTime', 'low')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        begin_datum = self.__convert_date(elements[0].get('value')) if elements else None

        # Lees de geboortedatum van de patient.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'subject', 'patient', 'subjectOf', 'leeftijd',
                 'value')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        leeftijd = int(elements[0].get('value')) if elements else None
        geboorte_datum = self.__leeftijd_geboorte_datum(begin_datum, leeftijd)

        # Lees het geslacht van de patient.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'subject', 'patient', 'patientPerson',
                 'administrativeGenderCode')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        geslacht_code = elements[0].get('code') if elements else None

        # Lees de AGB-code van de zorginstelling.
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'author', 'assignedOrganization', 'id')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        zorg_instelling_code = elements[0].get('extension') if elements else None

        # Lees alle zorgactiviteiten.
        zorg_activiteiten = list()
        parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'debit',
                 'zorgactiviteit')
        elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
        for element in elements:
            path = 'xmlns:code'
            sub_elements = element.xpath(path, namespaces=namespaces)
            zorg_activiteit_code = sub_elements[0].get('code') if sub_elements else None

            path = 'xmlns:repeatNumber'
            sub_elements = element.xpath(path, namespaces=namespaces)
            aantal = int(sub_elements[0].get('value')) if sub_elements else None

            zorg_activiteiten.append((zorg_activiteit_code, aantal))

        return {'subtraject_nummer':    os.path.basename(filename),
                'declaratie_code':      declaratie_code,
                'specialisme_code':     specialisme_code,
                'diagnose_code':        diagnose_code,
                'zorg_type_code':       zorg_type_code,
                'zorg_vraag_code':      zorg_vraag_code,
                'begin_datum':          begin_datum,
                'geboorte_datum':       geboorte_datum,
                'geslacht_code':        geslacht_code,
                'zorg_instelling_code': zorg_instelling_code,
                'zorg_activiteiten':    zorg_activiteiten}

    # ----------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def __write_subtraject(writer, subtraject: dict):
        """
        Schrijft het subtraject met alle zorgactiviteiten naar een CSV-bestand.

        :param writer: De handle naar de CSV writer.
        :param dict subtraject: De details van het subtract.
        """
        writer.writerow((subtraject['subtraject_nummer'],
                         subtraject['specialisme_code'],
                         subtraject['diagnose_code'],
                         subtraject['zorg_type_code'],
                         subtraject['zorg_vraag_code'],
                         subtraject['begin_datum'],
                         subtraject['geboorte_datum'],
                         subtraject['geslacht_code'],
                         subtraject['zorg_instelling_code'],
                         subtraject['declaratie_code']))

        for zorgactiviteit in subtraject['zorg_activiteiten']:
            writer.writerow((zorgactiviteit[0], zorgactiviteit[1]))

    # ----------------------------------------------------------------------------------------------------------------------
    def __extract_files(self, writer, filenames: list):
        """
        Extract de data van een lijst met XML-bestanden met test cases en schrijft deze data naar een CSV-bestand.

        :param writer:  De handle naar de CSV writer.
        :param list filenames: De lijst met bestandsnamen van XML-bestanden met test cases.
        """
        for filename in filenames:
            subtraject = self.__shred_xml_bestand(filename)
            self.__write_subtraject(writer, subtraject)

    # ------------------------------------------------------------------------------------------------------------------
    def handle(self):
        """
        Executes the command.
        """
        self.output = KerapuStyle(self.input, self.output)

        zip_filename = self.argument('testset-zip')
        csv_filename = self.argument('testset-csv')
        tmp_dir = '.kerapu-' + ''.join(random.choices(string.ascii_lowercase, k=12))

        os.mkdir(tmp_dir)

        self.__extract_zip_file(zip_filename, tmp_dir)
        files = self.__lees_test_cases_lijst(tmp_dir)

        with open(csv_filename, 'wt', encoding='utf-8') as handle:
            csv_writer = csv.writer(handle, lineterminator='\n', delimiter=',')
            self.__extract_files(csv_writer, files)

        shutil.rmtree(tmp_dir)

        return 0

# ----------------------------------------------------------------------------------------------------------------------


1			"""
2			Kerapu
3			"""
4			import csv
5			import datetime
6			import os
7			import random
8			import shutil
9			import string
10			import zipfile
11			from typing import Iterable
12
13			from cleo import Command
14			from lxml import etree
15
16			from kerapu.style.KerapuStyle import KerapuStyle
17
18
19			class TestShredderCommand(Command):
20			"""
21			Converteert XML-bestand met de testset naar een CSV-bestand
22
23			kerapu:test-shredder
24			{testset-zip : ZIP-bestand met de testset}
25			{testset-csv : Path waar het CSV-bestand met de tests moeten worden opgeslagen}
26			"""
27
28			# ------------------------------------------------------------------------------------------------------------------
29			def __extract_zip_file(self, zip_filename: str, tmp_dir: str):
30			"""
31			Extracts het ZIP-bestand met de testset in een folder.
32
33			:param str zip_filename: Het path naar het ZIP-bestand met de testset.
34			:param str tmp_dir: Path naar de folder.
35			"""
36			self.output.writeln('Uitpakken van <fso>{}</fso> in <fso>{}</fso>'.format(zip_filename, tmp_dir))
37
38			with zipfile.ZipFile(zip_filename, 'r') as zip_ref:
39			zip_ref.extractall(tmp_dir)
40
41			# ------------------------------------------------------------------------------------------------------------------
42			@staticmethod
43			def ordinal(path: str) -> int:
44			"""
45			Geeft het volgnummer van een test.
46
47			:param str path: Het path naar het XML-bestand met de test case.
48			"""
49			parts = os.path.basename(path).split('_')
50
51			return int(parts[6])
52
53			# ------------------------------------------------------------------------------------------------------------------
54			def __lees_test_cases_lijst(self, folder: str) -> list:
55			"""
56			Geeft een lijst met alle bestanden in een folder.
57
58			:param str folder: Het path naar de folder.
59			"""
60			entries = os.listdir(folder)
61			filenames = list()
62			for entry in entries:
63			path = os.path.join(folder, entry)
64			if os.path.isfile(path):
65			filenames.append(path)
66
67			self.output.writeln('Aantal gevonden test cases: {}'.format(len(filenames)))
68
69			return sorted(filenames, key=TestShredderCommand.ordinal)
70
71			# ------------------------------------------------------------------------------------------------------------------
72			@staticmethod
73			def __maak_xpath(parts: Iterable) -> str:
74			"""
75			Maakt een string met een xpath.
76
77			:param tuple parts: The onderdelen van het xpath.
78
79			:rtype: str
80			"""
81			xpath = ''
82			for part in parts:
83			if xpath:
84			xpath += '/'
85			xpath += 'xmlns:' + part
86
87			return xpath
88
89			# ------------------------------------------------------------------------------------------------------------------
90			@staticmethod
91			def __convert_date(date: str) -> str:
92			"""
93			Converteert een datum in YYYYMMDD formaat naar YYYY-MM-DD format.
94
95			:param str date: De datum in YYYYMMDD format.
96
97			:rtype: str
98			"""
99			return date[:4] + '-' + date[4:6] + '-' + date[6:8]
100
101			# ------------------------------------------------------------------------------------------------------------------
102			@staticmethod
103			def __leeftijd_geboorte_datum(date: str, leeftijd: int) -> str:
104			"""
105			Geeft de geboortedatum gegeven een datum en een leeftijd (en de persoon is niet jarig).
106
107			:param str date: De gegeven datum in YYYY-MM-DD format.
108			:param int leeftijd: De leeftijd in jaren.
109
110			:rtype: int
111			"""
112			date = datetime.date(int(date[:4]) - leeftijd, int(date[5:7]), int(date[8:10]))
113			date -= datetime.timedelta(days=1)
114
115			return date.isoformat()
116
117			# ------------------------------------------------------------------------------------------------------------------
118			def __shred_xml_bestand(self, filename: str) -> dict:
119			"""
120			Leest de relevante data in een XML-bestand met een test case.
121
122			:param str filename: De filenaam van het XML bestand.
123
124			:rtype: dict
125			"""
126			doc = etree.parse(filename)
127
128			xpath = '/soapenv:Envelope/soapenv:Body/xmlns:FICR_IN900101NL04'
129			namespaces = {'soapenv': 'http://schemas.xmlsoap.org/soap/envelope/',
130			'xmlns': 'urn:hl7-org:v3'}
131
132			# Lees declaratiecode.
133			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'id')
134			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
135			declaratie_code = elements[0].get('extension')
136
137			# Lees specialismecode.
138			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'derivedFrom',
139			'zorgtraject', 'responsibleParty', 'assignedPerson', 'code')
140			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
141			specialisme_code = elements[0].get('code')
142
143			# Lees diagnosecode.
144			parts = (
145			'ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'pertinentInformation1',
146			'typerendeDiagnose', 'value')
147			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
148			diagnose_code = elements[0].get('code')
149
150			# Lees zorgtypecode.
151			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'code')
152			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
153			zorg_type_code = elements[0].get('code') if elements else None
154
155			# Lees zorgvraagcode.
156			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'derivedFrom',
157			'zorgtraject', 'reason', 'zorgvraag', 'value')
158			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
159			zorg_vraag_code = elements[0].get('code') if elements else None
160
161			# Lees begindatum.
162			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'effectiveTime', 'low')
163			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
164			begin_datum = self.__convert_date(elements[0].get('value')) if elements else None
165
166			# Lees de geboortedatum van de patient.
167			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'subject', 'patient', 'subjectOf', 'leeftijd',
168			'value')
169			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
170			leeftijd = int(elements[0].get('value')) if elements else None
171			geboorte_datum = self.__leeftijd_geboorte_datum(begin_datum, leeftijd)
172
173			# Lees het geslacht van de patient.
174			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'subject', 'patient', 'patientPerson',
175			'administrativeGenderCode')
176			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
177			geslacht_code = elements[0].get('code') if elements else None
178
179			# Lees de AGB-code van de zorginstelling.
180			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'author', 'assignedOrganization', 'id')
181			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
182			zorg_instelling_code = elements[0].get('extension') if elements else None
183
184			# Lees alle zorgactiviteiten.
185			zorg_activiteiten = list()
186			parts = ('ControlActProcess', 'subject', 'Declaratiedataset', 'component', 'subtraject', 'debit',
187			'zorgactiviteit')
188			elements = doc.xpath(xpath + '/' + self.__maak_xpath(parts), namespaces=namespaces)
189			for element in elements:
190			path = 'xmlns:code'
191			sub_elements = element.xpath(path, namespaces=namespaces)
192			zorg_activiteit_code = sub_elements[0].get('code') if sub_elements else None
193
194			path = 'xmlns:repeatNumber'
195			sub_elements = element.xpath(path, namespaces=namespaces)
196			aantal = int(sub_elements[0].get('value')) if sub_elements else None
197
198			zorg_activiteiten.append((zorg_activiteit_code, aantal))
199
200			return {'subtraject_nummer': os.path.basename(filename),
201			'declaratie_code': declaratie_code,
202			'specialisme_code': specialisme_code,
203			'diagnose_code': diagnose_code,
204			'zorg_type_code': zorg_type_code,
205			'zorg_vraag_code': zorg_vraag_code,
206			'begin_datum': begin_datum,
207			'geboorte_datum': geboorte_datum,
208			'geslacht_code': geslacht_code,
209			'zorg_instelling_code': zorg_instelling_code,
210			'zorg_activiteiten': zorg_activiteiten}
211
212			# ----------------------------------------------------------------------------------------------------------------------
213			@staticmethod
214			def __write_subtraject(writer, subtraject: dict):
215			"""
216			Schrijft het subtraject met alle zorgactiviteiten naar een CSV-bestand.
217
218			:param writer: De handle naar de CSV writer.
219			:param dict subtraject: De details van het subtract.
220			"""
221			writer.writerow((subtraject['subtraject_nummer'],
222			subtraject['specialisme_code'],
223			subtraject['diagnose_code'],
224			subtraject['zorg_type_code'],
225			subtraject['zorg_vraag_code'],
226			subtraject['begin_datum'],
227			subtraject['geboorte_datum'],
228			subtraject['geslacht_code'],
229			subtraject['zorg_instelling_code'],
230			subtraject['declaratie_code']))
231
232			for zorgactiviteit in subtraject['zorg_activiteiten']:
233			writer.writerow((zorgactiviteit[0], zorgactiviteit[1]))
234
235			# ----------------------------------------------------------------------------------------------------------------------
236			def __extract_files(self, writer, filenames: list):
237			"""
238			Extract de data van een lijst met XML-bestanden met test cases en schrijft deze data naar een CSV-bestand.
239
240			:param writer: De handle naar de CSV writer.
241			:param list filenames: De lijst met bestandsnamen van XML-bestanden met test cases.
242			"""
243			for filename in filenames:
244			subtraject = self.__shred_xml_bestand(filename)
245			self.__write_subtraject(writer, subtraject)
246
247			# ------------------------------------------------------------------------------------------------------------------
248			def handle(self):
249			"""
250			Executes the command.
251			"""
252			self.output = KerapuStyle(self.input, self.output)
253
254			zip_filename = self.argument('testset-zip')
255			csv_filename = self.argument('testset-csv')
256			tmp_dir = '.kerapu-' + ''.join(random.choices(string.ascii_lowercase, k=12))
257
258			os.mkdir(tmp_dir)
259
260			self.__extract_zip_file(zip_filename, tmp_dir)
261			files = self.__lees_test_cases_lijst(tmp_dir)
262
263			with open(csv_filename, 'wt', encoding='utf-8') as handle:
264			csv_writer = csv.writer(handle, lineterminator='\n', delimiter=',')
265			self.__extract_files(csv_writer, files)
266
267			shutil.rmtree(tmp_dir)
268
269			return 0
270
271			# ----------------------------------------------------------------------------------------------------------------------
272

SetBased / py-kerapu

GitHub Access Token became invalid

Push — master ( be813a...3d8811 )

TestShredderCommand.__shred_xml_bestand() C

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like