Philipp/aki_prj23_transparenzregister
Philipp/aki_prj23_transparenzregister
1
0
Fork 0
mirror of https://github.com/fhswf/aki_prj23_transparenzregister.git synced 2025-04-22 22:12:53 +02:00
Code Issues Packages Projects Releases Wiki Activity

test(data-extraction): Cover transform.py

Browse Source
This commit is contained in:
TrisNol 2023-09-18 20:23:19 +02:00
parent 92ea72249c
commit 7269e7e6a4
4 changed files with 511 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
src/aki_prj23_transparenzregister/models/company.py
View File

@ -54,16 +54,24 @@ class CompanyTypeEnum(str, MultiValueEnum):
@dataclass
class CompanyID:
"""_summary_."""
class DistrictCourt:
"""DistrictCourt."""
district_court: str
name: str
city: str
@dataclass
class CompanyID:
"""CompanyID."""
district_court: DistrictCourt
hr_number: str
@dataclass
class Location:
"""_summary_."""
"""Location."""
city: str
street: str | None = None
@ -155,14 +163,29 @@ class YearlyResult:
kpis: dict[FinancialKPIEnum, float]
class CurrencyEnum(str, MultiValueEnum):
"""Enum of possible currencies."""
EURO = "EUR"
DEUTSCHE_MARK = "DM", "DEM"
KEINE_ANGABE = ""
class CapitalTypeEnum(str, Enum):
"""Enum of possible capital types."""
HAFTEINLAGE = "Hafteinlage"
STAMMKAPITAL = "Stammkapital"
GRUNDKAPITAL = "Grundkapital"
@dataclass
class Capital:
"""Capital of company."""
value: float
currency: str # TODO define Enum
# TODO define Enum
type: str # noqa: A003
currency: CurrencyEnum
type: CapitalTypeEnum # noqa: A003
@dataclass

5
src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/load.py
View File

@ -25,8 +25,9 @@ if __name__ == "__main__":
connector = MongoConnector(conn_string)
service = CompanyMongoService(connector)
for file in tqdm(glob.glob1("./data/Unternehmensregister/transformed", "*.json")):
path = os.path.join("./data/Unternehmensregister/transformed", file)
base_path = "./Jupyter/API-tests/Unternehmensregister/data/Unternehmensregister"
for file in tqdm(glob.glob1(f"{base_path}/transformed", "*.json")):
path = os.path.join(f"{base_path}/transformed", file)
with open(path, encoding="utf-8") as file_object:
data = json.loads(file_object.read())
company: Company = Company(**data)

160
src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/transform.py
View File

@ -11,12 +11,15 @@ from tqdm import tqdm
from aki_prj23_transparenzregister.models.company import (
Capital,
CapitalTypeEnum,
Company,
CompanyID,
CompanyRelationship,
CompanyRelationshipEnum,
CompanyToCompanyRelationship,
CompanyTypeEnum,
CurrencyEnum,
DistrictCourt,
Location,
PersonName,
PersonToCompanyRelationship,
@ -236,20 +239,21 @@ def map_rechtsform(company_name: str, data: dict) -> CompanyTypeEnum | None:
return None
def map_capital(data: dict, company_type: str) -> Capital | None:
def map_capital(data: dict, company_type: CompanyTypeEnum) -> Capital | None:
"""Extracts the company capital from the given Unternehmensregister export.
Args:
data (dict): Data export
company_type (str): Type of company (e.g., 'Gesellschaft mit beschränkter Haftung')
company_type (CompanyTypeEnum): Type of company (e.g., 'Gesellschaft mit beschränkter Haftung')
Returns:
Capital | None: Company Capital if found
"""
# Early return
if "Zusatzangaben" not in data["XJustiz_Daten"]["Fachdaten_Register"]:
return None
capital: dict = {"Zahl": 0.0, "Waehrung": ""}
if company_type == "Kommanditgesellschaft":
if "Zusatzangaben" not in data["XJustiz_Daten"]["Fachdaten_Register"]:
return None
if company_type == CompanyTypeEnum.KG:
capital_type = "Hafteinlage"
base = data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][
"Personengesellschaft"
@ -259,55 +263,50 @@ def map_capital(data: dict, company_type: str) -> Capital | None:
# TODO link to persons using Ref_Rollennummer then extract ["Hafteinlage"] as below
capital["Zahl"] = capital["Zahl"] + float(entry["Hafteinlage"]["Zahl"])
capital["Waehrung"] = entry["Hafteinlage"]["Waehrung"]
elif type(base) == "dict":
elif isinstance(base, dict):
capital = base["Hafteinlage"]
elif company_type in [
"Gesellschaft mit beschränkter Haftung",
"Europäische Aktiengesellschaft (SE)",
"Aktiengesellschaft",
"Kommanditgesellschaft auf Aktien",
"Rechtsform ausländischen Rechts HRB",
CompanyTypeEnum.GMBH,
CompanyTypeEnum.SE,
CompanyTypeEnum.AG,
CompanyTypeEnum.KGaA,
CompanyTypeEnum.AUSLAENDISCHE_RECHTSFORM,
CompanyTypeEnum.OHG,
]:
if "Zusatzangaben" not in data["XJustiz_Daten"]["Fachdaten_Register"]:
return None
if (
"Zusatz_GmbH"
in data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][
"Kapitalgesellschaft"
not in data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"]
):
base = data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][
"Personengesellschaft"
]
else:
base = data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][
"Kapitalgesellschaft"
]
):
if "Zusatz_GmbH" in base:
capital_type = "Stammkapital"
capital = data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][
"Kapitalgesellschaft"
]["Zusatz_GmbH"]["Stammkapital"]
elif (
"Zusatz_Aktiengesellschaft"
in data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][
"Kapitalgesellschaft"
]
):
capital = base["Zusatz_GmbH"]["Stammkapital"]
elif "Zusatz_Aktiengesellschaft" in base:
capital_type = "Grundkapital"
capital = data["XJustiz_Daten"]["Fachdaten_Register"]["Zusatzangaben"][
"Kapitalgesellschaft"
]["Zusatz_Aktiengesellschaft"]["Grundkapital"]["Hoehe"]
capital = base["Zusatz_Aktiengesellschaft"]["Grundkapital"]["Hoehe"]
elif company_type in [
"Einzelkaufmann",
"Einzelkauffrau",
"eingetragene Genossenschaft",
"Partnerschaft",
"Einzelkaufmann / Einzelkauffrau",
"Offene Handelsgesellschaft",
"Partnerschaftsgesellschaft",
CompanyTypeEnum.EINZELKAUFMANN,
CompanyTypeEnum.EG,
CompanyTypeEnum.PARTNERSCHAFT,
CompanyTypeEnum.PARTNERGESELLSCHAFT,
CompanyTypeEnum.PARTNERSCHAFTSGESELLSCHAFT,
None,
]:
return None
else:
# Catch entries having the dict but with null values
if not all(capital.values()):
return None
return Capital(
**{
"value": capital["Zahl"],
"currency": capital["Waehrung"],
"type": capital_type,
**{ # type: ignore
"value": float(capital["Zahl"]),
"currency": CurrencyEnum(capital["Waehrung"]),
"type": CapitalTypeEnum(capital_type),
}
)
@ -366,17 +365,6 @@ def map_founding_date(data: dict) -> str | None:
)
if len(entry_date) == 1:
return transform_date_to_iso(entry_date[0])
if "Eintragungstext" in data["XJustiz_Daten"]["Fachdaten_Register"]["Auszug"] and (
type(data["XJustiz_Daten"]["Fachdaten_Register"]["Auszug"]["Eintragungstext"])
== "list"
):
temp = data["XJustiz_Daten"]["Fachdaten_Register"]["Auszug"]["Eintragungstext"][
0
]["Text"]
results = re.findall(r"\d{1,2}\.\d{1,2}\.\d{2,4}", temp)
if len(temp) == 1:
return transform_date_to_iso(results[0])
if (
"Gruendungsmetadaten"
in data["XJustiz_Daten"]["Fachdaten_Register"]["Basisdaten_Register"]
@ -402,34 +390,50 @@ def map_company_id(data: dict) -> CompanyID:
"hr_number": data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Instanzdaten"
]["Aktenzeichen"],
"district_court": {
"name": data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]["Organisation"]["Bezeichnung"][
"Bezeichnung_Aktuell"
]
if "Organisation"
in data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]
else data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]["Natuerliche_Person"]["Voller_Name"]["Nachname"],
"city": data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]["Organisation"]["Sitz"]["Ort"]
if "Organisation"
in data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]
else data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]["Natuerliche_Person"]["Anschrift"]["Ort"],
},
"district_court": DistrictCourt(
**{
"name": data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]["Organisation"]["Bezeichnung"][
"Bezeichnung_Aktuell"
]
if "Organisation"
in data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]
else data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]["Natuerliche_Person"]["Voller_Name"][
"Nachname"
],
"city": data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]["Organisation"]["Sitz"]["Ort"]
if "Organisation"
in data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]
else data["XJustiz_Daten"]["Grunddaten"]["Verfahrensdaten"][
"Beteiligung"
][1]["Beteiligter"]["Natuerliche_Person"]["Anschrift"]["Ort"],
}
),
}
)
def map_last_update(data: dict) -> str:
"""Extract last update date from export.
Args:
data (dict): Unternehmensregister export
Returns:
str: Last update date
"""
return data["XJustiz_Daten"]["Fachdaten_Register"]["Auszug"]["letzte_Eintragung"]
def map_unternehmensregister_json(data: dict) -> Company:
"""Processes the Unternehmensregister structured export to a Company by using several helper methods.
@ -446,10 +450,8 @@ def map_unternehmensregister_json(data: dict) -> Company:
result["name"] = name_from_beteiligung(data)
result["location"] = loc_from_beteiligung(data)
result["last_update"] = data["XJustiz_Daten"]["Fachdaten_Register"]["Auszug"][
"letzte_Eintragung"
]
# TODO New features --> to be tested
result["last_update"] = map_last_update(data)
result["company_type"] = map_rechtsform(result["name"], data)
result["capital"] = map_capital(data, result["company_type"])
result["business_purpose"] = map_business_purpose(data)

397
tests/utils/data_extraction/unternehmensregister/transform_test.py
View File

@ -1,10 +1,18 @@
import json
import os
from tempfile import TemporaryDirectory
from unittest.mock import Mock, patch
from aki_prj23_transparenzregister.models.company import (
Capital,
CapitalTypeEnum,
Company,
CompanyID,
CompanyRelationshipEnum,
CompanyToCompanyRelationship,
CompanyTypeEnum,
CurrencyEnum,
DistrictCourt,
Location,
PersonName,
PersonToCompanyRelationship,
@ -206,3 +214,392 @@ def test_map_rechtsform_from_name() -> None:
for company_name, expected_result in data:
assert transform.map_rechtsform(company_name, {}) == expected_result
def test_map_capital_kg_single() -> None:
capital = Capital(
currency=CurrencyEnum.EURO, value=69000, type=CapitalTypeEnum.HAFTEINLAGE # type: ignore
)
data = {
"XJustiz_Daten": {
"Fachdaten_Register": {
"Zusatzangaben": {
"Personengesellschaft": {
"Zusatz_KG": {
"Daten_Kommanditist": {
"Hafteinlage": {
"Zahl": str(capital.value),
"Waehrung": capital.currency,
},
}
}
}
}
}
}
}
result = transform.map_capital(data, CompanyTypeEnum.KG) # type: ignore
assert result == capital
def test_map_capital_kg_sum() -> None:
capital = Capital(
currency=CurrencyEnum.EURO, value=20000, type=CapitalTypeEnum.HAFTEINLAGE # type: ignore
)
data = {
"XJustiz_Daten": {
"Fachdaten_Register": {
"Zusatzangaben": {
"Personengesellschaft": {
"Zusatz_KG": {
"Daten_Kommanditist": [
{
"Hafteinlage": {
"Zahl": str(10000),
"Waehrung": capital.currency,
}
},
{
"Hafteinlage": {
"Zahl": str(10000),
"Waehrung": capital.currency,
},
},
]
}
}
}
}
}
}
result = transform.map_capital(data, CompanyTypeEnum.KG) # type: ignore
assert result == capital
def test_map_capital_no_fachdaten() -> None:
data = {"XJustiz_Daten": {"Fachdaten_Register": {}}} # type: ignore
result = transform.map_capital(data, CompanyTypeEnum.KG) # type: ignore
assert result is None
def test_map_capital_gmbh() -> None:
capital = Capital(
currency=CurrencyEnum.DEUTSCHE_MARK, value=42, type=CapitalTypeEnum.STAMMKAPITAL # type: ignore
)
data = {
"XJustiz_Daten": {
"Fachdaten_Register": {
"Zusatzangaben": {
"Kapitalgesellschaft": {
"Zusatz_GmbH": {
"Stammkapital": {
"Zahl": str(capital.value),
"Waehrung": capital.currency,
},
}
}
}
}
}
}
result = transform.map_capital(data, CompanyTypeEnum.GMBH) # type: ignore
assert result == capital
def test_map_capital_ag() -> None:
capital = Capital(
currency=CurrencyEnum.DEUTSCHE_MARK, value=42, type=CapitalTypeEnum.GRUNDKAPITAL # type: ignore
)
data = {
"XJustiz_Daten": {
"Fachdaten_Register": {
"Zusatzangaben": {
"Kapitalgesellschaft": {
"Zusatz_Aktiengesellschaft": {
"Grundkapital": {
"Hoehe": {
"Zahl": str(capital.value),
"Waehrung": capital.currency,
}
},
}
}
}
}
}
}
result = transform.map_capital(data, CompanyTypeEnum.SE) # type: ignore
assert result == capital
def test_map_capital_personengesellschaft() -> None:
capital = Capital(
currency=CurrencyEnum.DEUTSCHE_MARK, value=42, type=CapitalTypeEnum.STAMMKAPITAL # type: ignore
)
data = {
"XJustiz_Daten": {
"Fachdaten_Register": {
"Zusatzangaben": {
"Personengesellschaft": {
"Zusatz_GmbH": {
"Stammkapital": {
"Zahl": str(capital.value),
"Waehrung": capital.currency,
},
}
}
}
}
}
}
result = transform.map_capital(data, CompanyTypeEnum.OHG) # type: ignore
assert result == capital
def test_map_capital_einzelkaufmann() -> None:
capital = Capital(
currency=CurrencyEnum.DEUTSCHE_MARK, value=42, type=CapitalTypeEnum.STAMMKAPITAL # type: ignore
)
data = {
"XJustiz_Daten": {
"Fachdaten_Register": {
"Zusatzangaben": {
"Personengesellschaft": {
"Zusatz_GmbH": {
"Stammkapital": {
"Zahl": str(capital.value),
"Waehrung": capital.currency,
},
}
}
}
}
}
}
result = transform.map_capital(data, CompanyTypeEnum.EINZELKAUFMANN) # type: ignore
assert result is None
def test_map_capital_partial_null_values() -> None:
capital = Capital(
currency=CurrencyEnum.DEUTSCHE_MARK, value=42, type=CapitalTypeEnum.STAMMKAPITAL # type: ignore
)
data = {
"XJustiz_Daten": {
"Fachdaten_Register": {
"Zusatzangaben": {
"Personengesellschaft": {
"Zusatz_GmbH": {
"Stammkapital": {
"Zahl": None,
"Waehrung": capital.currency,
},
}
}
}
}
}
}
result = transform.map_capital(data, CompanyTypeEnum.OHG) # type: ignore
assert result is None
def test_map_business_purpose() -> None:
business_purpose = "Handel mit Betäubungsmitteln aller Art"
data = {
"XJustiz_Daten": {
"Fachdaten_Register": {
"Basisdaten_Register": {
"Gegenstand_oder_Geschaeftszweck": business_purpose
}
}
}
}
result = transform.map_business_purpose(data)
assert result == business_purpose
def test_map_business_purpose_no_result() -> None:
data: dict = {"XJustiz_Daten": {}}
result = transform.map_business_purpose(data)
assert result is None
def test_transform_date_to_iso() -> None:
date = "10.10.1111"
expected_result = "1111-10-10"
result = transform.transform_date_to_iso(date)
assert result == expected_result
def test_transform_date_to_iso_2_char_year() -> None:
date = "10.10.98"
expected_result = "1998-10-10"
result = transform.transform_date_to_iso(date)
assert result == expected_result
def test_map_founding_date_from_tag_der_ersten_eintragung() -> None:
data = {
"some entry": "Tag der ersten Eintragung: 01.05.2004",
"some other entry": "hfjdoöiashföahöf iodsazo8 5z4o fdsha8oü gfdsö",
}
expected_result = "2004-05-01"
result = transform.map_founding_date(data)
assert result == expected_result
def test_map_founding_date_from_gesellschaftsvertrag() -> None:
data = {
"some entry": "hfjdoöiashföahöf iodsazo8 5z4o fdsha8oü gfdsö",
"some other entry": "Das Wesen der Rekursion ist der Selbstaufruf Gesellschaftsvertrag vom 22.12.1996 Hallo Welt",
}
expected_result = "1996-12-22"
result = transform.map_founding_date(data)
assert result == expected_result
def test_map_founding_date_from_gruendungsdatum() -> None:
data = {
"XJustiz_Daten": {
"Fachdaten_Register": {
"Basisdaten_Register": {
"Gruendungsmetadaten": {"Gruendungsdatum": "1998-01-01"}
}
}
}
}
expected_result = "1998-01-01"
result = transform.map_founding_date(data)
assert result == expected_result
def test_map_founding_date_no_result() -> None:
data: dict = {"XJustiz_Daten": {"Fachdaten_Register": {"Basisdaten_Register": {}}}}
result = transform.map_founding_date(data)
assert result is None
def test_map_company_id() -> None:
district_court = DistrictCourt("Amtsgericht Ulm", "Ulm")
company_id = CompanyID(district_court, "HRA 4711")
data = {
"XJustiz_Daten": {
"Grunddaten": {
"@XJustizVersion": "1.20.0",
"Verfahrensdaten": {
"Instanzdaten": {
"Aktenzeichen": company_id.hr_number,
},
"Beteiligung": [
{},
{
"Beteiligter": {
"Organisation": {
"Bezeichnung": {
"Bezeichnung_Aktuell": district_court.name
},
"Sitz": {
"Ort": district_court.city,
},
}
},
},
],
},
},
}
}
result = transform.map_company_id(data)
assert result == company_id
def test_map_last_update() -> None:
date = "2024-01-01"
data = {
"XJustiz_Daten": {"Fachdaten_Register": {"Auszug": {"letzte_Eintragung": date}}}
}
result = transform.map_last_update(data)
assert result == date
@patch(
"aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_company_id"
)
@patch(
"aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.name_from_beteiligung"
)
@patch(
"aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.loc_from_beteiligung"
)
@patch(
"aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_last_update"
)
@patch(
"aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_rechtsform"
)
@patch(
"aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_capital"
)
@patch(
"aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_business_purpose"
)
@patch(
"aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.map_founding_date"
)
@patch(
"aki_prj23_transparenzregister.utils.data_extraction.unternehmensregister.transform.parse_stakeholder"
)
def test_map_unternehmensregister_json( # noqa: PLR0913
mock_map_parse_stakeholder: Mock,
mock_map_founding_date: Mock,
mock_map_business_purpose: Mock,
mock_map_capital: Mock,
mock_map_rechtsform: Mock,
mock_map_last_update: Mock,
mock_loc_from_beteiligung: Mock,
mock_map_name_from_beteiligung: Mock,
mock_map_company_id: Mock,
) -> None:
expected_result = Company(
**{
"id": Mock(), # type: ignore
"name": Mock(),
"location": Mock(),
"last_update": Mock(),
"company_type": Mock(),
"capital": Mock(),
"business_purpose": Mock(),
"founding_date": Mock(),
"relationships": [Mock()],
}
)
mock_map_company_id.return_value = expected_result.id
mock_map_name_from_beteiligung.return_value = expected_result.name
mock_loc_from_beteiligung.return_value = expected_result.location
mock_map_last_update.return_value = expected_result.last_update
mock_map_rechtsform.return_value = expected_result.company_type
mock_map_capital.return_value = expected_result.capital
mock_map_business_purpose.return_value = expected_result.business_purpose
mock_map_founding_date.return_value = expected_result.founding_date
mock_map_parse_stakeholder.return_value = expected_result.relationships[0]
data: dict = {
"XJustiz_Daten": {
"Grunddaten": {"Verfahrensdaten": {"Beteiligung": [{}, {}, {}]}}
}
}
result = transform.map_unternehmensregister_json(data)
assert result == expected_result